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 017f0bb2 feat: upgrade to arrow 59, DataFusion 55 and pyo3 0.29 with
Lance kept on by default (#698)
017f0bb2 is described below
commit 017f0bb278ab882898dd78cd0b25a0a9db1c979e
Author: Xuanyi Li <[email protected]>
AuthorDate: Fri Oct 23 22:55:01 2026 -0700
feat: upgrade to arrow 59, DataFusion 55 and pyo3 0.29 with Lance kept on
by default (#698)
Co-authored-by: Y Ethan Guo <[email protected]>
---
AGENTS.md | 5 +-
Cargo.toml | 36 +-
README.md | 30 +-
benchmark/tpch/Cargo.toml | 12 +-
benchmark/tpch/src/datagen.rs | 12 +-
build-wrapper.sh | 29 +-
crates/core/Cargo.toml | 9 +-
crates/core/src/avro_to_arrow/schema.rs | 2 +-
crates/core/src/config/read.rs | 5 +
crates/core/src/file_group/base_file/lance.rs | 23 +-
.../core/src/file_group/base_file/lance_arrow.rs | 450 +++++++++++++++++++++
crates/core/src/file_group/base_file/mod.rs | 1 +
crates/core/src/file_group/base_file/parquet.rs | 8 +
crates/core/src/file_group/reader_v2/row_serde.rs | 2 +-
crates/datafusion/src/hudi_exec.rs | 9 +
crates/datafusion/src/lib.rs | 165 +++++++-
crates/datafusion/tests/plan_tests.rs | 39 +-
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 | 2 +-
22 files changed, 771 insertions(+), 76 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index 9cd1d0a3..5221f4a1 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -25,7 +25,10 @@ benchmark/filegroup/ file-group read benchmark (fg-bench /
fg-gen)
The [`Makefile`](./Makefile) is the canonical command surface — **prefer `make
<target>`** so agents
and humans run what CI runs. Cargo / maturin go through
[`build-wrapper.sh`](./build-wrapper.sh)
-(macOS 26+ SDK env vars).
+(macOS 26+ SDK env vars; on macOS arm64 it also builds dependencies at
opt-level 1, without which
+the `hudi-datafusion` test binaries and the Python extension fail to link, see
+[#844](https://github.com/apache/hudi-rs/issues/844)). On an arm64 Mac, run
`cargo test` for
+`hudi-datafusion` through the wrapper: `./build-wrapper.sh cargo test -p
hudi-datafusion`.
```bash
make setup-venv && source .venv/bin/activate
diff --git a/Cargo.toml b/Cargo.toml
index f6dfe768..0b6bda38 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -38,31 +38,31 @@ repository = "https://github.com/apache/hudi-rs";
[workspace.dependencies]
# arrow
-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" }
+arrow = { version = "59.2.0" }
+arrow-arith = { version = "59.2.0" }
+arrow-array = { version = "59.2.0" }
+arrow-buffer = { version = "59.2.0" }
+arrow-cast = { version = "59.2.0" }
+arrow-ipc = { version = "59.2.0" }
+arrow-json = { version = "59.2.0" }
+arrow-ord = { version = "59.2.0" }
+arrow-row = { version = "59.2.0" }
+arrow-schema = { version = "59.2.0", features = ["serde"] }
+arrow-select = { version = "59.2.0" }
object_store = { version = "0.13.2", features = ["aws", "azure", "gcp"] }
-parquet = { version = "58", features = ["async", "object_store"] }
+parquet = { version = "59.2.0", features = ["async", "object_store"] }
# avro
-arrow-avro = { version = "58" }
+arrow-avro = { version = "59.2.0" }
apache-avro = { version = "0.21", features = ["derive"] }
apache-avro-derive = { version = "0.21" }
# datafusion
-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" }
+datafusion = { version = "55.2" }
+datafusion-expr = { version = "55.2" }
+datafusion-common = { version = "55.2" }
+datafusion-physical-expr = { version = "55.2" }
+datafusion-ffi = { version = "55.2" }
# serde
percent-encoding = { version = "2" }
diff --git a/README.md b/README.md
index 4be8911f..3760aabc 100644
--- a/README.md
+++ b/README.md
@@ -364,7 +364,7 @@ silently redirect every later read.
| `hoodie.read.stream.batch_size` | `1024`
| table or read | Rows per batch for streaming reads.
|
| `hoodie.read.file.slice.read.concurrency` | `4`
| table or read | File slices read concurrently.
|
| `hoodie.read.scan.max.memory.size` | unset
| table or read | Total bytes a whole scan may use for concurrent slice
reads; when set, the concurrency is derived from it. Reaching the limit lowers
throughput, it never fails the read. |
-| `hoodie.read.input.partitions` | `0`
| table or read | How many partitions the DataFusion table provider buckets
the file slices into. `0` defers to DataFusion's `target_partitions`. |
+| `hoodie.read.input.partitions` | `0`
| table or read | How many partitions the DataFusion table provider buckets
the file slices into. `0` defers to DataFusion's `target_partitions`. On the
parquet-source path (copy-on-write, read-optimized merge-on-read) DataFusion
may split a bucket further once it exceeds
`datafusion.optimizer.repartition_file_min_size`, unless
`datafusion.optimizer.repartition_file_scans` is off. |
| `hoodie.merge.use.record.positions` | `false`
| table or read | Match a log record to the base row it updates by position
rather than by record key. Honored by reader version 2 only; a log block
written without positions is merged by key regardless. |
A table whose `hoodie.record.merge.mode` is `CUSTOM` needs a merger for its
payload class. When
@@ -392,12 +392,38 @@ Create a Hudi file group reader instance using its
constructor or the Hudi table
Enabling the `hudi` crate with `datafusion` feature will provide a
[DataFusion](https://datafusion.apache.org/)
extension to query Hudi tables.
+Each hudi-rs release targets one DataFusion major, and with it one arrow
major. The Rust `hudi`
+crate's `TableProvider` and the Python `hudi[datafusion]` extra both have to
match the DataFusion
+version of the application they plug into, so pick the hudi-rs release that
pairs with yours.
+Readers that take Arrow data out through the C data interface (Python
`read()`, the C++ and JVM
+bindings) are not tied to these versions. The `hudi[datafusion]` extra first
shipped in 0.5.
+
+| hudi-rs | DataFusion | arrow | Python `datafusion` package |
+|---------|------------|-------|-----------------------------|
+| 0.6 (unreleased) | 55 | 59 | `>=55.0.0,<56` |
+| 0.5 | 54 | 58 | `>=54.0.0,<55` |
+| 0.4 | 46 | 54 | not available |
+| 0.3 | 43 | 53 | not available |
+
+On macOS arm64, a debug build of an application that links `hudi` with the
`datafusion` feature
+can fail with `ld: B/BL out of range ... (max +/-128MB)`: unoptimized
dependencies push the
+binary's code section past the arm64 branch range. Optimize dependencies in
the dev profile:
+
+```toml
+[profile.dev.package."*"]
+opt-level = 1
+```
+
+This workspace applies the same setting from `build-wrapper.sh` on macOS arm64
only, since it
+slows compilation noticeably. It goes away once a Lance release shares the
workspace's arrow major
+([#844](https://github.com/apache/hudi-rs/issues/844)).
+
<details>
<summary>Add crate hudi with datafusion feature to your application to query a
Hudi table.</summary>
```shell
cargo new my_project --bin && cd my_project
-cargo add tokio@1 datafusion@54
+cargo add tokio@1 datafusion@55
cargo add hudi --features datafusion
```
diff --git a/benchmark/tpch/Cargo.toml b/benchmark/tpch/Cargo.toml
index 09977eed..91efe1d7 100644
--- a/benchmark/tpch/Cargo.toml
+++ b/benchmark/tpch/Cargo.toml
@@ -35,22 +35,12 @@ comfy-table = "8"
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 }
-# 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 198edc7e..fb4f6752 100644
--- a/benchmark/tpch/src/datagen.rs
+++ b/benchmark/tpch/src/datagen.rs
@@ -17,15 +17,15 @@
use std::sync::Arc;
-use arrow_gen::datatypes::SchemaRef;
-use arrow_gen::record_batch::RecordBatch;
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::RecordBatch;
use object_store::buffered::BufWriter;
use object_store::path::Path as ObjectPath;
use object_store::{ObjectStore, parse_url_opts};
-use parquet_gen::arrow::ArrowWriter;
-use parquet_gen::arrow::async_writer::AsyncArrowWriter;
-use parquet_gen::basic::Compression;
-use parquet_gen::file::properties::WriterProperties;
+use parquet::arrow::ArrowWriter;
+use parquet::arrow::async_writer::AsyncArrowWriter;
+use parquet::basic::Compression;
+use parquet::file::properties::WriterProperties;
use tpchgen::generators::{
CustomerGenerator, LineItemGenerator, NationGenerator, OrderGenerator,
PartGenerator,
PartSuppGenerator, RegionGenerator, SupplierGenerator,
diff --git a/build-wrapper.sh b/build-wrapper.sh
index de58e1e3..8eb5d90e 100755
--- a/build-wrapper.sh
+++ b/build-wrapper.sh
@@ -16,8 +16,8 @@
# specific language governing permissions and limitations
# under the License.
-# Setup script for macOS 26 (Tahoe) build environment
-# This script detects macOS version and sets SDK environment variables if
needed
+# Build environment wrapper for macOS: sets SDK environment variables on
+# macOS 26 (Tahoe) and later, and optimizes dependencies in dev builds on
arm64.
set -e
@@ -41,5 +41,30 @@ if [[ "$OSTYPE" == "darwin"* ]]; then
fi
fi
+# On macOS arm64, dependencies are compiled at opt-level 1 in dev builds.
+# hudi-core links two arrow majors while Lance stays on its own, and at
+# opt-level 0 that puts the code section of the hudi-datafusion test binaries
+# and the Python extension past the arm64 +/-128MB branch range, where the
+# linker cannot reach `ctor`'s startup section (pulled in through
+# lance-io -> opendal). Workspace crates stay at opt-level 0 for debugging.
+# The override is a global cargo flag, so it reaches cargo's built-in
+# subcommands and maturin, but not external ones such as `cargo tarpaulin`.
+# Remove with the Lance arrow bridge:
https://github.com/apache/hudi-rs/issues/844
+if [[ "$OSTYPE" == "darwin"* && "$(uname -m)" == "arm64" ]]; then
+ DEP_OPT_LEVEL='profile.dev.package."*".opt-level=1'
+ case "$1" in
+ cargo)
+ echo "Detected macOS arm64: dependencies build at opt-level 1"
+ set -- "$1" --config "$DEP_OPT_LEVEL" "${@:2}"
+ ;;
+ maturin)
+ if [[ "$2" == "build" || "$2" == "develop" ]]; then
+ echo "Detected macOS arm64: dependencies build at opt-level 1"
+ set -- "$1" "$2" --config "$DEP_OPT_LEVEL" "${@:3}"
+ fi
+ ;;
+ esac
+fi
+
# Run the command passed as arguments
exec "$@"
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index 62b657cf..b1acb775 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -32,7 +32,7 @@ unexpected_cfgs = { level = "warn", check-cfg =
['cfg(tarpaulin_include)'] }
[dependencies]
# arrow
-arrow = { workspace = true }
+arrow = { workspace = true, features = ["ffi"] }
arrow-arith = { workspace = true }
arrow-array = { workspace = true }
arrow-buffer = { workspace = true }
@@ -96,6 +96,13 @@ lance-core = { version = "11.0.0" }
lance-encoding = { version = "11.0.0" }
lance-file = { version = "11.0.0" }
lance-io = { version = "11.0.0" }
+# Lance 11 is built on arrow 58 while the workspace is on 59. These are the
+# arrow crates Lance's API speaks, renamed so the Lance reader can export its
+# batches through the Arrow C data interface and import them as the
+# workspace's arrow; see `file_group::base_file::lance_arrow`. Drop them once a
+# Lance release shares the workspace's arrow major.
+arrow-array-lance = { package = "arrow-array", version = "58", features =
["ffi"] }
+arrow-schema-lance = { package = "arrow-schema", version = "58", features =
["ffi"] }
# datafusion
datafusion = { workspace = true, optional = true }
diff --git a/crates/core/src/avro_to_arrow/schema.rs
b/crates/core/src/avro_to_arrow/schema.rs
index a342224c..3a63a9f9 100644
--- a/crates/core/src/avro_to_arrow/schema.rs
+++ b/crates/core/src/avro_to_arrow/schema.rs
@@ -126,7 +126,7 @@ fn schema_to_field_with_props(
.map(|s| schema_to_field_with_props(s, None, has_nullable,
None))
.collect::<Result<Vec<Field>>>()?;
let type_ids = 0_i8..fields.len() as i8;
- DataType::Union(UnionFields::new(type_ids, fields),
UnionMode::Dense)
+ DataType::Union(UnionFields::try_new(type_ids, fields)?,
UnionMode::Dense)
}
}
AvroSchema::Record(RecordSchema { fields, .. }) => {
diff --git a/crates/core/src/config/read.rs b/crates/core/src/config/read.rs
index 4370fe79..50cf3a41 100644
--- a/crates/core/src/config/read.rs
+++ b/crates/core/src/config/read.rs
@@ -114,6 +114,11 @@ pub enum HudiReadConfig {
/// Number of input partitions to read the data in parallel.
///
/// For processing 100 files, [InputPartitions] being 5 will produce 5
partitions, with each partition having 20 files.
+ ///
+ /// On the parquet-source path (copy-on-write, or read-optimized
merge-on-read)
+ /// DataFusion may split these buckets further once one exceeds
+ /// `datafusion.optimizer.repartition_file_min_size`, unless
+ /// `datafusion.optimizer.repartition_file_scans` is off.
InputPartitions,
/// When set to true, only base files will be read for optimized reads.
diff --git a/crates/core/src/file_group/base_file/lance.rs
b/crates/core/src/file_group/base_file/lance.rs
index b66614dc..9f41a67e 100644
--- a/crates/core/src/file_group/base_file/lance.rs
+++ b/crates/core/src/file_group/base_file/lance.rs
@@ -41,6 +41,7 @@ use object_store::ObjectStoreExt;
use object_store::path::Path as ObjPath;
use tokio::sync::OnceCell;
+use super::lance_arrow;
use super::reader::{BaseFileReadOptions, BaseFileReader, BaseFileStream};
use crate::statistics::{ColumnStatistics, StatisticsContainer,
StatsGranularity};
use crate::storage::Storage;
@@ -232,7 +233,13 @@ impl LanceBaseFileReader {
let (reader, _) = self
.open_file_reader_with_metadata(relative_path, None, None)
.await?;
- Ok(arrow_schema::Schema::from(reader.schema().as_ref()))
+ Self::file_schema(&reader)
+ }
+
+ /// The Lance file's schema in the workspace's arrow.
+ fn file_schema(reader: &FileReader) -> Result<arrow_schema::Schema> {
+ let lance_schema =
arrow_schema_lance::Schema::from(reader.schema().as_ref());
+ lance_arrow::schema_to_host(&lance_schema)
}
/// Build a single zero-column [`RecordBatch`] with the Lance file's
@@ -320,10 +327,12 @@ impl BaseFileReader for LanceBaseFileReader {
))
})?;
- let arrow_schema = lance_stream.schema().clone();
+ let arrow_schema: SchemaRef =
+
Arc::new(lance_arrow::schema_to_host(lance_stream.schema().as_ref())?);
+ let batch_schema = arrow_schema.clone();
let mapped_stream = lance_stream
- .map(|result| result.map_err(StorageError::from))
+ .map(move |result| lance_arrow::batch_to_host(result?,
batch_schema.clone()))
.boxed();
Ok(BaseFileStream::new(arrow_schema, mapped_stream))
@@ -368,12 +377,12 @@ impl BaseFileReader for LanceBaseFileReader {
// pruning falls back to "include".
let mut col_stats =
StatisticsContainer::new(StatsGranularity::File);
col_stats.num_rows = Some(num_rows);
- let file_schema = reader.schema();
- for field in &file_schema.fields {
- let field_name = field.name.clone();
+ let file_schema = Self::file_schema(&reader)?;
+ for field in file_schema.fields() {
+ let field_name = field.name().clone();
col_stats.columns.insert(
field_name.clone(),
- ColumnStatistics::new(field_name, field.data_type()),
+ ColumnStatistics::new(field_name,
field.data_type().clone()),
);
}
diff --git a/crates/core/src/file_group/base_file/lance_arrow.rs
b/crates/core/src/file_group/base_file/lance_arrow.rs
new file mode 100644
index 00000000..03229873
--- /dev/null
+++ b/crates/core/src/file_group/base_file/lance_arrow.rs
@@ -0,0 +1,450 @@
+/*
+ * 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.
+ */
+//! Carries Lance's record batches into the workspace's arrow through the
+//! Arrow C data interface.
+//!
+//! Lance is compiled against a different arrow major than the rest of
+//! hudi-core, and to rustc the two crates' `RecordBatch` types are unrelated,
+//! so a batch cannot be handed across directly. The C data interface structs
+//! (`ArrowArray`, `ArrowSchema`) are `#[repr(C)]` with a layout the Arrow
+//! specification fixes for every implementation, so a struct one arrow fills
+//! can be read by the other. The imported array wraps the exporter's buffers
+//! (`from_ffi` copies only a buffer that misses arrow's alignment, which
+//! Lance's decoders do not produce) and frees them through the exporter's
+//! `release` callback when the host batch drops.
+
+use std::mem::transmute;
+
+use arrow::array::{RecordBatch, RecordBatchOptions, make_array};
+use arrow::datatypes::{Schema, SchemaRef};
+use arrow::error::ArrowError;
+use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema, from_ffi};
+use arrow_array_lance::RecordBatch as LanceRecordBatch;
+use arrow_array_lance::ffi::{FFI_ArrowArray as LanceFfiArray, to_ffi};
+use arrow_schema_lance::Schema as LanceSchema;
+use arrow_schema_lance::ffi::FFI_ArrowSchema as LanceFfiSchema;
+
+use crate::storage::error::Result;
+
+/// Converts a schema from Lance's arrow into the workspace's arrow.
+///
+/// Field names, nullability, nested types and metadata (including extension
+/// type names) all travel in the C `ArrowSchema` struct.
+pub(super) fn schema_to_host(schema: &LanceSchema) -> Result<Schema> {
+ let exported = LanceFfiSchema::try_from(schema).map_err(export_error)?;
+ let imported = cross_schema(exported);
+ Ok(Schema::try_from(&imported)?)
+}
+
+/// Converts a batch from Lance's arrow into the workspace's arrow, sharing
+/// the column buffers.
+///
+/// `schema` is the stream's schema, already crossed by [`schema_to_host`].
+pub(super) fn batch_to_host(batch: LanceRecordBatch, schema: SchemaRef) ->
Result<RecordBatch> {
+ let mut columns = Vec::with_capacity(batch.num_columns());
+ for column in batch.columns() {
+ let (array, array_schema) =
to_ffi(&column.to_data()).map_err(export_error)?;
+ let array_schema = cross_schema(array_schema);
+ let array = cross_array(array);
+ // SAFETY: `array` and `array_schema` were produced together by
`to_ffi`
+ // from one valid array, and `cross_*` only relabeled their types.
+ let data = unsafe { from_ffi(array, &array_schema)? };
+ columns.push(make_array(data));
+ }
+ let options =
RecordBatchOptions::new().with_row_count(Some(batch.num_rows()));
+ Ok(RecordBatch::try_new_with_options(
+ schema, columns, &options,
+ )?)
+}
+
+/// Reads a C `ArrowSchema` struct filled by Lance's arrow as the workspace
+/// arrow's struct of the same name.
+fn cross_schema(schema: LanceFfiSchema) -> FFI_ArrowSchema {
+ // SAFETY: both types are `#[repr(C)]` definitions of the `ArrowSchema`
+ // struct from the Arrow C data interface specification, whose layout is
+ // fixed, so the bytes of one are a valid value of the other. `transmute`
+ // moves `schema` without running its `Drop`, so the `release` callback is
+ // owned, and later invoked exactly once, by the returned struct.
+ unsafe { transmute::<LanceFfiSchema, FFI_ArrowSchema>(schema) }
+}
+
+/// Reads a C `ArrowArray` struct filled by Lance's arrow as the workspace
+/// arrow's struct of the same name.
+fn cross_array(array: LanceFfiArray) -> FFI_ArrowArray {
+ // SAFETY: as in `cross_schema`, for the `ArrowArray` struct.
+ unsafe { transmute::<LanceFfiArray, FFI_ArrowArray>(array) }
+}
+
+/// Lance's arrow has its own `ArrowError` type; carry its message into ours.
+fn export_error(e: arrow_schema_lance::ArrowError) -> ArrowError {
+ ArrowError::CDataInterface(format!("Failed to export a Lance array: {e}"))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ use arrow::array::{Array, AsArray};
+ use arrow::datatypes::{DataType, Field, Int32Type, TimeUnit};
+ use arrow_array_lance::builder::{ListBuilder, StringBuilder,
StructBuilder};
+ use arrow_array_lance::types::Int32Type as LanceInt32Type;
+ use arrow_array_lance::{
+ ArrayRef as LanceArrayRef, DictionaryArray as LanceDictionaryArray,
+ Int32Array as LanceInt32Array, StringArray as LanceStringArray,
+ TimestampMicrosecondArray as LanceTimestampMicrosecondArray,
+ };
+ use arrow_schema_lance::{
+ DataType as LanceDataType, Field as LanceField, Fields as LanceFields,
+ TimeUnit as LanceTimeUnit,
+ };
+
+ fn to_host(batch: LanceRecordBatch) -> Result<RecordBatch> {
+ let schema = Arc::new(schema_to_host(batch.schema().as_ref())?);
+ batch_to_host(batch, schema)
+ }
+
+ /// A batch with the type shapes that take different paths through the C
+ /// data interface: a primitive with nulls, a variable-width string, a
+ /// timestamp with timezone, a dictionary, a list of structs, plus field
+ /// and schema metadata.
+ fn lance_batch() -> LanceRecordBatch {
+ let id = LanceInt32Array::from(vec![Some(1), None, Some(3)]);
+ let name = LanceStringArray::from(vec![Some("a"), Some("bb"), None]);
+ let ts = LanceTimestampMicrosecondArray::from(vec![Some(1_000_000),
Some(2_000_000), None])
+ .with_timezone("UTC");
+ let keys = LanceInt32Array::from(vec![Some(0), Some(1), Some(0)]);
+ let values = LanceStringArray::from(vec!["x", "y"]);
+ let dict = LanceDictionaryArray::<LanceInt32Type>::try_new(keys,
Arc::new(values)).unwrap();
+
+ let item_fields = LanceFields::from(vec![
+ LanceField::new("k", LanceDataType::Utf8, true),
+ LanceField::new("v", LanceDataType::Int32, true),
+ ]);
+ let mut items =
ListBuilder::new(StructBuilder::from_fields(item_fields.clone(), 4));
+ for row in 0..3 {
+ for i in 0..row {
+ let s = items.values();
+ s.field_builder::<StringBuilder>(0)
+ .unwrap()
+ .append_value(format!("k{i}"));
+ s.field_builder::<arrow_array_lance::builder::Int32Builder>(1)
+ .unwrap()
+ .append_value(i);
+ s.append(true);
+ }
+ items.append(row != 1);
+ }
+ let items = items.finish();
+
+ let mut field_meta = HashMap::new();
+ field_meta.insert("ARROW:extension:name".to_string(),
"hudi.test".to_string());
+ let schema = Arc::new(LanceSchema::new_with_metadata(
+ vec![
+ LanceField::new("id", LanceDataType::Int32,
true).with_metadata(field_meta),
+ LanceField::new("name", LanceDataType::Utf8, true),
+ LanceField::new(
+ "ts",
+ LanceDataType::Timestamp(LanceTimeUnit::Microsecond,
Some("UTC".into())),
+ true,
+ ),
+ LanceField::new(
+ "dict",
+ LanceDataType::Dictionary(
+ Box::new(LanceDataType::Int32),
+ Box::new(LanceDataType::Utf8),
+ ),
+ true,
+ ),
+ LanceField::new(
+ "items",
+ LanceDataType::List(Arc::new(LanceField::new(
+ "item",
+ LanceDataType::Struct(item_fields),
+ true,
+ ))),
+ true,
+ ),
+ ],
+ HashMap::from([("table".to_string(), "t1".to_string())]),
+ ));
+ let columns: Vec<LanceArrayRef> = vec![
+ Arc::new(id),
+ Arc::new(name),
+ Arc::new(ts),
+ Arc::new(dict),
+ Arc::new(items),
+ ];
+ LanceRecordBatch::try_new(schema, columns).unwrap()
+ }
+
+ #[test]
+ fn test_schema_to_host_preserves_fields_types_and_metadata() {
+ let batch = lance_batch();
+ let schema = schema_to_host(batch.schema().as_ref()).unwrap();
+
+ assert_eq!(
+ schema.metadata().get("table").map(String::as_str),
+ Some("t1")
+ );
+ let names: Vec<&str> = schema.fields().iter().map(|f|
f.name().as_str()).collect();
+ assert_eq!(names, ["id", "name", "ts", "dict", "items"]);
+
+ let id = schema.field(0);
+ assert_eq!(id.data_type(), &DataType::Int32);
+ assert!(id.is_nullable());
+ assert_eq!(
+ id.metadata()
+ .get("ARROW:extension:name")
+ .map(String::as_str),
+ Some("hudi.test")
+ );
+ assert_eq!(
+ schema.field(2).data_type(),
+ &DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()))
+ );
+ assert_eq!(
+ schema.field(3).data_type(),
+ &DataType::Dictionary(Box::new(DataType::Int32),
Box::new(DataType::Utf8))
+ );
+ let DataType::List(item) = schema.field(4).data_type() else {
+ panic!("items should be a list");
+ };
+ let DataType::Struct(item_fields) = item.data_type() else {
+ panic!("list item should be a struct");
+ };
+ assert_eq!(item_fields.len(), 2);
+ assert_eq!(item_fields[0].name(), "k");
+ assert_eq!(item_fields[1].data_type(), &DataType::Int32);
+ }
+
+ #[test]
+ fn test_batch_to_host_preserves_values_and_nulls() {
+ let host = to_host(lance_batch()).unwrap();
+ assert_eq!(host.num_rows(), 3);
+ assert_eq!(host.num_columns(), 5);
+ assert_eq!(host.schema().field(0).name(), "id");
+
+ let id = host.column(0).as_primitive::<Int32Type>();
+ assert_eq!(id.values(), &[1, 0, 3][..]);
+ assert!(id.is_null(1));
+
+ let name = host.column(1).as_string::<i32>();
+ assert_eq!(name.value(0), "a");
+ assert_eq!(name.value(1), "bb");
+ assert!(name.is_null(2));
+
+ let ts = host
+ .column(2)
+ .as_primitive::<arrow::datatypes::TimestampMicrosecondType>();
+ assert_eq!(ts.value(1), 2_000_000);
+ assert!(ts.is_null(2));
+
+ let dict = host.column(3).as_dictionary::<Int32Type>();
+ let dict_values = dict.values().as_string::<i32>();
+ assert_eq!(dict.keys().values(), &[0, 1, 0][..]);
+ assert_eq!(dict_values.value(1), "y");
+
+ let items = host.column(4).as_list::<i32>();
+ assert_eq!(items.value_length(0), 0);
+ assert!(items.is_null(1));
+ assert_eq!(items.value_length(2), 2);
+ // Row 2 holds the items (k0, 0) and (k1, 1).
+ let last = items.value(2);
+ let structs = last.as_struct();
+ assert_eq!(structs.column(0).as_string::<i32>().value(1), "k1");
+ assert_eq!(structs.column(1).as_primitive::<Int32Type>().value(1), 1);
+ }
+
+ #[test]
+ fn test_batch_to_host_carries_the_given_schema() {
+ let batch = lance_batch();
+ let schema =
Arc::new(schema_to_host(batch.schema().as_ref()).unwrap());
+ let host = batch_to_host(batch, schema.clone()).unwrap();
+ assert!(
+ Arc::ptr_eq(&host.schema(), &schema),
+ "the batch must share the stream's schema rather than cross its
own"
+ );
+ }
+
+ #[test]
+ fn test_batch_to_host_shares_column_buffers() {
+ let batch = lance_batch();
+ let source_ptr = batch.column(0).to_data().buffers()[0].as_ptr();
+ let host = to_host(batch).unwrap();
+ let host_ptr = host.column(0).to_data().buffers()[0].as_ptr();
+ assert_eq!(
+ source_ptr, host_ptr,
+ "the crossing must not copy column data"
+ );
+ }
+
+ #[test]
+ fn test_batch_to_host_keeps_buffers_alive_after_source_drops() {
+ let host = to_host(lance_batch()).unwrap();
+ // The source batch is gone; the imported arrays hold its buffers
+ // through the exporter's `release` callback.
+ let name = host.column(1).as_string::<i32>();
+ assert_eq!(name.value(1), "bb");
+ drop(host);
+ }
+
+ #[test]
+ fn test_batch_to_host_zero_rows_and_zero_columns() {
+ let one_field = Arc::new(LanceSchema::new(vec![LanceField::new(
+ "id",
+ LanceDataType::Int32,
+ true,
+ )]));
+ let no_rows = LanceRecordBatch::try_new(
+ one_field,
+ vec![Arc::new(LanceInt32Array::from(Vec::<i32>::new()))],
+ )
+ .unwrap();
+ let host = to_host(no_rows).unwrap();
+ assert_eq!(host.num_rows(), 0);
+ assert_eq!(host.num_columns(), 1);
+
+ let no_columns = LanceRecordBatch::try_new_with_options(
+ Arc::new(LanceSchema::empty()),
+ vec![],
+
&arrow_array_lance::RecordBatchOptions::new().with_row_count(Some(7)),
+ )
+ .unwrap();
+ let host = to_host(no_columns).unwrap();
+ assert_eq!(host.num_rows(), 7);
+ assert_eq!(host.num_columns(), 0);
+ }
+
+ #[test]
+ fn test_batch_to_host_handles_sliced_wide_and_fixed_width_types() {
+ use arrow_array_lance::{
+ Decimal128Array as LanceDecimal128Array, FixedSizeListArray as
LanceFixedSizeListArray,
+ LargeStringArray as LanceLargeStringArray, StructArray as
LanceStructArray,
+ };
+
+ // A slice carries a non-zero offset through the C interface.
+ let sliced = LanceInt32Array::from(vec![Some(1), None, Some(3),
Some(4)]).slice(1, 3);
+ let large = LanceLargeStringArray::from(vec![Some("x"), None,
Some("zz")]);
+ let decimal = LanceDecimal128Array::from(vec![Some(12345_i128), None,
Some(-1)])
+ .with_precision_and_scale(10, 2)
+ .unwrap();
+ let fixed = LanceFixedSizeListArray::try_new(
+ Arc::new(LanceField::new("item", LanceDataType::Int32, true)),
+ 2,
+ Arc::new(LanceInt32Array::from(vec![1, 2, 3, 4, 5, 6])),
+ None,
+ )
+ .unwrap();
+ let child_fields =
+ LanceFields::from(vec![LanceField::new("n", LanceDataType::Int32,
false)]);
+ let strukt = LanceStructArray::try_new(
+ child_fields.clone(),
+ vec![Arc::new(LanceInt32Array::from(vec![7, 8, 9])) as
LanceArrayRef],
+ None,
+ )
+ .unwrap();
+ let schema = Arc::new(LanceSchema::new(vec![
+ LanceField::new("sliced", LanceDataType::Int32, true),
+ LanceField::new("large", LanceDataType::LargeUtf8, true),
+ LanceField::new("decimal", LanceDataType::Decimal128(10, 2), true),
+ LanceField::new(
+ "fixed",
+ LanceDataType::FixedSizeList(
+ Arc::new(LanceField::new("item", LanceDataType::Int32,
true)),
+ 2,
+ ),
+ true,
+ ),
+ LanceField::new("strukt", LanceDataType::Struct(child_fields),
true),
+ ]));
+ let batch = LanceRecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(sliced),
+ Arc::new(large),
+ Arc::new(decimal),
+ Arc::new(fixed),
+ Arc::new(strukt),
+ ],
+ )
+ .unwrap();
+
+ let host = to_host(batch).unwrap();
+ assert_eq!(host.num_rows(), 3);
+
+ let sliced = host.column(0).as_primitive::<Int32Type>();
+ assert!(sliced.is_null(0));
+ assert_eq!(sliced.value(1), 3);
+ assert_eq!(sliced.value(2), 4);
+
+ let large = host.column(1).as_string::<i64>();
+ assert_eq!(large.value(2), "zz");
+ assert!(large.is_null(1));
+
+ let decimal = host
+ .column(2)
+ .as_primitive::<arrow::datatypes::Decimal128Type>();
+ assert_eq!(decimal.value(0), 12345);
+ assert_eq!(
+ host.schema().field(2).data_type(),
+ &DataType::Decimal128(10, 2)
+ );
+
+ let fixed = host.column(3).as_fixed_size_list();
+ assert_eq!(fixed.value_length(), 2);
+ assert_eq!(
+ fixed.value(2).as_primitive::<Int32Type>().values(),
+ &[5, 6][..]
+ );
+
+ let strukt = host.column(4).as_struct();
+ assert_eq!(strukt.column(0).as_primitive::<Int32Type>().value(2), 9);
+ let host_schema = host.schema();
+ let DataType::Struct(fields) = host_schema.field(4).data_type() else {
+ panic!("strukt should be a struct");
+ };
+ assert!(!fields[0].is_nullable(), "child nullability must survive");
+ }
+
+ #[test]
+ fn test_export_error_names_the_lance_side() {
+ let err = export_error(arrow_schema_lance::ArrowError::CDataInterface(
+ "bad".to_string(),
+ ));
+ let msg = err.to_string();
+ assert!(msg.contains("Failed to export a Lance array"), "{msg}");
+ assert!(msg.contains("bad"), "{msg}");
+ }
+
+ #[test]
+ fn test_host_field_type_round_trips_through_the_workspace_arrow() {
+ // A field built in the workspace arrow equals the crossed field, so
the
+ // two arrows agree on the type encoding the C interface carries.
+ let host = to_host(lance_batch()).unwrap();
+ let expected = Field::new(
+ "ts",
+ DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
+ true,
+ );
+ assert_eq!(host.schema().field(2), &expected);
+ }
+}
diff --git a/crates/core/src/file_group/base_file/mod.rs
b/crates/core/src/file_group/base_file/mod.rs
index 569d167b..32a8e3fd 100644
--- a/crates/core/src/file_group/base_file/mod.rs
+++ b/crates/core/src/file_group/base_file/mod.rs
@@ -18,6 +18,7 @@
*/
pub mod hfile;
pub mod lance;
+mod lance_arrow;
pub mod parquet;
pub mod reader;
diff --git a/crates/core/src/file_group/base_file/parquet.rs
b/crates/core/src/file_group/base_file/parquet.rs
index d6fe3ca4..e47ae066 100644
--- a/crates/core/src/file_group/base_file/parquet.rs
+++ b/crates/core/src/file_group/base_file/parquet.rs
@@ -25,6 +25,12 @@ use futures::future::BoxFuture;
use object_store::ObjectStoreExt;
use object_store::path::Path as ObjPath;
use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions};
+// `ParquetObjectReader` is deprecated as of parquet 59 in favour of
hand-rolling
+// an `AsyncFileReader` over `object_store`. Replacing it is its own change
+// (apache/hudi-rs#843): it has to carry the counting wrapper and the metadata
+// normalization below, so the deprecated reader is kept, allowed at each use
+// site rather than for the whole file.
+#[allow(deprecated)]
use parquet::arrow::async_reader::{AsyncFileReader, ParquetObjectReader};
use parquet::arrow::{ParquetRecordBatchStreamBuilder, RowNumber,
parquet_to_arrow_schema};
use parquet::file::metadata::ParquetMetaData;
@@ -89,6 +95,7 @@ impl<R: AsyncFileReader> AsyncFileReader for
CountingReader<R> {
/// The builder every read here is driven from: a parquet object reader with
the
/// read-volume counters wrapped around it.
+#[allow(deprecated)]
type CountedBuilder =
ParquetRecordBatchStreamBuilder<CountingReader<ParquetObjectReader>>;
/// Parquet implementation of [`BaseFileReader`].
@@ -134,6 +141,7 @@ impl ParquetBaseFileReader {
file_size: u64,
row_index_column: Option<&str>,
) -> Result<CountedBuilder> {
+ #[allow(deprecated)]
let mut reader =
ParquetObjectReader::new(self.storage.object_store.clone(), obj_path)
.with_file_size(file_size);
diff --git a/crates/core/src/file_group/reader_v2/row_serde.rs
b/crates/core/src/file_group/reader_v2/row_serde.rs
index ba9c314a..3b016790 100644
--- a/crates/core/src/file_group/reader_v2/row_serde.rs
+++ b/crates/core/src/file_group/reader_v2/row_serde.rs
@@ -80,7 +80,7 @@ pub fn to_binary_row_body(record: &RecordBatch) ->
Option<Vec<u8>> {
let generator = arrow_ipc::writer::IpcDataGenerator {};
let mut dict_tracker = arrow_ipc::writer::DictionaryTracker::new(false);
let opts = arrow_ipc::writer::IpcWriteOptions::default();
- let mut compression_context =
arrow_ipc::writer::CompressionContext::default();
+ let mut compression_context =
arrow_ipc::writer::IpcWriteContext::default();
let (dictionaries, encoded) = generator
.encode(record, &mut dict_tracker, &opts, &mut compression_context)
.ok()?;
diff --git a/crates/datafusion/src/hudi_exec.rs
b/crates/datafusion/src/hudi_exec.rs
index 73606154..d3a4ea48 100644
--- a/crates/datafusion/src/hudi_exec.rs
+++ b/crates/datafusion/src/hudi_exec.rs
@@ -36,7 +36,9 @@ use datafusion::physical_plan::{
};
use datafusion_common::DataFusionError::Execution;
use datafusion_common::stats::Precision;
+use datafusion_common::tree_node::TreeNodeRecursion;
use datafusion_common::{ColumnStatistics, DataFusionError, Result, Statistics};
+use datafusion_physical_expr::PhysicalExpr;
use futures::stream::{self, BoxStream, TryStreamExt};
use futures::{Stream, StreamExt};
@@ -269,6 +271,13 @@ impl ExecutionPlan for HudiScanExec {
vec![]
}
+ fn apply_expressions(
+ &self,
+ _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) ->
Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
fn with_new_children(
self: Arc<Self>,
children: Vec<Arc<dyn ExecutionPlan>>,
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/lib.rs
index 2e70738c..2900fbd9 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/lib.rs
@@ -45,7 +45,6 @@ use datafusion_common::stats::Precision;
use datafusion_common::{DataFusionError, Statistics};
use datafusion_expr::utils::split_conjunction;
use datafusion_expr::{CreateExternalTable, Expr, TableProviderFilterPushDown,
TableType};
-use datafusion_physical_expr::create_physical_expr;
use log::warn;
use crate::hudi_exec::HudiScanExec;
@@ -121,6 +120,7 @@ fn filter_field_matches_partition_column(filter_field:
&str, partition_column: &
#[derive(Clone)]
pub struct HudiDataSource {
table: Arc<HudiTable>,
+ options: HashMap<String, String>,
/// Cached table schema (with meta fields) for synchronous access in
`TableProvider::schema()`.
/// This provider is a construction-time metadata snapshot; create a new
/// provider to observe schema/stat changes from later commits.
@@ -231,6 +231,7 @@ impl HudiDataSource {
}
None => default_file_slice_read_concurrency(),
};
+ let options = all_options.iter().cloned().collect();
let table = HudiTable::new_with_options(base_uri, all_options)
.await
.map_err(|e| external_error("Failed to create Hudi table", e))?;
@@ -292,6 +293,7 @@ impl HudiDataSource {
Ok(Self {
table: Arc::new(table),
+ options,
schema,
partition_schema,
cached_stats,
@@ -302,6 +304,16 @@ impl HudiDataSource {
})
}
+ /// Returns the table location needed to reconstruct this provider.
+ pub fn base_uri(&self) -> String {
+ self.table.base_url().to_string()
+ }
+
+ /// Returns the caller-supplied options needed to reconstruct this
provider.
+ pub fn options(&self) -> &HashMap<String, String> {
+ &self.options
+ }
+
fn get_input_partitions(&self) -> usize {
self.input_partitions
}
@@ -408,7 +420,7 @@ impl HudiDataSource {
Some(format) => matches!(format, BaseFileFormatValue::Parquet),
None => Self::file_slices_are_parquet(file_slices)?,
};
- if !parquet_base_files {
+ if !parquet_base_files || !Self::file_slices_have_sizes(file_slices) {
return Ok(false);
}
if !self.table.is_mor() {
@@ -419,6 +431,35 @@ impl HudiDataSource {
.map_err(|e| external_error("Invalid read-optimized option", e))
}
+ /// The scan's routing once the file slices are known: a table whose
+ /// persisted format already settled on parquet still needs every base file
+ /// sized, and a table without a persisted format decides from the slices.
+ fn scan_uses_parquet_source(
+ &self,
+ format_decision: Option<bool>,
+ read_options: &ReadOptions,
+ file_slices: &[FileSlice],
+ ) -> Result<bool> {
+ match format_decision {
+ Some(false) => Ok(false),
+ Some(true) => Ok(Self::file_slices_have_sizes(file_slices)),
+ None => self.use_parquet_source(read_options, file_slices),
+ }
+ }
+
+ /// Whether every base file carries its size. The parquet source needs it:
+ /// DataFusion reads the footer at the size it is given, and its file-group
+ /// repartitioner assigns a zero-length file to no partition at all, so a
+ /// slice without one goes through `HudiScanExec`, which looks the size up.
+ fn file_slices_have_sizes(file_slices: &[FileSlice]) -> bool {
+ file_slices.iter().all(|slice| {
+ slice
+ .base_file
+ .as_ref()
+ .is_none_or(|base_file| base_file.file_metadata.is_some())
+ })
+ }
+
/// Check if the given expression can be pushed down to the Hudi table.
///
/// The expression can be pushed down if it is:
@@ -659,7 +700,12 @@ impl HudiDataSource {
.base_file
.as_ref()
.and_then(|b| b.file_metadata.as_ref())
- .map_or(0, |m| m.size);
+ .map(|m| m.size)
+ .ok_or_else(|| {
+ DataFusionError::Internal(format!(
+ "Base file {relative_path} has no known size"
+ ))
+ })?;
let partitioned_file = PartitionedFile::new(url.path(), size);
parquet_file_group_vec.push(partitioned_file);
}
@@ -682,7 +728,7 @@ impl HudiDataSource {
let filter = filters.iter().cloned().reduce(|acc, new| acc.and(new));
if let Some(expr) = filter {
let df_schema = DFSchema::try_from(table_schema.clone())?;
- let predicate = create_physical_expr(&expr, &df_schema,
state.execution_props())?;
+ let predicate = state.create_physical_expr(expr, &df_schema)?;
parquet_source = parquet_source.with_predicate(predicate)
}
@@ -854,16 +900,9 @@ impl TableProvider for HudiDataSource {
self.scan_read_options(all_pushdown_filters, read_optimized)?
};
- match
self.use_parquet_source_without_file_slices(&partition_read_options)? {
- Some(true) => {
- let flat_slices = self
- .table
- .get_file_slices(&partition_read_options)
- .await
- .map_err(|e| external_error("Failed to get file slices
from Hudi table", e))?;
- self.scan_parquet(state, projection, filters, limit,
flat_slices)
- .await
- }
+ let format_decision =
+
self.use_parquet_source_without_file_slices(&partition_read_options)?;
+ match format_decision {
Some(false) => {
let flat_slices = self
.table
@@ -879,14 +918,18 @@ impl TableProvider for HudiDataSource {
)
.await
}
- None => {
+ _ => {
let partition_flat_slices = self
.table
.get_file_slices(&partition_read_options)
.await
.map_err(|e| external_error("Failed to get file slices
from Hudi table", e))?;
- if self.use_parquet_source(&partition_read_options,
&partition_flat_slices)? {
+ if self.scan_uses_parquet_source(
+ format_decision,
+ &partition_read_options,
+ &partition_flat_slices,
+ )? {
self.scan_parquet(state, projection, filters, limit,
partition_flat_slices)
.await
} else {
@@ -1010,7 +1053,15 @@ impl TableProviderFactory for HudiTableFactory {
cmd: &CreateExternalTable,
) -> Result<Arc<dyn TableProvider>> {
let options = HudiTableFactory::resolve_options(state, cmd)?;
- let base_uri = cmd.location.as_str();
+ let base_uri = match cmd.locations.as_slice() {
+ [location] => location.as_str(),
+ locations => {
+ return Err(DataFusionError::Plan(format!(
+ "A Hudi table has exactly one location, got {}",
+ locations.len()
+ )));
+ }
+ };
let table_provider = HudiDataSource::new_with_options(base_uri,
options).await?;
Ok(Arc::new(table_provider))
}
@@ -1029,6 +1080,7 @@ mod tests {
use datafusion::logical_expr::BinaryExpr;
use datafusion::prelude::SessionContext;
+ use hudi_test::QuickstartTripsTable;
use hudi_test::SampleTable::{V6Nonpartitioned, V6SimplekeygenNonhivestyle,
V9LanceTxnsSimple};
use crate::HudiDataSource;
@@ -1081,6 +1133,85 @@ mod tests {
assert!(!HudiDataSource::file_slices_are_parquet(&[]).unwrap());
}
+ #[test]
+ fn test_file_slices_have_sizes_requires_metadata_on_every_base_file() {
+ use hudi_core::file_group::base_file::BaseFile;
+ use hudi_core::storage::file_metadata::FileMetadata;
+ use std::str::FromStr;
+
+ let name =
"0832997e-30d8-4a42-a52d-e2c540816593-0_0-43-78_20261016221427272.parquet";
+ let sized = {
+ let mut base_file = BaseFile::from_str(name).unwrap();
+ base_file.file_metadata = Some(FileMetadata::new(name.to_string(),
4096));
+ FileSlice::new(base_file, "city=sf".to_string())
+ };
+ let without_size = FileSlice::new(BaseFile::from_str(name).unwrap(),
"city=sf".to_string());
+
+ assert!(HudiDataSource::file_slices_have_sizes(&[]));
+ assert!(HudiDataSource::file_slices_have_sizes(
+ std::slice::from_ref(&sized)
+ ));
+ assert!(!HudiDataSource::file_slices_have_sizes(&[
+ sized,
+ without_size
+ ]));
+ }
+
+ #[tokio::test]
+ async fn
test_scan_routing_requires_sizes_even_when_the_table_persists_parquet() {
+ // This fixture persists `hoodie.table.base.file.format=PARQUET`, so a
+ // read-optimized read settles on the parquet source before any slice
is
+ // read.
+ let base_uri =
QuickstartTripsTable::MorDeleteOrdInt.path_to_mor_avro();
+ let hudi = HudiDataSource::new_with_options(base_uri.as_str(),
empty_options())
+ .await
+ .unwrap();
+ let read_options = hudi.scan_read_options(Vec::new(), true).unwrap();
+ let decision = hudi
+ .use_parquet_source_without_file_slices(&read_options)
+ .unwrap();
+ assert_eq!(
+ decision,
+ Some(true),
+ "the persisted format settles on parquet"
+ );
+
+ let mut slices =
hudi.table.get_file_slices(&read_options).await.unwrap();
+ assert!(!slices.is_empty());
+ assert!(
+ hudi.scan_uses_parquet_source(decision, &read_options, &slices)
+ .unwrap()
+ );
+
+ slices[0].base_file.as_mut().unwrap().file_metadata = None;
+ assert!(
+ !hudi
+ .scan_uses_parquet_source(decision, &read_options, &slices)
+ .unwrap(),
+ "a base file without a size must leave the parquet source"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_scan_parquet_rejects_a_base_file_without_size() {
+ let base_uri = V6Nonpartitioned.path_to_cow();
+ let hudi = HudiDataSource::new_with_options(base_uri.as_str(),
empty_options())
+ .await
+ .unwrap();
+ let read_options = hudi.scan_read_options(Vec::new(), false).unwrap();
+ let mut slices =
hudi.table.get_file_slices(&read_options).await.unwrap();
+ assert!(!slices.is_empty());
+ slices[0].base_file.as_mut().unwrap().file_metadata = None;
+
+ let ctx = SessionContext::new();
+ let state = ctx.state();
+ let err = hudi
+ .scan_parquet(&state, None, &[], None, slices)
+ .await
+ .expect_err("a sizeless base file must not reach DataFusion as
zero bytes");
+ assert!(err.to_string().contains("has no known size"), "{err}");
+ }
+
#[tokio::test]
async fn test_new_with_options_rejects_hfile_format_for_regular_scan() {
let result = HudiDataSource::new_with_options(
diff --git a/crates/datafusion/tests/plan_tests.rs
b/crates/datafusion/tests/plan_tests.rs
index 24b0c1af..e9fd0934 100644
--- a/crates/datafusion/tests/plan_tests.rs
+++ b/crates/datafusion/tests/plan_tests.rs
@@ -42,10 +42,20 @@ use hudi_test::{SampleTable, assert_arrow_field_names_eq};
// ============================================================================
async fn create_test_session() -> SessionContext {
- let config = SessionConfig::new().set(
- "datafusion.sql_parser.enable_ident_normalization",
- &ScalarValue::from(false),
- );
+ let config = SessionConfig::new()
+ .set(
+ "datafusion.sql_parser.enable_ident_normalization",
+ &ScalarValue::from(false),
+ )
+ // The plan assertions below check how the provider buckets file slices
+ // under `hoodie.read.input.partitions`. DataFusion may then split
those
+ // groups further for parallelism once they pass
+ // `repartition_file_min_size` (1 MiB since DataFusion 55), which the
+ // fixtures do, so that pass is turned off to keep the groups visible.
+ .set(
+ "datafusion.optimizer.repartition_file_scans",
+ &ScalarValue::from(false),
+ );
let table_factory: Arc<dyn TableProviderFactory> =
Arc::new(HudiTableFactory::default());
let session_state = SessionStateBuilder::new()
@@ -828,3 +838,24 @@ mod lance_tests {
assert_eq!(txn_types, ["reversal", "transfer", "debit", "debit"]);
}
}
+
+mod factory_tests {
+ use super::*;
+
+ #[tokio::test]
+ async fn test_create_external_table_rejects_multiple_locations() {
+ let ctx = create_test_session().await;
+ let path = SampleTable::V6Nonpartitioned.path_to_cow();
+ let sql = format!(
+ "CREATE EXTERNAL TABLE two_locations STORED AS HUDI LOCATION
('{path}', '{path}')"
+ );
+ let err = ctx
+ .sql(&sql)
+ .await
+ .expect_err("two locations must be rejected");
+ assert!(
+ err.to_string().contains("exactly one location, got 2"),
+ "unexpected error: {err}"
+ );
+ }
+}
diff --git a/demo/apps/datafusion/Cargo.toml b/demo/apps/datafusion/Cargo.toml
index bb9ea81b..aaaec3dd 100644
--- a/demo/apps/datafusion/Cargo.toml
+++ b/demo/apps/datafusion/Cargo.toml
@@ -25,5 +25,5 @@ edition = "2021"
[dependencies]
tokio = "1"
-datafusion = "54"
+datafusion = "55"
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 33df90fa..7c5b60c1 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 = "58" }
+arrow = { version = "59" }
hudi = { path = "../../../../crates/hudi" }
diff --git a/python/Cargo.toml b/python/Cargo.toml
index 09c6c0b9..8fbdb3e9 100644
--- a/python/Cargo.toml
+++ b/python/Cargo.toml
@@ -50,7 +50,7 @@ futures = { workspace = true }
tokio = { workspace = true }
[dependencies.pyo3]
-version = "0.28"
+version = "0.29"
features = ["extension-module", "abi3", "abi3-py310"]
[features]
diff --git a/python/pyproject.toml b/python/pyproject.toml
index 787530c3..77dbd399 100644
--- a/python/pyproject.toml
+++ b/python/pyproject.toml
@@ -56,7 +56,7 @@ lint = [
datafusion = [
# The FFI contract is per DataFusion major: keep the Python package on the
# same major as the Rust workspace's datafusion dependency.
- "datafusion>=54.0.0,<55",
+ "datafusion>=55.0.0,<56",
]
[tool.maturin]
diff --git a/python/src/datafusion_internal.rs
b/python/src/datafusion_internal.rs
index 96f46cdd..6c1d7006 100644
--- a/python/src/datafusion_internal.rs
+++ b/python/src/datafusion_internal.rs
@@ -91,6 +91,6 @@ impl HudiDataFusionDataSource {
let provider = Arc::new(self.table.clone());
let codec = extract_codec(session)?;
let ffi_provider = FFI_TableProvider::new_with_ffi_codec(provider,
false, None, codec);
- PyCapsule::new(py, ffi_provider,
Some(cr"datafusion_table_provider".into()))
+ PyCapsule::new_with_value(py, ffi_provider,
c"datafusion_table_provider")
}
}
