Visitar URL original
chore: Update to latest main on upstream DataFusion (#1662) · voidstackloop/datafusion-python@cc2ec5c · GitHub
Skip to content

Commit cc2ec5c

Browse files
timsaucertimsaucer
authored
chore: Update to latest main on upstream DataFusion (apache#1662)
* Update to main on the upstream datafusion to prepare for DF55 release * Deprecate singular external table location AI Disclosure: This code was written in part by an AI agent.: * Fix tests for updated DataFusion behavior AI Disclosure: This code was written in part by an AI agent.: * create table requires a location * Remove unused Python token from Substrait deserialization AI Disclosure: This code was written in part by an AI agent.: * test: cover external table locations binding AI Disclosure: This code was written in part by an AI agent.:
1 parent 4702915 commit cc2ec5c

35 files changed

Lines changed: 437 additions & 399 deletions

‎Cargo.lock‎

Lines changed: 237 additions & 264 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 & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -32,24 +32,24 @@ resolver = "3"
3232

3333
[workspace.dependencies]
3434
tokio = { version = "1.52" }
35-
pyo3 = { version = "0.28" }
36-
pyo3-async-runtimes = { version = "0.28" }
35+
pyo3 = { version = "0.29" }
36+
pyo3-async-runtimes = { version = "0.29" }
3737
pyo3-log = "0.13.3"
3838
chrono = { version = "0.4", default-features = false }
39-
arrow = { version = "58" }
40-
arrow-array = { version = "58" }
41-
arrow-schema = { version = "58" }
42-
arrow-select = { version = "58" }
43-
datafusion = { version = "54" }
44-
datafusion-substrait = { version = "54" }
45-
datafusion-proto = { version = "54" }
46-
datafusion-ffi = { version = "54" }
47-
datafusion-catalog = { version = "54", default-features = false }
48-
datafusion-common = { version = "54", default-features = false }
49-
datafusion-functions-aggregate = { version = "54" }
50-
datafusion-functions-window = { version = "54" }
51-
datafusion-spark = { version = "54" }
52-
datafusion-expr = { version = "54" }
39+
arrow = { version = "59" }
40+
arrow-array = { version = "59" }
41+
arrow-schema = { version = "59" }
42+
arrow-select = { version = "59" }
43+
datafusion = { version = "54.1.0" }
44+
datafusion-substrait = { version = "54.1.0" }
45+
datafusion-proto = { version = "54.1.0" }
46+
datafusion-ffi = { version = "54.1.0" }
47+
datafusion-catalog = { version = "54.1.0", default-features = false }
48+
datafusion-common = { version = "54.1.0", default-features = false }
49+
datafusion-functions-aggregate = { version = "54.1.0" }
50+
datafusion-functions-window = { version = "54.1.0" }
51+
datafusion-spark = { version = "54.1.0" }
52+
datafusion-expr = { version = "54.1.0" }
5353
prost = "0.14.3"
5454
serde_json = "1"
5555
uuid = { version = "1.23" }
@@ -62,7 +62,7 @@ url = "2"
6262
log = "0.4.29"
6363
parking_lot = "0.12"
6464
prost-types = "0.14.3" # keep in line with `datafusion-substrait`
65-
pyo3-build-config = "0.28"
65+
pyo3-build-config = "0.29"
6666
datafusion-python-util = { path = "crates/util", version = "54.0.0" }
6767

6868
[profile.release]
@@ -72,3 +72,13 @@ codegen-units = 2
7272
# We cannot publish to crates.io with any patches in the below section. Developers
7373
# must remove any entries in this section before creating a release candidate.
7474
[patch.crates-io]
75+
datafusion = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
76+
datafusion-substrait = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
77+
datafusion-proto = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
78+
datafusion-ffi = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
79+
datafusion-catalog = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
80+
datafusion-common = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
81+
datafusion-functions-aggregate = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
82+
datafusion-functions-window = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
83+
datafusion-spark = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }
84+
datafusion-expr = { git = "https://github.com/apache/datafusion", rev = "dbcb5c0f729e9ef6b0ab4c79253fe3b657929f48" }

‎crates/core/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ datafusion = { workspace = true, features = ["avro", "unicode_expressions"] }
5353
datafusion-substrait = { workspace = true, optional = true }
5454
datafusion-proto = { workspace = true }
5555
datafusion-ffi = { workspace = true }
56-
datafusion-spark = { workspace = true }
56+
datafusion-spark = { workspace = true, features = ["core"] }
5757
prost = { workspace = true } # keep in line with `datafusion-substrait`
5858
serde_json = { workspace = true }
5959
uuid = { workspace = true, features = ["v4"] }

‎crates/core/src/array.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,10 +63,10 @@ impl PyArrowArrayExportable {
6363
};
6464

6565
let ffi_schema = FFI_ArrowSchema::try_from(&field)?;
66-
let schema_capsule = PyCapsule::new(py, ffi_schema, Some(cr"arrow_schema".into()))?;
66+
let schema_capsule = PyCapsule::new_with_value(py, ffi_schema, cr"arrow_schema")?;
6767

6868
let ffi_array = FFI_ArrowArray::new(&self.array.to_data());
69-
let array_capsule = PyCapsule::new(py, ffi_array, Some(cr"arrow_array".into()))?;
69+
let array_capsule = PyCapsule::new_with_value(py, ffi_array, cr"arrow_array")?;
7070

7171
Ok((schema_capsule, array_capsule))
7272
}

‎crates/core/src/codec.rs‎

Lines changed: 23 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -100,9 +100,13 @@ use datafusion::logical_expr::{
100100
TypeSignature, Volatility, WindowUDF, WindowUDFImpl,
101101
};
102102
use datafusion::physical_expr::PhysicalExpr;
103+
use datafusion::physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
104+
use datafusion::physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
103105
use datafusion::physical_plan::ExecutionPlan;
104106
use datafusion_proto::logical_plan::{DefaultLogicalExtensionCodec, LogicalExtensionCodec};
105-
use datafusion_proto::physical_plan::{DefaultPhysicalExtensionCodec, PhysicalExtensionCodec};
107+
use datafusion_proto::physical_plan::{
108+
DefaultPhysicalExtensionCodec, PhysicalExtensionCodec, PhysicalProtoConverterExtension,
109+
};
106110
use pyo3::prelude::*;
107111
use pyo3::sync::PyOnceLock;
108112
use pyo3::types::{PyBytes, PyTuple};
@@ -483,12 +487,18 @@ impl PhysicalExtensionCodec for PythonPhysicalCodec {
483487
buf: &[u8],
484488
inputs: &[Arc<dyn ExecutionPlan>],
485489
ctx: &TaskContext,
490+
proto_converter: &dyn PhysicalProtoConverterExtension,
486491
) -> Result<Arc<dyn ExecutionPlan>> {
487-
self.inner.try_decode(buf, inputs, ctx)
492+
self.inner.try_decode(buf, inputs, ctx, proto_converter)
488493
}
489494

490-
fn try_encode(&self, node: Arc<dyn ExecutionPlan>, buf: &mut Vec<u8>) -> Result<()> {
491-
self.inner.try_encode(node, buf)
495+
fn try_encode(
496+
&self,
497+
node: Arc<dyn ExecutionPlan>,
498+
buf: &mut Vec<u8>,
499+
proto_converter: &dyn PhysicalProtoConverterExtension,
500+
) -> Result<()> {
501+
self.inner.try_encode(node, buf, proto_converter)
492502
}
493503

494504
fn try_encode_udf(&self, node: &ScalarUDF, buf: &mut Vec<u8>) -> Result<()> {
@@ -509,16 +519,22 @@ impl PhysicalExtensionCodec for PythonPhysicalCodec {
509519
self.inner.try_decode_udf(name, buf)
510520
}
511521

512-
fn try_encode_expr(&self, node: &Arc<dyn PhysicalExpr>, buf: &mut Vec<u8>) -> Result<()> {
513-
self.inner.try_encode_expr(node, buf)
522+
fn try_encode_expr(
523+
&self,
524+
node: &Arc<dyn PhysicalExpr>,
525+
buf: &mut Vec<u8>,
526+
ctx: &PhysicalExprEncodeCtx<'_>,
527+
) -> Result<()> {
528+
self.inner.try_encode_expr(node, buf, ctx)
514529
}
515530

516531
fn try_decode_expr(
517532
&self,
518533
buf: &[u8],
519534
inputs: &[Arc<dyn PhysicalExpr>],
535+
ctx: &PhysicalExprDecodeCtx<'_>,
520536
) -> Result<Arc<dyn PhysicalExpr>> {
521-
self.inner.try_decode_expr(buf, inputs)
537+
self.inner.try_decode_expr(buf, inputs, ctx)
522538
}
523539

524540
fn try_encode_udaf(&self, node: &AggregateUDF, buf: &mut Vec<u8>) -> Result<()> {

‎crates/core/src/context.rs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1371,12 +1371,10 @@ impl PySessionContext {
13711371
&self,
13721372
py: Python<'py>,
13731373
) -> PyResult<Bound<'py, PyCapsule>> {
1374-
let name = cr"datafusion_task_context_provider".into();
1375-
13761374
let ctx_provider = Arc::clone(&self.ctx) as Arc<dyn TaskContextProvider>;
13771375
let ffi_ctx_provider = FFI_TaskContextProvider::from(&ctx_provider);
13781376

1379-
PyCapsule::new(py, ffi_ctx_provider, Some(name))
1377+
PyCapsule::new_with_value(py, ffi_ctx_provider, cr"datafusion_task_context_provider")
13801378
}
13811379

13821380
pub fn __datafusion_logical_extension_codec__<'py>(

‎crates/core/src/dataframe.rs‎

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

1818
use std::collections::HashMap;
19-
use std::ffi::{CStr, CString};
19+
use std::ffi::CStr;
2020
use std::ptr::NonNull;
2121
use std::str::FromStr;
2222
use std::sync::Arc;
@@ -1237,8 +1237,7 @@ impl PyDataFrame {
12371237
// destructor provided by PyO3 will drop the stream unless ownership is
12381238
// transferred to PyArrow during import.
12391239
let stream = FFI_ArrowArrayStream::new(reader);
1240-
let name = CString::new(ARROW_ARRAY_STREAM_NAME.to_bytes()).unwrap();
1241-
let capsule = PyCapsule::new(py, stream, Some(name))?;
1240+
let capsule = PyCapsule::new_with_value(py, stream, ARROW_ARRAY_STREAM_NAME)?;
12421241
Ok(capsule)
12431242
}
12441243

@@ -1317,7 +1316,8 @@ impl PyDataFrame {
13171316
None => Vec::new(), // Empty vector means fill null for all columns
13181317
};
13191318

1320-
let df = self.df.as_ref().clone().fill_null(scalar_value.0, cols)?;
1319+
let cols = cols.iter().map(String::as_str).collect::<Vec<_>>();
1320+
let df = self.df.as_ref().fill_null(&scalar_value.0, &cols)?;
13211321
Ok(Self::new(df))
13221322
}
13231323
}

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ impl PyCreateExternalTable {
8888
let create = CreateExternalTable {
8989
schema: Arc::new(schema.into()),
9090
name: name.into(),
91-
location,
91+
locations: vec![location],
9292
file_type,
9393
table_partition_cols,
9494
if_not_exists,
@@ -118,8 +118,8 @@ impl PyCreateExternalTable {
118118
Ok(self.create.name.to_string())
119119
}
120120

121-
pub fn location(&self) -> String {
122-
self.create.location.clone()
121+
pub fn locations(&self) -> Vec<String> {
122+
self.create.locations.clone()
123123
}
124124

125125
pub fn file_type(&self) -> String {

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

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
use datafusion::logical_expr::dml::InsertOp;
1919
use datafusion::logical_expr::{DmlStatement, WriteOp};
2020
use pyo3::IntoPyObjectExt;
21+
use pyo3::exceptions::PyNotImplementedError;
2122
use pyo3::prelude::*;
2223

2324
use super::logical_node::LogicalNode;
@@ -71,8 +72,8 @@ impl PyDmlStatement {
7172
})
7273
}
7374

74-
pub fn op(&self) -> PyWriteOp {
75-
self.dml.op.clone().into()
75+
pub fn op(&self) -> PyResult<PyWriteOp> {
76+
self.dml.op.clone().try_into()
7677
}
7778

7879
pub fn input(&self) -> PyLogicalPlan {
@@ -112,16 +113,21 @@ pub enum PyWriteOp {
112113
Truncate,
113114
}
114115

115-
impl From<WriteOp> for PyWriteOp {
116-
fn from(write_op: WriteOp) -> Self {
116+
impl TryFrom<WriteOp> for PyWriteOp {
117+
type Error = PyErr;
118+
119+
fn try_from(write_op: WriteOp) -> Result<Self, Self::Error> {
117120
match write_op {
118-
WriteOp::Insert(InsertOp::Append) => PyWriteOp::Append,
119-
WriteOp::Insert(InsertOp::Overwrite) => PyWriteOp::Overwrite,
120-
WriteOp::Insert(InsertOp::Replace) => PyWriteOp::Replace,
121-
WriteOp::Update => PyWriteOp::Update,
122-
WriteOp::Delete => PyWriteOp::Delete,
123-
WriteOp::Ctas => PyWriteOp::Ctas,
124-
WriteOp::Truncate => PyWriteOp::Truncate,
121+
WriteOp::Insert(InsertOp::Append) => Ok(PyWriteOp::Append),
122+
WriteOp::Insert(InsertOp::Overwrite) => Ok(PyWriteOp::Overwrite),
123+
WriteOp::Insert(InsertOp::Replace) => Ok(PyWriteOp::Replace),
124+
WriteOp::Update => Ok(PyWriteOp::Update),
125+
WriteOp::Delete => Ok(PyWriteOp::Delete),
126+
WriteOp::Ctas => Ok(PyWriteOp::Ctas),
127+
WriteOp::Truncate => Ok(PyWriteOp::Truncate),
128+
unsupported => Err(PyNotImplementedError::new_err(format!(
129+
"DataFusion write operation {unsupported:?} is not supported"
130+
))),
125131
}
126132
}
127133
}

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,8 @@
1818
use std::fmt::{self, Display, Formatter};
1919
use std::sync::Arc;
2020

21-
use datafusion::common::SchemaReference;
21+
use datafusion::common::{SchemaReference, TableReference};
2222
use datafusion::logical_expr::DropCatalogSchema;
23-
use datafusion::sql::TableReference;
2423
use pyo3::IntoPyObjectExt;
2524
use pyo3::exceptions::PyValueError;
2625
use pyo3::prelude::*;

0 commit comments

Comments
 (0)