Skip to content

Commit 5eb35f9

Browse files
committed
Upgrade DataFusion to version 56
1 parent 878e5d3 commit 5eb35f9

23 files changed

Lines changed: 1307 additions & 303 deletions

‎Cargo.lock‎

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

‎Cargo.toml‎

Lines changed: 27 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -41,29 +41,29 @@ pyo3 = { version = "0.29" }
4141
pyo3-async-runtimes = { version = "0.29" }
4242
pyo3-log = "0.13.3"
4343
chrono = { version = "0.4", default-features = false }
44-
arrow = { version = "59" }
45-
arrow-array = { version = "59" }
46-
arrow-schema = { version = "59" }
47-
arrow-select = { version = "59" }
48-
datafusion = { version = "55.1.0" }
49-
datafusion-substrait = { version = "55.1.0" }
50-
datafusion-proto = { version = "55.1.0" }
51-
datafusion-ffi = { version = "55.1.0" }
52-
datafusion-catalog = { version = "55.1.0", default-features = false }
53-
datafusion-common = { version = "55.1.0", default-features = false }
54-
datafusion-functions-aggregate = { version = "55.1.0" }
55-
datafusion-functions-window = { version = "55.1.0" }
56-
datafusion-spark = { version = "55.1.0" }
57-
datafusion-expr = { version = "55.1.0" }
58-
datafusion-session = { version = "55.1.0" }
44+
arrow = { version = "60" }
45+
arrow-array = { version = "60" }
46+
arrow-schema = { version = "60" }
47+
arrow-select = { version = "60" }
48+
datafusion = { version = "56.0.0-dev" }
49+
datafusion-substrait = { version = "56.0.0-dev" }
50+
datafusion-proto = { version = "56.0.0-dev" }
51+
datafusion-ffi = { version = "56.0.0-dev" }
52+
datafusion-catalog = { version = "56.0.0-dev", default-features = false }
53+
datafusion-common = { version = "56.0.0-dev", default-features = false }
54+
datafusion-functions-aggregate = { version = "56.0.0-dev" }
55+
datafusion-functions-window = { version = "56.0.0-dev" }
56+
datafusion-spark = { version = "56.0.0-dev" }
57+
datafusion-expr = { version = "56.0.0-dev" }
58+
datafusion-session = { version = "56.0.0-dev" }
5959
prost = "0.14.3"
6060
serde_json = "1"
6161
uuid = { version = "1.23" }
6262
mimalloc = { version = "0.1", default-features = false }
6363
async-trait = "0.1.89"
6464
futures = "0.3"
6565
cstr = "0.2"
66-
object_store = { version = "0.13.1" }
66+
object_store = { version = "0.14" }
6767
url = "2"
6868
log = "0.4.29"
6969
parking_lot = "0.12"
@@ -78,3 +78,14 @@ codegen-units = 2
7878
# We cannot publish to crates.io with any patches in the below section. Developers
7979
# must remove any entries in this section before creating a release candidate.
8080
[patch.crates-io]
81+
datafusion = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
82+
datafusion-substrait = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
83+
datafusion-proto = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
84+
datafusion-ffi = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
85+
datafusion-catalog = { git = "https://github.com/apache/datafusion", branch = "branch-56" }
86+
datafusion-common = { git = "https://github.com/apache/datafusion", branch = "branch-56" }
87+
datafusion-functions-aggregate = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
88+
datafusion-functions-window = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
89+
datafusion-spark = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
90+
datafusion-expr = { git = "https://github.com/apache/datafusion", branch = "branch-56"}
91+
datafusion-session = { git = "https://github.com/apache/datafusion", branch = "branch-56"}

‎crates/core/src/context.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ use std::str::FromStr;
2222
use std::sync::Arc;
2323

2424
use arrow::array::RecordBatchReader;
25-
use arrow::ffi_stream::ArrowArrayStreamReader;
25+
use arrow::ffi_stream::RecordBatchStreamReader;
2626
use arrow::pyarrow::FromPyArrow;
2727
use datafusion::arrow::datatypes::{DataType, Schema, SchemaRef};
2828
use datafusion::arrow::pyarrow::PyArrowType;
@@ -695,7 +695,7 @@ impl PySessionContext {
695695
py: Python,
696696
) -> PyDataFusionResult<PyDataFrame> {
697697
let (schema, batches) =
698-
if let Ok(stream_reader) = ArrowArrayStreamReader::from_pyarrow_bound(&data) {
698+
if let Ok(stream_reader) = RecordBatchStreamReader::from_pyarrow_bound(&data) {
699699
// Works for any object that implements __arrow_c_stream__ in pycapsule.
700700

701701
let schema = stream_reader.schema().as_ref().to_owned();

‎crates/core/src/dataframe.rs‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,9 @@ use datafusion::arrow::pyarrow::{PyArrowType, ToPyArrow};
3333
use datafusion::arrow::util::pretty;
3434
use datafusion::catalog::TableProvider;
3535
use datafusion::common::UnnestOptions;
36+
use datafusion::common::parquet_config::{
37+
DFParquetCompression, DFParquetEncoding, DFParquetStatistics, DFParquetWriterVersion,
38+
};
3639
use datafusion::config::{CsvOptions, ParquetColumnOptions, ParquetOptions, TableParquetOptions};
3740
use datafusion::dataframe::{DataFrame, DataFrameWriteOptions};
3841
use datafusion::error::DataFusionError;
@@ -229,8 +232,22 @@ impl PyParquetWriterOptions {
229232
maximum_buffered_record_batches_per_stream: usize,
230233
) -> PyResult<Self> {
231234
let writer_version =
232-
datafusion::common::parquet_config::DFParquetWriterVersion::from_str(writer_version)
233-
.map_err(py_datafusion_err)?;
235+
DFParquetWriterVersion::from_str(writer_version).map_err(py_datafusion_err)?;
236+
let compression = compression
237+
.as_deref()
238+
.map(DFParquetCompression::from_str)
239+
.transpose()
240+
.map_err(py_datafusion_err)?;
241+
let statistics_enabled = statistics_enabled
242+
.as_deref()
243+
.map(DFParquetStatistics::from_str)
244+
.transpose()
245+
.map_err(py_datafusion_err)?;
246+
let encoding = encoding
247+
.as_deref()
248+
.map(DFParquetEncoding::from_str)
249+
.transpose()
250+
.map_err(py_datafusion_err)?;
234251
Ok(Self {
235252
options: ParquetOptions {
236253
data_pagesize_limit,
@@ -1120,7 +1137,7 @@ impl PyDataFrame {
11201137
}
11211138

11221139
let mut options = TableParquetOptions::default();
1123-
options.global.compression = Some(compression_string);
1140+
options.global.compression = Some(DFParquetCompression::from_str(&compression_string)?);
11241141
let write_options = write_options
11251142
.map(DataFrameWriteOptions::from)
11261143
.unwrap_or_default();

‎crates/core/src/dataset.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ impl TableProvider for Dataset {
8989
async fn scan(
9090
&self,
9191
_ctx: &dyn Session,
92-
projection: Option<&Vec<usize>>,
92+
projection: Option<&[usize]>,
9393
filters: &[Expr],
9494
// limit can be used to reduce the amount scanned
9595
// from the datasource as a performance optimization.
@@ -99,7 +99,7 @@ impl TableProvider for Dataset {
9999
) -> DFResult<Arc<dyn ExecutionPlan>> {
100100
Python::attach(|py| {
101101
let plan: Arc<dyn ExecutionPlan> = Arc::new(
102-
DatasetExec::new(py, self.dataset.bind(py), projection.cloned(), filters)
102+
DatasetExec::new(py, self.dataset.bind(py), projection, filters)
103103
.map_err(|err| DataFusionError::External(Box::new(err)))?,
104104
);
105105
Ok(plan)

‎crates/core/src/dataset_exec.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ impl DatasetExec {
7878
pub fn new(
7979
py: Python,
8080
dataset: &Bound<'_, PyAny>,
81-
projection: Option<Vec<usize>>,
81+
projection: Option<&[usize]>,
8282
filters: &[Expr],
8383
) -> PyDataFusionResult<Self> {
8484
let columns: Option<PyDataFusionResult<Vec<String>>> = projection.map(|p| {

‎crates/core/src/expr.rs‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@ pub mod aggregate;
6363
pub mod aggregate_expr;
6464
pub mod alias;
6565
pub mod analyze;
66+
pub mod asof_join;
6667
pub mod between;
6768
pub mod binary_expr;
6869
pub mod bool_expr;
@@ -73,14 +74,15 @@ pub mod conditional_expr;
7374
pub mod copy_to;
7475
pub mod create_catalog;
7576
pub mod create_catalog_schema;
77+
pub mod create_external_catalog;
7678
pub mod create_external_table;
7779
pub mod create_function;
7880
pub mod create_index;
7981
pub mod create_memory_table;
8082
pub mod create_view;
81-
pub mod describe_table;
8283
pub mod distinct;
8384
pub mod dml;
85+
pub mod drop_catalog;
8486
pub mod drop_catalog_schema;
8587
pub mod drop_function;
8688
pub mod drop_table;
@@ -101,13 +103,15 @@ pub mod like;
101103
pub mod limit;
102104
pub mod literal;
103105
pub mod logical_node;
106+
pub mod materialized_cte;
104107
pub mod placeholder;
105108
pub mod projection;
106109
pub mod recursive_query;
107110
pub mod repartition;
108111
pub mod scalar_subquery;
109112
pub mod scalar_variable;
110113
pub mod set_comparison;
114+
pub mod show;
111115
pub mod signature;
112116
pub mod sort;
113117
pub mod sort_expr;
@@ -1091,13 +1095,19 @@ pub(crate) fn init_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
10911095
m.add_class::<copy_to::PyFileType>()?;
10921096
m.add_class::<create_catalog::PyCreateCatalog>()?;
10931097
m.add_class::<create_catalog_schema::PyCreateCatalogSchema>()?;
1098+
m.add_class::<create_external_catalog::PyCreateExternalCatalog>()?;
10941099
m.add_class::<create_external_table::PyCreateExternalTable>()?;
10951100
m.add_class::<create_function::PyCreateFunction>()?;
10961101
m.add_class::<create_function::PyOperateFunctionArg>()?;
10971102
m.add_class::<create_function::PyCreateFunctionBody>()?;
10981103
m.add_class::<create_index::PyCreateIndex>()?;
1099-
m.add_class::<describe_table::PyDescribeTable>()?;
1104+
m.add_class::<show::PyShow>()?;
1105+
m.add_class::<asof_join::PyAsOfJoin>()?;
1106+
m.add_class::<asof_join::PyAsOfMatch>()?;
1107+
m.add_class::<materialized_cte::PyMaterializedCte>()?;
1108+
m.add_class::<materialized_cte::PyMaterializedCteScan>()?;
11001109
m.add_class::<dml::PyDmlStatement>()?;
1110+
m.add_class::<drop_catalog::PyDropCatalog>()?;
11011111
m.add_class::<drop_catalog_schema::PyDropCatalogSchema>()?;
11021112
m.add_class::<drop_function::PyDropFunction>()?;
11031113
m.add_class::<drop_view::PyDropView>()?;

‎crates/core/src/expr/asof_join.rs‎

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,175 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
use std::fmt::{self, Display, Formatter};
19+
20+
use datafusion::logical_expr::logical_plan::{AsOfJoin, AsOfMatch};
21+
use pyo3::IntoPyObjectExt;
22+
use pyo3::prelude::*;
23+
24+
use crate::common::df_schema::PyDFSchema;
25+
use crate::expr::PyExpr;
26+
use crate::expr::join::PyJoinConstraint;
27+
use crate::expr::logical_node::LogicalNode;
28+
use crate::sql::logical::PyLogicalPlan;
29+
30+
#[derive(Debug, Clone)]
31+
#[pyclass(from_py_object, frozen, name = "AsOfMatch", module = "datafusion.expr")]
32+
pub struct PyAsOfMatch {
33+
match_condition: AsOfMatch,
34+
}
35+
36+
impl From<AsOfMatch> for PyAsOfMatch {
37+
fn from(match_condition: AsOfMatch) -> PyAsOfMatch {
38+
PyAsOfMatch { match_condition }
39+
}
40+
}
41+
42+
impl From<PyAsOfMatch> for AsOfMatch {
43+
fn from(match_condition: PyAsOfMatch) -> Self {
44+
match_condition.match_condition
45+
}
46+
}
47+
48+
#[pymethods]
49+
impl PyAsOfMatch {
50+
fn left(&self) -> PyResult<PyExpr> {
51+
Ok(self.match_condition.left.clone().into())
52+
}
53+
54+
fn op(&self) -> PyResult<String> {
55+
Ok(self.match_condition.op.to_string())
56+
}
57+
58+
fn right(&self) -> PyResult<PyExpr> {
59+
Ok(self.match_condition.right.clone().into())
60+
}
61+
62+
fn __repr__(&self) -> PyResult<String> {
63+
Ok(format!("{}", self.match_condition))
64+
}
65+
}
66+
67+
impl Display for PyAsOfMatch {
68+
fn fmt(&self, f: &mut Formatter) -> fmt::Result {
69+
write!(f, "{}", self.match_condition)
70+
}
71+
}
72+
73+
#[pyclass(
74+
from_py_object,
75+
frozen,
76+
name = "AsOfJoin",
77+
module = "datafusion.expr",
78+
subclass
79+
)]
80+
#[derive(Clone)]
81+
pub struct PyAsOfJoin {
82+
join: AsOfJoin,
83+
}
84+
85+
impl From<AsOfJoin> for PyAsOfJoin {
86+
fn from(join: AsOfJoin) -> PyAsOfJoin {
87+
PyAsOfJoin { join }
88+
}
89+
}
90+
91+
impl From<PyAsOfJoin> for AsOfJoin {
92+
fn from(join: PyAsOfJoin) -> Self {
93+
join.join
94+
}
95+
}
96+
97+
impl Display for PyAsOfJoin {
98+
fn fmt(&self, f: &mut Formatter) -> fmt::Result {
99+
write!(
100+
f,
101+
"AsOfJoin
102+
Left: {:?}
103+
Right: {:?}
104+
On: {:?}
105+
MatchCondition: {:?}
106+
JoinConstraint: {:?}
107+
Schema: {:?}",
108+
self.join.left,
109+
self.join.right,
110+
self.join.on,
111+
self.join.match_condition,
112+
self.join.join_constraint,
113+
self.join.schema,
114+
)
115+
}
116+
}
117+
118+
#[pymethods]
119+
impl PyAsOfJoin {
120+
/// Retrieves the left input `LogicalPlan` to this `AsOfJoin` node
121+
fn left(&self) -> PyResult<PyLogicalPlan> {
122+
Ok(self.join.left.as_ref().clone().into())
123+
}
124+
125+
/// Retrieves the right input `LogicalPlan` to this `AsOfJoin` node
126+
fn right(&self) -> PyResult<PyLogicalPlan> {
127+
Ok(self.join.right.as_ref().clone().into())
128+
}
129+
130+
/// Retrieves the equality keys of this `AsOfJoin` node
131+
fn on(&self) -> PyResult<Vec<(PyExpr, PyExpr)>> {
132+
Ok(self
133+
.join
134+
.on
135+
.iter()
136+
.map(|(l, r)| (PyExpr::from(l.clone()), PyExpr::from(r.clone())))
137+
.collect())
138+
}
139+
140+
/// Retrieves the `AsOfMatch` condition of this `AsOfJoin` node
141+
fn match_condition(&self) -> PyResult<PyAsOfMatch> {
142+
Ok(self.join.match_condition.as_ref().clone().into())
143+
}
144+
145+
/// Retrieves the `JoinConstraint` to this `AsOfJoin` node
146+
fn join_constraint(&self) -> PyResult<PyJoinConstraint> {
147+
Ok(self.join.join_constraint.into())
148+
}
149+
150+
/// Resulting Schema for this `AsOfJoin` node instance
151+
fn schema(&self) -> PyResult<PyDFSchema> {
152+
Ok(self.join.schema.as_ref().clone().into())
153+
}
154+
155+
fn __repr__(&self) -> PyResult<String> {
156+
Ok(format!("AsOfJoin({self})"))
157+
}
158+
159+
fn __name__(&self) -> PyResult<String> {
160+
Ok("AsOfJoin".to_string())
161+
}
162+
}
163+
164+
impl LogicalNode for PyAsOfJoin {
165+
fn inputs(&self) -> Vec<PyLogicalPlan> {
166+
vec![
167+
PyLogicalPlan::from((*self.join.left).clone()),
168+
PyLogicalPlan::from((*self.join.right).clone()),
169+
]
170+
}
171+
172+
fn to_variant<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
173+
self.clone().into_bound_py_any(py)
174+
}
175+
}

0 commit comments

Comments
 (0)