Distributed hive-style join testing with superset satisfaction

Cargo.toml

@@ -2,8 +2,15 @@
 members = ["benchmarks"]
 
 [workspace.dependencies]
-datafusion = { version = "51.0.0", default-features = false }
-datafusion-proto = { version = "51.0.0" }
+# Using PR #19304: Hash partitioning satisfies subset
+# https://github.com/apache/datafusion/pull/19304
+# This PR includes:
+# - Hash subset satisfaction logic (eliminates unnecessary repartitions)
+# - File partitioning preservation (from PR #19124)
+# datafusion = { git = "https://github.com/gene-bordegaray/datafusion.git", branch = "gene.bordegaray/2025/12/hash_partitioning_satisfies_subset", default-features = false }
+# datafusion-proto = { git = "https://github.com/gene-bordegaray/datafusion.git", branch = "gene.bordegaray/2025/12/hash_partitioning_satisfies_subset" }
+  datafusion = { git = "https://github.com/gene-bordegaray/datafusion.git", branch = "gene.bordegaray/2025/12/hash_superset_satisfies_partitioning", default-features = false }
+  datafusion-proto = { git = "https://github.com/gene-bordegaray/datafusion.git", branch = "gene.bordegaray/2025/12/hash_superset_satisfies_partitioning" }
 
 [package]
 name = "datafusion-distributed"
@@ -68,7 +75,7 @@ insta = { version = "1.43.1", features = ["filters"] }
 tpchgen = { git = "https://github.com/clflushopt/tpchgen-rs", rev = "e83365a5a9101906eb9f78c5607b83bc59849acf" }
 tpchgen-arrow = { git = "https://github.com/clflushopt/tpchgen-rs", rev = "e83365a5a9101906eb9f78c5607b83bc59849acf" }
 parquet = "57.0.0"
-arrow = "57.0.0"
+arrow = "57.0.1"
 tokio-stream = "0.1.17"
 hyper-util = "0.1.16"
 pretty_assertions = "1.4"

src/metrics/proto.rs

@@ -288,7 +288,7 @@ pub fn df_metric_to_proto(metric: Arc<Metric>) -> Result<MetricProto, DataFusion
             labels,
         }),
         MetricValue::Custom { .. } => internal_err!("{}", CUSTOM_METRICS_NOT_SUPPORTED),
-        MetricValue::OutputBytes(_) | MetricValue::PruningMetrics { .. } | MetricValue::Ratio { .. } => {
+        MetricValue::OutputBytes(_) | MetricValue::PruningMetrics { .. } | MetricValue::Ratio { .. } | MetricValue::OutputBatches(_) => {
             // TODO: Support these metrics
             internal_err!("{}", UNSUPPORTED_METRICS)
         }

src/protobuf/errors/datafusion_error.rs

@@ -173,6 +173,9 @@ impl DataFusionErrorProto {
                     DataFusionErrorProto::from_datafusion_error(err.as_ref()),
                 ))),
             },
+            DataFusionError::Ffi(err) => DataFusionErrorProto {
+                inner: Some(DataFusionErrorInnerProto::Plan(err.clone())),
+            },
         }
     }
 

testdata/join/csv/dim/d_dkey=A/data0.csv

@@ -0,0 +1,3 @@
+env,service,host
+dev,log,host-y
+

testdata/join/csv/dim/d_dkey=B/data0.csv

@@ -0,0 +1,3 @@
+env,service,host
+prod,log,host-x
+

testdata/join/csv/dim/d_dkey=C/data0.csv

@@ -0,0 +1,3 @@
+env,service,host
+dev,trace,host-z
+

testdata/join/csv/dim/d_dkey=D/data0.csv

@@ -0,0 +1,3 @@
+env,service,host
+prod,trace,host-x
+

testdata/join/csv/fact/f_dkey=A/data0.csv

@@ -0,0 +1,9 @@
+timestamp,value
+2023-01-01T09:00:00,95.5
+2023-01-01T09:00:10,102.3
+2023-01-01T09:00:20,98.7
+2023-01-01T09:12:20,105.1
+2023-01-01T09:12:30,100.0
+2023-01-01T09:12:40,150.0
+2023-01-01T09:12:50,120.8
+

testdata/join/csv/fact/f_dkey=B/data0.csv

@@ -0,0 +1,9 @@
+timestamp,value
+2023-01-01T09:00:00,75.2
+2023-01-01T09:00:10,82.4
+2023-01-01T09:00:20,78.9
+2023-01-01T09:00:30,85.6
+2023-01-01T09:12:30,80.0
+2023-01-01T09:12:40,120.0
+2023-01-01T09:12:50,92.3
+

testdata/join/csv/fact/f_dkey=B/data1.csv

@@ -0,0 +1,8 @@
+timestamp,value
+2023-01-01T10:00:00,88.5
+2023-01-01T10:00:10,91.2
+2023-01-01T10:00:20,87.3
+2023-01-01T10:00:30,94.1
+2023-01-01T10:12:30,89.5
+2023-01-01T10:12:40,95.8
+

testdata/join/csv/fact/f_dkey=B/data2.csv

@@ -0,0 +1,7 @@
+timestamp,value
+2023-01-01T11:00:00,72.8
+2023-01-01T11:00:10,79.4
+2023-01-01T11:00:20,76.1
+2023-01-01T11:00:30,83.7
+2023-01-01T11:12:30,77.2
+

testdata/join/csv/fact/f_dkey=C/data0.csv

@@ -0,0 +1,9 @@
+timestamp,value
+2023-01-01T10:00:00,310.5
+2023-01-01T10:00:10,225.7
+2023-01-01T10:00:20,380.2
+2023-01-01T10:00:30,205.8
+2023-01-01T10:00:40,350.0
+2023-01-01T10:12:40,200.0
+2023-01-01T10:12:50,205.4
+

testdata/join/csv/fact/f_dkey=C/data1.csv

@@ -0,0 +1,9 @@
+timestamp,value
+2023-01-01T11:00:00,295.3
+2023-01-01T11:00:10,318.6
+2023-01-01T11:00:20,342.9
+2023-01-01T11:00:30,287.4
+2023-01-01T11:00:40,365.2
+2023-01-01T11:12:40,310.8
+2023-01-01T11:12:50,298.1
+

testdata/join/csv/fact/f_dkey=D/data0.csv

@@ -0,0 +1,5 @@
+timestamp,value
+2023-01-01T10:00:00,24.8
+2023-01-01T10:00:10,72.1
+2023-01-01T10:00:20,42.5
+

testdata/join/generate_parquet_from_csv.sql

@@ -0,0 +1,48 @@
+-- datafusion-cli -f testdata/join/generate_parquet_from_csv.sql
+
+-- Generate parquet dim files from csv files.
+COPY (SELECT * FROM "testdata/join/csv/dim/d_dkey=A/data0.csv")
+TO "testdata/join/parquet/dim/d_dkey=A/data0.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/dim/d_dkey=B/data0.csv")
+TO "testdata/join/parquet/dim/d_dkey=B/data0.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/dim/d_dkey=C/data0.csv")
+TO "testdata/join/parquet/dim/d_dkey=C/data0.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/dim/d_dkey=D/data0.csv")
+TO "testdata/join/parquet/dim/d_dkey=D/data0.parquet"
+STORED AS PARQUET;
+
+-- Generate parquet fact files from csv files.
+COPY (SELECT * FROM "testdata/join/csv/fact/f_dkey=A/data0.csv")
+TO "testdata/join/parquet/fact/f_dkey=A/data0.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/fact/f_dkey=B/data0.csv")
+TO "testdata/join/parquet/fact/f_dkey=B/data0.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/fact/f_dkey=B/data1.csv")
+TO "testdata/join/parquet/fact/f_dkey=B/data1.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/fact/f_dkey=B/data2.csv")
+TO "testdata/join/parquet/fact/f_dkey=B/data2.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/fact/f_dkey=C/data0.csv")
+TO "testdata/join/parquet/fact/f_dkey=C/data0.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/fact/f_dkey=C/data1.csv")
+TO "testdata/join/parquet/fact/f_dkey=C/data1.parquet"
+STORED AS PARQUET;
+
+COPY (SELECT * FROM "testdata/join/csv/fact/f_dkey=D/data0.csv")
+TO "testdata/join/parquet/fact/f_dkey=D/data0.parquet"
+STORED AS PARQUET;
+