CopyTo plans from a 54 client silently decode as the wrong file format on a 55 scheduler

[andygrove]

Describe the bug

#2348 changed BallistaLogicalExtensionCodec to delegate try_encode_file_format / try_decode_file_format to DataFusion's DefaultLogicalExtensionCodec, removing Ballista's private FileFormatProto. This is a breaking wire-format change for CopyTo plans, and the two encodings are structurally identical, so the mismatch is not detected as a decode error.

	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 the old message happily and the format tag is simply reinterpreted with the wrong meaning.

Decoding every old position with the new codec gives:

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

Parquet fails loudly. CSV and JSON happen to line up numerically. Arrow and Avro silently decode as the wrong format: ArrowLogicalExtensionCodec::try_encode_file_format writes an empty payload, and the Parquet decoder accepts empty bytes as "all defaults", so COPY ... STORED AS ARROW from a 54 client is executed as Parquet by a 55 scheduler with no error at all.

To Reproduce

The failure is already reproduced in CI by the new job added in #2374, where a released 54.0.0 Python client runs against a cluster built from the branch:

FAILED python/tests/test_context.py::test_write_parquet - Exception: DataFusion error: Arrow error:
External error: Execution error: Status { code: InvalidArgument, message: "Could not parse plan:
DataFusion error: This feature is not implemented: Unspecified file format kind" }

The full matrix above can be reproduced with a unit test in ballista-core that feeds the old wire bytes (field 1 = varint position, empty blob, i.e. [0x08, pos]) to BallistaLogicalExtensionCodec::default().try_decode_file_format.

Expected behavior

Either the 54 encoding keeps working against a 55 scheduler, or the client is rejected with a clear version-mismatch error. Silently writing Arrow output as Parquet is the outcome to avoid.

Additional context

This matters for the 55.0.0 release (#2369) because client/cluster skew is the steady state rather than an edge case: pyballista re-exports datafusion-python types, so the Python bindings cannot move to 55 until there is a matching datafusion-python release. Until then every Python user runs a 54 client against a 55 cluster.

Possible directions:

• Keep the new delegation but make the old positional encoding detectable, so an old client gets a clear error rather than a mis-decode.
• Bump BALLISTA_PROTOCOL_VERSION and reject 54 clients outright. This is the same gap described in Validate BALLISTA_PROTOCOL_VERSION for clients, not just executors #2370, where the protocol version is validated for executors but not for clients.

At minimum the Arrow and Avro cases should not fail silently.

Related: #2348, #2374, #2369, #2370

[andygrove]

@milenkovicm we may need to revert #2348 and rethink the approach

[milenkovicm]

Please do @andygrove