Add protoc support for ArrowScanExecNode (apache#20280) (apache#20284)

## Which issue does this PR close?

- Closes apache#20280.

## Rationale for this change

Physical plans that read Arrow files (.arrow / IPC) could not be
serialized or deserialized via the proto layer. PhysicalPlanNode already
had scan nodes for Parquet, CSV, JSON, Avro, and in-memory sources, but
not for Arrow, so a DataSourceExec using ArrowSource was not
round-trippable. That blocked use cases like distributing plans that
scan Arrow files (e.g. Ballista). This change adds Arrow scan to the
proto layer so those plans can be serialized and deserialized like the
other file formats.

## What changes are included in this PR?

Proto: Added ArrowScanExecNode (with FileScanExecConf base_conf) and
arrow_scan = 38 to the PhysicalPlanNode oneof in datafusion.proto.

Generated code: Updated prost.rs and pbjson.rs to include
ArrowScanExecNode and the ArrowScan variant (manual edits; protoc was
not run).

To-proto: In try_from_data_source_exec, when the data source is a
FileScanConfig whose file source is ArrowSource, it is now serialized as
ArrowScanExecNode.

From-proto: Implemented try_into_arrow_scan_physical_plan to deserialize
ArrowScanExecNode into DataSourceExec with ArrowSource; missing
base_conf returns an explicit error (no .unwrap()).

Test: Added roundtrip_arrow_scan in roundtrip_physical_plan.rs to assert
Arrow scan plans round-trip correctly.

## Are these changes tested?

Yes. A new test roundtrip_arrow_scan builds a physical plan that scans
Arrow files, serializes it to bytes and deserializes it back, and
asserts the round-tripped plan matches the original. The full cargo test
-p datafusion-proto suite (150 tests: unit, integration, and doc tests)
passes, including all existing roundtrip and serialization tests.

## Are there any user-facing changes?

No. This only extends the existing physical-plan proto support to Arrow
scan. Callers that already serialize/deserialize physical plans (e.g.
for distributed execution) can now round-trip plans that read Arrow
files in addition to Parquet, CSV, JSON, and Avro, with no API or
behavioral changes for existing usage.

Author: JoshElkind

datafusion/proto/proto/datafusion.proto

@@ -751,6 +751,7 @@ message PhysicalPlanNode {
     MemoryScanExecNode memory_scan = 35;
     AsyncFuncExecNode async_func = 36;
     BufferExecNode buffer = 37;
+    ArrowScanExecNode arrow_scan = 38;
   }
 }
 
@@ -1106,6 +1107,10 @@ message AvroScanExecNode {
   FileScanExecConf base_conf = 1;
 }
 
+message ArrowScanExecNode {
+  FileScanExecConf base_conf = 1;
+}
+
 message MemoryScanExecNode {
   repeated bytes partitions = 1;
   datafusion_common.Schema schema = 2;

datafusion/proto/src/generated/pbjson.rs

@@ -34,6 +34,7 @@ use datafusion_datasource::file_compression_type::FileCompressionType;
 use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
 use datafusion_datasource::sink::DataSinkExec;
 use datafusion_datasource::source::{DataSource, DataSourceExec};
+use datafusion_datasource_arrow::source::ArrowSource;
 #[cfg(feature = "avro")]
 use datafusion_datasource_avro::source::AvroSource;
 use datafusion_datasource_csv::file_format::CsvSink;
@@ -199,6 +200,9 @@ impl protobuf::PhysicalPlanNode {
             PhysicalPlanType::MemoryScan(scan) => {
                 self.try_into_memory_scan_physical_plan(scan, ctx, codec, proto_converter)
             }
+            PhysicalPlanType::ArrowScan(scan) => {
+                self.try_into_arrow_scan_physical_plan(scan, ctx, codec, proto_converter)
+            }
             PhysicalPlanType::CoalesceBatches(coalesce_batches) => self
                 .try_into_coalesce_batches_physical_plan(
                     coalesce_batches,
@@ -774,6 +778,27 @@ impl protobuf::PhysicalPlanNode {
         Ok(DataSourceExec::from_data_source(scan_conf))
     }
 
+    fn try_into_arrow_scan_physical_plan(
+        &self,
+        scan: &protobuf::ArrowScanExecNode,
+        ctx: &TaskContext,
+        codec: &dyn PhysicalExtensionCodec,
+        proto_converter: &dyn PhysicalProtoConverterExtension,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let base_conf = scan.base_conf.as_ref().ok_or_else(|| {
+            internal_datafusion_err!("base_conf in ArrowScanExecNode is missing.")
+        })?;
+        let table_schema = parse_table_schema_from_proto(base_conf)?;
+        let scan_conf = parse_protobuf_file_scan_config(
+            base_conf,
+            ctx,
+            codec,
+            proto_converter,
+            Arc::new(ArrowSource::new_file_source(table_schema)),
+        )?;
+        Ok(DataSourceExec::from_data_source(scan_conf))
+    }
+
     #[cfg_attr(not(feature = "parquet"), expect(unused_variables))]
     fn try_into_parquet_scan_physical_plan(
         &self,
@@ -2867,6 +2892,23 @@ impl protobuf::PhysicalPlanNode {
             }
         }
 
+        if let Some(scan_conf) = data_source.as_any().downcast_ref::<FileScanConfig>() {
+            let source = scan_conf.file_source();
+            if let Some(_arrow_source) = source.as_any().downcast_ref::<ArrowSource>() {
+                return Ok(Some(protobuf::PhysicalPlanNode {
+                    physical_plan_type: Some(PhysicalPlanType::ArrowScan(
+                        protobuf::ArrowScanExecNode {
+                            base_conf: Some(serialize_file_scan_config(
+                                scan_conf,
+                                codec,
+                                proto_converter,
+                            )?),
+                        },
+                    )),
+                }));
+            }
+        }
+
         #[cfg(feature = "parquet")]
         if let Some((maybe_parquet, conf)) =
             data_source_exec.downcast_to_file_source::<ParquetSource>()

datafusion/proto/src/generated/prost.rs

@@ -36,8 +36,8 @@ use datafusion::datasource::listing::{
 };
 use datafusion::datasource::object_store::ObjectStoreUrl;
 use datafusion::datasource::physical_plan::{
-    FileGroup, FileOutputMode, FileScanConfigBuilder, FileSinkConfig, ParquetSource,
-    wrap_partition_type_in_dict, wrap_partition_value_in_dict,
+    ArrowSource, FileGroup, FileOutputMode, FileScanConfigBuilder, FileSinkConfig,
+    ParquetSource, wrap_partition_type_in_dict, wrap_partition_value_in_dict,
 };
 use datafusion::datasource::sink::DataSinkExec;
 use datafusion::datasource::source::DataSourceExec;
@@ -929,6 +929,30 @@ fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> {
     roundtrip_test(DataSourceExec::from_data_source(scan_config))
 }
 
+#[test]
+fn roundtrip_arrow_scan() -> Result<()> {
+    let file_schema =
+        Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)]));
+
+    let table_schema = TableSchema::new(file_schema.clone(), vec![]);
+    let file_source = Arc::new(ArrowSource::new_file_source(table_schema));
+
+    let scan_config =
+        FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source)
+            .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::new(
+                "/path/to/file.arrow".to_string(),
+                1024,
+            )])])
+            .with_statistics(Statistics {
+                num_rows: Precision::Inexact(100),
+                total_byte_size: Precision::Inexact(1024),
+                column_statistics: Statistics::unknown_column(&file_schema),
+            })
+            .build();
+
+    roundtrip_test(DataSourceExec::from_data_source(scan_config))
+}
+
 #[tokio::test]
 async fn roundtrip_parquet_exec_with_table_partition_cols() -> Result<()> {
     let mut file_group =