Upgrade materialized views to DataFusion 53 (#46)

Author: xudong963

Cargo.toml

@@ -25,27 +25,27 @@ authors = ["Matthew Cramerus <matt@polygon.io>"]
 license = "Apache-2.0"
 description = "Materialized Views & Query Rewriting in DataFusion"
 keywords = ["arrow", "arrow-rs", "datafusion"]
-rust-version = "1.85.1"
+rust-version = "1.88.0"
 
 [dependencies]
 aquamarine = "0.6.0"
-arrow = "57.1.0"
-arrow-schema = "57.1.0"
+arrow = "58.0.0"
+arrow-schema = "58.0.0"
 async-trait = "0.1.89"
 dashmap = "6"
-datafusion = "52"
-datafusion-common = "52"
-datafusion-expr = "52"
-datafusion-functions = "52"
-datafusion-functions-aggregate = "52"
-datafusion-optimizer = "52"
-datafusion-physical-expr = "52"
-datafusion-physical-plan = "52"
-datafusion-sql = "52"
+datafusion = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-common = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-expr = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-functions = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-functions-aggregate = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-optimizer = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-physical-expr = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-physical-plan = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
+datafusion-sql = { git = "https://github.com/massive-com/arrow-datafusion", rev = "f069fa60c2ff13411922fca07f1e4980571e80fb" }
 futures = "0.3"
 itertools = "0.14"
 log = "0.4"
-object_store = "0.12.4"
+object_store = "0.13.1"
 ordered-float = "5.0.0"
 
 [dev-dependencies]

src/materialized/dependencies.rs

@@ -42,13 +42,13 @@ use datafusion::{
     catalog::{CatalogProviderList, TableFunctionImpl},
     config::{CatalogOptions, ConfigOptions},
     datasource::{provider_as_source, TableProvider, ViewTable},
-    prelude::{flatten, get_field, make_array},
+    prelude::{flatten, make_array},
 };
 use datafusion_common::{
     alias::AliasGenerator,
     internal_err,
     tree_node::{Transformed, TreeNode},
-    DFSchema, DataFusionError, Result, ScalarValue,
+    Column as DFColumn, DFSchema, DataFusionError, Result, ScalarValue,
 };
 use datafusion_expr::{
     col, lit, utils::split_conjunction, Expr, LogicalPlan, LogicalPlanBuilder, TableScan,
@@ -400,17 +400,25 @@ pub fn mv_dependencies_plan(
         .into_iter()
         .find(|c| c.name.starts_with(META_COLUMN))
         .ok_or_else(|| DataFusionError::Plan(format!("Plan contains no {META_COLUMN} column")))?;
-    let files_col = Expr::Column(files.clone());
+    let meta_table_catalog =
+        Expr::Column(DFColumn::from_name(format!("{}.table_catalog", files.name)));
+    let meta_table_schema =
+        Expr::Column(DFColumn::from_name(format!("{}.table_schema", files.name)));
+    let meta_table_name = Expr::Column(DFColumn::from_name(format!("{}.table_name", files.name)));
+    let meta_source_uri = Expr::Column(DFColumn::from_name(format!("{}.source_uri", files.name)));
+    let meta_last_modified =
+        Expr::Column(DFColumn::from_name(format!("{}.last_modified", files.name)));
 
     LogicalPlanBuilder::from(pruned_plan_with_source_files)
+        .unnest_column(files.clone())?
         .unnest_column(files)?
         .project(vec![
             construct_target_path_from_static_partition_columns(materialized_view).alias("target"),
-            get_field(files_col.clone(), "table_catalog").alias("source_table_catalog"),
-            get_field(files_col.clone(), "table_schema").alias("source_table_schema"),
-            get_field(files_col.clone(), "table_name").alias("source_table_name"),
-            get_field(files_col.clone(), "source_uri").alias("source_uri"),
-            get_field(files_col.clone(), "last_modified").alias("source_last_modified"),
+            meta_table_catalog.alias("source_table_catalog"),
+            meta_table_schema.alias("source_table_schema"),
+            meta_table_name.alias("source_table_name"),
+            meta_source_uri.alias("source_uri"),
+            meta_last_modified.alias("source_last_modified"),
         ])?
         .distinct()?
         .build()

src/materialized/file_metadata.rs

@@ -43,7 +43,7 @@ use futures::stream::{self, BoxStream};
 use futures::{future, Future, FutureExt, StreamExt, TryStreamExt};
 use itertools::Itertools;
 use log::debug;
-use object_store::{ObjectMeta, ObjectStore};
+use object_store::{ObjectMeta, ObjectStore, ObjectStoreExt};
 use std::any::Any;
 use std::sync::Arc;
 
@@ -137,7 +137,7 @@ impl TableProvider for FileMetadata {
 /// An [`ExecutionPlan`] that scans object store metadata.
 pub struct FileMetadataExec {
     table_schema: SchemaRef,
-    plan_properties: PlanProperties,
+    plan_properties: Arc<PlanProperties>,
     projection: Option<Vec<usize>>,
     filters: Vec<Arc<dyn PhysicalExpr>>,
     limit: Option<usize>,
@@ -170,7 +170,7 @@ impl FileMetadataExec {
 
         let exec = Self {
             table_schema,
-            plan_properties,
+            plan_properties: Arc::new(plan_properties),
             projection,
             filters,
             limit,
@@ -192,7 +192,7 @@ impl ExecutionPlan for FileMetadataExec {
         "FileMetadataExec"
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.plan_properties
     }