Revert "remove redundant logic from ballista logical codec (#2348)"

[andygrove]

Which issue does this PR close?

Closes #2376.

Rationale for this change

#2348 delegated file format serde to DataFusion's DefaultLogicalExtensionCodec. That is a good cleanup in isolation, but it changed the CopyTo wire format in a way that older clients cannot detect, and the two encodings are structurally identical so nothing errors at the transport layer.

	field 1	field 2
Ballista <= 54	encoder_position: u32, an index into [Parquet, Csv, Json, Arrow, Avro]	blob
Ballista post-#2348	kind: FileFormatKind (UNSPECIFIED=0, CSV=1, JSON=2, PARQUET=3, ARROW=4, AVRO=5)	encoded_file_format

Both are a varint followed by a bytes field, so prost decodes an old message happily and the format tag is simply reinterpreted:

pos=0 (Parquet) -> Err(This feature is not implemented: Unspecified file format kind)
pos=1 (Csv)     -> Ok(csv)       lines up by coincidence
pos=2 (Json)    -> Ok(json)      lines up by coincidence
pos=3 (Arrow)   -> Ok(parquet)   silently wrong format
pos=4 (Avro)    -> Ok(arrow)     silently wrong format

Parquet at least fails loudly, which is what the new job in #2374 caught when a released 54.0.0 Python client ran against a cluster built from the branch. Arrow is the one that worries me: ArrowLogicalExtensionCodec::try_encode_file_format writes an empty payload and the Parquet decoder treats empty bytes as all defaults, so COPY ... STORED AS ARROW from a 54 client executes as Parquet with no error at all.

This is not transient skew. pyballista re-exports datafusion-python types, so the Python bindings cannot move to 55 until there is a matching datafusion-python release, and until then every Python user runs a 54 client against a 55 cluster.

Worth noting that DataFusion deliberately kept its own logical proto backward compatible across 54 to 55: removed fields became reserved 8; // was bool collect_stat, string location = 2 was kept and marked deprecated alongside a new repeated string locations = 16, and everything else is additive. So a 54 client's plan is meant to be readable by a 55 scheduler, and 73 of the 75 Python tests in #2374 confirm it is. This encoding change was the one thing that broke it.

What changes are included in this PR?

• Reverts 6fcc736, restoring the encoder_position based FileFormatProto.
• Adds decodes_file_format_bytes_from_a_54_client, which pins the wire format by re-declaring the legacy layout locally and asserting that those bytes still decode to the right format.

The existing file_format_serialization_roundtrip test could not catch this because it encodes and decodes with the same codec, so it passes for any self-consistent encoding. The new test fails on the pre-revert code with exactly the error CI hit, and passes after the revert.

Avro is deliberately left out of the new test. DataFusion's AvroLogicalExtensionCodec::try_decode_file_format returns an ArrowFormatFactory in both 54 and 55, which looks like an upstream bug and is unrelated to this change.

Are these changes tested?

• New regression test above, verified to fail before the revert and pass after.
• cargo test -p ballista-core --lib serde:: passes, 32 tests.
• cargo clippy --all-targets --workspace --all-features -- -D warnings clean.
• cargo fmt --all -- --check clean.
• The real check is ci: run Python client tests against a cluster built from the branch #2374, whose Python client vs branch-built cluster job should go green once this lands.

Are there any user-facing changes?

Yes, in the sense that it restores compatibility. Released 54 clients can once again run CopyTo against a scheduler built from main.

I am not proposing we carry the duplicated codec forever. The better end state is client side validation of BALLISTA_PROTOCOL_VERSION (#2370), after which #2348 can land again on a bumped protocol version and old clients get a clear error instead of a mis-decode. That is a larger change than I would want to rush before 55.0.0 (#2369), and on its own it would not help the Python client, since it would turn silent breakage into a hard block that users cannot act on until datafusion-python 55 exists.

@milenkovicm sorry to revert your cleanup, happy to help re-land it behind a version check once the client handshake is in place.

[milenkovicm]

no problem @andygrove
I did not really expect this is going to break this way, but good time to catch it