Skip to content

Commit a47de78

Browse files
matthewmturnermatthewmturner
authored
DF54 (#378)
1 parent d7cf5fd commit a47de78

18 files changed

Lines changed: 2382 additions & 1801 deletions

File tree

‎Cargo.lock‎

Lines changed: 2269 additions & 1663 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,12 @@ version = "0.2.2"
1616
[dependencies]
1717
arrow-flight = { features = [
1818
"flight-sql-experimental",
19-
], optional = true, version = "57" }
19+
], optional = true, version = "58" }
2020
axum = { features = ["macros"], optional = true, version = "0.7.9" }
2121
clap = { features = ["derive"], version = "4.5.27" }
2222
color-eyre = "0.6.3"
2323
crossterm = { features = ["event-stream"], optional = true, version = "0.29" }
24-
datafusion = { version = "52" }
24+
datafusion = { version = "54" }
2525
datafusion-app = { path = "crates/datafusion-app", version = "0.1.0" }
2626
directories = "5.0.1"
2727
env_logger = "0.11.5"
@@ -35,8 +35,8 @@ lazy_static = "1.4.0"
3535
log = "0.4.22"
3636
metrics = { optional = true, version = "0.24.0" }
3737
metrics-exporter-prometheus = { optional = true, version = "0.16.0" }
38-
object_store = "0.12"
39-
parquet = "57"
38+
object_store = "0.13"
39+
parquet = "58"
4040
pin-project-lite = { version = "0.2.14" }
4141
prost = "0.14"
4242
ratatui = { optional = true, version = "0.30" }
@@ -68,10 +68,10 @@ tui-logger = { features = [
6868
], optional = true, version = "0.18" }
6969
url = { features = ["serde"], version = "2.5.2" }
7070
uuid = { optional = true, version = "1.10.0" }
71-
vortex = { optional = true, version = "0.66" }
72-
vortex-datafusion = { optional = true, version = "0.66" }
73-
vortex-file = { optional = true, version = "0.66" }
74-
vortex-session = { optional = true, version = "0.66" }
71+
vortex = { optional = true, version = "0.76" }
72+
vortex-datafusion = { optional = true, version = "0.76" }
73+
vortex-file = { optional = true, version = "0.76" }
74+
vortex-session = { optional = true, version = "0.76" }
7575

7676
[dev-dependencies]
7777
assert_cmd = "2.0.16"
@@ -134,3 +134,11 @@ path = "src/main.rs"
134134

135135
[lints.clippy]
136136
clone_on_ref_ptr = "deny"
137+
138+
# tpchgen has no arrow 58 compatible release (2.0.2 is arrow 57, 3.0.0 is arrow 59), so we
139+
# use a fork of v2.0.2 with arrow bumped to 58.
140+
# TODO: remove this patch once tpchgen and DataFusion align on an arrow version again
141+
# (tpchgen-arrow 3.0 / arrow 59 should line up with DataFusion 55).
142+
[patch.crates-io]
143+
tpchgen = { git = "https://github.com/matthewmturner/tpchgen-rs", rev = "1e07d12f2677da2ee3b9b502fed8760fac3b7e89" }
144+
tpchgen-arrow = { git = "https://github.com/matthewmturner/tpchgen-rs", rev = "1e07d12f2677da2ee3b9b502fed8760fac3b7e89" }

‎crates/datafusion-app/Cargo.toml‎

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -8,35 +8,41 @@ version = "0.1.0"
88
[dependencies]
99
arrow-flight = { features = [
1010
"flight-sql-experimental",
11-
], optional = true, version = "57" }
11+
], optional = true, version = "58" }
1212
async-trait = "0.1.80"
1313
base64 = { optional = true, version = "0.22.1" }
1414
color-eyre = "0.6.3"
15-
datafusion = "52"
16-
datafusion-functions-json = { optional = true, version = "0.52" }
15+
datafusion = "54"
16+
datafusion-functions-json = { optional = true, version = "0.54" }
1717
datafusion-functions-parquet = { optional = true, path = "../datafusion-functions-parquet", version = "0.1.0" }
1818
datafusion-table-providers = { features = [
1919
"clickhouse",
20-
], optional = true, version = "0.10.1" }
20+
], optional = true, version = "0.12.0" }
2121
datafusion-udfs-wasm = { features = [
2222
"serde",
2323
], optional = true, path = "../datafusion-udfs-wasm", version = "0.1.0" }
24+
# TODO: revert to crates.io once deltalake releases with DataFusion 54 support.
25+
# Pinned to the head of https://github.com/delta-io/delta-rs/pull/4582 because earlier
26+
# revs map the `s3`/`rustls` features to kernel features removed in buoyant_kernel 0.24.
2427
deltalake = { features = [
2528
"datafusion",
2629
"s3",
27-
], optional = true, version = "0.31" }
30+
], git = "https://github.com/delta-io/delta-rs", optional = true, rev = "b1a18b5aa6d1" }
2831
directories = "5.0.1"
2932
futures = "0.3.30"
3033
indexmap = { features = ["serde"], version = "2.8.0" }
3134
itertools = "0.13.0"
3235
log = "0.4.22"
3336
metrics = { optional = true, version = "0.24.0" }
3437
num_cpus = "1.16.0"
35-
object_store = { features = ["aws"], optional = true, version = "0.12" }
36-
object_store_opendal = { optional = true, version = "0.54" }
38+
object_store = { features = ["aws"], optional = true, version = "0.13" }
39+
# TODO: revert to crates.io once object_store_opendal releases with DataFusion 54 support.
40+
# Both crates must come from the same rev because an opendal `Operator` is passed to
41+
# `OpendalStore::new` in the huggingface extension.
42+
object_store_opendal = { git = "https://github.com/apache/opendal", optional = true, rev = "24aaff9c62b1" }
3743
opendal = { features = [
3844
"services-huggingface",
39-
], optional = true, version = "0.54.1" }
45+
], git = "https://github.com/apache/opendal", optional = true, rev = "24aaff9c62b1" }
4046
parking_lot = "0.12.3"
4147
rustls = { default-features = false, features = [
4248
"aws-lc-rs",
@@ -52,7 +58,7 @@ tokio-tungstenite = { features = [
5258
], optional = true, version = "0.29" }
5359
tonic = { optional = true, version = "0.14" }
5460
url = { optional = true, version = "2.5.2" }
55-
vortex-datafusion = { optional = true, version = "0.66" }
61+
vortex-datafusion = { optional = true, version = "0.76" }
5662

5763
[dev-dependencies]
5864
criterion = { features = ["async_tokio"], version = "0.5.1" }

‎crates/datafusion-app/src/catalog/clickhouse.rs‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
//! [`CatalogProvider`] and [`SchemaProvider`] implementations backed by a ClickHouse instance
1919
20-
use std::any::Any;
2120
use std::collections::HashMap;
2221
use std::sync::Arc;
2322

@@ -69,10 +68,6 @@ impl ClickHouseCatalogProvider {
6968
}
7069

7170
impl CatalogProvider for ClickHouseCatalogProvider {
72-
fn as_any(&self) -> &dyn Any {
73-
self
74-
}
75-
7671
fn schema_names(&self) -> Vec<String> {
7772
self.schemas.keys().cloned().collect()
7873
}
@@ -95,10 +90,6 @@ pub struct ClickHouseSchemaProvider {
9590

9691
#[async_trait::async_trait]
9792
impl SchemaProvider for ClickHouseSchemaProvider {
98-
fn as_any(&self) -> &dyn Any {
99-
self
100-
}
101-
10293
fn table_names(&self) -> Vec<String> {
10394
self.tables.clone()
10495
}

‎crates/datafusion-app/src/extensions/huggingface.rs‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -70,11 +70,9 @@ impl Extension for HuggingFaceExtension {
7070
if let Some(repo_id) = &huggingface_config.repo_id {
7171
hf_builder = hf_builder.repo_id(repo_id);
7272

73-
let operator = Operator::new(hf_builder)
74-
.map_err(|e| {
75-
datafusion::common::error::DataFusionError::External(e.to_string().into())
76-
})?
77-
.finish();
73+
let operator = Operator::new(hf_builder).map_err(|e| {
74+
datafusion::common::error::DataFusionError::External(e.to_string().into())
75+
})?;
7876

7977
let store = object_store_opendal::OpendalStore::new(operator);
8078

‎crates/datafusion-app/src/flightsql.rs‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -303,7 +303,10 @@ impl FlightSQLContext {
303303
_opts: ExecOptions,
304304
) -> DFResult<ExecResult> {
305305
if let Some(ref mut client) = *self.client.lock().await {
306-
let flight_info = client.execute(sql.to_string(), None).await?;
306+
let flight_info = client
307+
.execute(sql.to_string(), None)
308+
.await
309+
.map_err(|e| DataFusionError::External(Box::new(e)))?;
307310
if flight_info.endpoint.len() != 1 {
308311
return Err(DataFusionError::External("More than one endpoint".into()));
309312
}
@@ -341,7 +344,7 @@ impl FlightSQLContext {
341344
client
342345
.get_catalogs()
343346
.await
344-
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
347+
.map_err(|e| DataFusionError::External(Box::new(e)))
345348
} else {
346349
Err(DataFusionError::External(
347350
"No FlightSQL client configured. Add one in `~/.config/dft/config.toml`".into(),
@@ -364,7 +367,7 @@ impl FlightSQLContext {
364367
client
365368
.get_db_schemas(cmd)
366369
.await
367-
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
370+
.map_err(|e| DataFusionError::External(Box::new(e)))
368371
} else {
369372
Err(DataFusionError::External(
370373
"No FlightSQL client configured. Add one in `~/.config/dft/config.toml`".into(),
@@ -393,7 +396,7 @@ impl FlightSQLContext {
393396
client
394397
.get_tables(cmd)
395398
.await
396-
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
399+
.map_err(|e| DataFusionError::External(Box::new(e)))
397400
} else {
398401
Err(DataFusionError::External(
399402
"No FlightSQL client configured. Add one in `~/.config/dft/config.toml`".into(),
@@ -408,7 +411,7 @@ impl FlightSQLContext {
408411
client
409412
.get_table_types()
410413
.await
411-
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
414+
.map_err(|e| DataFusionError::External(Box::new(e)))
412415
} else {
413416
Err(DataFusionError::External(
414417
"No FlightSQL client configured. Add one in `~/.config/dft/config.toml`".into(),
@@ -430,7 +433,7 @@ impl FlightSQLContext {
430433
client
431434
.get_sql_info(sql_info_list)
432435
.await
433-
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
436+
.map_err(|e| DataFusionError::External(Box::new(e)))
434437
} else {
435438
Err(DataFusionError::External(
436439
"No FlightSQL client configured. Add one in `~/.config/dft/config.toml`".into(),
@@ -449,7 +452,7 @@ impl FlightSQLContext {
449452
client
450453
.get_xdbc_type_info(cmd)
451454
.await
452-
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
455+
.map_err(|e| DataFusionError::External(Box::new(e)))
453456
} else {
454457
Err(DataFusionError::External(
455458
"No FlightSQL client configured. Add one in `~/.config/dft/config.toml`".into(),
@@ -467,7 +470,7 @@ impl FlightSQLContext {
467470
let stream = client
468471
.do_get(ticket.into_request())
469472
.await
470-
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
473+
.map_err(|e| DataFusionError::External(Box::new(e)))?;
471474
streams.push(stream);
472475
} else {
473476
debug!("No ticket for endpoint: {endpoint}");

‎crates/datafusion-app/src/stats.rs‎

Lines changed: 22 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,10 @@
1616
// under the License.
1717

1818
use datafusion::{
19-
datasource::{physical_plan::ParquetSource, source::DataSourceExec},
19+
datasource::{
20+
physical_plan::{FileScanConfig, ParquetSource},
21+
source::DataSourceExec,
22+
},
2023
physical_plan::{
2124
aggregates::AggregateExec,
2225
filter::FilterExec,
@@ -354,12 +357,16 @@ impl PlanIOVisitor {
354357
self.time_opening = metrics.sum_by_name("time_elapsed_opening");
355358
self.time_scanning = metrics.sum_by_name("time_elapsed_scanning_total");
356359

357-
if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
360+
if let Some(data_source_exec) = plan.downcast_ref::<DataSourceExec>() {
358361
if data_source_exec
359362
.data_source()
360-
.as_any()
361-
.downcast_ref::<ParquetSource>()
362-
.is_some()
363+
.downcast_ref::<FileScanConfig>()
364+
.is_some_and(|config| {
365+
config
366+
.file_source()
367+
.downcast_ref::<ParquetSource>()
368+
.is_some()
369+
})
363370
{
364371
self.parquet_output_rows = metrics.output_rows();
365372
self.parquet_rg_pruned_stats =
@@ -675,34 +682,28 @@ impl PlanComputeVisitor {
675682
}
676683

677684
fn is_filter_plan(plan: &dyn ExecutionPlan) -> bool {
678-
plan.as_any().downcast_ref::<FilterExec>().is_some()
685+
plan.downcast_ref::<FilterExec>().is_some()
679686
}
680687

681688
fn is_sort_plan(plan: &dyn ExecutionPlan) -> bool {
682-
plan.as_any().downcast_ref::<SortExec>().is_some()
683-
|| plan
684-
.as_any()
685-
.downcast_ref::<SortPreservingMergeExec>()
686-
.is_some()
689+
plan.downcast_ref::<SortExec>().is_some()
690+
|| plan.downcast_ref::<SortPreservingMergeExec>().is_some()
687691
}
688692

689693
fn is_projection_plan(plan: &dyn ExecutionPlan) -> bool {
690-
plan.as_any().downcast_ref::<ProjectionExec>().is_some()
694+
plan.downcast_ref::<ProjectionExec>().is_some()
691695
}
692696

693697
fn is_join_plan(plan: &dyn ExecutionPlan) -> bool {
694-
plan.as_any().downcast_ref::<HashJoinExec>().is_some()
695-
|| plan.as_any().downcast_ref::<CrossJoinExec>().is_some()
696-
|| plan.as_any().downcast_ref::<SortMergeJoinExec>().is_some()
697-
|| plan.as_any().downcast_ref::<NestedLoopJoinExec>().is_some()
698-
|| plan
699-
.as_any()
700-
.downcast_ref::<SymmetricHashJoinExec>()
701-
.is_some()
698+
plan.downcast_ref::<HashJoinExec>().is_some()
699+
|| plan.downcast_ref::<CrossJoinExec>().is_some()
700+
|| plan.downcast_ref::<SortMergeJoinExec>().is_some()
701+
|| plan.downcast_ref::<NestedLoopJoinExec>().is_some()
702+
|| plan.downcast_ref::<SymmetricHashJoinExec>().is_some()
702703
}
703704

704705
fn is_aggregate_plan(plan: &dyn ExecutionPlan) -> bool {
705-
plan.as_any().downcast_ref::<AggregateExec>().is_some()
706+
plan.downcast_ref::<AggregateExec>().is_some()
706707
}
707708

708709
impl From<PlanComputeVisitor> for ExecutionComputeStats {

‎crates/datafusion-app/src/tables/map_table.rs‎

Lines changed: 7 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
// specific language governing permissions and limitations
1616
// under the License.
1717

18-
use std::{any::Any, collections::HashMap, sync::Arc};
18+
use std::{collections::HashMap, sync::Arc};
1919

2020
use async_trait::async_trait;
2121
use datafusion::{
@@ -137,10 +137,6 @@ impl MapTable {
137137

138138
#[async_trait]
139139
impl TableProvider for MapTable {
140-
fn as_any(&self) -> &dyn Any {
141-
self
142-
}
143-
144140
fn schema(&self) -> SchemaRef {
145141
Arc::clone(&self.schema)
146142
}
@@ -180,7 +176,7 @@ struct MapExec {
180176
projected_schema: SchemaRef,
181177
// Sort information: one or more equivalent orderings
182178
_sort_information: Vec<LexOrdering>,
183-
cache: PlanProperties,
179+
cache: Arc<PlanProperties>,
184180
}
185181

186182
impl MapExec {
@@ -210,14 +206,14 @@ impl MapExec {
210206
orderings: &[LexOrdering],
211207
constraints: Constraints,
212208
partitions: &[Vec<RecordBatch>],
213-
) -> PlanProperties {
214-
PlanProperties::new(
209+
) -> Arc<PlanProperties> {
210+
Arc::new(PlanProperties::new(
215211
EquivalenceProperties::new_with_orderings(schema, orderings.iter().cloned())
216212
.with_constraints(constraints),
217213
Partitioning::UnknownPartitioning(partitions.len()),
218214
EmissionType::Incremental,
219215
Boundedness::Bounded,
220-
)
216+
))
221217
}
222218
}
223219

@@ -249,11 +245,7 @@ impl ExecutionPlan for MapExec {
249245
"MapExec"
250246
}
251247

252-
fn as_any(&self) -> &dyn Any {
253-
self
254-
}
255-
256-
fn properties(&self) -> &PlanProperties {
248+
fn properties(&self) -> &Arc<PlanProperties> {
257249
&self.cache
258250
}
259251

@@ -538,8 +530,7 @@ mod test {
538530
"| | TableScan: test projection=[id, val] |",
539531
"| physical_plan | FilterExec: id@0 = 2 |",
540532
"| | RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 |",
541-
"| | CooperativeExec |",
542-
"| | MapExec: partitions=1, projection=Some([0, 1]) |",
533+
"| | MapExec: partitions=1, projection=Some([0, 1]) |",
543534
"| | |",
544535
"+---------------+------------------------------------------------------------------------+",
545536
];

0 commit comments

Comments
 (0)