chore: bump arrow version to 51

Cargo.toml

@@ -12,7 +12,7 @@ include = ["src/**/*.rs", "Cargo.toml"]
 rust-version = "1.70"
 
 [dependencies]
-arrow = { version = "50", features = ["prettyprint"] }
+arrow = { version = "51", features = ["prettyprint"] }
 bytes = "1.4"
 chrono = { version = "0.4.37", default-features = false, features = ["std"] }
 chrono-tz = "0.8.6"
@@ -38,12 +38,12 @@ tokio = { version = "1.28", optional = true, features = [
 zstd = "0.12"
 
 [dev-dependencies]
-arrow-json = "50.0.0"
+arrow-json = "51.0.0"
 async-trait = "0.1.77"
 criterion = { version = "0.5", default-features = false, features = ["async_tokio"] }
-datafusion = "36.0.0"
-datafusion-expr = "36.0.0"
-datafusion-physical-expr = "36.0.0"
+datafusion = "37.0.0"
+datafusion-expr = "37.0.0"
+datafusion-physical-expr = "37.0.0"
 object_store = "0.9.0"
 pretty_assertions = "1.3.0"
 serde_json = { version = "1.0", default-features = false, features = ["std"] }

examples/datafusion_integration.rs

@@ -17,11 +17,12 @@ use datafusion::error::Result;
 use datafusion::execution::context::{SessionState, TaskContext};
 use datafusion::execution::object_store::ObjectStoreUrl;
 use datafusion::physical_plan::{
-    DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, SendableRecordBatchStream,
+    DisplayAs, DisplayFormatType, ExecutionMode, ExecutionPlan, Partitioning, PlanProperties,
+    SendableRecordBatchStream,
 };
 use datafusion::prelude::*;
 use datafusion_expr::{Expr, LogicalPlanBuilder};
-use datafusion_physical_expr::PhysicalSortExpr;
+use datafusion_physical_expr::EquivalenceProperties;
 use orc_rust::ArrowReaderBuilder;
 
 use async_trait::async_trait;
@@ -101,11 +102,20 @@ impl TableProvider for OrcDataSource {
 struct OrcExec {
     orc: OrcDataSource,
     schema: SchemaRef,
+    properties: PlanProperties,
 }
 
 impl OrcExec {
     fn new(orc: OrcDataSource, schema: SchemaRef) -> Result<Self> {
-        Ok(Self { orc, schema })
+        Ok(Self {
+            orc,
+            schema: schema.clone(),
+            properties: PlanProperties::new(
+                EquivalenceProperties::new(schema),
+                Partitioning::UnknownPartitioning(1),
+                ExecutionMode::Bounded,
+            ),
+        })
     }
 }
 
@@ -135,12 +145,8 @@ impl ExecutionPlan for OrcExec {
         self.schema.clone()
     }
 
-    fn output_partitioning(&self) -> Partitioning {
-        Partitioning::UnknownPartitioning(1)
-    }
-
-    fn output_ordering(&self) -> Option<&[PhysicalSortExpr]> {
-        None
+    fn properties(&self) -> &PlanProperties {
+        &self.properties
     }
 
     fn execute(