This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new eb95599c feat(deps): upgrade to arrow 58, DataFusion 54, object_store
0.13, and Lance 11 (#711)
eb95599c is described below
commit eb95599c29dc9859c62294335e51d302a59d10e7
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Sep 3 14:35:17 2026 -0700
feat(deps): upgrade to arrow 58, DataFusion 54, object_store 0.13, and
Lance 11 (#711)
---
Cargo.toml | 38 ++++++------
README.md | 2 +-
benchmark/filegroup/src/gen.rs | 2 +-
benchmark/tpch/Cargo.toml | 17 +++++-
benchmark/tpch/src/datagen.rs | 12 ++--
cpp/src/util.rs | 39 +++++++++++++
crates/core/Cargo.toml | 8 +--
crates/core/src/file_group/base_file/lance.rs | 11 ++--
crates/core/src/file_group/base_file/parquet.rs | 1 +
crates/core/src/file_group/reader_v2/engine.rs | 2 +-
crates/core/src/storage/counting.rs | 31 +++++-----
crates/core/src/storage/mod.rs | 2 +-
crates/core/src/storage/reader.rs | 2 +-
crates/core/tests/statistics_tests.rs | 2 +-
crates/datafusion/src/hudi_exec.rs | 35 +++--------
crates/datafusion/src/lib.rs | 7 ---
crates/datafusion/tests/plan_tests.rs | 60 ++++++++++++-------
demo/apps/datafusion/Cargo.toml | 2 +-
demo/apps/hudi-table-api/rust/Cargo.toml | 2 +-
python/Cargo.toml | 2 +-
python/pyproject.toml | 2 +-
python/src/datafusion_internal.rs | 32 +++++-----
python/src/internal.rs | 10 ++--
python/tests/test_datafusion_ffi.py | 77 +++++++++++++++++++++++++
24 files changed, 266 insertions(+), 132 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index d469c460..4a380c5d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -38,31 +38,31 @@ repository = "https://github.com/apache/hudi-rs";
[workspace.dependencies]
# arrow
-arrow = { version = "57" }
-arrow-arith = { version = "57" }
-arrow-array = { version = "57" }
-arrow-buffer = { version = "57" }
-arrow-cast = { version = "57" }
-arrow-ipc = { version = "57" }
-arrow-json = { version = "57" }
-arrow-ord = { version = "57" }
-arrow-row = { version = "57" }
-arrow-schema = { version = "57", features = ["serde"] }
-arrow-select = { version = "57" }
-object_store = { version = "0.12", features = ["aws", "azure", "gcp"] }
-parquet = { version = "57", features = ["async", "object_store"] }
+arrow = { version = "58" }
+arrow-arith = { version = "58" }
+arrow-array = { version = "58" }
+arrow-buffer = { version = "58" }
+arrow-cast = { version = "58" }
+arrow-ipc = { version = "58" }
+arrow-json = { version = "58" }
+arrow-ord = { version = "58" }
+arrow-row = { version = "58" }
+arrow-schema = { version = "58", features = ["serde"] }
+arrow-select = { version = "58" }
+object_store = { version = "0.13.2", features = ["aws", "azure", "gcp"] }
+parquet = { version = "58", features = ["async", "object_store"] }
# avro
-arrow-avro = { version = "57" }
+arrow-avro = { version = "58" }
apache-avro = { version = "0.21", features = ["derive"] }
apache-avro-derive = { version = "0.21" }
# datafusion
-datafusion = { version = "52" }
-datafusion-expr = { version = "52" }
-datafusion-common = { version = "52" }
-datafusion-physical-expr = { version = "52" }
-datafusion-ffi = { version = "52" }
+datafusion = { version = "54.1" }
+datafusion-expr = { version = "54.1" }
+datafusion-common = { version = "54.1" }
+datafusion-physical-expr = { version = "54.1" }
+datafusion-ffi = { version = "54.1" }
# serde
percent-encoding = { version = "2" }
diff --git a/README.md b/README.md
index 2541e7b8..aab198cd 100644
--- a/README.md
+++ b/README.md
@@ -362,7 +362,7 @@ extension to query Hudi tables.
```shell
cargo new my_project --bin && cd my_project
-cargo add tokio@1 datafusion@52
+cargo add tokio@1 datafusion@54
cargo add hudi --features datafusion
```
diff --git a/benchmark/filegroup/src/gen.rs b/benchmark/filegroup/src/gen.rs
index ceba15d1..f2a5f8fd 100644
--- a/benchmark/filegroup/src/gen.rs
+++ b/benchmark/filegroup/src/gen.rs
@@ -252,7 +252,7 @@ fn batch(from: u64, rows: usize) -> RecordBatch {
fn write_one(path: &Path, target_bytes: u64, row_group_rows: usize, start_key:
u64) -> u64 {
let props = WriterProperties::builder()
.set_compression(Compression::UNCOMPRESSED)
- .set_max_row_group_size(row_group_rows)
+ .set_max_row_group_row_count(Some(row_group_rows))
.build();
let file = fs::File::create(path).expect("create parquet");
let mut writer = ArrowWriter::try_new(file, schema(),
Some(props)).expect("writer");
diff --git a/benchmark/tpch/Cargo.toml b/benchmark/tpch/Cargo.toml
index 218b899e..42cc918b 100644
--- a/benchmark/tpch/Cargo.toml
+++ b/benchmark/tpch/Cargo.toml
@@ -34,11 +34,22 @@ comfy-table = "7"
datafusion = { workspace = true }
hudi = { path = "../../crates/hudi", features = ["datafusion"] }
object_store = { workspace = true }
-parquet = { workspace = true }
+
serde = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true }
-tpchgen = "2"
-tpchgen-arrow = "2"
+# The TPC-H generator has no arrow-58 line: tpchgen-arrow ships 2.x on arrow 57
+# and 3.x on arrow 59, and the workspace sits on 58. Generation is a
+# self-contained step — `datagen::run_generate` takes a scale factor and an
+# output path and hands nothing arrow-shaped back — so it builds against its
own
+# arrow and parquet rather than holding the workspace back. `object_store` is
+# not duplicated: parquet 59 wants the same 0.13 the workspace does.
+tpchgen = "3"
+tpchgen-arrow = "3"
+arrow-gen = { package = "arrow", version = "59" }
+parquet-gen = { package = "parquet", version = "59", features = [
+ "async",
+ "object_store",
+] }
serde_yaml = "0.9"
url = { workspace = true }
diff --git a/benchmark/tpch/src/datagen.rs b/benchmark/tpch/src/datagen.rs
index fb4f6752..198edc7e 100644
--- a/benchmark/tpch/src/datagen.rs
+++ b/benchmark/tpch/src/datagen.rs
@@ -17,15 +17,15 @@
use std::sync::Arc;
-use arrow::datatypes::SchemaRef;
-use arrow::record_batch::RecordBatch;
+use arrow_gen::datatypes::SchemaRef;
+use arrow_gen::record_batch::RecordBatch;
use object_store::buffered::BufWriter;
use object_store::path::Path as ObjectPath;
use object_store::{ObjectStore, parse_url_opts};
-use parquet::arrow::ArrowWriter;
-use parquet::arrow::async_writer::AsyncArrowWriter;
-use parquet::basic::Compression;
-use parquet::file::properties::WriterProperties;
+use parquet_gen::arrow::ArrowWriter;
+use parquet_gen::arrow::async_writer::AsyncArrowWriter;
+use parquet_gen::basic::Compression;
+use parquet_gen::file::properties::WriterProperties;
use tpchgen::generators::{
CustomerGenerator, LineItemGenerator, NationGenerator, OrderGenerator,
PartGenerator,
PartSuppGenerator, RegionGenerator, SupplierGenerator,
diff --git a/cpp/src/util.rs b/cpp/src/util.rs
index 5a30cf3d..2bbf59bc 100644
--- a/cpp/src/util.rs
+++ b/cpp/src/util.rs
@@ -31,3 +31,42 @@ pub fn create_raw_pointer_for_record_batches(
let raw_ptr = Box::into_raw(Box::new(ffi_array_stream));
raw_ptr as *mut ffi::ArrowArrayStream
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::datatypes::{DataType, Field, Schema};
+ use arrow_array::ffi_stream::ArrowArrayStreamReader;
+ use arrow_array::{Int32Array, StringArray};
+ use std::sync::Arc;
+
+ /// The C++ caller only ever sees batches as a C Data Interface stream, so
an
+ /// arrow upgrade that changed how that stream is exported would break the
+ /// binding without failing anything else: running the C++ side needs Arrow
+ /// C++, which the Rust test suite does not have. Importing the exported
+ /// pointer back keeps the surface covered here instead.
+ #[test]
+ fn exported_stream_round_trips_through_the_c_data_interface() {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("name", DataType::Utf8, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["a", "b", "c"])),
+ ],
+ )
+ .unwrap();
+
+ let raw = create_raw_pointer_for_record_batches(vec![batch.clone()],
schema);
+ // SAFETY: the pointer is the one box
`create_raw_pointer_for_record_batches`
+ // leaks, taken back here rather than by the C++ caller that normally
frees it.
+ let stream = unsafe { Box::from_raw(raw as *mut FFI_ArrowArrayStream)
};
+
+ let mut reader = ArrowArrayStreamReader::try_new(*stream).unwrap();
+ assert_eq!(reader.next().unwrap().unwrap(), batch);
+ assert!(reader.next().is_none());
+ }
+}
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index b3f733b7..62b657cf 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -92,10 +92,10 @@ flate2 = { workspace = true }
# prost-build at compile time, which shells out to `protoc`. Any build of
# hudi-core (CI, Docker, local dev, downstream consumers) must therefore
# have `protoc` available on PATH — there is no opt-out.
-lance-core = { version = "4.0.1" }
-lance-encoding = { version = "4.0.1" }
-lance-file = { version = "4.0.1" }
-lance-io = { version = "4.0.1" }
+lance-core = { version = "11.0.0" }
+lance-encoding = { version = "11.0.0" }
+lance-file = { version = "11.0.0" }
+lance-io = { version = "11.0.0" }
# datafusion
datafusion = { workspace = true, optional = true }
diff --git a/crates/core/src/file_group/base_file/lance.rs
b/crates/core/src/file_group/base_file/lance.rs
index 5e3d396d..b66614dc 100644
--- a/crates/core/src/file_group/base_file/lance.rs
+++ b/crates/core/src/file_group/base_file/lance.rs
@@ -32,10 +32,12 @@ use
lance_core::utils::tokio::get_num_compute_intensive_cpus;
use lance_encoding::decoder::{DecoderPlugins, FilterExpression};
use lance_file::LanceEncodingsIo;
use lance_file::reader::{CachedFileMetadata, FileReader, FileReaderOptions,
ReaderProjection};
+use lance_file::versions::reader_projection_from_column_names;
use lance_io::ReadBatchParams;
use lance_io::object_store::{ObjectStore as LanceObjectStore,
ObjectStoreRegistry};
use lance_io::scheduler::{FileScheduler, ScanScheduler, SchedulerConfig};
use lance_io::utils::CachedFileSize;
+use object_store::ObjectStoreExt;
use object_store::path::Path as ObjPath;
use tokio::sync::OnceCell;
@@ -81,7 +83,7 @@ impl LanceBaseFileReader {
),
);
// `ObjectStoreParams::object_store` is marked deprecated in
lance-io
- // 4.0.x in favor of implementing `ObjectStoreProvider`. We
still set
+ // 11.x in favor of implementing `ObjectStoreProvider`. We
still set
// it here so Lance reuses the ObjectStore we already built
(with
// hudi-rs's storage options applied) and skips re-resolving
// credentials. `storage_options_accessor` itself is current
API.
@@ -154,7 +156,7 @@ impl LanceBaseFileReader {
return Ok(None);
}
let col_refs: Vec<&str> = col_names.iter().map(|s|
s.as_str()).collect();
- ReaderProjection::from_column_names(
+ reader_projection_from_column_names(
metadata.version(),
metadata.file_schema.as_ref(),
&col_refs,
@@ -304,13 +306,14 @@ impl BaseFileReader for LanceBaseFileReader {
ReadBatchParams::RangeFull,
batch_size,
batch_readahead,
- // lance-file 4.0.x decoders do not act on
FilterExpression:
+ // lance-file 11.x decoders do not act on FilterExpression:
// per lance-encoding, "the core decoders do not currently
// take advantage of filtering in any way." Callers needing
// row-level predicates must apply them on the returned
// record batches.
FilterExpression::no_filter(),
)
+ .await
.map_err(|e| {
StorageError::Creation(format!(
"Failed to create Lance read stream for
{relative_path}: {e}"
@@ -359,7 +362,7 @@ impl BaseFileReader for LanceBaseFileReader {
num_records: num_rows,
};
- // lance-file 4.0.x v2 format does not expose per-column min/max
via
+ // lance-file 11.x v2 format does not expose per-column min/max via
// `FileReader` (only num_pages and size_bytes via
`FileStatistics`).
// Populate an entry per file column with empty bounds so
column-level
// pruning falls back to "include".
diff --git a/crates/core/src/file_group/base_file/parquet.rs
b/crates/core/src/file_group/base_file/parquet.rs
index c2b5e494..d6fe3ca4 100644
--- a/crates/core/src/file_group/base_file/parquet.rs
+++ b/crates/core/src/file_group/base_file/parquet.rs
@@ -22,6 +22,7 @@ use std::sync::Arc;
use futures::StreamExt;
use futures::future::BoxFuture;
+use object_store::ObjectStoreExt;
use object_store::path::Path as ObjPath;
use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions};
use parquet::arrow::async_reader::{AsyncFileReader, ParquetObjectReader};
diff --git a/crates/core/src/file_group/reader_v2/engine.rs
b/crates/core/src/file_group/reader_v2/engine.rs
index f5f089a3..4483abe2 100644
--- a/crates/core/src/file_group/reader_v2/engine.rs
+++ b/crates/core/src/file_group/reader_v2/engine.rs
@@ -1428,7 +1428,7 @@ mod tests {
use parquet::arrow::ArrowWriter;
use parquet::file::properties::WriterProperties;
let props = WriterProperties::builder()
- .set_max_row_group_size(rows_per_group)
+ .set_max_row_group_row_count(Some(rows_per_group))
.build();
let file = std::fs::File::create(dir.join(name)).unwrap();
let mut writer = ArrowWriter::try_new(file, batch.schema(),
Some(props)).unwrap();
diff --git a/crates/core/src/storage/counting.rs
b/crates/core/src/storage/counting.rs
index fb72eaa6..5ae6c153 100644
--- a/crates/core/src/storage/counting.rs
+++ b/crates/core/src/storage/counting.rs
@@ -32,7 +32,7 @@ use async_trait::async_trait;
use futures::stream::BoxStream;
use object_store::path::Path;
use object_store::{
- GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
ObjectStore,
+ CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload,
ObjectMeta, ObjectStore,
PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
};
@@ -101,17 +101,22 @@ impl ObjectStore for CountingObjectStore {
}
async fn get_opts(&self, location: &Path, options: GetOptions) ->
Result<GetResult> {
- self.counts.gets.fetch_add(1, Ordering::Relaxed);
+ // A metadata-only lookup reaches the store as a `get_opts` carrying
+ // `head`, not as its own trait method, so the two request kinds are
+ // told apart here.
+ if options.head {
+ self.counts.heads.fetch_add(1, Ordering::Relaxed);
+ } else {
+ self.counts.gets.fetch_add(1, Ordering::Relaxed);
+ }
self.inner.get_opts(location, options).await
}
- async fn head(&self, location: &Path) -> Result<ObjectMeta> {
- self.counts.heads.fetch_add(1, Ordering::Relaxed);
- self.inner.head(location).await
- }
-
- async fn delete(&self, location: &Path) -> Result<()> {
- self.inner.delete(location).await
+ fn delete_stream(
+ &self,
+ locations: BoxStream<'static, Result<Path>>,
+ ) -> BoxStream<'static, Result<Path>> {
+ self.inner.delete_stream(locations)
}
fn list(&self, prefix: Option<&Path>) -> BoxStream<'static,
Result<ObjectMeta>> {
@@ -122,11 +127,7 @@ impl ObjectStore for CountingObjectStore {
self.inner.list_with_delimiter(prefix).await
}
- async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
- self.inner.copy(from, to).await
- }
-
- async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
- self.inner.copy_if_not_exists(from, to).await
+ async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) ->
Result<()> {
+ self.inner.copy_opts(from, to, options).await
}
}
diff --git a/crates/core/src/storage/mod.rs b/crates/core/src/storage/mod.rs
index 69e7474a..a025878f 100644
--- a/crates/core/src/storage/mod.rs
+++ b/crates/core/src/storage/mod.rs
@@ -26,7 +26,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use async_recursion::async_recursion;
use bytes::Bytes;
use object_store::path::Path as ObjPath;
-use object_store::{ObjectStore, parse_url_opts};
+use object_store::{ObjectStore, ObjectStoreExt, parse_url_opts};
use url::Url;
use crate::config::HudiConfigs;
diff --git a/crates/core/src/storage/reader.rs
b/crates/core/src/storage/reader.rs
index 7527a372..c0687a50 100644
--- a/crates/core/src/storage/reader.rs
+++ b/crates/core/src/storage/reader.rs
@@ -19,7 +19,7 @@
use crate::config::HudiConfigs;
use bytes::Bytes;
use object_store::path::Path as ObjPath;
-use object_store::{ObjectMeta, ObjectStore};
+use object_store::{ObjectMeta, ObjectStore, ObjectStoreExt};
use std::io::{Error, ErrorKind, Result};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
diff --git a/crates/core/tests/statistics_tests.rs
b/crates/core/tests/statistics_tests.rs
index 0186b2a0..c694f0fb 100644
--- a/crates/core/tests/statistics_tests.rs
+++ b/crates/core/tests/statistics_tests.rs
@@ -69,7 +69,7 @@ fn write_parquet_file_multiple_row_groups(batches:
&[RecordBatch], path: &std::p
let props = WriterProperties::builder()
.set_compression(Compression::SNAPPY)
.set_statistics_enabled(parquet::file::properties::EnabledStatistics::Page)
- .set_max_row_group_size(3) // Force smaller row groups
+ .set_max_row_group_row_count(Some(3)) // Force smaller row groups
.build();
let mut writer = ArrowWriter::try_new(file, batches[0].schema(),
Some(props)).unwrap();
for batch in batches {
diff --git a/crates/datafusion/src/hudi_exec.rs
b/crates/datafusion/src/hudi_exec.rs
index 9c23af1f..73606154 100644
--- a/crates/datafusion/src/hudi_exec.rs
+++ b/crates/datafusion/src/hudi_exec.rs
@@ -19,7 +19,6 @@
//! Custom DataFusion execution plan for reading Hudi tables through
//! [`FileGroupReader`], supporting all base file formats and MOR log merging.
-use std::any::Any;
use std::fmt;
use std::pin::Pin;
use std::sync::Arc;
@@ -63,7 +62,7 @@ pub struct HudiScanExec {
projected_schema: SchemaRef,
projection: Option<Vec<usize>>,
limit: Option<usize>,
- properties: PlanProperties,
+ properties: Arc<PlanProperties>,
metrics: ExecutionPlanMetricsSet,
}
@@ -152,7 +151,7 @@ impl HudiScanExec {
projected_schema,
projection,
limit,
- properties,
+ properties: Arc::new(properties),
metrics: ExecutionPlanMetricsSet::new(),
}
}
@@ -262,11 +261,7 @@ impl ExecutionPlan for HudiScanExec {
"HudiScanExec"
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
- fn properties(&self) -> &PlanProperties {
+ fn properties(&self) -> &Arc<PlanProperties> {
&self.properties
}
@@ -394,11 +389,7 @@ impl ExecutionPlan for HudiScanExec {
)))
}
- fn statistics(&self) -> Result<Statistics> {
- Ok(self.aggregate_file_slice_statistics())
- }
-
- fn partition_statistics(&self, partition: Option<usize>) ->
Result<Statistics> {
+ fn partition_statistics(&self, partition: Option<usize>) ->
Result<Arc<Statistics>> {
let column_statistics =
vec![ColumnStatistics::new_unknown();
self.projected_schema.fields().len()];
@@ -410,11 +401,14 @@ impl ExecutionPlan for HudiScanExec {
),
Some(idx) => match self.file_slice_partitions.get(idx) {
Some(slices) => Box::new(std::iter::once(slices.as_slice())),
- None => return
Ok(Statistics::new_unknown(&self.projected_schema)),
+ None => return
Ok(Arc::new(Statistics::new_unknown(&self.projected_schema))),
},
};
- Ok(Self::aggregate_partitions(partitions, column_statistics))
+ Ok(Arc::new(Self::aggregate_partitions(
+ partitions,
+ column_statistics,
+ )))
}
fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn
ExecutionPlan>> {
@@ -440,17 +434,6 @@ impl ExecutionPlan for HudiScanExec {
}
impl HudiScanExec {
- fn aggregate_file_slice_statistics(&self) -> Statistics {
- let column_statistics =
- vec![ColumnStatistics::new_unknown();
self.projected_schema.fields().len()];
- Self::aggregate_partitions(
- self.file_slice_partitions
- .iter()
- .map(|slices| slices.as_slice()),
- column_statistics,
- )
- }
-
fn aggregate_partitions<'a, I>(
partitions: I,
column_statistics: Vec<ColumnStatistics>,
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/lib.rs
index 9d8ebc0b..2e70738c 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/lib.rs
@@ -20,7 +20,6 @@
pub(crate) mod hudi_exec;
pub(crate) mod util;
-use std::any::Any;
use std::collections::HashMap;
use std::error::Error;
use std::fmt::Debug;
@@ -787,10 +786,6 @@ impl HudiDataSource {
#[async_trait]
impl TableProvider for HudiDataSource {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> SchemaRef {
self.schema.clone()
}
@@ -1158,7 +1153,6 @@ mod tests {
let state = ctx.state();
let plan = hudi.scan(&state, None, filters, None).await.unwrap();
let exec = plan
- .as_any()
.downcast_ref::<HudiScanExec>()
.expect("scan should route to HudiScanExec");
@@ -1431,7 +1425,6 @@ mod tests {
.await
.unwrap();
let exec = plan
- .as_any()
.downcast_ref::<HudiScanExec>()
.expect("MOR snapshot scan should use HudiScanExec");
diff --git a/crates/datafusion/tests/plan_tests.rs
b/crates/datafusion/tests/plan_tests.rs
index 419d2ced..24b0c1af 100644
--- a/crates/datafusion/tests/plan_tests.rs
+++ b/crates/datafusion/tests/plan_tests.rs
@@ -148,26 +148,46 @@ async fn verify_plan(
plan.contains("SortExec: TopK(fetch=10)"),
"Plan should contain TopK sort"
);
+ // The projection, struct field access included, is pushed into the Parquet
+ // source rather than planned as its own `ProjectionExec`. Keeping the
whole
+ // column list in the anchor is what makes this catch a scan that stopped
+ // projecting; only the `structField@N` index varies per table, so the
alias
+ // is matched separately.
assert!(
- plan.contains(&format!(
- "ProjectionExec: expr=[id@0 as id, name@1 as name, isActive@2 as
isActive, \
- get_field(structField@3, field2) as
{table_name}.structField[field2]]"
- )),
- "Plan should contain expected projection"
+ plan.contains("projection=[id, name, isActive, get_field(structField@")
+ && plan.contains(&format!(", field2) as
{table_name}.structField[field2]")),
+ "Plan should project the struct field"
);
- // With pushdown_filters enabled, simple predicates (id % 2 = 0, name !=
Alice)
- // are pushed into the Parquet source. Only non-pushable predicates like
- // struct field access remain in FilterExec.
+ // Simple predicates (id % 2 = 0, name != Alice) and the struct field
access
+ // alike are pushed into the Parquet source.
assert!(
- plan.contains("get_field(structField@3, field2) > 30"),
- "Plan should contain struct field filter (either in FilterExec or
DataSourceExec)"
+ scan_predicate(&plan).contains(", field2) > 30"),
+ "Scan predicate should contain the struct field filter"
);
+ // `hoodie.read.input.partitions` decides how many groups the scan is split
+ // into. One group renders in the singular, which this prefix covers too.
assert!(
-
plan.contains(&format!("input_partitions={planned_input_partitioned}")),
- "Plan should contain expected
input_partitions={planned_input_partitioned}"
+ plan.contains(&format!("file_groups={{{planned_input_partitioned}
group")),
+ "Plan should scan {planned_input_partitioned} file group(s)"
);
}
+/// The predicate the scan evaluates per row, which it renders as `,
predicate=`.
+///
+/// The scan also renders a `, pruning_predicate=` derived from that one, and a
+/// bare substring search cannot tell the two apart. Only the former says what
+/// actually reached the source.
+fn scan_predicate(plan: &str) -> &str {
+ let predicate = plan
+ .split(", predicate=")
+ .nth(1)
+ .expect("plan should carry a scan predicate");
+ predicate
+ .split(", pruning_predicate=")
+ .next()
+ .unwrap_or(predicate)
+}
+
async fn verify_data(ctx: &SessionContext, sql: &str, table_name: &str) {
let df = ctx.sql(sql).await.unwrap();
let rb = df.collect().await.unwrap();
@@ -385,16 +405,16 @@ mod v8_tests {
"Should have TopK sort"
);
assert!(
- plan_lines[2].contains("ProjectionExec"),
- "Should have ProjectionExec"
+ plan_lines[2].starts_with("DataSourceExec"),
+ "Should scan through DataSourceExec"
);
assert!(
- plan.contains("FilterExec"),
- "Should have FilterExec for non-partition filters"
+ scan_predicate(&plan).contains(", field2) > 30"),
+ "Non-partition filters should reach the source"
);
assert!(
- plan.contains("input_partitions=2"),
- "Should have input_partitions=2"
+ plan.contains("file_groups={2 groups:"),
+ "Should scan 2 file groups"
);
// Verify data
@@ -430,8 +450,8 @@ mod v8_tests {
let plan = get_str_column(explaining_rb, "plan").join("");
assert!(
- plan.contains("input_partitions=2"),
- "Complex keygen table should have input_partitions=2"
+ plan.contains("file_groups={2 groups:"),
+ "Complex keygen table should scan 2 file groups"
);
assert!(
plan.contains("DataSourceExec"),
diff --git a/demo/apps/datafusion/Cargo.toml b/demo/apps/datafusion/Cargo.toml
index cabebd1c..bb9ea81b 100644
--- a/demo/apps/datafusion/Cargo.toml
+++ b/demo/apps/datafusion/Cargo.toml
@@ -25,5 +25,5 @@ edition = "2021"
[dependencies]
tokio = "1"
-datafusion = "52"
+datafusion = "54"
hudi = { path = "../../../crates/hudi", features = ["datafusion"] }
diff --git a/demo/apps/hudi-table-api/rust/Cargo.toml
b/demo/apps/hudi-table-api/rust/Cargo.toml
index cca2e3fe..33df90fa 100644
--- a/demo/apps/hudi-table-api/rust/Cargo.toml
+++ b/demo/apps/hudi-table-api/rust/Cargo.toml
@@ -25,6 +25,6 @@ edition = "2021"
[dependencies]
tokio = "1"
-arrow = { version = "57" }
+arrow = { version = "58" }
hudi = { path = "../../../../crates/hudi" }
diff --git a/python/Cargo.toml b/python/Cargo.toml
index df673504..09c6c0b9 100644
--- a/python/Cargo.toml
+++ b/python/Cargo.toml
@@ -50,7 +50,7 @@ futures = { workspace = true }
tokio = { workspace = true }
[dependencies.pyo3]
-version = "0.26"
+version = "0.28"
features = ["extension-module", "abi3", "abi3-py310"]
[features]
diff --git a/python/pyproject.toml b/python/pyproject.toml
index 35a5d47a..a6b345e7 100644
--- a/python/pyproject.toml
+++ b/python/pyproject.toml
@@ -53,7 +53,7 @@ lint = [
"mypy==2.3.1",
]
datafusion = [
- "datafusion==52.3.0",
+ "datafusion==54.0.0",
]
[tool.maturin]
diff --git a/python/src/datafusion_internal.rs
b/python/src/datafusion_internal.rs
index 242bd372..f25167d0 100644
--- a/python/src/datafusion_internal.rs
+++ b/python/src/datafusion_internal.rs
@@ -40,23 +40,29 @@ fn extract_codec(session: Bound<PyAny>) ->
PyResult<FFI_LogicalExtensionCodec> {
} else {
session
};
- let capsule = capsule_obj.downcast::<PyCapsule>()?;
- if let Some(name) = capsule.name()? {
- let name = name
- .to_str()
- .map_err(|e| PyValueError::new_err(format!("{e}")))?;
- if name != "datafusion_logical_extension_codec" {
- return Err(PyValueError::new_err(format!(
- "Expected PyCapsule name 'datafusion_logical_extension_codec',
got '{name}'"
- )));
- }
- }
- let codec = unsafe { capsule.reference::<FFI_LogicalExtensionCodec>() };
+ let capsule = capsule_obj.cast::<PyCapsule>()?;
+ // `pointer_checked` is the name check and the pointer read in one step: it
+ // refuses any capsule not carrying exactly this name. An unnamed capsule
no
+ // longer passes, where the previous name-then-read pair let one through.
+ let codec = capsule
+ .pointer_checked(Some(c"datafusion_logical_extension_codec"))
+ .map_err(|e| {
+ PyValueError::new_err(format!(
+ "Expected a PyCapsule named
'datafusion_logical_extension_codec': {e}"
+ ))
+ })?;
+ // SAFETY: DataFusion puts that name only on a capsule holding an
+ // `FFI_LogicalExtensionCodec`, so the check above settles what the capsule
+ // holds. It does not settle which `datafusion-ffi` built it, because the
+ // name carries no version: a `datafusion` wheel built against another
major
+ // would present the same name over a different layout. Holding the two in
+ // step is what the pinned `datafusion` extra in pyproject.toml is for.
+ let codec = unsafe { codec.cast::<FFI_LogicalExtensionCodec>().as_ref() };
Ok(codec.clone())
}
#[cfg(not(tarpaulin_include))]
-#[pyclass(name = "HudiDataFusionDataSource")]
+#[pyclass(name = "HudiDataFusionDataSource", from_py_object)]
#[derive(Clone)]
pub struct HudiDataFusionDataSource {
table: InternalDataFusionHudiDataSource,
diff --git a/python/src/internal.rs b/python/src/internal.rs
index 0d2be9bd..98639e12 100644
--- a/python/src/internal.rs
+++ b/python/src/internal.rs
@@ -84,7 +84,7 @@ impl From<PythonError> for PyErr {
/// Python wrapper around [`hudi::table::QueryType`].
#[cfg(not(tarpaulin_include))]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
-#[pyclass(eq)]
+#[pyclass(eq, from_py_object)]
pub struct HudiQueryType {
inner: QueryType,
}
@@ -121,7 +121,7 @@ impl HudiQueryType {
#[cfg(not(tarpaulin_include))]
#[derive(Clone, Debug, Default)]
-#[pyclass]
+#[pyclass(from_py_object)]
pub struct HudiReadOptions {
inner: ReadOptions,
}
@@ -347,7 +347,7 @@ impl HudiRecordBatchStream {
#[cfg(not(tarpaulin_include))]
#[derive(Clone, Debug)]
-#[pyclass]
+#[pyclass(from_py_object)]
pub struct HudiFileGroupReader {
inner: FileGroupReader,
}
@@ -488,7 +488,7 @@ impl HudiFileGroupReader {
#[cfg(not(tarpaulin_include))]
#[derive(Clone, Debug)]
-#[pyclass]
+#[pyclass(from_py_object)]
pub struct HudiFileSlice {
#[pyo3(get)]
file_id: String,
@@ -632,7 +632,7 @@ impl From<&FileSlice> for HudiFileSlice {
#[cfg(not(tarpaulin_include))]
#[derive(Clone, Debug)]
-#[pyclass]
+#[pyclass(from_py_object)]
pub struct HudiInstant {
inner: Instant,
}
diff --git a/python/tests/test_datafusion_ffi.py
b/python/tests/test_datafusion_ffi.py
new file mode 100644
index 00000000..51628fba
--- /dev/null
+++ b/python/tests/test_datafusion_ffi.py
@@ -0,0 +1,77 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Reads through the FFI table provider, against the real DataFusion Python
package.
+
+This is the one path where a version skew between the `datafusion-ffi` this
+crate is built against and the `datafusion` wheel installed alongside it shows
+up, and it shows up as an ABI mismatch rather than a compile error, so nothing
+else in the suite would catch it.
+"""
+
+import pytest
+
+datafusion = pytest.importorskip("datafusion")
+
+from hudi._internal import HudiDataFusionDataSource, get_test_table_path #
noqa: E402
+
+
[email protected]
+def v6_table() -> str:
+ return get_test_table_path("v6_simplekeygen_nonhivestyle", "cow")
+
+
+def test_datafusion_ffi_reads_through_the_table_provider(v6_table):
+ ctx = datafusion.SessionContext()
+ ctx.register_table("hudi_t", HudiDataFusionDataSource(v6_table))
+
+ rows = []
+ for batch in ctx.sql(
+ 'SELECT id, name, "isActive" FROM hudi_t ORDER BY id'
+ ).collect():
+ columns = batch.to_pydict()
+ for i in range(batch.num_rows):
+ rows.append((columns["id"][i], columns["name"][i],
columns["isActive"][i]))
+
+ assert rows == [
+ (1, "Alice", False),
+ (2, "Bob", False),
+ (3, "Carol", True),
+ (4, "Diana", True),
+ ]
+
+
+def test_datafusion_ffi_pushes_the_filter_into_the_provider(v6_table):
+ ctx = datafusion.SessionContext()
+ ctx.register_table("hudi_t", HudiDataFusionDataSource(v6_table))
+
+ batch = ctx.sql("SELECT count(*) AS n FROM hudi_t WHERE id % 2 =
0").collect()[0]
+ assert batch.to_pydict()["n"] == [2]
+
+
+def test_datafusion_ffi_plan_goes_through_the_ffi_boundary(v6_table):
+ ctx = datafusion.SessionContext()
+ ctx.register_table("hudi_t", HudiDataFusionDataSource(v6_table))
+
+ plans = []
+ for batch in ctx.sql("EXPLAIN SELECT id FROM hudi_t").collect():
+ columns = batch.to_pydict()
+ for i in range(batch.num_rows):
+ if columns["plan_type"][i] == "physical_plan":
+ plans.append(columns["plan"][i])
+
+ assert any("FFI_ExecutionPlan" in plan for plan in plans), plans
