Arrow IPC stream format is silently lost when a scan is serialized to protobuf

[adriangb]

Filed with AI assistance (Claude Code); the reproduction below was run against main.

Describe the bug

ArrowSource carries a format: ArrowFormat field distinguishing the Arrow IPC file format from the Arrow IPC stream format. That field is not represented on the wire — ArrowScanExecNode only carries a base_conf — so a stream-format scan silently round-trips into a file-format scan:

• Encode emits PhysicalPlanType::ArrowScan for both ArrowFormat::File and ArrowFormat::Stream.
• Decode unconditionally rebuilds the source with ArrowSource::new_file_source(table_schema).

This is long-standing behaviour and predates the proto-hook migration, but it is now stated outright in ArrowSource::try_from_proto — "Defaults to the IPC file format because protobuf does not distinguish it from the IPC stream format" — so it seems worth tracking separately rather than leaving as a doc comment.

To Reproduce

Note that the usual roundtrip_test helper does not catch this. I checked: the before/after Debug strings compare equal, which is exactly the caveat in that helper's own doc comment ("the string representation of a plan often only shows a subset of state"). The loss has to be asserted by downcasting.

Dropping this into datafusion/proto/tests/cases/roundtrip_physical_plan.rs prints FILE_TYPE AFTER: arrow for a source that was built as a stream source:

#[test]
fn roundtrip_arrow_stream_scan() -> Result<()> {
    let file_schema =
        Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)]));
    let file_source = Arc::new(ArrowSource::new_stream_file_source(TableSchema::from(
        &file_schema,
    )));
    let scan_config =
        FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source)
            .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::new(
                "/path/to/file.arrows".to_string(),
                1024,
            )])])
            .build();

    let ctx = SessionContext::new();
    let after = roundtrip_test_and_return(
        DataSourceExec::from_data_source(scan_config),
        &ctx,
        &DefaultPhysicalExtensionCodec {},
        &DefaultPhysicalProtoConverter {},
    )?;

    let exec = after.downcast_ref::<DataSourceExec>().unwrap();
    let fsc = exec
        .data_source()
        .downcast_ref::<FileScanConfig>()
        .unwrap();
    // prints "arrow"; the source was built as "arrow_stream"
    println!("FILE_TYPE AFTER: {}", fsc.file_source().file_type());
    Ok(())
}

Expected behavior

Either the format survives the round trip, or serialization refuses to silently downgrade it:

1. Add a format discriminator to ArrowScanExecNode and thread it through ArrowSource::try_to_proto / try_from_proto. This is an additive wire change — a new optional field defaulting to the file format keeps existing payloads decoding exactly as they do today.
2. Return an error from ArrowSource::try_to_proto when format == ArrowFormat::Stream, so callers get a loud failure instead of a plan that quietly reads with the wrong opener.

Option 1 is the real fix; option 2 is a safe stopgap. Whichever is chosen, the test above should assert the format rather than relying on roundtrip_test.

Additional context

Why this matters beyond the format label: ArrowSource::repartitioned() branches on self.format and refuses range-based parallel reading for the stream format, because the stream format has no footer to locate record-batch boundaries. A stream scan decoded as a file scan can therefore be repartitioned into byte ranges that are not valid boundaries.

Noticed while reviewing #24189, which moves this code into datafusion-datasource-arrow unchanged. Part of the epic in #23494.

[buraksenn]

take