Skip to content

Commit 11ab0db

Browse files
sgrebnovsgrebnov
authored
Upgrade to DataFusion 47 and Arrow 55 (#355)
1 parent dddeb8c commit 11ab0db

15 files changed

Lines changed: 310 additions & 196 deletions

File tree

‎Cargo.toml‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -6,11 +6,11 @@ edition = "2021"
66
repository = "https://github.com/datafusion-contrib/datafusion-table-providers"
77

88
[dependencies]
9-
arrow = "54"
10-
arrow-array = { version = "54", optional = true }
11-
arrow-flight = { version = "54", optional = true, features = ["flight-sql-experimental", "tls"] }
12-
arrow-schema = { version = "54", optional = true, features = ["serde"] }
13-
arrow-json = "54"
9+
arrow = "55"
10+
arrow-array = { version = "55", optional = true }
11+
arrow-flight = { version = "55", optional = true, features = ["flight-sql-experimental", "tls"] }
12+
arrow-schema = { version = "55", optional = true, features = ["serde"] }
13+
arrow-json = "55"
1414
async-stream = { version = "0.3.5", optional = true }
1515
async-trait = "0.1.80"
1616
num-bigint = "0.4.4"
@@ -20,11 +20,11 @@ byte-unit = { version = "5.1.4", optional = true }
2020
bigdecimal = "0.4.5"
2121
byteorder = "1.5.0"
2222
chrono = "0.4"
23-
datafusion = "46"
24-
datafusion-expr = { version = "46", optional = true }
25-
datafusion-physical-expr = { version = "46", optional = true }
26-
datafusion-physical-plan = { version = "46", optional = true }
27-
datafusion-proto = { version = "46", optional = true }
23+
datafusion = "47"
24+
datafusion-expr = { version = "47", optional = true }
25+
datafusion-physical-expr = { version = "47", optional = true }
26+
datafusion-physical-plan = { version = "47", optional = true }
27+
datafusion-proto = { version = "47", optional = true }
2828
duckdb = { version = "1", features = [
2929
"bundled",
3030
"r2d2",
@@ -59,7 +59,7 @@ pem = { version = "3.0.4", optional = true }
5959
tokio-rusqlite = { version = "0.5.1", optional = true }
6060
tonic = { version = "0.12.2", optional = true }
6161
datafusion-federation = "0.1"
62-
datafusion-federation-sql = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "a4c752a172904a0c214fc37ab2d62e907eb16c8f" }
62+
datafusion-federation-sql = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "dc11bd6edec00fb42ea48b8c5256970c8a533aa2" }
6363
itertools = "0.13.0"
6464
dyn-clone = { version = "1.0.17", optional = true }
6565
geo-types = "0.7.13"
@@ -77,7 +77,7 @@ rstest = "0.22.0"
7777
geozero = { version = "0.13.0", features = ["with-wkb"] }
7878
tokio-stream = { version = "0.1.15", features = ["net"] }
7979
insta = { version = "1.40.0", features = ["filters"] }
80-
datafusion-physical-plan = { version = "46" }
80+
datafusion-physical-plan = { version = "47" }
8181
tempfile = "3.8.1"
8282

8383
[features]
@@ -103,11 +103,11 @@ sqlite-federation = ["sqlite"]
103103
postgres-federation = ["postgres"]
104104

105105
[patch.crates-io]
106-
datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "a4c752a172904a0c214fc37ab2d62e907eb16c8f" } # spiceai-46
107-
duckdb = { git = "https://github.com/spiceai/duckdb-rs.git", rev = "2e24b958e44ec7419290249e27a15f1a19703fff" } # spiceai-1.1.3-backported-arrow-54
106+
datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "dc11bd6edec00fb42ea48b8c5256970c8a533aa2" } # spiceai-47
107+
duckdb = { git = "https://github.com/spiceai/duckdb-rs.git", rev = "69ae7518ee093a1b070e9e4e6f011ef353431086" } # spiceai-1.1.3-backported-arrow-55
108108

109-
datafusion = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" } # spiceai-46
110-
datafusion-expr = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" } # spiceai-46
111-
datafusion-physical-expr = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" } # spiceai-46
112-
datafusion-physical-plan = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" } # spiceai-46
113-
datafusion-proto = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" } # spiceai-46
109+
datafusion = { git = "https://github.com/spiceai/datafusion.git", rev = "b5c62f29d2c70c5331ff50015b67b5e1cafcd578" } # spiceai-47
110+
datafusion-expr = { git = "https://github.com/spiceai/datafusion.git", rev = "b5c62f29d2c70c5331ff50015b67b5e1cafcd578" } # spiceai-47
111+
datafusion-physical-expr = { git = "https://github.com/spiceai/datafusion.git", rev = "b5c62f29d2c70c5331ff50015b67b5e1cafcd578" } # spiceai-47
112+
datafusion-physical-plan = { git = "https://github.com/spiceai/datafusion.git", rev = "b5c62f29d2c70c5331ff50015b67b5e1cafcd578" } # spiceai-47
113+
datafusion-proto = { git = "https://github.com/spiceai/datafusion.git", rev = "b5c62f29d2c70c5331ff50015b67b5e1cafcd578" } # spiceai-47

‎src/duckdb.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -156,7 +156,9 @@ pub enum Error {
156156
#[snafu(display("Failed to register Arrow scan view for DuckDB ingestion: {source}"))]
157157
UnableToRegisterArrowScanView { source: duckdb::Error },
158158

159-
#[snafu(display("Failed to register Arrow scan view to build table creation statement: {source}"))]
159+
#[snafu(display(
160+
"Failed to register Arrow scan view to build table creation statement: {source}"
161+
))]
160162
UnableToRegisterArrowScanViewForTableCreation { source: duckdb::Error },
161163

162164
#[snafu(display("Failed to drop Arrow scan view for DuckDB ingestion: {source}"))]

‎src/duckdb/creator.rs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -409,9 +409,8 @@ impl TableManager {
409409
tx.register_arrow_scan_view(&view_name, &stream)
410410
.context(super::UnableToRegisterArrowScanViewForTableCreationSnafu)?;
411411

412-
let sql = format!(
413-
r#"CREATE TABLE IF NOT EXISTS "{table_name}" AS SELECT * FROM "{view_name}""#,
414-
);
412+
let sql =
413+
format!(r#"CREATE TABLE IF NOT EXISTS "{table_name}" AS SELECT * FROM "{view_name}""#,);
415414
tracing::debug!("{sql}");
416415

417416
tx.execute(&sql, [])
@@ -750,10 +749,11 @@ pub(crate) mod tests {
750749
use arrow::array::RecordBatch;
751750
use datafusion::{
752751
common::SchemaExt,
752+
datasource::sink::DataSink,
753753
execution::{SendableRecordBatchStream, TaskContext},
754754
logical_expr::dml::InsertOp,
755755
parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder,
756-
physical_plan::{insert::DataSink, memory::MemoryStream},
756+
physical_plan::memory::MemoryStream,
757757
};
758758
use tracing::subscriber::DefaultGuard;
759759
use tracing_subscriber::EnvFilter;

‎src/duckdb/write.rs‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,17 +15,14 @@ use arrow_schema::ArrowError;
1515
use async_trait::async_trait;
1616
use datafusion::catalog::Session;
1717
use datafusion::common::{Constraints, SchemaExt};
18+
use datafusion::datasource::sink::{DataSink, DataSinkExec};
1819
use datafusion::logical_expr::dml::InsertOp;
1920
use datafusion::{
2021
datasource::{TableProvider, TableType},
2122
error::DataFusionError,
2223
execution::{SendableRecordBatchStream, TaskContext},
2324
logical_expr::Expr,
24-
physical_plan::{
25-
insert::{DataSink, DataSinkExec},
26-
metrics::MetricsSet,
27-
DisplayAs, DisplayFormatType, ExecutionPlan,
28-
},
25+
physical_plan::{metrics::MetricsSet, DisplayAs, DisplayFormatType, ExecutionPlan},
2926
};
3027
use duckdb::Transaction;
3128
use futures::StreamExt;
@@ -642,6 +639,7 @@ impl RecordBatchReader for RecordBatchReaderFromStream {
642639
#[cfg(test)]
643640
mod test {
644641
use arrow::array::{Int64Array, StringArray};
642+
use datafusion::datasource::sink::DataSink;
645643
use datafusion_physical_plan::memory::MemoryStream;
646644

647645
use super::*;

‎src/flight/exec.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -240,7 +240,7 @@ async fn try_fetch_stream(
240240
impl DisplayAs for FlightExec {
241241
fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
242242
match t {
243-
DisplayFormatType::Default => write!(
243+
DisplayFormatType::Default | DisplayFormatType::TreeRender => write!(
244244
f,
245245
"FlightExec: origin={}, streams={}",
246246
self.config.origin,

‎src/mysql/mysql_window.rs‎

Lines changed: 21 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
use datafusion::sql::sqlparser::ast::ObjectNamePart;
12
use datafusion::sql::sqlparser::ast::{Expr, Function, Ident, VisitorMut, WindowType};
23
use std::ops::ControlFlow;
34

@@ -41,7 +42,7 @@ impl VisitorMut for MySQLWindowVisitor {
4142

4243
fn pre_visit_expr(&mut self, expr: &mut Expr) -> ControlFlow<Self::Break> {
4344
if let Expr::Function(func) = expr {
44-
if let Some(Ident { value, .. }) = func.name.0.first() {
45+
if let Some(ObjectNamePart::Identifier(Ident { value, .. })) = func.name.0.first() {
4546
// match for some scalars that support window functions
4647
// all of them need to remove nulls first/last, but only rank removes the frame clause
4748
if let Some(func_type) = FunctionType::from_str(value) {
@@ -58,7 +59,7 @@ impl MySQLWindowVisitor {
5859
pub fn remove_nulls_first_last(func: &mut Function) {
5960
if let Some(WindowType::WindowSpec(spec)) = func.over.as_mut() {
6061
for order_by in &mut spec.order_by {
61-
order_by.nulls_first = None; // nulls first/last are not supported in MySQL
62+
order_by.options.nulls_first = None; // nulls first/last are not supported in MySQL
6263
}
6364
}
6465
}
@@ -75,27 +76,24 @@ mod test {
7576
use datafusion::sql::sqlparser::{
7677
self,
7778
ast::{self, helpers::attached_token::AttachedToken, ObjectName, WindowFrame},
78-
tokenizer::Span,
7979
};
8080

8181
use super::*;
8282

8383
#[test]
8484
fn test_remove_frame_clause() {
8585
let mut func = Function {
86-
name: ObjectName(vec![Ident {
87-
value: "RANK".to_string(),
88-
quote_style: None,
89-
span: Span::empty(),
90-
}]),
86+
name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("RANK"))]),
9187
args: ast::FunctionArguments::None,
9288
over: Some(WindowType::WindowSpec(ast::WindowSpec {
9389
window_name: None,
9490
partition_by: vec![],
9591
order_by: vec![sqlparser::ast::OrderByExpr {
9692
expr: sqlparser::ast::Expr::Wildcard(AttachedToken::empty()),
97-
asc: None,
98-
nulls_first: Some(true),
93+
options: sqlparser::ast::OrderByOptions {
94+
asc: None,
95+
nulls_first: Some(true),
96+
},
9997
with_fill: None,
10098
}],
10199
window_frame: Some(WindowFrame {
@@ -116,8 +114,10 @@ mod test {
116114
partition_by: vec![],
117115
order_by: vec![sqlparser::ast::OrderByExpr {
118116
expr: sqlparser::ast::Expr::Wildcard(AttachedToken::empty()),
119-
asc: None,
120-
nulls_first: Some(true),
117+
options: sqlparser::ast::OrderByOptions {
118+
asc: None,
119+
nulls_first: Some(true),
120+
},
121121
with_fill: None,
122122
}],
123123
window_frame: None,
@@ -131,19 +131,17 @@ mod test {
131131
#[test]
132132
fn test_remove_nulls_first_last() {
133133
let mut func = Function {
134-
name: ObjectName(vec![Ident {
135-
value: "RANK".to_string(),
136-
quote_style: None,
137-
span: Span::empty(),
138-
}]),
134+
name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("RANK"))]),
139135
args: sqlparser::ast::FunctionArguments::None,
140136
over: Some(WindowType::WindowSpec(sqlparser::ast::WindowSpec {
141137
window_name: None,
142138
partition_by: vec![],
143139
order_by: vec![sqlparser::ast::OrderByExpr {
144140
expr: sqlparser::ast::Expr::Wildcard(AttachedToken::empty()),
145-
asc: None,
146-
nulls_first: Some(true),
141+
options: sqlparser::ast::OrderByOptions {
142+
asc: None,
143+
nulls_first: Some(true),
144+
},
147145
with_fill: None,
148146
}],
149147
window_frame: Some(WindowFrame {
@@ -164,8 +162,10 @@ mod test {
164162
partition_by: vec![],
165163
order_by: vec![sqlparser::ast::OrderByExpr {
166164
expr: sqlparser::ast::Expr::Wildcard(AttachedToken::empty()),
167-
asc: None,
168-
nulls_first: None,
165+
options: sqlparser::ast::OrderByOptions {
166+
asc: None,
167+
nulls_first: None,
168+
},
169169
with_fill: None,
170170
}],
171171
window_frame: Some(WindowFrame {

‎src/postgres/write.rs‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,13 @@ use async_trait::async_trait;
66
use datafusion::{
77
catalog::Session,
88
common::{Constraints, SchemaExt},
9-
datasource::{TableProvider, TableType},
9+
datasource::{
10+
sink::{DataSink, DataSinkExec},
11+
TableProvider, TableType,
12+
},
1013
execution::{SendableRecordBatchStream, TaskContext},
1114
logical_expr::{dml::InsertOp, Expr},
12-
physical_plan::{
13-
insert::{DataSink, DataSinkExec},
14-
metrics::MetricsSet,
15-
DisplayAs, DisplayFormatType, ExecutionPlan,
16-
},
15+
physical_plan::{metrics::MetricsSet, DisplayAs, DisplayFormatType, ExecutionPlan},
1716
};
1817
use futures::StreamExt;
1918
use snafu::prelude::*;

0 commit comments

Comments
 (0)