Use upstream arrow-rs record_batch! and create_array! macros

Removes DataFusion's custom `record_batch!` and `create_array!` macro
implementations in favor of the upstream versions from arrow-rs added in
apache/arrow-rs#6588.

Changes:
- Replace custom macro definitions with re-exports from arrow::array
- Update syntax from vec![...] to array literal [...] across 67 usages
- Add arrow_schema aliases in test modules for macro compatibility
- Add record_batch_vec! legacy macro for cases where upstream macro
  cannot be used (variables, vec repetition syntax)
- Use record_batch_vec! in examples and tests that pass variables
- Add datafusion-common dependency to ffi_example_table_provider

The record_batch_vec! macro is a temporary solution until arrow-rs
supports variables and repetition syntax. See upstream issue:
apache/arrow-rs#6553

Closes #13037

Author: ByteBaker

Cargo.lock

@@ -25,6 +25,7 @@ publish = false
 abi_stable = "0.11.3"
 arrow = { workspace = true }
 datafusion = { workspace = true }
+datafusion-common = { workspace = true }
 datafusion-ffi = { workspace = true }
 ffi_module_interface = { path = "../ffi_module_interface" }
 

datafusion-examples/examples/ffi/ffi_example_table_provider/Cargo.toml

@@ -20,7 +20,7 @@ use std::sync::Arc;
 use abi_stable::{export_root_module, prefix_type::PrefixTypeTrait};
 use arrow::array::RecordBatch;
 use arrow::datatypes::{DataType, Field, Schema};
-use datafusion::{common::record_batch, datasource::MemTable};
+use datafusion::datasource::MemTable;
 use datafusion_ffi::table_provider::FFI_TableProvider;
 use ffi_module_interface::{TableProviderModule, TableProviderModuleRef};
 
@@ -29,7 +29,10 @@ fn create_record_batch(start_value: i32, num_values: usize) -> RecordBatch {
     let a_vals: Vec<i32> = (start_value..end_value).collect();
     let b_vals: Vec<f64> = a_vals.iter().map(|v| *v as f64).collect();
 
-    record_batch!(("a", Int32, a_vals), ("b", Float64, b_vals)).unwrap()
+    // TODO: Use arrow::record_batch! once it supports variables
+    // See https://github.com/apache/arrow-rs/issues/6553
+    datafusion_common::record_batch_old!(("a", Int32, a_vals), ("b", Float64, b_vals))
+        .unwrap()
 }
 
 /// Here we only wish to create a simple table provider as an example.

datafusion-examples/examples/ffi/ffi_example_table_provider/src/lib.rs

@@ -24,10 +24,11 @@
 //! - Handle memory pressure by spilling to disk
 //! - Release memory when done
 
-use arrow::record_batch::RecordBatch;
+use arrow::array::RecordBatch;
+use arrow::datatypes::{DataType, Field, Schema};
 use arrow_schema::SchemaRef;
-use datafusion::common::record_batch;
 use datafusion::common::{exec_datafusion_err, internal_err};
+use datafusion::common::{record_batch, record_batch_old};
 use datafusion::datasource::{memory::MemTable, DefaultTableSource};
 use datafusion::error::Result;
 use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation};
@@ -57,30 +58,35 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let config = SessionConfig::new().with_coalesce_batches(false);
     let ctx = SessionContext::new_with_config_rt(config, runtime.clone());
 
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Int32, true),
+        Field::new("name", DataType::Utf8, true),
+    ]));
+
     // Create smaller batches to ensure we get multiple RecordBatches from the scan
     // Make each batch smaller than the memory limit to force multiple batches
-    let batch1 = record_batch!(
+    // TODO: Use arrow::record_batch! once it supports vec repetition syntax
+    // See https://github.com/apache/arrow-rs/issues/6553
+    let batch1 = record_batch_old!(
         ("id", Int32, vec![1; 800]),
         ("name", Utf8, vec!["Alice"; 800])
     )?;
 
-    let batch2 = record_batch!(
+    let batch2 = record_batch_old!(
         ("id", Int32, vec![2; 800]),
         ("name", Utf8, vec!["Bob"; 800])
     )?;
 
-    let batch3 = record_batch!(
+    let batch3 = record_batch_old!(
         ("id", Int32, vec![3; 800]),
         ("name", Utf8, vec!["Charlie"; 800])
     )?;
 
-    let batch4 = record_batch!(
+    let batch4 = record_batch_old!(
         ("id", Int32, vec![4; 800]),
         ("name", Utf8, vec!["David"; 800])
     )?;
 
-    let schema = batch1.schema();
-
     // Create a single MemTable with all batches in one partition to preserve order but ensure streaming
     let mem_table = Arc::new(MemTable::try_new(
         Arc::clone(&schema),
@@ -257,6 +263,9 @@ impl ExecutionPlan for BufferingExecutionPlan {
         partition: usize,
         context: Arc<TaskContext>,
     ) -> Result<SendableRecordBatchStream> {
+        // Required for arrow_array::record_batch! macro to work
+        use arrow::datatypes as arrow_schema;
+
         // Register memory consumer with the context's memory pool
         let reservation = MemoryConsumer::new("MyExternalBatchBufferer")
             .with_can_spill(true)
@@ -286,10 +295,9 @@ impl ExecutionPlan for BufferingExecutionPlan {
 
                 // Since this is a simplified example, return an empty batch
                 // In a real implementation, you would create a batch stream from the processed results
-                record_batch!(("id", Int32, vec![5]), ("name", Utf8, vec!["Eve"]))
-                    .map_err(|e| {
-                        exec_datafusion_err!("Failed to create final RecordBatch: {e}")
-                    })
+                record_batch!(("id", Int32, [5]), ("name", Utf8, ["Eve"])).map_err(|e| {
+                    exec_datafusion_err!("Failed to create final RecordBatch: {e}")
+                })
             }),
         )))
     }

datafusion-examples/examples/memory_pool_execution_plan.rs

@@ -64,6 +64,8 @@ pub mod utils;
 
 /// Reexport arrow crate
 pub use arrow;
+/// Reexport arrow-rs macros for creating arrays and record batches
+pub use arrow::array::{create_array, record_batch};
 pub use column::Column;
 pub use dfschema::{
     qualified_name, DFSchema, DFSchemaRef, ExprSchema, SchemaExt, ToDFSchema,