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 a3196cd7 perf(core): decode table read file slices on runtime worker
threads (#890)
a3196cd7 is described below
commit a3196cd7be938444d9924472b8dedb797012231d
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 00:00:55 2026 -0700
perf(core): decode table read file slices on runtime worker threads (#890)
`Table::read` read its file slices with `try_join_all` on the awaiting
task, so all decoding ran on one thread, which for the Python binding is the
Python thread. Each slice is now read on its own runtime task, in input order
and within the existing scan memory budget; the task is aborted if the read is
dropped, and the slice is read in place when there is no multi-thread runtime.
When `hoodie.read.file.slice.read.concurrency` is unset, a `Table` read now
defaults to the available cor [...]
Full-table reads through the Python binding got 3 to 6 times faster on
local tables of 1M to 10M rows (for example 2.56 s to 0.42 s for a 10M-row
copy-on-write table).
---
README.md | 2 +-
crates/core/src/config/read.rs | 4 +-
crates/core/src/table/mod.rs | 91 +++++++++++++++----
crates/core/src/util/concurrency.rs | 71 ++++++++++++++-
crates/core/tests/table_read_runtime_tests.rs | 120 ++++++++++++++++++++++++++
5 files changed, 268 insertions(+), 20 deletions(-)
diff --git a/README.md b/README.md
index 6e95716e..ce078cd1 100644
--- a/README.md
+++ b/README.md
@@ -465,7 +465,7 @@ silently redirect every later read.
| `hoodie.read.file.group.reader.version` | `2`
| table or read | Which file group reader merges a slice. Version 2 is the
default; a read it cannot serve falls back to version 1. |
| `hoodie.read.use.read_optimized.mode` | `false`
| table or read | Read base files only, skipping the log files, on MOR
tables. |
| `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.file.slice.read.concurrency` | `4`
| table or read | File slices read concurrently per DataFusion partition; a
`Table` read defaults to the available cores. |
| `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`. 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. |
diff --git a/crates/core/src/config/read.rs b/crates/core/src/config/read.rs
index 50cf3a41..bec51459 100644
--- a/crates/core/src/config/read.rs
+++ b/crates/core/src/config/read.rs
@@ -140,7 +140,9 @@ pub enum HudiReadConfig {
/// This controls the batch size when using streaming APIs.
StreamBatchSize,
- /// Maximum number of file-slice streams to poll concurrently within one
scan partition.
+ /// Maximum number of file slices to read concurrently within one scan
partition.
+ /// Defaults to 4 per DataFusion partition; a `Table` read, which is one
+ /// partition covering the whole scan, defaults to the available cores.
FileSliceReadConcurrency,
/// Total memory, in bytes, a whole scan may use for concurrent file-slice
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index 20d2c48a..c96ffb59 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -102,6 +102,7 @@ pub use crate::write::append::AppendResult;
pub use crate::write::{UpsertOptions, WriteResult};
use crate::Result;
+use crate::config::ConfigParser;
use crate::config::HudiConfigs;
use crate::config::internal::HudiInternalConfig;
use crate::config::read::HudiReadConfig;
@@ -1000,8 +1001,12 @@ impl Table {
/// batches at once — and under file group reader version 2 each of those
also
/// carries its own merge map and, on spill, its own RocksDB instance.
///
- /// The same derivation the DataFusion plan uses, so the two paths cannot
- /// bound a scan differently. This is the **only** way to obtain a
+ /// The same budget derivation the DataFusion plan uses, so the two paths
+ /// cannot budget a scan differently. Only the default ceiling differs: a
+ /// `Table` read is the whole scan, with no caller dividing it into
+ /// partitions, so it defaults to the available cores, while the DataFusion
+ /// plan applies the config's own default per partition, where the
partitions
+ /// already spread the scan across cores. This is the **only** way to
obtain a
/// concurrency for a read: the raw
`hoodie.read.file.slice.read.concurrency`
/// lookup lives inside it rather than in a method of its own, so a read
/// cannot reach the ceiling while bypassing the budget. A test can miss
that
@@ -1014,7 +1019,9 @@ impl Table {
/// tested elsewhere in isolation, which says nothing about whether
anything
/// calls it.
fn bounded_read_concurrency(&self, file_slices: &[FileSlice]) -> usize {
+ let cores = std::thread::available_parallelism().map_or(4,
std::num::NonZeroUsize::get);
self.bounded_read_concurrency_from_log_bytes(
+ cores,
file_slices.iter().map(FileSlice::log_size_bytes),
)
}
@@ -1025,8 +1032,18 @@ impl Table {
/// holds `&FileSlice` rather than owning them. The budget must see that
/// subset, not the whole partition: routing a single key to one shard
should
/// admit on that shard's cost.
+ ///
+ /// Keeps the config's own default ceiling: these reads are polled on the
+ /// caller's task, so a wider fan-out would open more readers without
+ /// adding cores.
pub(crate) fn bounded_read_concurrency_for(&self, file_slices:
&[&FileSlice]) -> usize {
-
self.bounded_read_concurrency_from_log_bytes(file_slices.iter().map(|s|
s.log_size_bytes()))
+ let config_default = HudiReadConfig::FileSliceReadConcurrency
+ .default_value()
+ .map_or(1, usize::from);
+ self.bounded_read_concurrency_from_log_bytes(
+ config_default,
+ file_slices.iter().map(|s| s.log_size_bytes()),
+ )
}
/// The one place the ceiling and the budget meet.
@@ -1036,13 +1053,14 @@ impl Table {
/// to own or copy one.
fn bounded_read_concurrency_from_log_bytes(
&self,
+ default_ceiling: usize,
slice_log_bytes: impl Iterator<Item = Option<u64>>,
) -> usize {
- let slice_log_bytes: Vec<Option<u64>> = slice_log_bytes.collect();
- let ceiling: usize = self
+ let ceiling = self
.hudi_configs
- .get_or_default(HudiReadConfig::FileSliceReadConcurrency)
- .into();
+ .get(HudiReadConfig::FileSliceReadConcurrency)
+ .map_or(default_ceiling, usize::from);
+ let slice_log_bytes: Vec<Option<u64>> = slice_log_bytes.collect();
crate::file_group::admission::slices_in_flight(
self.scan_max_memory_size(),
1,
@@ -1054,22 +1072,37 @@ impl Table {
/// Read `file_slices` with at most [`Self::bounded_read_concurrency`] in
/// flight, or fewer when a scan memory budget admits fewer, in slice
order.
///
+ /// Each slice is read on its own runtime task: decoding is CPU-bound, and
+ /// futures polled together on the caller's task share a single thread.
/// Order is preserved (`buffered`, not `buffer_unordered`) because a
caller
/// that concatenates these batches should not see its row order shift with
- /// scheduling. Any single failure aborts the whole read, as `try_join_all`
- /// did — a partially-read table is not a useful answer.
+ /// scheduling. Any single failure aborts the whole read: a partially-read
+ /// table is not a useful answer.
async fn read_file_slices_bounded(
&self,
fg_reader: &FileGroupReader,
file_slices: &[FileSlice],
fg_options: &ReadOptions,
) -> Result<Vec<RecordBatch>> {
- crate::util::concurrency::bounded_in_order(
- file_slices,
- self.bounded_read_concurrency(file_slices),
- |file_slice| fg_reader.read_file_slice(file_slice, fg_options),
- )
- .await
+ use futures::{StreamExt, TryStreamExt};
+ let concurrency = self.bounded_read_concurrency(file_slices).max(1);
+ // Collected rather than mapped lazily: a stream over a borrowing
+ // iterator cannot be proven `Send`, and callers spawn this read.
+ let reads: Vec<_> = file_slices
+ .iter()
+ .map(|file_slice| {
+ let file_slice = file_slice.clone();
+ let fg_reader = fg_reader.clone();
+ let fg_options = fg_options.clone();
+ crate::util::concurrency::spawn_aborting(async move {
+ fg_reader.read_file_slice(&file_slice, &fg_options).await
+ })
+ })
+ .collect();
+ futures::stream::iter(reads)
+ .buffered(concurrency)
+ .try_collect()
+ .await
}
/// Warn when an incremental window reaches below the active timeline.
@@ -1647,7 +1680,8 @@ mod tests {
}
/// A scan memory budget lowers the core fan-out's concurrency; without one
- /// the configured ceiling stands.
+ /// the ceiling stands: the configured value when set, the core count
+ /// otherwise.
///
/// Pins the wiring rather than the arithmetic — `slices_in_flight` has its
/// own tests, and passing those tells you nothing about whether this path
@@ -1677,11 +1711,34 @@ mod tests {
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let unbounded = Table::new(base_url.path()).await.unwrap();
+ let cores = std::thread::available_parallelism().unwrap().get();
assert_eq!(
unbounded.bounded_read_concurrency(&slices),
+ cores,
+ "with no budget and no configured ceiling, the core count stands"
+ );
+ let slice_refs: Vec<&FileSlice> = slices.iter().collect();
+ assert_eq!(
+ unbounded.bounded_read_concurrency_for(&slice_refs),
4,
+ "the metadata read keeps the config's own default"
+ );
+ let configured = Table::new_with_options(
+ base_url.path(),
+ [(HudiReadConfig::FileSliceReadConcurrency.as_ref(), "3")],
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ configured.bounded_read_concurrency(&slices),
+ 3,
"with no budget the configured ceiling stands"
);
+ assert_eq!(
+ configured.bounded_read_concurrency_for(&slice_refs),
+ 3,
+ "a configured ceiling applies to the metadata read too"
+ );
// Each slice estimates at 33 MiB + 4 x 8 MiB = 65 MiB, so 128 MiB
admits one.
let bounded = Table::new_with_options(
@@ -1693,7 +1750,7 @@ mod tests {
assert_eq!(
bounded.bounded_read_concurrency(&slices),
1,
- "a 128 MiB budget must admit one 65 MiB slice, not the ceiling of
4"
+ "a 128 MiB budget must admit one 65 MiB slice, not the ceiling"
);
}
diff --git a/crates/core/src/util/concurrency.rs
b/crates/core/src/util/concurrency.rs
index 56f86856..14acaf80 100644
--- a/crates/core/src/util/concurrency.rs
+++ b/crates/core/src/util/concurrency.rs
@@ -19,8 +19,41 @@
//! flight at once.
use crate::Result;
+use crate::error::CoreError;
use std::future::Future;
+/// Run `fut` as a task on the current multi-thread runtime, so it can make
+/// progress on another worker thread than the one awaiting it. Without one (no
+/// runtime, or a current-thread runtime with no other thread to use) `fut` is
+/// awaited in place.
+///
+/// Dropping the returned future aborts the task, so a caller that gives up on
a
+/// read (an error elsewhere, a dropped stream) does not leave it running. A
+/// panic in a spawned task comes back as an error rather than unwinding the
+/// caller.
+pub(crate) async fn spawn_aborting<R, Fut>(fut: Fut) -> Result<R>
+where
+ R: Send + 'static,
+ Fut: Future<Output = Result<R>> + Send + 'static,
+{
+ struct AbortOnDrop<R>(tokio::task::JoinHandle<R>);
+ impl<R> Drop for AbortOnDrop<R> {
+ fn drop(&mut self) {
+ self.0.abort();
+ }
+ }
+ let handle = match tokio::runtime::Handle::try_current() {
+ Ok(h) if h.runtime_flavor() ==
tokio::runtime::RuntimeFlavor::MultiThread => h,
+ _ => return fut.await,
+ };
+ let mut task = AbortOnDrop(handle.spawn(fut));
+ (&mut task.0).await.unwrap_or_else(|e| {
+ Err(CoreError::ReadFileSliceError(format!(
+ "read task failed: {e}"
+ )))
+ })
+}
+
/// Apply `op` to every item, at most `concurrency` at a time, keeping input
/// order in the output.
///
@@ -62,7 +95,43 @@ where
#[cfg(test)]
mod tests {
use super::*;
- use std::sync::atomic::{AtomicUsize, Ordering};
+ use std::sync::Arc;
+ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+ /// Dropping the wrapper stops the task instead of leaving it running.
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn spawn_aborting_aborts_the_task_on_drop() {
+ let finished = Arc::new(AtomicBool::new(false));
+ let flag = finished.clone();
+ let read = spawn_aborting(async move {
+ tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+ flag.store(true, Ordering::SeqCst);
+ Ok(())
+ });
+ // Start the task, then give up on it.
+ let _ = tokio::time::timeout(std::time::Duration::from_millis(10),
read).await;
+ tokio::time::sleep(std::time::Duration::from_millis(400)).await;
+ assert!(
+ !finished.load(Ordering::SeqCst),
+ "the dropped task kept running"
+ );
+ }
+
+ /// Outside any runtime the future is awaited in place instead of panicking
+ /// on a missing reactor.
+ #[test]
+ fn spawn_aborting_runs_in_place_without_a_runtime() {
+ let got = futures::executor::block_on(spawn_aborting(async { Ok(7)
})).unwrap();
+ assert_eq!(got, 7);
+ }
+
+ /// A panic in the task comes back as an error.
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn spawn_aborting_reports_a_panic_as_an_error() {
+ let result: Result<()> = spawn_aborting(async { panic!("boom")
}).await;
+ let err = result.expect_err("a panicking task must fail the read");
+ assert!(err.to_string().contains("read task failed"), "{err}");
+ }
/// Output order follows input order, not completion order.
#[tokio::test]
diff --git a/crates/core/tests/table_read_runtime_tests.rs
b/crates/core/tests/table_read_runtime_tests.rs
new file mode 100644
index 00000000..784d339f
--- /dev/null
+++ b/crates/core/tests/table_read_runtime_tests.rs
@@ -0,0 +1,120 @@
+/*
+ * 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.
+ */
+//! Where a table read decodes its file slices.
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::cell::Cell;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use arrow_array::{Int64Array, RecordBatch, StringArray};
+use arrow_schema::{DataType, Field, Schema};
+use hudi_core::table::{ReadOptions, Table};
+use tempfile::tempdir;
+
+/// Counts the bytes allocated by threads that set [`COUNT_ALLOCS`].
+struct CountingAllocator;
+
+thread_local! {
+ static COUNT_ALLOCS: Cell<bool> = const { Cell::new(false) };
+}
+
+static MARKED_THREAD_BYTES: AtomicUsize = AtomicUsize::new(0);
+
+unsafe impl GlobalAlloc for CountingAllocator {
+ unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+ // `try_with` because the flag may already be gone while a thread
exits.
+ if COUNT_ALLOCS.try_with(Cell::get).unwrap_or(false) {
+ MARKED_THREAD_BYTES.fetch_add(layout.size(), Ordering::Relaxed);
+ }
+ unsafe { System.alloc(layout) }
+ }
+
+ unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+ unsafe { System.dealloc(ptr, layout) }
+ }
+}
+
+#[global_allocator]
+static GLOBAL: CountingAllocator = CountingAllocator;
+
+const GROUPS: usize = 4;
+const ROWS: usize = 50_000;
+
+fn batch(group: usize) -> RecordBatch {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ let ids: Vec<String> = (0..ROWS).map(|i|
format!("key{group}-{i:08}")).collect();
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(ids)),
+ Arc::new(Int64Array::from_iter_values(
+ (0..ROWS as i64).map(|i| i * 7),
+ )),
+ ],
+ )
+ .unwrap()
+}
+
+/// A caller that blocks on a read, as the Python binding does, must not decode
+/// the file slices on its own thread: that serializes every slice onto one
+/// core. Measured by the bytes the calling thread allocates during the read:
+/// decoding there costs several times the batches read, while planning alone
+/// stays well below them.
+#[test]
+fn test_read_decodes_file_slices_off_the_calling_thread() {
+ let rt = tokio::runtime::Builder::new_multi_thread()
+ .worker_threads(2)
+ .enable_all()
+ .build()
+ .unwrap();
+ let dir = tempdir().unwrap();
+ let table = rt.block_on(async {
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("trips")
+ .with_record_key_fields(["id"])
+ .with_option("hoodie.metadata.enable", "false")
+ // One file group per append, so the read has several slices.
+ .with_option("hoodie.parquet.small.file.limit", "0")
+ .create()
+ .await
+ .unwrap();
+ for group in 0..GROUPS {
+ table.append([batch(group)]).await.unwrap();
+ }
+ Table::new(dir.path().to_str().unwrap()).await.unwrap()
+ });
+
+ COUNT_ALLOCS.with(|c| c.set(true));
+ let batches = rt.block_on(table.read(&ReadOptions::new())).unwrap();
+ COUNT_ALLOCS.with(|c| c.set(false));
+
+ let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
+ assert_eq!(rows, GROUPS * ROWS);
+ let decoded: usize =
batches.iter().map(RecordBatch::get_array_memory_size).sum();
+ let caller = MARKED_THREAD_BYTES.load(Ordering::Relaxed);
+ assert!(
+ caller < decoded,
+ "the calling thread allocated {caller} bytes reading {decoded} bytes
of batches; \
+ the file slices were decoded on it"
+ );
+}
