This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 3f21239 perf(scan): stop split planning after limit coverage (#470)
3f21239 is described below
commit 3f21239eca56cfc2a7a44c0de52a22ef8938fdca
Author: Huang Qiwei <[email protected]>
AuthorDate: Tue Jul 7 20:46:57 2026 +0800
perf(scan): stop split planning after limit coverage (#470)
Push safe LIMIT planning into split construction so limited scans can stop
building candidate splits once known row counts cover the requested limit. Keep
manifest planning complete, preserve predicate and row-range safety gates, and
expose early-stop state in scan traces.
---
crates/integration_tests/tests/read_tables.rs | 231 +++++++++++++++++-
.../datafusion/tests/scan_pruning_trace.rs | 14 +-
crates/paimon/src/table/scan_trace.rs | 25 +-
crates/paimon/src/table/table_scan.rs | 266 ++++++++++++++++++---
4 files changed, 495 insertions(+), 41 deletions(-)
diff --git a/crates/integration_tests/tests/read_tables.rs
b/crates/integration_tests/tests/read_tables.rs
index c4b7677..11a390f 100644
--- a/crates/integration_tests/tests/read_tables.rs
+++ b/crates/integration_tests/tests/read_tables.rs
@@ -21,13 +21,15 @@ use arrow_array::{
Array, ArrowPrimitiveType, Int32Array, Int64Array, ListArray, MapArray,
PrimitiveArray,
RecordBatch, StringArray, StructArray,
};
+use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
use futures::TryStreamExt;
use paimon::api::ConfigResponse;
use paimon::catalog::{Identifier, RESTCatalog};
use paimon::common::Options;
use paimon::spec::{DataType, IntType, Predicate, Schema, VarCharType};
-use paimon::{Catalog, CatalogOptions, Error, FileSystemCatalog, Plan};
+use paimon::{Catalog, CatalogOptions, Error, FileSystemCatalog, Plan,
ScanTrace};
use std::collections::{HashMap, HashSet};
+use std::sync::{Arc, OnceLock};
#[path = "../../paimon/tests/mock_server.rs"]
mod mock_server;
@@ -74,10 +76,158 @@ async fn get_table_from_catalog<C: Catalog + ?Sized>(
table_name: &str,
) -> paimon::Table {
let identifier = Identifier::new("default", table_name);
+ match catalog.get_table(&identifier).await {
+ Ok(table) => table,
+ Err(Error::TableNotExist { .. }) if table_name ==
"data_evolution_table" => {
+ ensure_data_evolution_table(catalog).await;
+ catalog
+ .get_table(&identifier)
+ .await
+ .expect("Failed to get generated data_evolution_table")
+ }
+ Err(err) => panic!("Failed to get table: {err:?}"),
+ }
+}
+
+const DATA_EVOLUTION_FIXTURE_MARKER: &str = "paimon-rust.integration-fixture";
+const DATA_EVOLUTION_FIXTURE_VERSION: &str = "data-evolution-limit-v2";
+
+static DATA_EVOLUTION_FIXTURE_LOCK: OnceLock<tokio::sync::Mutex<()>> =
OnceLock::new();
+
+async fn ensure_data_evolution_table<C: Catalog + ?Sized>(catalog: &C) {
+ let lock = DATA_EVOLUTION_FIXTURE_LOCK.get_or_init(||
tokio::sync::Mutex::new(()));
+ let _guard = lock.lock().await;
+
+ let identifier = Identifier::new("default", "data_evolution_table");
+ if let Ok(table) = catalog.get_table(&identifier).await {
+ let options = table.schema().options();
+ let fixture_marker = options
+ .get(DATA_EVOLUTION_FIXTURE_MARKER)
+ .map(String::as_str);
+ if fixture_marker == Some(DATA_EVOLUTION_FIXTURE_VERSION) ||
fixture_marker.is_none() {
+ return;
+ }
+ catalog
+ .drop_table(&identifier, true)
+ .await
+ .expect("Failed to replace stale generated data_evolution_table
fixture");
+ }
+
catalog
+ .create_database("default", true, HashMap::new())
+ .await
+ .expect("Failed to create default database for data_evolution_table
fixture");
+ catalog
+ .create_table(&identifier, data_evolution_fixture_schema(), true)
+ .await
+ .expect("Failed to create data_evolution_table fixture");
+
+ let table = catalog
.get_table(&identifier)
.await
- .expect("Failed to get table")
+ .expect("Failed to load data_evolution_table fixture");
+
+ append_data_evolution_batch(
+ &table,
+ data_evolution_fixture_batch(
+ vec![1, 2, 3],
+ vec!["alice", "bob", "carol"],
+ vec![100, 200, 300],
+ ),
+ )
+ .await;
+ append_data_evolution_batch(
+ &table,
+ data_evolution_fixture_batch(vec![4, 5], vec!["dave", "eve"],
vec![400, 500]),
+ )
+ .await;
+
+ let wb = table.new_write_builder();
+ let mut update = wb
+ .new_update(vec!["name".to_string()])
+ .expect("Failed to create data-evolution update writer");
+ update
+ .add_matched_batch(data_evolution_update_batch())
+ .expect("Failed to add data-evolution update batch");
+ wb.new_commit()
+ .commit(
+ update
+ .prepare_commit()
+ .await
+ .expect("Failed to prepare data-evolution update commit"),
+ )
+ .await
+ .expect("Failed to commit data-evolution update fixture");
+}
+
+fn data_evolution_fixture_schema() -> Schema {
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "name",
+
DataType::VarChar(VarCharType::new(VarCharType::MAX_LENGTH).unwrap()),
+ )
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("source.split.target-size", "1b")
+ .option("source.split.open-file-cost", "1b")
+ .option(
+ DATA_EVOLUTION_FIXTURE_MARKER,
+ DATA_EVOLUTION_FIXTURE_VERSION,
+ )
+ .build()
+ .unwrap()
+}
+
+fn data_evolution_fixture_batch(ids: Vec<i32>, names: Vec<&str>, values:
Vec<i32>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("name", ArrowDataType::Utf8, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(StringArray::from(names)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap()
+}
+
+fn data_evolution_update_batch() -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("_ROW_ID", ArrowDataType::Int64, false),
+ ArrowField::new("name", ArrowDataType::Utf8, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int64Array::from(vec![0, 2, 3])),
+ Arc::new(StringArray::from(vec!["alice-v2", "carol-v2", "dave"])),
+ ],
+ )
+ .unwrap()
+}
+
+async fn append_data_evolution_batch(table: &paimon::Table, batch:
RecordBatch) {
+ let wb = table.new_write_builder();
+ let mut write = wb.new_write().expect("Failed to create fixture writer");
+ write
+ .write_arrow_batch(&batch)
+ .await
+ .expect("Failed to write fixture batch");
+ wb.new_commit()
+ .commit(
+ write
+ .prepare_commit()
+ .await
+ .expect("Failed to prepare fixture commit"),
+ )
+ .await
+ .expect("Failed to commit fixture batch");
}
fn create_file_system_catalog() -> FileSystemCatalog {
@@ -1081,6 +1231,17 @@ async fn plan_table(table: &paimon::Table, limit:
Option<usize>) -> Plan {
scan.plan().await.expect("Failed to plan scan")
}
+async fn plan_table_with_trace(table: &paimon::Table, limit: Option<usize>) ->
(Plan, ScanTrace) {
+ let mut read_builder = table.new_read_builder();
+ if let Some(limit) = limit {
+ read_builder.with_limit(limit);
+ }
+ let scan = read_builder.new_scan();
+ scan.plan_with_trace()
+ .await
+ .expect("Failed to plan scan with trace")
+}
+
/// Test limit pushdown: when limit is smaller than total rows, fewer data
files may be generated.
#[tokio::test]
async fn test_limit_pushdown() {
@@ -1090,11 +1251,11 @@ async fn test_limit_pushdown() {
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
// Get full plan without limit
- let full_plan = plan_table(&table, None).await;
+ let (full_plan, full_trace) = plan_table_with_trace(&table, None).await;
let full_data_split_count: usize = full_plan.splits().iter().count();
// Get the plan with limit = 2
- let limited_plan = plan_table(&table, Some(2)).await;
+ let (limited_plan, limited_trace) = plan_table_with_trace(&table,
Some(2)).await;
let limited_data_split_count: usize = limited_plan.splits().iter().count();
// For data evolution tables, limit pushdown at split level uses
merged_row_count
@@ -1103,20 +1264,35 @@ async fn test_limit_pushdown() {
limited_data_split_count < full_data_split_count,
"Limit pushdown should reduce data split count for data evolution
table: limited={limited_data_split_count}, full={full_data_split_count}"
);
+ assert!(
+ limited_trace.limit_early_stopped,
+ "Limit pushdown should stop split construction early:
{limited_trace:?}"
+ );
+ assert!(
+ limited_trace.split_candidates_built <
full_trace.split_candidates_built,
+ "Limit pushdown should build fewer split candidates:
limited={limited_trace:?}, full={full_trace:?}"
+ );
- // Verify data evolution splits have merged_row_count
+ // Verify data evolution splits have merged_row_count. Overlapping row-id
+ // groups reduce the physical row count, while non-overlapping groups can
+ // have equal physical and merged counts.
+ let mut saw_overlapping_group = false;
for split in full_plan.splits() {
let merged_count = split.merged_row_count().expect(
"Data evolution table should have merged_row_count (all files
should have first_row_id)",
);
- // merged_row_count should be < row_count (overlapping ranges reduce
count)
assert!(
- merged_count < split.row_count(),
- "merged_row_count ({}) should be < row_count ({})",
+ merged_count <= split.row_count(),
+ "merged_row_count ({}) should be <= row_count ({})",
merged_count,
split.row_count()
);
+ saw_overlapping_group |= merged_count < split.row_count();
}
+ assert!(
+ saw_overlapping_group,
+ "Expected at least one overlapping data-evolution split with
merged_row_count < row_count"
+ );
}
// ---------------------------------------------------------------------------
@@ -3160,6 +3336,45 @@ async fn
test_read_data_evolution_table_with_full_row_ranges() {
assert_eq!(filtered_rows, full_rows);
}
+#[tokio::test]
+async fn test_limit_pushdown_disabled_with_row_ranges() {
+ use paimon::RowRange;
+
+ let catalog = create_file_system_catalog();
+ let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
+
+ let mut row_range_reader = table.new_read_builder();
+ row_range_reader.with_row_ranges(vec![RowRange::new(0, i64::MAX)]);
+ let (row_range_plan, row_range_trace) = row_range_reader
+ .new_scan()
+ .plan_with_trace()
+ .await
+ .expect("Failed to plan row-range scan");
+
+ let mut limited_reader = table.new_read_builder();
+ limited_reader.with_row_ranges(vec![RowRange::new(0, i64::MAX)]);
+ limited_reader.with_limit(2);
+ let (limited_plan, limited_trace) = limited_reader
+ .new_scan()
+ .plan_with_trace()
+ .await
+ .expect("Failed to plan limited row-range scan");
+
+ assert_eq!(
+ limited_plan.splits().len(),
+ row_range_plan.splits().len(),
+ "Positive limit pushdown should be disabled when effective row_ranges
exist"
+ );
+ assert!(
+ !limited_trace.limit_early_stopped,
+ "Row ranges should disable limit early stop: {limited_trace:?}"
+ );
+ assert_eq!(
+ limited_trace.split_candidates_built,
row_range_trace.split_candidates_built,
+ "Row-range limited scan should build the same split candidates as
row-range scan"
+ );
+}
+
#[tokio::test]
async fn test_read_data_evolution_table_with_row_id_projection() {
let catalog = create_file_system_catalog();
diff --git a/crates/integrations/datafusion/tests/scan_pruning_trace.rs
b/crates/integrations/datafusion/tests/scan_pruning_trace.rs
index c92ee8a..ad5d573 100644
--- a/crates/integrations/datafusion/tests/scan_pruning_trace.rs
+++ b/crates/integrations/datafusion/tests/scan_pruning_trace.rs
@@ -238,12 +238,22 @@ async fn
test_scan_trace_records_bucket_limit_and_time_travel() {
"bucket-key predicate should prune manifest entries by bucket:
{bucket_trace:?}"
);
+ let (_full_plan, full_trace) = table
+ .new_read_builder()
+ .new_scan()
+ .plan_with_trace()
+ .await
+ .unwrap();
let mut limit_reader = table.new_read_builder();
limit_reader.with_limit(1);
let (_limit_plan, limit_trace) =
limit_reader.new_scan().plan_with_trace().await.unwrap();
assert!(
- limit_trace.splits_after_limit < limit_trace.splits_before_limit,
- "LIMIT should reduce planned splits when no data residual exists:
{limit_trace:?}"
+ limit_trace.limit_early_stopped,
+ "LIMIT should stop split construction early when no data residual
exists: {limit_trace:?}"
+ );
+ assert!(
+ limit_trace.split_candidates_built < full_trace.split_candidates_built,
+ "LIMIT should build fewer split candidates: limited={limit_trace:?},
full={full_trace:?}"
);
let snapshot_one_table = table.copy_with_options(HashMap::from([(
diff --git a/crates/paimon/src/table/scan_trace.rs
b/crates/paimon/src/table/scan_trace.rs
index d6e9e46..9dd616a 100644
--- a/crates/paimon/src/table/scan_trace.rs
+++ b/crates/paimon/src/table/scan_trace.rs
@@ -45,6 +45,8 @@ pub struct ScanTrace {
pub data_evolution_groups_before_stats: usize,
pub data_evolution_groups_pruned_by_stats: usize,
pub data_evolution_groups_pruned_by_row_ranges: usize,
+ pub split_candidates_built: usize,
+ pub limit_early_stopped: bool,
pub splits_before_limit: usize,
pub splits_after_limit: usize,
pub final_splits: usize,
@@ -65,6 +67,25 @@ impl ScanTrace {
splits: usize,
files: usize,
) {
+ self.record_final_plan_with_limit(
+ splits_before_limit,
+ splits_before_limit,
+ splits,
+ files,
+ false,
+ );
+ }
+
+ pub(crate) fn record_final_plan_with_limit(
+ &mut self,
+ split_candidates_built: usize,
+ splits_before_limit: usize,
+ splits: usize,
+ files: usize,
+ limit_early_stopped: bool,
+ ) {
+ self.split_candidates_built = split_candidates_built;
+ self.limit_early_stopped = limit_early_stopped;
self.splits_before_limit = splits_before_limit;
self.splits_after_limit = splits;
self.final_splits = splits;
@@ -76,7 +97,7 @@ impl fmt::Display for ScanTrace {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
- "snapshot={:?}, manifests={}/{}, entries_read={},
bucket_pruned={}, partition_pruned={}, data_stats_pruned={},
cross_schema_pruned={}, splits_before_limit={}, splits_after_limit={},
files={}",
+ "snapshot={:?}, manifests={}/{}, entries_read={},
bucket_pruned={}, partition_pruned={}, data_stats_pruned={},
cross_schema_pruned={}, split_candidates_built={}, limit_early_stopped={},
splits_before_limit={}, splits_after_limit={}, files={}",
self.snapshot_id,
self.manifest_files_after_partition_pruning,
self.manifest_files_before_partition_pruning,
@@ -85,6 +106,8 @@ impl fmt::Display for ScanTrace {
self.manifest_entries_pruned_by_partition,
self.manifest_entries_pruned_by_data_stats,
self.manifest_entries_pruned_by_cross_schema_stats,
+ self.split_candidates_built,
+ self.limit_early_stopped,
self.splits_before_limit,
self.splits_after_limit,
self.final_files
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index cab4781..b69097c 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -345,6 +345,66 @@ pub(super) fn can_push_down_limit_hint_for_scan(
data_predicates.is_empty() && row_ranges.is_none()
}
+#[derive(Debug)]
+struct LimitPushdownResult {
+ splits: Vec<DataSplit>,
+ split_candidates_built: usize,
+ limit_early_stopped: bool,
+}
+
+#[derive(Debug)]
+struct LimitPushdownAccumulator {
+ limit: usize,
+ fallback_splits: Vec<DataSplit>,
+ limited_splits: Vec<DataSplit>,
+ scanned_row_count: i64,
+ limit_early_stopped: bool,
+}
+
+impl LimitPushdownAccumulator {
+ fn new(limit: usize) -> Self {
+ Self {
+ limit,
+ fallback_splits: Vec::new(),
+ limited_splits: Vec::new(),
+ scanned_row_count: 0,
+ limit_early_stopped: limit == 0,
+ }
+ }
+
+ fn push(&mut self, split: DataSplit) -> bool {
+ if self.limit_early_stopped {
+ return true;
+ }
+
+ if let Some(merged_count) = split.merged_row_count() {
+ self.fallback_splits.push(split.clone());
+ self.limited_splits.push(split);
+ self.scanned_row_count += merged_count;
+ self.limit_early_stopped = self.scanned_row_count >= self.limit as
i64;
+ } else {
+ self.fallback_splits.push(split);
+ }
+
+ self.limit_early_stopped
+ }
+
+ fn finish(self) -> LimitPushdownResult {
+ let split_candidates_built = self.fallback_splits.len();
+ let splits = if self.limit_early_stopped {
+ self.limited_splits
+ } else {
+ self.fallback_splits
+ };
+
+ LimitPushdownResult {
+ splits,
+ split_candidates_built,
+ limit_early_stopped: self.limit_early_stopped,
+ }
+ }
+}
+
type BucketDataFileGroups = HashMap<(Vec<u8>, i32), (i32, Vec<DataFileMeta>)>;
fn global_index_detail_data_ranges(groups: &BucketDataFileGroups) ->
Vec<RowRange> {
@@ -735,33 +795,21 @@ impl<'a> TableScan<'a> {
/// committed only once the accumulated known row count reaches the limit;
/// if it never does, the original split list is returned unchanged. The
/// caller or query engine must still enforce the final LIMIT.
+ #[cfg(test)]
fn apply_limit_pushdown(&self, splits: Vec<DataSplit>) -> Vec<DataSplit> {
let limit = match self.limit {
Some(l) => l,
None => return splits,
};
- if limit == 0 {
- return Vec::new();
- }
-
- if splits.is_empty() {
- return splits;
- }
-
- let mut limited_splits = Vec::new();
- let mut scanned_row_count: i64 = 0;
+ let mut accumulator = LimitPushdownAccumulator::new(limit);
- for split in &splits {
- if let Some(merged_count) = split.merged_row_count() {
- limited_splits.push(split.clone());
- scanned_row_count += merged_count;
- if scanned_row_count >= limit as i64 {
- return limited_splits;
- }
+ for split in splits {
+ if accumulator.push(split) {
+ break;
}
}
- splits
+ accumulator.finish().splits
}
/// Read all manifest entries from a snapshot, applying filters and
merging.
@@ -944,6 +992,13 @@ impl<'a> TableScan<'a> {
}
}
+ if matches!(self.limit, Some(0)) {
+ if let Some(trace) = trace {
+ trace.record_final_plan_with_limit(0, 0, 0, 0, true);
+ }
+ return Ok(Plan::new(Vec::new()));
+ }
+
// Group by (partition, bucket), decomposing entries to avoid cloning
partition.
let mut groups: BucketDataFileGroups =
HashMap::with_capacity(entries.len());
for e in entries {
@@ -1039,7 +1094,15 @@ impl<'a> TableScan<'a> {
};
let mut data_file_field_ids_cache = DataFileFieldIdsCache::new();
- for ((partition, bucket), (total_buckets, data_files)) in groups {
+ let can_push_down_limit =
self.can_push_down_limit_hint(effective_row_ranges.as_deref());
+ let mut limit_accumulator = match self.limit {
+ Some(limit) if limit > 0 && can_push_down_limit => {
+ Some(LimitPushdownAccumulator::new(limit))
+ }
+ _ => None,
+ };
+
+ 'groups: for ((partition, bucket), (total_buckets, data_files)) in
groups {
let partition_row = BinaryRow::from_serialized_bytes(&partition)?;
let bucket_path = if let Some(ref computer) = partition_computer {
let partition_path =
computer.generate_partition_path(&partition_row)?;
@@ -1220,21 +1283,39 @@ impl<'a> TableScan<'a> {
if let Some(row_ranges) = split_row_ranges {
builder = builder.with_row_ranges(row_ranges);
}
- splits.push(builder.build()?);
+ let split = builder.build()?;
+ if let Some(accumulator) = limit_accumulator.as_mut() {
+ if accumulator.push(split) {
+ break 'groups;
+ }
+ } else {
+ splits.push(split);
+ }
}
}
- // With data predicates or row_ranges, merged_row_count() reflects
pre-filter
- // row counts, so stopping early could return fewer rows than the
limit.
- let splits_before_limit = splits.len();
- let splits = if
self.can_push_down_limit_hint(effective_row_ranges.as_deref()) {
- self.apply_limit_pushdown(splits)
- } else {
- splits
- };
+ let (splits, split_candidates_built, limit_early_stopped) =
+ if let Some(accumulator) = limit_accumulator {
+ let result = accumulator.finish();
+ (
+ result.splits,
+ result.split_candidates_built,
+ result.limit_early_stopped,
+ )
+ } else {
+ let split_candidates_built = splits.len();
+ (splits, split_candidates_built, false)
+ };
+ let splits_before_limit = split_candidates_built;
if let Some(trace) = trace {
let final_files = splits.iter().map(|split|
split.data_files().len()).sum();
- trace.record_final_plan(splits_before_limit, splits.len(),
final_files);
+ trace.record_final_plan_with_limit(
+ split_candidates_built,
+ splits_before_limit,
+ splits.len(),
+ final_files,
+ limit_early_stopped,
+ );
}
Ok(Plan::new(splits))
@@ -1244,7 +1325,8 @@ impl<'a> TableScan<'a> {
#[cfg(test)]
mod tests {
use super::{
- prune_data_evolution_group_by_read_fields,
should_skip_level_zero_for_scan, TableScan,
+ prune_data_evolution_group_by_read_fields,
should_skip_level_zero_for_scan,
+ LimitPushdownAccumulator, TableScan,
};
use crate::catalog::Identifier;
use crate::io::FileIOBuilder;
@@ -1484,6 +1566,13 @@ mod tests {
)
}
+ fn scan_trace_small_split_table(table_path: &str) -> Table {
+ scan_trace_test_table(table_path).copy_with_options(HashMap::from([
+ ("source.split.target-size".to_string(), "1b".to_string()),
+ ("source.split.open-file-cost".to_string(), "1b".to_string()),
+ ]))
+ }
+
async fn setup_scan_trace_dirs(table: &Table) {
table
.file_io()
@@ -1648,6 +1737,50 @@ mod tests {
assert_eq!(split_file_names(&pruned), vec!["a.parquet"]);
}
+ #[test]
+ fn
test_incremental_limit_accumulator_stops_after_known_count_reaches_limit() {
+ let mut accumulator = LimitPushdownAccumulator::new(3);
+
+ assert!(!accumulator.push(limit_test_split("a.parquet", 2)));
+ assert!(
+ !accumulator.push(limit_test_split_with_unknown_merged_row_count(
+ "b.parquet",
+ 4
+ ))
+ );
+ assert!(accumulator.push(limit_test_split("c.parquet", 3)));
+
+ let result = accumulator.finish();
+ assert!(result.limit_early_stopped);
+ assert_eq!(result.split_candidates_built, 3);
+ assert_eq!(
+ split_file_names(&result.splits),
+ vec!["a.parquet", "c.parquet"]
+ );
+ }
+
+ #[test]
+ fn
test_incremental_limit_accumulator_returns_fallback_when_limit_not_reached() {
+ let mut accumulator = LimitPushdownAccumulator::new(100);
+
+ assert!(!accumulator.push(limit_test_split("a.parquet", 2)));
+ assert!(
+ !accumulator.push(limit_test_split_with_unknown_merged_row_count(
+ "b.parquet",
+ 4
+ ))
+ );
+ assert!(!accumulator.push(limit_test_split("c.parquet", 3)));
+
+ let result = accumulator.finish();
+ assert!(!result.limit_early_stopped);
+ assert_eq!(result.split_candidates_built, 3);
+ assert_eq!(
+ split_file_names(&result.splits),
+ vec!["a.parquet", "b.parquet", "c.parquet"]
+ );
+ }
+
#[test]
fn test_first_row_skips_level_zero_by_default() {
assert!(should_skip_level_zero_for_scan(
@@ -2019,6 +2152,79 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn
test_plan_with_trace_records_limit_early_stop_during_split_construction() {
+ let table_path =
+
"memory:/test_plan_with_trace_records_limit_early_stop_during_split_construction";
+ let table = scan_trace_small_split_table(table_path);
+ setup_scan_trace_dirs(&table).await;
+
+ TableCommit::new(table.clone(), "scan-trace-limit-test".to_string())
+ .commit(vec![CommitMessage::new(
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ vec![
+ stats_trace_file("limit-1.parquet", 1, 1),
+ stats_trace_file("limit-2.parquet", 2, 2),
+ stats_trace_file("limit-3.parquet", 3, 3),
+ ],
+ )])
+ .await
+ .unwrap();
+
+ let (_full_plan, full_trace) = table
+ .new_read_builder()
+ .new_scan()
+ .plan_with_trace()
+ .await
+ .unwrap();
+ let mut limited_reader = table.new_read_builder();
+ limited_reader.with_limit(2);
+ let (_limited_plan, limited_trace) =
+ limited_reader.new_scan().plan_with_trace().await.unwrap();
+
+ assert_eq!(
+ full_trace.final_splits, 3,
+ "fixture should build three splits: {full_trace:?}"
+ );
+ assert!(!full_trace.limit_early_stopped);
+ assert_eq!(full_trace.split_candidates_built, full_trace.final_splits);
+ assert!(limited_trace.limit_early_stopped);
+ assert!(
+ limited_trace.split_candidates_built <
full_trace.split_candidates_built,
+ "limited trace should show construction-time stop:
full={full_trace:?}, limited={limited_trace:?}"
+ );
+ assert_eq!(limited_trace.final_splits, 1);
+ }
+
+ #[tokio::test]
+ async fn test_plan_with_trace_zero_limit_records_no_split_candidates() {
+ let table_path =
"memory:/test_plan_with_trace_zero_limit_records_no_split_candidates";
+ let table = scan_trace_small_split_table(table_path);
+ setup_scan_trace_dirs(&table).await;
+
+ TableCommit::new(table.clone(),
"scan-trace-zero-limit-test".to_string())
+ .commit(vec![CommitMessage::new(
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ vec![
+ stats_trace_file("zero-1.parquet", 1, 1),
+ stats_trace_file("zero-2.parquet", 2, 2),
+ ],
+ )])
+ .await
+ .unwrap();
+
+ let mut reader = table.new_read_builder();
+ reader.with_limit(0);
+ let (plan, trace) = reader.new_scan().plan_with_trace().await.unwrap();
+
+ assert!(plan.splits().is_empty());
+ assert!(trace.limit_early_stopped);
+ assert_eq!(trace.split_candidates_built, 0);
+ assert_eq!(trace.final_splits, 0);
+ }
+
#[test]
fn test_data_file_matches_in_prunes_when_all_literals_out_of_range() {
let fields = int_field();
