chore(deps): bump datafusion to 55 and arrow to 59

Cargo.toml

@@ -30,25 +30,25 @@ liquid-cache-datafusion = { path = "src/datafusion", version = "0.1.13" }
 liquid-cache-common = { path = "src/common", version = "0.1.13" }
 liquid-cache = { path = "src/core", version = "0.1.13" }
 liquid-cache-datafusion-local = { path = "src/datafusion-local", version = "0.1.13" }
-arrow = { version = "58.3.0", default-features = false, features = [
+arrow = { version = "59.2", default-features = false, features = [
 	"prettyprint",
 	"ipc",
 ] }
-arrow-flight = { version = "58.3.0", features = ["flight-sql-experimental"] }
-arrow-schema = { version = "58.3.0", features = ["serde"] }
-parquet = { version = "58.3.0", features = [
+arrow-flight = { version = "59.2", features = ["flight-sql-experimental"] }
+arrow-schema = { version = "59.2", features = ["serde"] }
+parquet = { version = "59.2", features = [
 	"async",
 	"experimental",
 	"variant_experimental",
 ] }
-parquet-variant-json = { version = "58.3.0" }
-parquet-variant-compute = { version = "58.3.0" }
-datafusion = { version = "54.0.0" }
-datafusion-common = { version = "54.0.0" }
-datafusion-expr-common = { version = "54.0.0" }
-datafusion-physical-expr = { version = "54.0.0" }
-datafusion-physical-expr-common = { version = "54.0.0" }
-datafusion-proto = { version = "54.0.0" }
+parquet-variant-json = { version = "59.2" }
+parquet-variant-compute = { version = "59.2" }
+datafusion = { version = "55" }
+datafusion-common = { version = "55" }
+datafusion-expr-common = { version = "55" }
+datafusion-physical-expr = { version = "55" }
+datafusion-physical-expr-common = { version = "55" }
+datafusion-proto = { version = "55" }
 async-trait = "0.1.89"
 futures = { version = "0.3.32", default-features = false, features = ["std"] }
 tokio = { version = "1.52.3", features = ["rt-multi-thread"] }

src/core/src/cache/policies/squeeze.rs

@@ -244,7 +244,7 @@ pub(crate) fn try_variant_squeeze(
         shredded_array = Some(shredded_struct);
     }
 
-    let typed_root = variant_array.typed_value_field()?;
+    let typed_root = variant_array.typed_value_column()?;
     let typed_root = typed_root.as_any().downcast_ref::<StructArray>()?;
 
     let mut collected = Vec::new();
@@ -676,7 +676,9 @@ mod tests {
                 inner
                     .column_by_name("metadata")
                     .cloned()
-                    .unwrap_or_else(|| Arc::new(base_variant.metadata_field().clone()) as ArrayRef),
+                    .unwrap_or_else(|| {
+                        Arc::new(base_variant.metadata_column().clone()) as ArrayRef
+                    }),
                 inner.column_by_name("value").cloned().unwrap_or_else(|| {
                     Arc::new(BinaryViewArray::from(vec![None::<&[u8]>; inner.len()])) as ArrayRef
                 }),

src/datafusion-client/src/client_exec.rs

@@ -10,7 +10,7 @@ use arrow_flight::flight_service_client::FlightServiceClient;
 use arrow_schema::{Schema, SchemaRef};
 use datafusion::catalog::memory::DataSourceExec;
 use datafusion::common::internal_err;
-use datafusion::common::tree_node::{Transformed, TreeNode};
+use datafusion::common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
 use datafusion::config::ConfigOptions;
 use datafusion::datasource::physical_plan::{FileSource, ParquetSource};
 use datafusion::execution::object_store::ObjectStoreUrl;
@@ -140,6 +140,15 @@ impl ExecutionPlan for LiquidCacheClientExec {
         vec![&self.remote_plan]
     }
 
+    /// The client node holds no expressions of its own; the wrapped remote plan
+    /// is visited as a child.
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+    ) -> Result<TreeNodeRecursion> {
+        Ok(TreeNodeRecursion::Continue)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,

src/datafusion-client/src/lib.rs

@@ -9,6 +9,7 @@ mod metrics;
 mod optimizer;
 pub use client_exec::LiquidCacheClientExec;
 use datafusion::{
+    common::config::ConfigNonZeroUsize,
     error::{DataFusionError, Result},
     execution::{SessionStateBuilder, object_store::ObjectStoreUrl, runtime_env::RuntimeEnv},
     prelude::*,
@@ -91,7 +92,7 @@ impl LiquidCacheClientBuilder {
             .execution
             .parquet
             .binary_as_string = true;
-        session_config.options_mut().execution.batch_size = 8192 * 2;
+        session_config.options_mut().execution.batch_size = ConfigNonZeroUsize::try_new(8192 * 2)?;
         // Dynamic filters (e.g. a hash join's runtime build-side filter) are pushed
         // into scan predicates by DataFusion. In distributed mode those scans are
         // serialized and executed on a remote server that can never receive the

src/datafusion-client/src/optimizer.rs

@@ -5,6 +5,7 @@ use datafusion::{
     execution::object_store::ObjectStoreUrl, physical_optimizer::PhysicalOptimizerRule,
     physical_plan::ExecutionPlan, physical_plan::aggregates::AggregateExec,
     physical_plan::aggregates::AggregateMode, physical_plan::repartition::RepartitionExec,
+    physical_plan::replace_children_if_necessary,
 };
 
 use liquid_cache_datafusion::optimizers::SqueezeHintMap;
@@ -69,22 +70,13 @@ impl PushdownOptimizer {
 
         // Otherwise, recurse into children
         let mut new_children = Vec::with_capacity(plan.children().len());
-        let mut children_changed = false;
 
         for child in plan.children() {
-            let new_child = self.optimize_plan(child.clone(), hints)?;
-            if !Arc::ptr_eq(child, &new_child) {
-                children_changed = true;
-            }
-            new_children.push(new_child);
+            new_children.push(self.optimize_plan(child.clone(), hints)?);
         }
 
-        // If any children were changed, create a new plan with the updated children
-        if children_changed {
-            plan.with_new_children(new_children)
-        } else {
-            Ok(plan)
-        }
+        // Returns the plan untouched when every child is unchanged.
+        replace_children_if_necessary(plan, new_children)
     }
 }
 

src/datafusion-client/src/tests/snapshots/liquid_cache_datafusion_client__tests__tpch_q1.snap

@@ -10,13 +10,6 @@ expression: displayable.tree_render().to_string()
 │       ASC NULLS LAST      │
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
-│          SortExec         │
-│    --------------------   │
-│  l_returnflag@0 ASC NULLS │
-│    LAST, l_linestatus@1   │
-│       ASC NULLS LAST      │
-└─────────────┬─────────────┘
-┌─────────────┴─────────────┐
 │       ProjectionExec      │
 │    --------------------   │
 │         avg_disc:         │
@@ -51,6 +44,13 @@ expression: displayable.tree_render().to_string()
 │            ...            │
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
+│          SortExec         │
+│    --------------------   │
+│  l_returnflag@0 ASC NULLS │
+│    LAST, l_linestatus@1   │
+│       ASC NULLS LAST      │
+└─────────────┬─────────────┘
+┌─────────────┴─────────────┐
 │       AggregateExec       │
 │    --------------------   │
 │           aggr:           │
@@ -62,14 +62,13 @@ expression: displayable.tree_render().to_string()
 │     .l_extendedprice *    │
 │     Int64(1) - lineitem   │
 │     .l_discount), sum     │
-│     (__common_expr_1 *    │
-│       Some(1),20,0 +      │
-│     lineitem.l_tax) as    │
-│        sum(lineitem       │
-│       .l_extendedpri      │
-│  ce * Int64(1) - lineitem │
-│  .l_discount * Int64(1)   │
-│   + lineitem.l_tax), avg  │
+│   (__common_expr_1 * 1 +  │
+│   lineitem.l_tax) as sum  │
+│         (lineitem         │
+│        .l_extendedp       │
+│ rice * Int64(1) - lineitem│
+│  .l_discount * Int64(1) + │
+│    lineitem.l_tax), avg   │
 │   (lineitem.l_quantity),  │
 │        avg(lineitem       │
 │       .l_extendedpri      │
@@ -105,14 +104,13 @@ expression: displayable.tree_render().to_string()
 │     .l_extendedprice *    │
 │     Int64(1) - lineitem   │
 │     .l_discount), sum     │
-│     (__common_expr_1 *    │
-│       Some(1),20,0 +      │
-│     lineitem.l_tax) as    │
-│        sum(lineitem       │
-│       .l_extendedpri      │
-│  ce * Int64(1) - lineitem │
-│  .l_discount * Int64(1)   │
-│   + lineitem.l_tax), avg  │
+│   (__common_expr_1 * 1 +  │
+│   lineitem.l_tax) as sum  │
+│         (lineitem         │
+│        .l_extendedp       │
+│ rice * Int64(1) - lineitem│
+│  .l_discount * Int64(1) + │
+│    lineitem.l_tax), avg   │
 │   (lineitem.l_quantity),  │
 │        avg(lineitem       │
 │       .l_extendedpri      │
@@ -129,8 +127,8 @@ expression: displayable.tree_render().to_string()
 │       ProjectionExec      │
 │    --------------------   │
 │      __common_expr_1:     │
-│ l_extendedprice * (Some(1)│
-│    ,20,0 - l_discount)    │
+│   l_extendedprice * (1 -  │
+│         l_discount)       │
 │                           │
 │        l_discount:        │
 │         l_discount        │

src/datafusion-client/src/tests/snapshots/liquid_cache_datafusion_client__tests__tpch_q10.snap

@@ -10,13 +10,6 @@ expression: displayable.tree_render().to_string()
 │        revenue DESC       │
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
-│       SortExec(TopK)      │
-│    --------------------   │
-│         limit: 20         │
-│                           │
-│       revenue@2 DESC      │
-└─────────────┬─────────────┘
-┌─────────────┴─────────────┐
 │       ProjectionExec      │
 │    --------------------   │
 │    c_acctbal: c_acctbal   │
@@ -34,16 +27,26 @@ expression: displayable.tree_render().to_string()
 │        .l_discount)       │
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
+│       SortExec(TopK)      │
+│    --------------------   │
+│         limit: 20         │
+│                           │
+│        sum(lineitem       │
+│        .l_extendedp       │
+│ rice * Int64(1) - lineitem│
+│    .l_discount)@7 DESC    │
+└─────────────┬─────────────┘
+┌─────────────┴─────────────┐
 │       AggregateExec       │
 │    --------------------   │
 │           aggr:           │
 │        sum(lineitem       │
 │        .l_extendedp       │
-│   rice * Some(1),20,0 -   │
-│     lineitem.l_discount   │
-│     ) as sum(lineitem     │
-│     .l_extendedprice      │
-│   * Int64(1) - lineitem   │
+│    rice * 1 - lineitem    │
+│      .l_discount) as      │
+│        sum(lineitem       │
+│      .l_extendedprice     │
+│    * Int64(1) - lineitem  │
 │        .l_discount)       │
 │                           │
 │         group_by:         │
@@ -73,11 +76,11 @@ expression: displayable.tree_render().to_string()
 │           aggr:           │
 │        sum(lineitem       │
 │        .l_extendedp       │
-│   rice * Some(1),20,0 -   │
-│     lineitem.l_discount   │
-│     ) as sum(lineitem     │
-│     .l_extendedprice      │
-│   * Int64(1) - lineitem   │
+│    rice * 1 - lineitem    │
+│      .l_discount) as      │
+│        sum(lineitem       │
+│      .l_extendedprice     │
+│    * Int64(1) - lineitem  │
 │        .l_discount)       │
 │                           │
 │         group_by:         │

src/datafusion-client/src/tests/snapshots/liquid_cache_datafusion_client__tests__tpch_q11.snap

@@ -19,23 +19,21 @@ expression: displayable.tree_render().to_string()
 │                           │                                                          │     Decimal128(38, 15))   │
 └─────────────┬─────────────┘                                                          └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐                                                          ┌─────────────┴─────────────┐
-│          SortExec         │                                                          │       AggregateExec       │
+│       ProjectionExec      │                                                          │       AggregateExec       │
 │    --------------------   │                                                          │    --------------------   │
-│        value@1 DESC       │                                                          │           aggr:           │
-│                           │                                                          │ sum(partsupp.ps_supplycost│
+│        ps_partkey:        │                                                          │           aggr:           │
+│         ps_partkey        │                                                          │ sum(partsupp.ps_supplycost│
 │                           │                                                          │   * partsupp.ps_availqty) │
-│                           │                                                          │                           │
-│                           │                                                          │        mode: Final        │
+│           value:          │                                                          │                           │
+│ sum(partsupp.ps_supplycost│                                                          │        mode: Final        │
+│   * partsupp.ps_availqty) │                                                          │                           │
 └─────────────┬─────────────┘                                                          └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐                                                          ┌─────────────┴─────────────┐
-│       ProjectionExec      │                                                          │   CoalescePartitionsExec  │
+│          SortExec         │                                                          │   CoalescePartitionsExec  │
 │    --------------------   │                                                          │                           │
-│        ps_partkey:        │                                                          │                           │
-│         ps_partkey        │                                                          │                           │
-│                           │                                                          │                           │
-│           value:          │                                                          │                           │
 │ sum(partsupp.ps_supplycost│                                                          │                           │
 │   * partsupp.ps_availqty) │                                                          │                           │
+│          @1 DESC          │                                                          │                           │
 └─────────────┬─────────────┘                                                          └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐                                                          ┌─────────────┴─────────────┐
 │         FilterExec        │                                                          │       AggregateExec       │

src/datafusion-client/src/tests/snapshots/liquid_cache_datafusion_client__tests__tpch_q12.snap

@@ -8,11 +8,6 @@ expression: displayable.tree_render().to_string()
 │ l_shipmode ASC NULLS LAST │
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
-│          SortExec         │
-│    --------------------   │
-│l_shipmode@0 ASC NULLS LAST│
-└─────────────┬─────────────┘
-┌─────────────┴─────────────┐
 │       ProjectionExec      │
 │    --------------------   │
 │      high_line_count:     │
@@ -37,6 +32,11 @@ expression: displayable.tree_render().to_string()
 │            END)           │
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
+│          SortExec         │
+│    --------------------   │
+│l_shipmode@0 ASC NULLS LAST│
+└─────────────┬─────────────┘
+┌─────────────┴─────────────┐
 │       AggregateExec       │
 │    --------------------   │
 │           aggr:           │

src/datafusion-client/src/tests/snapshots/liquid_cache_datafusion_client__tests__tpch_q13.snap

@@ -8,12 +8,6 @@ expression: displayable.tree_render().to_string()
 │custdist DESC, c_count DESC│
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
-│          SortExec         │
-│    --------------------   │
-│ custdist@1 DESC, c_count@0│
-│            DESC           │
-└─────────────┬─────────────┘
-┌─────────────┴─────────────┐
 │       ProjectionExec      │
 │    --------------------   │
 │      c_count: c_count     │
@@ -22,6 +16,12 @@ expression: displayable.tree_render().to_string()
 │      count(Int64(1))      │
 └─────────────┬─────────────┘
 ┌─────────────┴─────────────┐
+│          SortExec         │
+│    --------------------   │
+│  count(Int64(1))@1 DESC,  │
+│       c_count@0 DESC      │
+└─────────────┬─────────────┘
+┌─────────────┴─────────────┐
 │       AggregateExec       │
 │    --------------------   │
 │       aggr: count(1)      │