Skip to content

Commit 25e5ccc

Browse files
xudong963xudong963
authored
Upgrade to DF49 (#75)
* Upgrade to DF49 * fix licenses * use 49 * resolve comments
1 parent d4cc10f commit 25e5ccc

File tree

8 files changed

+60
-53
lines changed

8 files changed

+60
-53
lines changed

‎Cargo.toml‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -28,19 +28,19 @@ keywords = ["arrow", "arrow-rs", "datafusion"]
2828
rust-version = "1.80"
2929

3030
[dependencies]
31-
arrow = "55.1.0"
32-
arrow-schema = "55.1.0"
31+
arrow = "55.2.0"
32+
arrow-schema = "55.2.0"
3333
async-trait = "0.1"
3434
dashmap = "6"
35-
datafusion = "48"
36-
datafusion-common = "48"
37-
datafusion-expr = "48"
38-
datafusion-functions = "48"
39-
datafusion-functions-aggregate = "48"
40-
datafusion-optimizer = "48"
41-
datafusion-physical-expr = "48"
42-
datafusion-physical-plan = "48"
43-
datafusion-sql = "48"
35+
datafusion = "49"
36+
datafusion-common = "49"
37+
datafusion-expr = "49"
38+
datafusion-functions = "49"
39+
datafusion-functions-aggregate = "49"
40+
datafusion-optimizer = "49"
41+
datafusion-physical-expr = "49"
42+
datafusion-physical-plan = "49"
43+
datafusion-sql = "49"
4444
futures = "0.3"
4545
itertools = "0.14"
4646
log = "0.4"

‎deny.toml‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ allow = [
2424
"BSD-3-Clause",
2525
"CC0-1.0",
2626
"Unicode-3.0",
27-
"Zlib"
27+
"Zlib",
28+
"ISC",
29+
"bzip2-1.0.6"
2830
]
2931
version = 2

‎src/materialized/dependencies.rs‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1447,7 +1447,7 @@ mod test {
14471447
.enumerate()
14481448
.filter_map(|(i, c)| case.partition_cols.contains(&c.name.as_str()).then_some(i))
14491449
.collect();
1450-
println!("indices: {:?}", partition_col_indices);
1450+
println!("indices: {partition_col_indices:?}");
14511451
let analyzed = pushdown_projection_inexact(plan.clone(), &partition_col_indices)?;
14521452
println!(
14531453
"inexact projection pushdown:\n{}",
@@ -1720,19 +1720,19 @@ mod test {
17201720
",
17211721
projection: &["year"],
17221722
expected_plan: vec![
1723-
"+--------------+--------------------------------------------------+",
1724-
"| plan_type | plan |",
1725-
"+--------------+--------------------------------------------------+",
1726-
"| logical_plan | Union |",
1727-
"| | Projection: coalesce(t1.year, t2.year) AS year |",
1728-
"| | Full Join: Using t1.year = t2.year |",
1729-
"| | SubqueryAlias: t1 |",
1730-
"| | Projection: t1.column1 AS year |",
1731-
"| | TableScan: t1 projection=[column1] |",
1732-
"| | SubqueryAlias: t2 |",
1733-
"| | TableScan: t2 projection=[year] |",
1734-
"| | TableScan: t3 projection=[year] |",
1735-
"+--------------+--------------------------------------------------+",
1723+
"+--------------+--------------------------------------------------------------------+",
1724+
"| plan_type | plan |",
1725+
"+--------------+--------------------------------------------------------------------+",
1726+
"| logical_plan | Union |",
1727+
"| | Projection: coalesce(CAST(t1.year AS Utf8View), t2.year) AS year |",
1728+
"| | Full Join: Using CAST(t1.year AS Utf8View) = t2.year |",
1729+
"| | SubqueryAlias: t1 |",
1730+
"| | Projection: t1.column1 AS year |",
1731+
"| | TableScan: t1 projection=[column1] |",
1732+
"| | SubqueryAlias: t2 |",
1733+
"| | TableScan: t2 projection=[year] |",
1734+
"| | TableScan: t3 projection=[year] |",
1735+
"+--------------+--------------------------------------------------------------------+",
17361736
],
17371737
expected_output: vec![
17381738
"+------+",

‎src/materialized/file_metadata.rs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -226,7 +226,7 @@ impl ExecutionPlan for FileMetadataExec {
226226
.map(|record_batch| {
227227
record_batch
228228
.project(&projection)
229-
.map_err(|e| DataFusionError::ArrowError(e, None))
229+
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
230230
})
231231
.collect::<Vec<_>>();
232232
}
@@ -858,7 +858,7 @@ mod test {
858858
.await?;
859859

860860
ctx.sql(
861-
"INSERT INTO t1 VALUES
861+
"INSERT INTO t1 VALUES
862862
(1, '2021'),
863863
(2, '2022'),
864864
(3, '2023'),
@@ -882,7 +882,7 @@ mod test {
882882
.await?;
883883

884884
ctx.sql(
885-
"INSERT INTO private.t1 VALUES
885+
"INSERT INTO private.t1 VALUES
886886
(1, '2021', '01'),
887887
(2, '2022', '02'),
888888
(3, '2023', '03'),
@@ -906,7 +906,7 @@ mod test {
906906
.await?;
907907

908908
ctx.sql(
909-
"INSERT INTO datafusion_mv.public.t3 VALUES
909+
"INSERT INTO datafusion_mv.public.t3 VALUES
910910
(1, '2021-01-01'),
911911
(2, '2022-02-02'),
912912
(3, '2023-03-03'),
@@ -929,8 +929,8 @@ mod test {
929929
ctx.sql(
930930
// Remove timestamps and trim (randomly generated) file names since they're not stable in tests
931931
"CREATE VIEW file_metadata_test_view AS SELECT
932-
* EXCLUDE(file_path, last_modified),
933-
regexp_replace(file_path, '/[^/]*$', '/') AS file_path
932+
* EXCLUDE(file_path, last_modified),
933+
regexp_replace(file_path, '/[^/]*$', '/') AS file_path
934934
FROM file_metadata",
935935
)
936936
.await

‎src/materialized/row_metadata.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ impl RowMetadataRegistry {
9898
.get(&table.to_string())
9999
.map(|o| Arc::clone(o.value()))
100100
.or_else(|| self.default_source.clone())
101-
.ok_or_else(|| DataFusionError::Internal(format!("No metadata source for {}", table)))
101+
.ok_or_else(|| DataFusionError::Internal(format!("No metadata source for {table}")))
102102
}
103103
}
104104

‎src/rewrite/exploitation.rs‎

Lines changed: 16 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ use datafusion::catalog::TableProvider;
2323
use datafusion::datasource::provider_as_source;
2424
use datafusion::execution::context::SessionState;
2525
use datafusion::execution::{SendableRecordBatchStream, TaskContext};
26-
use datafusion::physical_expr::{LexRequirement, PhysicalSortExpr, PhysicalSortRequirement};
26+
use datafusion::physical_expr::{PhysicalSortExpr, PhysicalSortRequirement};
2727
use datafusion::physical_expr_common::sort_expr::format_physical_sort_requirement_list;
2828
use datafusion::physical_optimizer::PhysicalOptimizerRule;
2929
use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
@@ -32,6 +32,7 @@ use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion, Tre
3232
use datafusion_common::{DataFusionError, Result, TableReference};
3333
use datafusion_expr::{Extension, LogicalPlan, UserDefinedLogicalNode, UserDefinedLogicalNodeCore};
3434
use datafusion_optimizer::OptimizerRule;
35+
use datafusion_physical_expr::OrderingRequirements;
3536
use itertools::Itertools;
3637
use ordered_float::OrderedFloat;
3738

@@ -316,7 +317,7 @@ pub struct OneOfExec {
316317
// Optionally declare a required input ordering
317318
// This will inform DataFusion to add sorts to children,
318319
// which will improve cost estimation of candidates
319-
required_input_ordering: Option<LexRequirement>,
320+
required_input_ordering: Option<OrderingRequirements>,
320321
// Index of the candidate with the best cost
321322
best: usize,
322323
// Cost function to use in optimization
@@ -337,7 +338,7 @@ impl OneOfExec {
337338
/// Create a new `OneOfExec`
338339
pub fn try_new(
339340
candidates: Vec<Arc<dyn ExecutionPlan>>,
340-
required_input_ordering: Option<LexRequirement>,
341+
required_input_ordering: Option<OrderingRequirements>,
341342
cost: CostFn,
342343
) -> Result<Self> {
343344
if candidates.is_empty() {
@@ -366,7 +367,7 @@ impl OneOfExec {
366367

367368
/// Modify this plan's required input ordering.
368369
/// Used for sort pushdown
369-
pub fn with_required_input_ordering(self, requirement: Option<LexRequirement>) -> Self {
370+
pub fn with_required_input_ordering(self, requirement: Option<OrderingRequirements>) -> Self {
370371
Self {
371372
required_input_ordering: requirement,
372373
..self
@@ -387,7 +388,7 @@ impl ExecutionPlan for OneOfExec {
387388
self.candidates[self.best].properties()
388389
}
389390

390-
fn required_input_ordering(&self) -> Vec<Option<LexRequirement>> {
391+
fn required_input_ordering(&self) -> Vec<Option<OrderingRequirements>> {
391392
vec![self.required_input_ordering.clone(); self.children().len()]
392393
}
393394

@@ -455,12 +456,16 @@ impl DisplayAs for OneOfExec {
455456
format_physical_sort_requirement_list(
456457
&self
457458
.required_input_ordering
458-
.clone()
459-
.unwrap_or_default()
460-
.into_iter()
461-
.map(PhysicalSortExpr::from)
462-
.map(PhysicalSortRequirement::from)
463-
.collect_vec()
459+
.as_ref()
460+
.map(|req| {
461+
req.clone()
462+
.into_single()
463+
.into_iter()
464+
.map(PhysicalSortExpr::from)
465+
.map(PhysicalSortRequirement::from)
466+
.collect_vec()
467+
})
468+
.unwrap_or_default(),
464469
)
465470
)
466471
}

‎src/rewrite/normal_form.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -988,8 +988,7 @@ mod test {
988988
let ctx = SessionContext::new_with_config(
989989
SessionConfig::new()
990990
.set_bool("datafusion.execution.parquet.pushdown_filters", true)
991-
.set_bool("datafusion.explain.logical_plan_only", true)
992-
.set_bool("datafusion.sql_parser.map_varchar_to_utf8view", false),
991+
.set_bool("datafusion.explain.logical_plan_only", true),
993992
);
994993

995994
let t1_path = tempdir()?;

‎tests/materialized_listing_table.rs‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ async fn setup() -> Result<TestContext> {
185185
.await?;
186186

187187
ctx.sql(
188-
"INSERT INTO t1 VALUES
188+
"INSERT INTO t1 VALUES
189189
(1, '2023-01-01', 'A'),
190190
(2, '2023-01-02', 'B'),
191191
(3, '2023-01-03', 'C'),
@@ -251,7 +251,7 @@ async fn test_materialized_listing_table_incremental_maintenance() -> Result<()>
251251

252252
// Insert another row into the source table
253253
ctx.sql(
254-
"INSERT INTO t1 VALUES
254+
"INSERT INTO t1 VALUES
255255
(7, '2024-12-07', 'W')",
256256
)
257257
.await?
@@ -352,12 +352,13 @@ impl MaterializedListingTable {
352352
file_sort_order: opts.file_sort_order,
353353
});
354354

355+
let mut listing_table_config = ListingTableConfig::new(config.table_path);
356+
if let Some(options) = options {
357+
listing_table_config = listing_table_config.with_listing_options(options);
358+
}
359+
listing_table_config = listing_table_config.with_schema(Arc::new(file_schema));
355360
Ok(MaterializedListingTable {
356-
inner: ListingTable::try_new(ListingTableConfig {
357-
table_paths: vec![config.table_path],
358-
file_schema: Some(Arc::new(file_schema)),
359-
options,
360-
})?,
361+
inner: ListingTable::try_new(listing_table_config)?,
361362
query: normalized_query,
362363
schema: normalized_schema,
363364
})

0 commit comments

Comments
 (0)