feat!: upgrade DataFusion to 55

Cargo.toml

@@ -100,13 +100,13 @@ rust.unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
 # See for more detaiils: https://github.com/rust-lang/cargo/issues/11329
 ahash = { version = "0.8", features = ["compile-time-rng"] }
 aquamarine = "0.6"
-arrow = { version = "58.3", features = ["prettyprint"] }
-arrow-array = { version = "58.3", default-features = false, features = ["chrono-tz"] }
-arrow-buffer = "58.3"
-arrow-cast = "58.3"
-arrow-flight = "58.3"
-arrow-ipc = { version = "58.3", default-features = false, features = ["lz4", "zstd"] }
-arrow-schema = { version = "58.3", features = ["serde"] }
+arrow = { version = "=59.2.0", features = ["prettyprint"] }
+arrow-array = { version = "=59.2.0", default-features = false, features = ["chrono-tz"] }
+arrow-buffer = "=59.2.0"
+arrow-cast = "=59.2.0"
+arrow-flight = "=59.2.0"
+arrow-ipc = { version = "=59.2.0", default-features = false, features = ["lz4", "zstd"] }
+arrow-schema = { version = "=59.2.0", features = ["serde"] }
 async-stream = "0.3"
 async-trait = "0.1"
 # Remember to update axum-extra, axum-macros when updating axum
@@ -128,22 +128,22 @@ const_format = "0.2"
 criterion = "0.7"
 crossbeam-utils = "0.8"
 dashmap = "6.1"
-datafusion = "=53.1.0"
-datafusion-common = "=53.1.0"
-datafusion-datasource = "=53.1.0"
-datafusion-expr = "=53.1.0"
-datafusion-expr-common = "=53.1.0"
-datafusion-functions = "=53.1.0"
-datafusion-functions-aggregate-common = "=53.1.0"
-datafusion-functions-window-common = "=53.1.0"
-datafusion-optimizer = "=53.1.0"
-datafusion-orc = { git = "https://github.com/datafusion-contrib/datafusion-orc.git", rev = "6c07fa282dc8d62db2aa4ded06ab55485efc811a" }
-datafusion-pg-catalog = "0.17.3"
-datafusion-physical-expr = "=53.1.0"
-datafusion-physical-plan = "=53.1.0"
-datafusion-proto = "=53.1.0"
-datafusion-sql = "=53.1.0"
-datafusion-substrait = "=53.1.0"
+datafusion = "=55.0.0"
+datafusion-common = "=55.0.0"
+datafusion-datasource = "=55.0.0"
+datafusion-expr = "=55.0.0"
+datafusion-expr-common = "=55.0.0"
+datafusion-functions = "=55.0.0"
+datafusion-functions-aggregate-common = "=55.0.0"
+datafusion-functions-window-common = "=55.0.0"
+datafusion-optimizer = "=55.0.0"
+datafusion-orc = "0.10.0"
+datafusion-pg-catalog = "0.18.3"
+datafusion-physical-expr = "=55.0.0"
+datafusion-physical-plan = "=55.0.0"
+datafusion-proto = "=55.0.0"
+datafusion-sql = "=55.0.0"
+datafusion-substrait = "=55.0.0"
 datafusion_object_store = { package = "object_store", version = "0.13.2" }
 deadpool = "0.12"
 deadpool-postgres = "0.14"
@@ -195,10 +195,10 @@ otel-arrow-rust = { git = "https://github.com/GreptimeTeam/otel-arrow", rev = "5
     "server",
 ] }
 parking_lot = "0.12"
-parquet = { version = "58.3", default-features = false, features = ["arrow", "async", "object_store"] }
-parquet-variant = "58.3"
-parquet-variant-compute = "58.3"
-parquet-variant-json = "58.3"
+parquet = { version = "=59.2.0", default-features = false, features = ["arrow", "async", "object_store"] }
+parquet-variant = "=59.2.0"
+parquet-variant-compute = "=59.2.0"
+parquet-variant-json = "=59.2.0"
 paste = "1.0"
 pin-project = "1.0"
 pretty_assertions = "1.4.0"
@@ -239,7 +239,7 @@ simd-json = "0.15"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
-sqlparser = { version = "0.61.0", default-features = false, features = ["std", "visitor", "serde"] }
+sqlparser = { version = "0.62.0", default-features = false, features = ["std", "visitor", "serde"] }
 sqlx = { version = "0.8", default-features = false, features = [
     "any",
     "macros",
@@ -352,22 +352,24 @@ git = "https://github.com/GreptimeTeam/greptime-meter.git"
 rev = "5618e779cf2bb4755b499c630fba4c35e91898cb"
 
 [patch.crates-io]
-datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-proto = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "a281d9f0ea4b3fb2ec88bc3553d8dcef307e072a" }
-sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "2aefa08a8d69c96eec2d6d6703598a009bba6e4c" }                           # on branch v0.61.x
+datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-proto = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7ac19906c191e51433c0c3d8810d4fd47ab1529a" }
+datafusion-pg-catalog = { git = "https://github.com/GreptimeTeam/datafusion-postgres.git", rev = "3c77e6c32b8db80635a0d2f4b318a36b31170bc3" }
+arrow-pg = { git = "https://github.com/GreptimeTeam/datafusion-postgres.git", rev = "3c77e6c32b8db80635a0d2f4b318a36b31170bc3" }
+sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "9e9019bb1c7040ed956f654e39378dd42ab17884" }
 
 [profile.release]
 debug = 1

src/catalog/src/information_extension.rs

@@ -157,10 +157,6 @@ impl ExecutionPlan for DistributedInspectExec {
         "DistributedInspectExec"
     }
 
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn schema(&self) -> ArrowSchemaRef {
         self.arrow_schema.clone()
     }
@@ -173,6 +169,17 @@ impl ExecutionPlan for DistributedInspectExec {
         vec![]
     }
 
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(
+            &Arc<dyn datafusion::physical_expr::PhysicalExpr>,
+        ) -> datafusion::common::Result<
+            datafusion::common::tree_node::TreeNodeRecursion,
+        >,
+    ) -> datafusion::common::Result<datafusion::common::tree_node::TreeNodeRecursion> {
+        Ok(datafusion::common::tree_node::TreeNodeRecursion::Continue)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,

src/catalog/src/system_schema/information_schema/ssts.rs

@@ -336,7 +336,7 @@ mod tests {
         };
         let plan = table.scan_to_plan(request).unwrap().unwrap();
 
-        assert!(plan.as_any().is::<EmptyExec>());
+        assert!(plan.as_ref().is::<EmptyExec>());
         assert_eq!(1, plan.schema().fields().len());
     }
 }

src/catalog/src/table_source/dummy_catalog.rs

@@ -14,7 +14,6 @@
 
 //! Dummy catalog for region server.
 
-use std::any::Any;
 use std::fmt;
 use std::sync::Arc;
 
@@ -64,10 +63,6 @@ impl fmt::Debug for DummyCatalogList {
 }
 
 impl CatalogProviderList for DummyCatalogList {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn register_catalog(
         &self,
         _name: String,
@@ -98,10 +93,6 @@ struct DummyCatalogProvider {
 }
 
 impl CatalogProvider for DummyCatalogProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema_names(&self) -> Vec<String> {
         vec![]
     }
@@ -135,10 +126,6 @@ struct DummySchemaProvider {
 
 #[async_trait]
 impl SchemaProvider for DummySchemaProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn table_names(&self) -> Vec<String> {
         vec![]
     }

src/cmd/src/bin/query_perf_fixture/inspect_footer.rs

@@ -14,18 +14,21 @@
 
 use std::collections::BTreeSet;
 use std::fs;
+use std::ops::Range;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use clap::Args as ClapArgs;
 use datafusion_object_store::path::Path as StorePath;
-use datafusion_object_store::{ObjectMeta, ObjectStore};
-use futures::{StreamExt, TryStreamExt};
+use datafusion_object_store::{ObjectMeta, ObjectStore, ObjectStoreExt};
+use futures::future::BoxFuture;
+use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt};
 use object_store::config::ObjectStoreConfig;
 use object_store::factory::new_raw_object_store;
 use object_store::services::Fs;
-use parquet::arrow::async_reader::ParquetObjectReader;
-use parquet::file::metadata::ParquetMetaDataReader;
+use parquet::arrow::async_reader::AsyncFileReader;
+use parquet::errors::{ParquetError, Result as ParquetResult};
+use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
 use serde::{Deserialize, Serialize};
 
 /// Same shape as `query_regression_runner::model::DestinationConfig`. The two
@@ -100,6 +103,70 @@ struct ListedFile {
     relative_path: String,
 }
 
+/// An asynchronous Parquet reader backed directly by an object store.
+///
+/// The file size is retained from the listing so footer reads use bounded
+/// ranges without an additional stat/head request.
+#[derive(Clone, Debug)]
+struct ObjectStoreReader {
+    store: Arc<dyn ObjectStore>,
+    path: StorePath,
+    file_size: u64,
+}
+
+impl ObjectStoreReader {
+    fn new(store: Arc<dyn ObjectStore>, path: StorePath, file_size: u64) -> Self {
+        Self {
+            store,
+            path,
+            file_size,
+        }
+    }
+}
+
+fn to_parquet_error(error: datafusion_object_store::Error) -> ParquetError {
+    ParquetError::External(Box::new(error))
+}
+
+impl AsyncFileReader for ObjectStoreReader {
+    fn get_bytes(
+        &mut self,
+        range: Range<u64>,
+    ) -> BoxFuture<'_, ParquetResult<prost::bytes::Bytes>> {
+        self.store
+            .get_range(&self.path, range)
+            .map_err(to_parquet_error)
+            .boxed()
+    }
+
+    fn get_byte_ranges(
+        &mut self,
+        ranges: Vec<Range<u64>>,
+    ) -> BoxFuture<'_, ParquetResult<Vec<prost::bytes::Bytes>>> {
+        async move {
+            self.store
+                .get_ranges(&self.path, &ranges)
+                .await
+                .map_err(to_parquet_error)
+        }
+        .boxed()
+    }
+
+    fn get_metadata<'a>(
+        &'a mut self,
+        _options: Option<&'a parquet::arrow::arrow_reader::ArrowReaderOptions>,
+    ) -> BoxFuture<'a, ParquetResult<Arc<ParquetMetaData>>> {
+        let file_size = self.file_size;
+        async move {
+            let metadata = ParquetMetaDataReader::new()
+                .load_and_finish(self, file_size)
+                .await?;
+            Ok(Arc::new(metadata))
+        }
+        .boxed()
+    }
+}
+
 pub(super) async fn run_inspect_footer(
     args: InspectFooterArgs,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -217,11 +284,8 @@ async fn inspect_file(
     file: &ListedFile,
     column: &str,
 ) -> Result<FooterFileReport, Box<dyn std::error::Error + Send + Sync>> {
-    let mut reader =
-        ParquetObjectReader::new(store, file.location.clone()).with_file_size(file.size);
-    let metadata = ParquetMetaDataReader::new()
-        .load_and_finish(&mut reader, file.size)
-        .await?;
+    let mut reader = ObjectStoreReader::new(store, file.location.clone(), file.size);
+    let metadata = reader.get_metadata(None).await?;
     let file_metadata = metadata.file_metadata();
     let row_groups = metadata.row_groups();
     let mut columns = Vec::new();

src/common/datasource/Cargo.toml

@@ -34,7 +34,7 @@ futures.workspace = true
 lazy_static.workspace = true
 object-store.workspace = true
 object_store_opendal.workspace = true
-orc-rust = { version = "0.8", default-features = false, features = ["async"] }
+orc-rust = { version = "0.9", default-features = false, features = ["async"] }
 parquet.workspace = true
 paste.workspace = true
 regex.workspace = true

src/common/datasource/src/file_format.rs

@@ -35,7 +35,7 @@ use datafusion::datasource::file_format::file_compression_type::FileCompressionT
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::object_store::ObjectStoreUrl;
 use datafusion::datasource::physical_plan::{
-    FileGroup, FileOpenFuture, FileScanConfigBuilder, FileSource, FileStream,
+    FileGroup, FileOpenFuture, FileScanConfigBuilder, FileSource, FileStreamBuilder,
 };
 use datafusion::error::{DataFusionError, Result as DataFusionResult};
 use datafusion::physical_plan::SendableRecordBatchStream;
@@ -321,7 +321,11 @@ pub async fn file_to_stream(
 
     let store = Arc::new(object_store_opendal::OpendalStore::new(store.clone()));
     let file_opener = config.file_source().create_file_opener(store, &config, 0)?;
-    let stream = FileStream::new(&config, 0, file_opener, &ExecutionPlanMetricsSet::new())?;
+    let stream = FileStreamBuilder::new(&config)
+        .with_partition(0)
+        .with_file_opener(file_opener)
+        .with_metrics(&ExecutionPlanMetricsSet::new())
+        .build()?;
 
     Ok(Box::pin(stream))
 }

src/common/datasource/src/file_format/tests.rs

@@ -19,7 +19,7 @@ use std::{assert_matches, vec};
 use common_test_util::find_workspace_path;
 use datafusion::assert_batches_eq;
 use datafusion::datasource::physical_plan::{
-    CsvSource, FileScanConfig, FileSource, FileStream, JsonSource, ParquetSource,
+    CsvSource, FileScanConfig, FileSource, FileStreamBuilder, JsonSource, ParquetSource,
 };
 use datafusion::datasource::source::DataSourceExec;
 use datafusion::execution::context::TaskContext;
@@ -50,16 +50,15 @@ impl Test<'_> {
             .create_file_opener(store, &self.config, 0)
             .unwrap();
 
-        let result = FileStream::new(
-            &self.config,
-            0,
-            file_opener,
-            &ExecutionPlanMetricsSet::new(),
-        )
-        .unwrap()
-        .map(|b| b.unwrap())
-        .collect::<Vec<_>>()
-        .await;
+        let result = FileStreamBuilder::new(&self.config)
+            .with_partition(0)
+            .with_file_opener(file_opener)
+            .with_metrics(&ExecutionPlanMetricsSet::new())
+            .build()
+            .unwrap()
+            .map(|b| b.unwrap())
+            .collect::<Vec<_>>()
+            .await;
 
         assert_batches_eq!(self.expected, &result);
     }