fix: preserve None projection semantics across FFI boundary in ForeignTableProvider::scan (#20393)

## Which issue does this PR close?

N/A (newly discovered bug)

This is originally found in apache/sedona-db when working on a custom
plan node:
apache/sedona-db#611 (comment)

## Rationale for this change

`ForeignTableProvider::scan()` converts a `None` projection (meaning
"return all columns") into an empty `RVec<usize>` before passing it
across the FFI boundary. On the receiving side, `scan_fn_wrapper` always
wraps the received `RVec` in `Some(...)`, passing `Some(&vec![])` to the
inner `TableProvider::scan()`. This means "project zero columns" — the
exact opposite of the intended "project all columns."

The root cause is that the `FFI_TableProvider::scan` function signature
uses `RVec<usize>` for the projections parameter. Since `RVec<usize>`
cannot represent `None`, the `None` vs. empty-vec distinction is lost at
the FFI boundary.

## What changes are included in this PR?

Three coordinated changes in `datafusion/ffi/src/table_provider.rs`:

1. **FFI struct definition**: Changed `scan` function pointer signature
from `RVec<usize>` to `ROption<RVec<usize>>` for the projections
parameter, matching how `limit` already uses `ROption<usize>` for the
same `None`-vs-value distinction.

2. **Receiver side** (`scan_fn_wrapper`): Converts
`ROption<RVec<usize>>` via `.into_option().map(...)` and passes
`projections.as_ref()` to the inner provider, preserving `None`
semantics.

3. **Sender side** (`ForeignTableProvider::scan`): Converts
`Option<&Vec<usize>>` to `ROption<RVec<usize>>` via `.into()` instead of
using `unwrap_or_default()`.

Plus a new unit test
`test_scan_with_none_projection_returns_all_columns` that directly
exercises the FFI round-trip with `projection=None` and verifies all 3
columns are returned.

Also fixed the existing `test_aggregation` test to set
`library_marker_id = mock_foreign_marker_id` so it actually exercises
the FFI path instead of taking the local bypass.

## How are these changes tested?

- New test `test_scan_with_none_projection_returns_all_columns`: creates
a 3-column MemTable, wraps it through FFI with the foreign marker set,
calls `scan(None)`, and asserts 3 columns are returned (previously
returned 0).

## Are these changes safe?

This is a **breaking FFI ABI change** to the `FFI_TableProvider::scan`
function pointer signature. However:

- The `abi_stable` crate's `#[derive(StableAbi)]` generates layout
checks at dylib load time, so mismatched dylibs will be caught at load
rather than causing silent corruption.
- All existing providers construct `FFI_TableProvider` via `::new()` or
`::new_with_ffi_codec()`, which internally wire up `scan_fn_wrapper` —
nobody constructs the `scan` function pointer manually.

Author: Kontinuation

datafusion/ffi/src/table_provider.rs

@@ -108,7 +108,7 @@ pub struct FFI_TableProvider {
     scan: unsafe extern "C" fn(
         provider: &Self,
         session: FFI_SessionRef,
-        projections: RVec<usize>,
+        projections: ROption<RVec<usize>>,
         filters_serialized: RVec<u8>,
         limit: ROption<usize>,
     ) -> FfiFuture<FFIResult<FFI_ExecutionPlan>>,
@@ -232,7 +232,7 @@ unsafe extern "C" fn supports_filters_pushdown_fn_wrapper(
 unsafe extern "C" fn scan_fn_wrapper(
     provider: &FFI_TableProvider,
     session: FFI_SessionRef,
-    projections: RVec<usize>,
+    projections: ROption<RVec<usize>>,
     filters_serialized: RVec<u8>,
     limit: ROption<usize>,
 ) -> FfiFuture<FFIResult<FFI_ExecutionPlan>> {
@@ -269,11 +269,12 @@ unsafe extern "C" fn scan_fn_wrapper(
             }
         };
 
-        let projections: Vec<_> = projections.into_iter().collect();
+        let projections: Option<Vec<usize>> =
+            projections.into_option().map(|p| p.into_iter().collect());
 
         let plan = rresult_return!(
             internal_provider
-                .scan(session, Some(&projections), &filters, limit.into())
+                .scan(session, projections.as_ref(), &filters, limit.into())
                 .await
         );
 
@@ -461,8 +462,9 @@ impl TableProvider for ForeignTableProvider {
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let session = FFI_SessionRef::new(session, None, self.0.logical_codec.clone());
 
-        let projections: Option<RVec<usize>> =
-            projection.map(|p| p.iter().map(|v| v.to_owned()).collect());
+        let projections: ROption<RVec<usize>> = projection
+            .map(|p| p.iter().map(|v| v.to_owned()).collect())
+            .into();
 
         let codec: Arc<dyn LogicalExtensionCodec> = (&self.0.logical_codec).into();
         let filter_list = LogicalExprList {
@@ -474,7 +476,7 @@ impl TableProvider for ForeignTableProvider {
             let maybe_plan = (self.0.scan)(
                 &self.0,
                 session,
-                projections.unwrap_or_default(),
+                projections,
                 filters_serialized,
                 limit.into(),
             )
@@ -658,8 +660,9 @@ mod tests {
 
         let provider = Arc::new(MemTable::try_new(schema, vec![vec![batch1]])?);
 
-        let ffi_provider =
+        let mut ffi_provider =
             FFI_TableProvider::new(provider, true, None, task_ctx_provider, None);
+        ffi_provider.library_marker_id = crate::mock_foreign_marker_id;
 
         let foreign_table_provider: Arc<dyn TableProvider> = (&ffi_provider).into();
 
@@ -712,4 +715,62 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_scan_with_none_projection_returns_all_columns() -> Result<()> {
+        use arrow::datatypes::Field;
+        use datafusion::arrow::array::Float32Array;
+        use datafusion::arrow::datatypes::DataType;
+        use datafusion::arrow::record_batch::RecordBatch;
+        use datafusion::datasource::MemTable;
+        use datafusion::physical_plan::collect;
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Float32, false),
+            Field::new("b", DataType::Float32, false),
+            Field::new("c", DataType::Float32, false),
+        ]));
+
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![
+                Arc::new(Float32Array::from(vec![1.0, 2.0])),
+                Arc::new(Float32Array::from(vec![3.0, 4.0])),
+                Arc::new(Float32Array::from(vec![5.0, 6.0])),
+            ],
+        )?;
+
+        let provider =
+            Arc::new(MemTable::try_new(Arc::clone(&schema), vec![vec![batch]])?);
+
+        let ctx = Arc::new(SessionContext::new());
+        let task_ctx_provider = Arc::clone(&ctx) as Arc<dyn TaskContextProvider>;
+        let task_ctx_provider = FFI_TaskContextProvider::from(&task_ctx_provider);
+
+        // Wrap in FFI and force the foreign path (not local bypass)
+        let mut ffi_provider =
+            FFI_TableProvider::new(provider, true, None, task_ctx_provider, None);
+        ffi_provider.library_marker_id = crate::mock_foreign_marker_id;
+
+        let foreign_table_provider: Arc<dyn TableProvider> = (&ffi_provider).into();
+
+        // Call scan with projection=None, meaning "return all columns"
+        let plan = foreign_table_provider
+            .scan(&ctx.state(), None, &[], None)
+            .await?;
+        assert_eq!(
+            plan.schema().fields().len(),
+            3,
+            "scan(projection=None) should return all columns; got {}",
+            plan.schema().fields().len()
+        );
+
+        // Also verify we can execute and get correct data
+        let batches = collect(plan, ctx.task_ctx()).await?;
+        assert_eq!(batches.len(), 1);
+        assert_eq!(batches[0].num_columns(), 3);
+        assert_eq!(batches[0].num_rows(), 2);
+
+        Ok(())
+    }
 }