chore: upgrade DataFusion family, again

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 = "57.0", features = ["prettyprint"] }
-arrow-array = { version = "57.0", default-features = false, features = ["chrono-tz"] }
-arrow-buffer = "57.0"
-arrow-cast = "57.0"
-arrow-flight = "57.0"
-arrow-ipc = { version = "57.0", default-features = false, features = ["lz4", "zstd"] }
-arrow-schema = { version = "57.0", features = ["serde"] }
+arrow = { version = "57.2", features = ["prettyprint"] }
+arrow-array = { version = "57.2", default-features = false, features = ["chrono-tz"] }
+arrow-buffer = "57.2"
+arrow-cast = "57.2"
+arrow-flight = "57.2"
+arrow-ipc = { version = "57.2", default-features = false, features = ["lz4", "zstd"] }
+arrow-schema = { version = "57.2", features = ["serde"] }
 async-stream = "0.3"
 async-trait = "0.1"
 # Remember to update axum-extra, axum-macros when updating axum
@@ -127,20 +127,20 @@ const_format = "0.2"
 criterion = "0.7"
 crossbeam-utils = "0.8"
 dashmap = "6.1"
-datafusion = "51.0"
-datafusion-common = "51.0"
-datafusion-datasource = "51.0"
-datafusion-expr = "51.0"
-datafusion-functions = "51.0"
-datafusion-functions-aggregate-common = "51.0"
-datafusion-functions-window-common = "51.0"
-datafusion-optimizer = "51.0"
-datafusion-orc = { git = "https://github.com/GreptimeTeam/datafusion-orc.git", rev = "35f2e04bf81f2ab7b6f86c0450d6a77b7098d43e" }
-datafusion-pg-catalog = "0.13"
-datafusion-physical-expr = "51.0"
-datafusion-physical-plan = "51.0"
-datafusion-sql = "51.0"
-datafusion-substrait = "51.0"
+datafusion = "52.1"
+datafusion-common = "52.1"
+datafusion-datasource = "52.1"
+datafusion-expr = "52.1"
+datafusion-functions = "52.1"
+datafusion-functions-aggregate-common = "52.1"
+datafusion-functions-window-common = "52.1"
+datafusion-optimizer = "52.1"
+datafusion-orc = "0.7"
+datafusion-pg-catalog = "0.15.1"
+datafusion-physical-expr = "52.1"
+datafusion-physical-plan = "52.1"
+datafusion-sql = "52.1"
+datafusion-substrait = "52.1"
 deadpool = "0.12"
 deadpool-postgres = "0.14"
 derive_builder = "0.20"
@@ -187,7 +187,7 @@ otel-arrow-rust = { git = "https://github.com/GreptimeTeam/otel-arrow", rev = "5
     "server",
 ] }
 parking_lot = "0.12"
-parquet = { version = "57.0", default-features = false, features = ["arrow", "async", "object_store"] }
+parquet = { version = "57.2", default-features = false, features = ["arrow", "async", "object_store"] }
 paste = "1.0"
 pin-project = "1.0"
 pretty_assertions = "1.4.0"
@@ -226,7 +226,7 @@ simd-json = "0.15"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
-sqlparser = { version = "0.59.0", default-features = false, features = ["std", "visitor", "serde"] }
+sqlparser = { version = "0.61.0", default-features = false, features = ["std", "visitor", "serde"] }
 sqlx = { version = "0.8", default-features = false, features = [
     "any",
     "macros",
@@ -331,21 +331,21 @@ git = "https://github.com/GreptimeTeam/greptime-meter.git"
 rev = "5618e779cf2bb4755b499c630fba4c35e91898cb"
 
 [patch.crates-io]
-datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-pg-catalog = { git = "https://github.com/GreptimeTeam/datafusion-postgres.git", rev = "f675927a79cd714a8eeb438b0d3015cd54d4e60a" }
-datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7143b2fc4492a7970774583ed0997a459f3e5c05" }
-sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "d7d95a44889e099e32d78e9bad9bc00598faef28" }                           # on branch v0.59.x
+datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "02b82535e0160c4545667f36a03e1ff9d1d2e51f" }
+sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "2aefa08a8d69c96eec2d6d6703598a009bba6e4c" }                           # on branch v0.61.x
+
 [profile.release]
 debug = 1
 

flake.nix

@@ -15,6 +15,7 @@
       let
         pkgs = nixpkgs.legacyPackages.${system};
         buildInputs = with pkgs; [
+          libz.out
         ];
         lib = nixpkgs.lib;
         rustToolchain = fenix.packages.${system}.fromToolchainName {
@@ -53,6 +54,7 @@
           NIX_HARDENING_ENABLE = "";
           LD_LIBRARY_PATH = with pkgs; lib.makeLibraryPath [
             stdenv.cc.cc.lib
+            libz
           ];
         };
       });

src/common/datasource/Cargo.toml

@@ -10,7 +10,7 @@ workspace = true
 [dependencies]
 arrow.workspace = true
 arrow-schema.workspace = true
-async-compression = { version = "0.3", features = [
+async-compression = { version = "0.4", features = [
     "bzip2",
     "gzip",
     "xz",

src/common/datasource/src/error.rs

@@ -203,8 +203,8 @@ pub enum Error {
         error: parquet::errors::ParquetError,
     },
 
-    #[snafu(display("Failed to build file stream"))]
-    BuildFileStream {
+    #[snafu(transparent)]
+    DataFusion {
         #[snafu(implicit)]
         location: Location,
         #[snafu(source)]
@@ -241,8 +241,9 @@ impl ErrorExt for Error {
             | ReadRecordBatch { .. }
             | WriteRecordBatch { .. }
             | EncodeRecordBatch { .. }
-            | BuildFileStream { .. }
             | OrcReader { .. } => StatusCode::Unexpected,
+
+            DataFusion { .. } => StatusCode::Internal,
         }
     }
 

src/common/datasource/src/file_format.rs

@@ -313,16 +313,13 @@ pub async fn file_to_stream(
                 filename.to_string(),
                 0,
             )]))
-            .with_projection_indices(projection)
+            .with_projection_indices(projection)?
             .with_file_compression_type(df_compression)
             .build();
 
     let store = Arc::new(OpendalStore::new(store.clone()));
-    let file_opener = file_source
-        .with_projection(&config)
-        .create_file_opener(store, &config, 0);
-    let stream = FileStream::new(&config, 0, file_opener, &ExecutionPlanMetricsSet::new())
-        .context(error::BuildFileStreamSnafu)?;
+    let file_opener = config.file_source().create_file_opener(store, &config, 0)?;
+    let stream = FileStream::new(&config, 0, file_opener, &ExecutionPlanMetricsSet::new())?;
 
     Ok(Box::pin(stream))
 }

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

@@ -46,7 +46,10 @@ struct Test<'a> {
 impl Test<'_> {
     async fn run(self, store: &ObjectStore) {
         let store = Arc::new(object_store_opendal::OpendalStore::new(store.clone()));
-        let file_opener = self.file_source.create_file_opener(store, &self.config, 0);
+        let file_opener = self
+            .file_source
+            .create_file_opener(store, &self.config, 0)
+            .unwrap();
 
         let result = FileStream::new(
             &self.config,

src/common/datasource/src/test_util.rs

@@ -104,6 +104,7 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
         schema.clone(),
         FileCompressionType::UNCOMPRESSED,
         Arc::new(object_store_opendal::OpendalStore::new(store.clone())),
+        true,
     );
 
     let size = store.read(origin_path).await.unwrap().len();
@@ -154,11 +155,13 @@ pub async fn setup_stream_to_csv_test(
     let config = scan_config(None, origin_path, csv_source.clone());
     let size = store.read(origin_path).await.unwrap().len();
 
-    let csv_opener = csv_source.create_file_opener(
-        Arc::new(object_store_opendal::OpendalStore::new(store.clone())),
-        &config,
-        0,
-    );
+    let csv_opener = csv_source
+        .create_file_opener(
+            Arc::new(object_store_opendal::OpendalStore::new(store.clone())),
+            &config,
+            0,
+        )
+        .unwrap();
     let stream = FileStream::new(&config, 0, csv_opener, &ExecutionPlanMetricsSet::new()).unwrap();
 
     let (tmp_store, dir) = test_tmp_store("test_stream_to_csv");

src/common/function/src/aggrs/aggr_wrapper/tests.rs

@@ -61,18 +61,18 @@ use crate::scalars::uddsketch_calc::UddSketchCalcFunction;
 pub struct MockInputExec {
     input: Vec<RecordBatch>,
     schema: SchemaRef,
-    properties: PlanProperties,
+    properties: Arc<PlanProperties>,
 }
 
 impl MockInputExec {
     pub fn new(input: Vec<RecordBatch>, schema: SchemaRef) -> Self {
         Self {
-            properties: PlanProperties::new(
+            properties: Arc::new(PlanProperties::new(
                 EquivalenceProperties::new(schema.clone()),
                 Partitioning::UnknownPartitioning(1),
                 EmissionType::Incremental,
                 Boundedness::Bounded,
-            ),
+            )),
             input,
             schema,
         }
@@ -94,7 +94,7 @@ impl ExecutionPlan for MockInputExec {
         self
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
 

src/common/function/src/aggrs/count_hash.rs

@@ -428,7 +428,7 @@ impl Accumulator for CountHashAccumulator {
             &self.random_state,
             &mut self.batch_hashes,
         )?;
-        for hash in hashes.as_slice() {
+        for hash in hashes {
             self.values.insert(*hash);
         }
         Ok(())

src/common/query/src/logical_plan.rs

@@ -48,7 +48,7 @@ pub fn rename_logical_plan_columns(
             plan.schema().qualified_field_from_column(&old_column)?;
 
         for (qualifier, field) in plan.schema().iter() {
-            if qualifier.eq(&qualifier_rename) && field.as_ref() == field_rename {
+            if qualifier.eq(&qualifier_rename) && field == field_rename {
                 projection.push(col(Column::from((qualifier, field))).alias(new_name));
             }
         }

src/common/query/src/stream.rs

@@ -32,7 +32,7 @@ pub struct StreamScanAdapter {
     stream: Mutex<Option<SendableRecordBatchStream>>,
     schema: SchemaRef,
     arrow_schema: ArrowSchemaRef,
-    properties: PlanProperties,
+    properties: Arc<PlanProperties>,
     output_ordering: Option<Vec<PhysicalSortExpr>>,
 }
 
@@ -49,12 +49,12 @@ impl StreamScanAdapter {
     pub fn new(stream: SendableRecordBatchStream) -> Self {
         let schema = stream.schema();
         let arrow_schema = schema.arrow_schema().clone();
-        let properties = PlanProperties::new(
+        let properties = Arc::new(PlanProperties::new(
             EquivalenceProperties::new(arrow_schema.clone()),
             Partitioning::UnknownPartitioning(1),
             EmissionType::Incremental,
             Boundedness::Bounded,
-        );
+        ));
 
         Self {
             stream: Mutex::new(Some(stream)),
@@ -91,7 +91,7 @@ impl ExecutionPlan for StreamScanAdapter {
         self.arrow_schema.clone()
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
 

src/datatypes/src/value.rs

@@ -1229,7 +1229,8 @@ impl TryFrom<ScalarValue> for Value {
             | ScalarValue::Utf8View(_)
             | ScalarValue::BinaryView(_)
             | ScalarValue::Map(_)
-            | ScalarValue::Date64(_) => {
+            | ScalarValue::Date64(_)
+            | ScalarValue::RunEndEncoded(_, _, _) => {
                 return error::UnsupportedArrowTypeSnafu {
                     arrow_type: v.data_type(),
                 }

src/datatypes/src/vectors/helper.rs

@@ -256,7 +256,8 @@ impl Helper {
             | ScalarValue::Utf8View(_)
             | ScalarValue::BinaryView(_)
             | ScalarValue::Map(_)
-            | ScalarValue::Date64(_) => {
+            | ScalarValue::Date64(_)
+            | ScalarValue::RunEndEncoded(_, _, _) => {
                 return error::ConversionSnafu {
                     from: format!("Unsupported scalar value: {value}"),
                 }

src/file-engine/src/error.rs

@@ -178,8 +178,8 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Failed to extract column from filter"))]
-    ExtractColumnFromFilter {
+    #[snafu(transparent)]
+    DataFusion {
         #[snafu(source)]
         error: DataFusionError,
         #[snafu(implicit)]
@@ -225,8 +225,9 @@ impl ErrorExt for Error {
             | ManifestExists { .. }
             | BuildStream { .. }
             | ParquetScanPlan { .. }
-            | UnexpectedEngine { .. }
-            | ExtractColumnFromFilter { .. } => StatusCode::Unexpected,
+            | UnexpectedEngine { .. } => StatusCode::Unexpected,
+
+            DataFusion { .. } => StatusCode::Internal,
         }
     }
 

src/file-engine/src/query.rs

@@ -37,10 +37,7 @@ use snafu::{GenerateImplicitData, ResultExt, ensure};
 use store_api::storage::ScanRequest;
 
 use self::file_stream::ScanPlanConfig;
-use crate::error::{
-    BuildBackendSnafu, ExtractColumnFromFilterSnafu, ProjectSchemaSnafu,
-    ProjectionOutOfBoundsSnafu, Result,
-};
+use crate::error::{BuildBackendSnafu, ProjectSchemaSnafu, ProjectionOutOfBoundsSnafu, Result};
 use crate::region::FileRegion;
 
 impl FileRegion {
@@ -126,8 +123,7 @@ impl FileRegion {
 
         let mut aux_column_set = HashSet::new();
         for scan_filter in scan_filters {
-            df_logical_expr_utils::expr_to_columns(scan_filter, &mut aux_column_set)
-                .context(ExtractColumnFromFilterSnafu)?;
+            df_logical_expr_utils::expr_to_columns(scan_filter, &mut aux_column_set)?;
 
             let all_file_columns = aux_column_set
                 .iter()

src/file-engine/src/query/file_stream.rs

@@ -56,7 +56,7 @@ fn build_record_batch_stream(
 
     let config =
         FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source.clone())
-            .with_projection_indices(scan_plan_config.projection.cloned())
+            .with_projection_indices(scan_plan_config.projection.cloned())?
             .with_limit(limit)
             .with_file_group(FileGroup::new(files))
             .build();
@@ -65,7 +65,7 @@ fn build_record_batch_stream(
         scan_plan_config.store.clone(),
     ));
 
-    let file_opener = file_source.create_file_opener(store, &config, 0);
+    let file_opener = config.file_source().create_file_opener(store, &config, 0)?;
     let stream = FileStream::new(
         &config,
         0, // partition: hard-code
@@ -146,7 +146,7 @@ fn new_parquet_stream_with_exec_plan(
     let file_scan_config =
         FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), Arc::new(parquet_source))
             .with_file_group(file_group)
-            .with_projection_indices(projection.cloned())
+            .with_projection_indices(projection.cloned())?
             .with_limit(*limit)
             .build();