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 a156316 fix: merge primary-key rows correctly when bucket files span
multiple splits (#374)
a156316 is described below
commit a1563168c35ee893675bb90e7dbdf5fb2f992137
Author: chaoyang <[email protected]>
AuthorDate: Sat Jun 13 11:05:08 2026 +0800
fix: merge primary-key rows correctly when bucket files span multiple
splits (#374)
plan_snapshot bin-packed a bucket's files purely by size, so files whose
primary-key ranges overlap could land in different splits. Each split
runs its own sort-merge reader, so the same key surfaced once per split
instead of merging to a single row (reproducible with three commits of
one key under source.split.target-size=1b).
Port the Java MergeTreeSplitGenerator/IntervalPartition algorithm: sort
files by decoded (min_key, max_key), group transitively overlapping
files into sections, then bin-pack whole sections via pack_for_ordered.
Keys are decoded with the trimmed-PK data types and compared through
datum_cmp — BinaryRow stores fields little-endian, so raw byte
comparison would order int 256 before int 1. Undecodable key ranges
collapse the bucket into one section, trading parallelism for
correctness. Append-only tables keep the file-level bin pack.
---
crates/integrations/datafusion/tests/pk_tables.rs | 242 ++++++++-
crates/paimon/src/table/data_evolution_writer.rs | 3 +
crates/paimon/src/table/kv_file_reader.rs | 17 +-
.../paimon/src/table/merge_tree_split_generator.rs | 591 +++++++++++++++++++++
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/source.rs | 167 +++++-
crates/paimon/src/table/table_read.rs | 112 +++-
crates/paimon/src/table/table_scan.rs | 159 +++++-
crates/paimon/src/table/table_write.rs | 135 +++++
9 files changed, 1376 insertions(+), 51 deletions(-)
diff --git a/crates/integrations/datafusion/tests/pk_tables.rs
b/crates/integrations/datafusion/tests/pk_tables.rs
index bf1a317..473413c 100644
--- a/crates/integrations/datafusion/tests/pk_tables.rs
+++ b/crates/integrations/datafusion/tests/pk_tables.rs
@@ -19,7 +19,8 @@
//!
//! Covers: basic write+read, dedup within/across commits, partitioned PK
tables,
//! multi-bucket, column projection, FirstRow merge engine, sequence.field,
-//! INSERT OVERWRITE, filter pushdown, and error cases.
+//! INSERT OVERWRITE, filter pushdown, cross-split merge correctness, and
+//! error cases.
//!
//! Dynamic bucket and cross-partition tests are in separate files:
//! - `dynamic_bucket_tables.rs`
@@ -28,8 +29,8 @@
mod common;
use common::{
- collect_id_name, collect_id_value, create_sql_context, create_test_env,
row_count,
- setup_sql_context,
+ collect_id_name, collect_id_value, collect_int_int_str,
create_sql_context, create_test_env,
+ row_count, setup_sql_context,
};
use datafusion::arrow::array::{Array, Int32Array, StringArray};
use paimon::catalog::Identifier;
@@ -1979,3 +1980,238 @@ async fn test_pk_dv_deduplicate_read_no_error() {
result.err()
);
}
+
+// ======================= Cross-Split Merge Correctness
=======================
+
+/// Regression: a 1-byte split target forces every data file into its own
+/// split candidate. Files holding versions of the same key overlap on key
+/// range and must still be merged into a single row — previously each split
+/// emitted its own (stale) version.
+#[tokio::test]
+async fn test_pk_dedup_merges_across_tiny_splits() {
+ let (_tmp, sql_context) = setup_sql_context().await;
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.t_tiny_split (
+ id INT NOT NULL, value INT,
+ PRIMARY KEY (id)
+ ) WITH (
+ 'bucket' = '1',
+ 'source.split.target-size' = '1b',
+ 'source.split.open-file-cost' = '1b'
+ )",
+ )
+ .await
+ .unwrap();
+
+ for value in [10, 20, 30] {
+ sql_context
+ .sql(&format!(
+ "INSERT INTO paimon.test_db.t_tiny_split VALUES (1, {value})"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ }
+
+ let rows = collect_id_value(
+ &sql_context,
+ "SELECT id, value FROM paimon.test_db.t_tiny_split",
+ )
+ .await;
+ assert_eq!(rows, vec![(1, 30)]);
+}
+
+/// LIMIT must not be starved by merge-needed splits: the three versions of
+/// key 1 share one split whose physical row count (3) overstates its single
+/// logical row. Such splits report an unknown merged row count, so limit
+/// pushdown cannot stop before the split holding key 2.
+#[tokio::test]
+async fn test_pk_limit_not_starved_by_merge_splits() {
+ let (_tmp, sql_context) = setup_sql_context().await;
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.t_tiny_split_limit (
+ id INT NOT NULL, value INT,
+ PRIMARY KEY (id)
+ ) WITH (
+ 'bucket' = '1',
+ 'source.split.target-size' = '1b',
+ 'source.split.open-file-cost' = '1b'
+ )",
+ )
+ .await
+ .unwrap();
+
+ // Three versions of key 1 (one overlapping section), then key 2.
+ for value in [10, 20, 30] {
+ sql_context
+ .sql(&format!(
+ "INSERT INTO paimon.test_db.t_tiny_split_limit VALUES (1,
{value})"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ }
+ sql_context
+ .sql("INSERT INTO paimon.test_db.t_tiny_split_limit VALUES (2, 200)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ // Two logical rows exist; LIMIT 2 must return both.
+ let returned = row_count(
+ &sql_context,
+ "SELECT id, value FROM paimon.test_db.t_tiny_split_limit LIMIT 2",
+ )
+ .await;
+ assert_eq!(returned, 2, "LIMIT 2 must yield 2 rows");
+
+ // COUNT(*) must reflect logical rows, not the physical (pre-merge) count
+ // that DataFusion could otherwise read from exact scan statistics.
+ let batches = sql_context
+ .sql("SELECT COUNT(*) FROM paimon.test_db.t_tiny_split_limit")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let count = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<datafusion::arrow::array::Int64Array>()
+ .unwrap()
+ .value(0);
+ assert_eq!(count, 2, "COUNT(*) must count merged rows");
+}
+
+/// Partial-update files can hold several physical rows of one key (the
+/// writer keeps all rows for read-side field-wise merge), so their splits
+/// must never report physical row counts as merged row counts: COUNT(*) has
+/// to count merged rows and LIMIT must not be starved.
+#[tokio::test]
+async fn test_pk_partial_update_count_and_limit_see_merged_rows() {
+ let (_tmp, sql_context) = setup_sql_context().await;
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.t_pu_count (
+ id INT NOT NULL, v_int INT, v_str STRING,
+ PRIMARY KEY (id)
+ ) WITH (
+ 'bucket' = '1',
+ 'merge-engine' = 'partial-update',
+ 'source.split.target-size' = '1b',
+ 'source.split.open-file-cost' = '1b'
+ )",
+ )
+ .await
+ .unwrap();
+
+ // One INSERT writes three partial updates of key 1 into a single file,
+ // plus an independent key 2.
+ sql_context
+ .sql(
+ "INSERT INTO paimon.test_db.t_pu_count VALUES
+ (1, 10, CAST(NULL AS STRING)),
+ (1, CAST(NULL AS INT), 'hello'),
+ (1, 100, CAST(NULL AS STRING)),
+ (2, 200, 'world')",
+ )
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ // COUNT(*) must count merged rows, not the physical rows a single file
+ // holds (DataFusion may answer COUNT(*) from exact scan statistics).
+ let batches = sql_context
+ .sql("SELECT COUNT(*) FROM paimon.test_db.t_pu_count")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let count = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<datafusion::arrow::array::Int64Array>()
+ .unwrap()
+ .value(0);
+ assert_eq!(count, 2, "COUNT(*) must count merged rows");
+
+ // Two logical rows exist; LIMIT 2 must not be starved by the
+ // multi-version file of key 1.
+ let returned = row_count(
+ &sql_context,
+ "SELECT id, v_int FROM paimon.test_db.t_pu_count LIMIT 2",
+ )
+ .await;
+ assert_eq!(returned, 2, "LIMIT 2 must yield 2 rows");
+}
+
+/// Same regression for the partial-update engine: per-column updates of one
+/// key spread over three commits/files must merge into a single row even
+/// when the split target would otherwise separate the files.
+#[tokio::test]
+async fn test_pk_partial_update_merges_across_tiny_splits() {
+ let (_tmp, sql_context) = setup_sql_context().await;
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.t_tiny_split_pu (
+ id INT NOT NULL, v_int INT, v_str STRING,
+ PRIMARY KEY (id)
+ ) WITH (
+ 'bucket' = '1',
+ 'merge-engine' = 'partial-update',
+ 'source.split.target-size' = '1b',
+ 'source.split.open-file-cost' = '1b'
+ )",
+ )
+ .await
+ .unwrap();
+
+ sql_context
+ .sql("INSERT INTO paimon.test_db.t_tiny_split_pu VALUES (1, 10,
CAST(NULL AS STRING))")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ sql_context
+ .sql("INSERT INTO paimon.test_db.t_tiny_split_pu VALUES (1, CAST(NULL
AS INT), 'hello')")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ sql_context
+ .sql("INSERT INTO paimon.test_db.t_tiny_split_pu VALUES (1, 100,
CAST(NULL AS STRING))")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let batches = sql_context
+ .sql("SELECT id, v_int, v_str FROM paimon.test_db.t_tiny_split_pu")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ collect_int_int_str(&batches),
+ vec![(1, 100, "hello".to_string())]
+ );
+}
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index f63ac24..4afe198 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -222,6 +222,8 @@ impl DataEvolutionWriter {
rb.with_projection(&col_refs);
let read = rb.new_read()?;
+ // Base + partial-column files share row-id ranges, so physical
+ // row counts overcount the group's logical rows.
let split = DataSplitBuilder::new()
.with_snapshot(file_range.snapshot_id)
.with_partition(BinaryRow::from_serialized_bytes(&file_range.partition)?)
@@ -229,6 +231,7 @@ impl DataEvolutionWriter {
.with_bucket_path(file_range.bucket_path.clone())
.with_total_buckets(file_range.total_buckets)
.with_data_files(file_range.files.clone())
+ .with_raw_convertible(file_range.files.len() == 1)
.build()?;
let stream = read.to_arrow(&[split])?;
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 128b2da..9a6f007 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -37,7 +37,7 @@ use crate::spec::{
use crate::table::schema_manager::SchemaManager;
use crate::table::ArrowRecordBatchStream;
use crate::{DataSplit, Error};
-use arrow_array::RecordBatch;
+use arrow_array::{RecordBatch, RecordBatchOptions};
use async_stream::try_stream;
use futures::StreamExt;
@@ -339,11 +339,16 @@ impl KeyValueFileReader {
.iter()
.map(|&src| batch.column(src).clone())
.collect();
- let reordered =
RecordBatch::try_new(output_schema.clone(), columns)
- .map_err(|e| Error::UnexpectedError {
- message: format!("Failed to reorder merged
RecordBatch: {e}"),
- source: Some(Box::new(e)),
- })?;
+ // An explicit row count keeps empty projections working
+ // (e.g. COUNT(*) reads no columns).
+ let options =
+
RecordBatchOptions::new().with_row_count(Some(batch.num_rows()));
+ let reordered =
+
RecordBatch::try_new_with_options(output_schema.clone(), columns, &options)
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to reorder merged
RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ })?;
yield reordered;
}
}
diff --git a/crates/paimon/src/table/merge_tree_split_generator.rs
b/crates/paimon/src/table/merge_tree_split_generator.rs
new file mode 100644
index 0000000..276aea5
--- /dev/null
+++ b/crates/paimon/src/table/merge_tree_split_generator.rs
@@ -0,0 +1,591 @@
+// 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.
+
+//! Split generation for primary-key (merge-tree) tables.
+//!
+//! Files whose primary-key ranges overlap must be read by the same
+//! sort-merge reader, so they have to stay in the same split. This module
+//! groups a bucket's files into key-range "sections" and then bin-packs
+//! whole sections into splits, mirroring the Java implementation.
+//!
+//! References:
+//!
[MergeTreeSplitGenerator](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/MergeTreeSplitGenerator.java),
+//!
[IntervalPartition](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/IntervalPartition.java)
+
+use super::bin_pack::pack_for_ordered;
+use crate::spec::{datum_cmp, BinaryRow, DataFileMeta, DataType, Datum,
TableSchema};
+use std::cmp::{self, Ordering};
+
+/// Compares serialized `BinaryRow` keys field-by-field using the trimmed
+/// primary-key data types.
+///
+/// BinaryRow stores fields little-endian, so raw byte comparison would order
+/// e.g. int 256 (`[00 01 00 00]`) before int 1 (`[01 00 00 00]`); keys must
+/// be decoded before comparing.
+pub(crate) struct KeyComparator {
+ key_types: Vec<DataType>,
+}
+
+/// A decoded key: one `Option<Datum>` per trimmed primary-key field
+/// (`None` = SQL NULL).
+type DecodedKey = Vec<Option<Datum>>;
+
+impl KeyComparator {
+ pub(crate) fn new(key_types: Vec<DataType>) -> Self {
+ Self { key_types }
+ }
+
+ /// Build a comparator over a table's trimmed primary keys, matching the
+ /// key layout the kv writer uses for min/max keys. Returns `None` for
+ /// tables without primary keys.
+ pub(crate) fn from_table_schema(schema: &TableSchema) -> Option<Self> {
+ let trimmed_pks = schema.trimmed_primary_keys();
+ if trimmed_pks.is_empty() {
+ return None;
+ }
+ let fields = schema.fields();
+ let key_types: Vec<DataType> = trimmed_pks
+ .iter()
+ .filter_map(|name| {
+ fields
+ .iter()
+ .find(|f| f.name() == name)
+ .map(|f| f.data_type().clone())
+ })
+ .collect();
+ // A PK name missing from the fields (should not happen) leaves the
+ // arity short; decode then fails and callers degrade safely.
+ Some(Self::new(key_types))
+ }
+
+ /// Decode a serialized min/max key. Returns `None` when the key is empty
+ /// or malformed, letting callers degrade to the safe "treat everything as
+ /// overlapping" path instead of failing the scan.
+ fn decode(&self, key: &[u8]) -> Option<DecodedKey> {
+ if key.is_empty() {
+ return None;
+ }
+ let row = BinaryRow::from_serialized_bytes(key).ok()?;
+ if (row.arity() as usize) != self.key_types.len() {
+ return None;
+ }
+ self.key_types
+ .iter()
+ .enumerate()
+ .map(|(pos, dt)| row.get_datum(pos, dt).ok())
+ .collect()
+ }
+}
+
+/// Compare decoded keys field-by-field. NULL sorts first; fields that
+/// `datum_cmp` cannot order (e.g. float NaN) compare as equal, which forces
+/// the files into the same section — conservative but never incorrect.
+fn compare_decoded(a: &DecodedKey, b: &DecodedKey) -> Ordering {
+ for (fa, fb) in a.iter().zip(b.iter()) {
+ let ord = match (fa, fb) {
+ (None, None) => Ordering::Equal,
+ (None, Some(_)) => Ordering::Less,
+ (Some(_), None) => Ordering::Greater,
+ (Some(da), Some(db)) => datum_cmp(da,
db).unwrap_or(Ordering::Equal),
+ };
+ if ord != Ordering::Equal {
+ return ord;
+ }
+ }
+ Ordering::Equal
+}
+
+/// A file paired with its decoded min/max keys.
+struct KeyedFile {
+ file: DataFileMeta,
+ min: DecodedKey,
+ max: DecodedKey,
+}
+
+/// Decode every file's key range up front. Returns `None` if any file lacks
+/// a usable key range, in which case callers must assume full overlap.
+fn decode_all(
+ files: Vec<DataFileMeta>,
+ comparator: &KeyComparator,
+) -> Result<Vec<KeyedFile>, Vec<DataFileMeta>> {
+ let mut keyed = Vec::with_capacity(files.len());
+ let mut undecodable = false;
+ for file in &files {
+ match (
+ comparator.decode(&file.min_key),
+ comparator.decode(&file.max_key),
+ ) {
+ (Some(min), Some(max)) if !undecodable => keyed.push(KeyedFile {
+ file: file.clone(),
+ min,
+ max,
+ }),
+ _ => undecodable = true,
+ }
+ }
+ if undecodable {
+ Err(files)
+ } else {
+ Ok(keyed)
+ }
+}
+
+/// Group files into sections by primary-key range overlap.
+///
+/// Files are sorted by `(min_key, max_key)`; a running upper bound tracks the
+/// max key seen in the current section, and a file whose min key exceeds the
+/// bound starts a new section. Sections never overlap each other, while files
+/// inside one section all transitively overlap and must be merged together.
+///
+/// Files with empty or undecodable key ranges collapse everything into one
+/// section: no parallelism, but never a missed merge.
+pub(crate) fn interval_partition(
+ files: Vec<DataFileMeta>,
+ comparator: &KeyComparator,
+) -> Vec<Vec<DataFileMeta>> {
+ if files.len() <= 1 {
+ return if files.is_empty() {
+ Vec::new()
+ } else {
+ vec![files]
+ };
+ }
+
+ let mut keyed = match decode_all(files, comparator) {
+ Ok(keyed) => keyed,
+ Err(files) => return vec![files],
+ };
+ keyed.sort_by(|a, b| {
+ compare_decoded(&a.min, &b.min).then_with(|| compare_decoded(&a.max,
&b.max))
+ });
+
+ let mut sections: Vec<Vec<DataFileMeta>> = Vec::new();
+ let mut current: Vec<DataFileMeta> = Vec::new();
+ let mut bound: Option<DecodedKey> = None;
+
+ for kf in keyed {
+ if let Some(ref b) = bound {
+ if compare_decoded(&kf.min, b) == Ordering::Greater {
+ sections.push(std::mem::take(&mut current));
+ bound = None;
+ }
+ }
+ match bound {
+ Some(ref b) if compare_decoded(&kf.max, b) != Ordering::Greater =>
{}
+ _ => bound = Some(kf.max),
+ }
+ current.push(kf.file);
+ }
+ if !current.is_empty() {
+ sections.push(current);
+ }
+ sections
+}
+
+/// Bin-pack whole sections into splits. A section is atomic: its files
+/// overlap on primary key and must never be separated, even when the section
+/// alone exceeds `target_split_size`.
+///
+/// Mirrors Java `MergeTreeSplitGenerator#packSplits`: a section's weight is
+/// `max(total file size, open_file_cost)` — the open-file cost is charged
+/// once per section, not per file.
+pub(crate) fn pack_sections(
+ sections: Vec<Vec<DataFileMeta>>,
+ target_split_size: i64,
+ open_file_cost: i64,
+) -> Vec<Vec<DataFileMeta>> {
+ pack_for_ordered(
+ sections,
+ |section| {
+ cmp::max(
+ section.iter().map(|f| f.file_size).sum::<i64>(),
+ open_file_cost,
+ )
+ },
+ target_split_size,
+ )
+ .into_iter()
+ .map(|sections| sections.into_iter().flatten().collect())
+ .collect()
+}
+
+/// A group of files forming one split, plus whether the split can be read
+/// raw — without the sort-merge reader — so its physical row count equals
+/// its logical row count.
+///
+/// Mirrors Java `SplitGenerator.SplitGroup`.
+#[derive(Debug)]
+pub(crate) struct SplitGroup {
+ pub(crate) files: Vec<DataFileMeta>,
+ pub(crate) raw_convertible: bool,
+}
+
+/// Whether a file is known to contain no DELETE rows.
+///
+/// Mirrors Java `MergeTreeSplitGenerator#withoutDeleteRow`: a missing
+/// `delete_row_count` is treated as "no deletes" for compatibility with files
+/// written by old versions.
+fn without_delete_row(file: &DataFileMeta) -> bool {
+ file.delete_row_count.is_none_or(|count| count == 0)
+}
+
+/// Generate batch splits for a merge-tree (primary-key) bucket.
+///
+/// Mirrors Java `MergeTreeSplitGenerator#splitForBatch` for the merging read
+/// path (deletion-vector and first-row tables are routed to plain size-based
+/// packing before reaching this function, matching Java's
+/// `alwaysRawConvertible` fast path):
+///
+/// * If every file is compacted (level != 0), has no delete rows, and all
+/// files sit on a single level, no two files can overlap on key range, so
+/// the files are bin-packed individually and every group is raw
+/// convertible.
+/// * Otherwise files are sectioned by key-range overlap and whole sections
+/// are bin-packed; a group is raw convertible only when it holds exactly
+/// one file without delete rows.
+///
+/// `file_keys_unique` is a deliberate deviation from Java: raw convertibility
+/// additionally assumes a file never holds two rows of one key. Java's
+/// `MergeTreeWriter#flushWriteBuffer` runs the merge function before flushing,
+/// so that holds for every engine; the Rust writer only deduplicates at flush
+/// for deduplicate/first-row, while partial-update keeps all rows for
+/// read-side field-wise merge (`kv_file_writer.rs`, `select_flush_indices`).
+/// Callers pass `false` for engines without that write-time guarantee, forcing
+/// every group non-raw-convertible. Can be relaxed once the writer merges on
+/// flush like Java.
+pub(crate) fn merge_tree_split_for_batch(
+ files: Vec<DataFileMeta>,
+ comparator: &KeyComparator,
+ target_split_size: i64,
+ open_file_cost: i64,
+ file_keys_unique: bool,
+) -> Vec<SplitGroup> {
+ let raw_convertible = files.iter().all(|f| f.level != 0 &&
without_delete_row(f));
+ let one_level = {
+ let mut levels: Vec<i32> = files.iter().map(|f| f.level).collect();
+ levels.sort_unstable();
+ levels.dedup();
+ levels.len() == 1
+ };
+
+ if raw_convertible && one_level {
+ return pack_for_ordered(
+ files,
+ |f| cmp::max(f.file_size, open_file_cost),
+ target_split_size,
+ )
+ .into_iter()
+ .map(|files| SplitGroup {
+ files,
+ raw_convertible: file_keys_unique,
+ })
+ .collect();
+ }
+
+ pack_sections(
+ interval_partition(files, comparator),
+ target_split_size,
+ open_file_cost,
+ )
+ .into_iter()
+ .map(|files| {
+ let raw_convertible = file_keys_unique && files.len() == 1 &&
without_delete_row(&files[0]);
+ SplitGroup {
+ files,
+ raw_convertible,
+ }
+ })
+ .collect()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::stats::BinaryTableStats;
+ use crate::spec::{BinaryRowBuilder, IntType};
+ use chrono::{DateTime, Utc};
+
+ fn int_key(value: i32) -> Vec<u8> {
+ let mut builder = BinaryRowBuilder::new(1);
+ builder.write_int(0, value);
+ builder.build_serialized()
+ }
+
+ fn keyed_file(name: &str, min: i32, max: i32, file_size: i64, level: i32)
-> DataFileMeta {
+ DataFileMeta {
+ file_name: name.to_string(),
+ file_size,
+ row_count: 100,
+ min_key: int_key(min),
+ max_key: int_key(max),
+ key_stats: BinaryTableStats::new(Vec::new(), Vec::new(),
Vec::new()),
+ value_stats: BinaryTableStats::new(Vec::new(), Vec::new(),
Vec::new()),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id: 0,
+ level,
+ extra_files: Vec::new(),
+ creation_time: DateTime::<Utc>::from_timestamp(0, 0),
+ delete_row_count: None,
+ embedded_index: None,
+ first_row_id: None,
+ write_cols: None,
+ external_path: None,
+ file_source: None,
+ value_stats_cols: None,
+ }
+ }
+
+ fn int_comparator() -> KeyComparator {
+ KeyComparator::new(vec![DataType::Int(IntType::new())])
+ }
+
+ fn section_names(sections: &[Vec<DataFileMeta>]) -> Vec<Vec<&str>> {
+ sections
+ .iter()
+ .map(|s| s.iter().map(|f| f.file_name.as_str()).collect())
+ .collect()
+ }
+
+ /// Int keys must be ordered numerically, not by little-endian bytes
+ /// (byte-wise, 256 = [00 01 00 00] would sort before 1 = [01 00 00 00]).
+ #[test]
+ fn key_comparator_orders_ints_numerically() {
+ let comparator = int_comparator();
+ let one = comparator.decode(&int_key(1)).unwrap();
+ let two = comparator.decode(&int_key(2)).unwrap();
+ let big = comparator.decode(&int_key(256)).unwrap();
+ assert_eq!(compare_decoded(&one, &two), Ordering::Less);
+ assert_eq!(compare_decoded(&two, &big), Ordering::Less);
+ assert_eq!(compare_decoded(&big, &one), Ordering::Greater);
+ }
+
+ #[test]
+ fn interval_partition_groups_overlapping_files() {
+ let files = vec![
+ keyed_file("a", 1, 10, 100, 0),
+ keyed_file("b", 5, 15, 100, 0),
+ keyed_file("c", 20, 30, 100, 0),
+ keyed_file("d", 25, 28, 100, 0),
+ ];
+ let sections = interval_partition(files, &int_comparator());
+ assert_eq!(
+ section_names(§ions),
+ vec![vec!["a", "b"], vec!["c", "d"]]
+ );
+ }
+
+ #[test]
+ fn interval_partition_keeps_disjoint_files_separate() {
+ let files = vec![
+ keyed_file("b", 3, 4, 100, 0),
+ keyed_file("a", 1, 2, 100, 0),
+ keyed_file("c", 5, 6, 100, 0),
+ ];
+ let sections = interval_partition(files, &int_comparator());
+ assert_eq!(
+ section_names(§ions),
+ vec![vec!["a"], vec!["b"], vec!["c"]]
+ );
+ }
+
+ /// A later file can extend the section bound past an earlier file's max:
+ /// [1,100] chains [50,60] and [90,110] into one section with [105,120].
+ #[test]
+ fn interval_partition_tracks_running_bound() {
+ let files = vec![
+ keyed_file("a", 1, 100, 100, 0),
+ keyed_file("b", 50, 60, 100, 0),
+ keyed_file("c", 90, 110, 100, 0),
+ keyed_file("d", 105, 120, 100, 0),
+ keyed_file("e", 121, 130, 100, 0),
+ ];
+ let sections = interval_partition(files, &int_comparator());
+ assert_eq!(
+ section_names(§ions),
+ vec![vec!["a", "b", "c", "d"], vec!["e"]]
+ );
+ }
+
+ #[test]
+ fn interval_partition_empty_key_degrades_to_single_section() {
+ let mut no_key = keyed_file("a", 1, 2, 100, 0);
+ no_key.min_key = Vec::new();
+ no_key.max_key = Vec::new();
+ let files = vec![no_key, keyed_file("b", 10, 20, 100, 0)];
+ let sections = interval_partition(files, &int_comparator());
+ assert_eq!(section_names(§ions), vec![vec!["a", "b"]]);
+ }
+
+ #[test]
+ fn pack_sections_respects_target_size() {
+ let sections = vec![
+ vec![keyed_file("a", 1, 2, 100, 0)],
+ vec![keyed_file("b", 3, 4, 100, 0)],
+ vec![keyed_file("c", 5, 6, 100, 0)],
+ ];
+ let splits = pack_sections(sections, 250, 1);
+ assert_eq!(section_names(&splits), vec![vec!["a", "b"], vec!["c"]]);
+ }
+
+ #[test]
+ fn pack_sections_never_splits_a_section() {
+ let sections = vec![vec![
+ keyed_file("a", 1, 10, 100, 0),
+ keyed_file("b", 5, 15, 100, 0),
+ ]];
+ let splits = pack_sections(sections, 50, 1);
+ assert_eq!(section_names(&splits), vec![vec!["a", "b"]]);
+ }
+
+ #[test]
+ fn pack_sections_applies_open_file_cost() {
+ let sections = vec![
+ vec![keyed_file("a", 1, 1, 2, 0)],
+ vec![keyed_file("b", 2, 2, 2, 0)],
+ vec![keyed_file("c", 3, 3, 2, 0)],
+ ];
+ // Weight per section is max(total file size=2, open_file_cost=100) =
100.
+ let splits = pack_sections(sections, 150, 100);
+ assert_eq!(
+ section_names(&splits),
+ vec![vec!["a"], vec!["b"], vec!["c"]]
+ );
+ }
+
+ /// The open-file cost is charged once per section, not per file (Java
+ /// `packSplits`): two 3-file sections weigh max(6, 100) = 100 each and
+ /// share one split under a 250 target, where a per-file charge (3 × 100)
+ /// would split them apart.
+ #[test]
+ fn pack_sections_charges_open_file_cost_per_section() {
+ let sections = vec![
+ vec![
+ keyed_file("a1", 1, 2, 2, 0),
+ keyed_file("a2", 1, 2, 2, 0),
+ keyed_file("a3", 1, 2, 2, 0),
+ ],
+ vec![
+ keyed_file("b1", 3, 4, 2, 0),
+ keyed_file("b2", 3, 4, 2, 0),
+ keyed_file("b3", 3, 4, 2, 0),
+ ],
+ ];
+ let splits = pack_sections(sections, 250, 100);
+ assert_eq!(
+ section_names(&splits),
+ vec![vec!["a1", "a2", "a3", "b1", "b2", "b3"]]
+ );
+ }
+
+ fn group_names(groups: &[SplitGroup]) -> Vec<Vec<&str>> {
+ groups
+ .iter()
+ .map(|g| g.files.iter().map(|f| f.file_name.as_str()).collect())
+ .collect()
+ }
+
+ /// All files compacted on one level: the fast path bin-packs files
+ /// individually and every group is raw convertible, even multi-file ones
+ /// (same-level files never overlap).
+ #[test]
+ fn split_for_batch_one_level_fast_path_is_raw_convertible() {
+ let comparator = int_comparator();
+ let files = vec![
+ keyed_file("a", 1, 10, 100, 5),
+ keyed_file("b", 11, 20, 100, 5),
+ keyed_file("c", 21, 30, 100, 5),
+ ];
+ let groups = merge_tree_split_for_batch(files, &comparator, 250, 1,
true);
+ assert_eq!(group_names(&groups), vec![vec!["a", "b"], vec!["c"]]);
+ assert!(groups.iter().all(|g| g.raw_convertible));
+ }
+
+ /// A delete-row file disables the fast path; after sectioning, only
+ /// single-file groups without delete rows stay raw convertible.
+ #[test]
+ fn split_for_batch_delete_rows_disable_raw_conversion() {
+ let comparator = int_comparator();
+ let mut with_deletes = keyed_file("del", 1, 10, 100, 5);
+ with_deletes.delete_row_count = Some(3);
+ let files = vec![with_deletes, keyed_file("clean", 11, 20, 100, 5)];
+ // Large target size packs both disjoint sections into one split.
+ let groups = merge_tree_split_for_batch(files, &comparator, 1000, 1,
true);
+ assert_eq!(group_names(&groups), vec![vec!["del", "clean"]]);
+ assert!(!groups[0].raw_convertible, "multi-file group is never raw");
+
+ // Tiny target size keeps each section alone; the delete-row file is
+ // still not raw convertible, the clean one is.
+ let mut with_deletes = keyed_file("del", 1, 10, 100, 5);
+ with_deletes.delete_row_count = Some(3);
+ let files = vec![with_deletes, keyed_file("clean", 11, 20, 100, 5)];
+ let groups = merge_tree_split_for_batch(files, &comparator, 1, 1,
true);
+ assert_eq!(group_names(&groups), vec![vec!["del"], vec!["clean"]]);
+ assert!(!groups[0].raw_convertible);
+ assert!(groups[1].raw_convertible);
+ }
+
+ /// Engines whose writer does not deduplicate at flush (partial-update
+ /// keeps all rows of a key in one file) cannot prove file-internal key
+ /// uniqueness: every group must stay non-raw-convertible on both the fast
+ /// path and the sectioned path, so physical row counts are never reported
+ /// as merged row counts.
+ #[test]
+ fn split_for_batch_without_unique_keys_never_raw_convertible() {
+ let comparator = int_comparator();
+
+ // Fast-path shape: all compacted, one level, no delete rows.
+ let files = vec![
+ keyed_file("a", 1, 10, 100, 5),
+ keyed_file("b", 11, 20, 100, 5),
+ ];
+ let groups = merge_tree_split_for_batch(files, &comparator, 250, 1,
false);
+ assert_eq!(group_names(&groups), vec![vec!["a", "b"]]);
+ assert!(groups.iter().all(|g| !g.raw_convertible));
+
+ // Sectioned shape: a disjoint single compacted file would be raw for
+ // deduplicate, but not without the write-time uniqueness guarantee.
+ let files = vec![
+ keyed_file("l0", 1, 50, 100, 0),
+ keyed_file("solo", 100, 120, 100, 2),
+ ];
+ let groups = merge_tree_split_for_batch(files, &comparator, 1, 1,
false);
+ assert_eq!(group_names(&groups), vec![vec!["l0"], vec!["solo"]]);
+ assert!(groups.iter().all(|g| !g.raw_convertible));
+ }
+
+ /// Level-0 or cross-level files take the sectioning path; overlapping
+ /// files share a non-raw-convertible group while a disjoint single file
+ /// stays raw convertible. Missing delete_row_count counts as "no deletes"
+ /// (old-version files).
+ #[test]
+ fn split_for_batch_sections_overlapping_files() {
+ let comparator = int_comparator();
+ let files = vec![
+ keyed_file("l0", 1, 50, 100, 0),
+ keyed_file("l1", 40, 90, 100, 1),
+ keyed_file("solo", 100, 120, 100, 2),
+ ];
+ let groups = merge_tree_split_for_batch(files, &comparator, 1, 1,
true);
+ assert_eq!(group_names(&groups), vec![vec!["l0", "l1"], vec!["solo"]]);
+ assert!(
+ !groups[0].raw_convertible,
+ "overlapping versions must merge"
+ );
+ assert!(groups[1].raw_convertible, "disjoint single compacted file");
+ }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 6c4901a..3872c6a 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -37,6 +37,7 @@ mod full_text_search_builder;
pub(crate) mod global_index_scanner;
mod kv_file_reader;
mod kv_file_writer;
+pub(crate) mod merge_tree_split_generator;
mod partition_filter;
mod postpone_file_writer;
mod prepared_files;
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 573d2f0..7d45742 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -425,6 +425,10 @@ pub struct DataSplit {
/// `None` at index `i` means no deletion file for `data_files[i]`
(matches Java getDeletionFiles() / List<DeletionFile> with null elements).
data_deletion_files: Option<Vec<Option<DeletionFile>>>,
row_ranges: Option<Vec<RowRange>>,
+ /// Whether the split can be read raw, without the merge reader: its
+ /// physical rows are exactly its logical rows (modulo deletion files).
+ /// Mirrors Java `DataSplit#rawConvertible`.
+ raw_convertible: bool,
}
impl DataSplit {
@@ -457,6 +461,12 @@ impl DataSplit {
self.row_ranges.as_deref()
}
+ /// Whether this split can be read raw (no sort-merge needed); see the
+ /// field doc. Mirrors Java `DataSplit#rawConvertible`.
+ pub fn raw_convertible(&self) -> bool {
+ self.raw_convertible
+ }
+
/// Returns the deletion file for the data file at the given index, if
any. `None` at that index means no deletion file.
pub fn deletion_file_for_data_file_index(&self, index: usize) ->
Option<&DeletionFile> {
self.data_deletion_files
@@ -487,21 +497,33 @@ impl DataSplit {
/// Returns the merged row count if it can be computed.
///
- /// Two paths:
- /// 1. If all files have `first_row_id` (data evolution mode): merge
overlapping
- /// row ID ranges and take max row count per group.
- /// 2. Otherwise: row_count() minus deleted rows (if deletion file
cardinality is known).
+ /// Two paths, checked in the same order as Java:
+ /// 1. Raw convertible splits (with all deletion-file cardinalities known):
+ /// physical row counts equal logical row counts, so sum `row_count`
+ /// minus deleted rows. Splits that need the sort-merge reader may
+ /// collapse multiple versions of a key into one row, so their physical
+ /// counts are only an upper bound and are never reported.
+ /// 2. If all files have `first_row_id` (data evolution mode): merge
+ /// overlapping row ID ranges and take max row count per group.
///
- /// Returns `None` if deletion files exist but cardinality is unknown.
+ /// Returns `None` otherwise.
///
/// Reference:
[DataSplit.mergedRowCount()](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java#L133)
pub fn merged_row_count(&self) -> Option<i64> {
- // Try data evolution merged row count first (all files have
first_row_id)
- if let Some(count) = self.data_evolution_merged_row_count() {
+ if let Some(count) = self.raw_merged_row_count() {
return Some(count);
}
+ self.data_evolution_merged_row_count()
+ }
- // Fallback: row_count - deleted rows
+ /// Physical row count minus deletions, valid only for raw convertible
+ /// splits with all deletion-file cardinalities known.
+ ///
+ /// Mirrors Java `rawMergedRowCountAvailable` + `rawMergedRowCount`.
+ fn raw_merged_row_count(&self) -> Option<i64> {
+ if !self.raw_convertible {
+ return None;
+ }
match &self.data_deletion_files {
None => Some(self.row_count()),
Some(deletion_files) => {
@@ -565,6 +587,7 @@ pub struct DataSplitBuilder {
/// Same length as data_files; `None` at index i = no deletion file for
data_files[i].
data_deletion_files: Option<Vec<Option<DeletionFile>>>,
row_ranges: Option<Vec<RowRange>>,
+ raw_convertible: bool,
}
impl DataSplitBuilder {
@@ -578,6 +601,10 @@ impl DataSplitBuilder {
data_files: None,
data_deletion_files: None,
row_ranges: None,
+ // Splits with no merge semantics (append tables, single-file
+ // utility splits) are raw by nature; the merge-tree and
+ // data-evolution scan paths set this explicitly per split group.
+ raw_convertible: true,
}
}
@@ -620,6 +647,12 @@ impl DataSplitBuilder {
self
}
+ /// Mark whether the split can be read raw; see
[`DataSplit::raw_convertible`].
+ pub fn with_raw_convertible(mut self, raw_convertible: bool) -> Self {
+ self.raw_convertible = raw_convertible;
+ self
+ }
+
pub fn build(self) -> crate::Result<DataSplit> {
if self.snapshot_id == -1 {
return Err(crate::Error::UnexpectedError {
@@ -672,6 +705,7 @@ impl DataSplitBuilder {
data_files,
data_deletion_files: self.data_deletion_files,
row_ranges: self.row_ranges,
+ raw_convertible: self.raw_convertible,
})
}
}
@@ -700,3 +734,120 @@ impl Plan {
&self.splits
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::stats::BinaryTableStats;
+
+ fn file(name: &str, row_count: i64, first_row_id: Option<i64>) ->
DataFileMeta {
+ DataFileMeta {
+ file_name: name.to_string(),
+ file_size: 128,
+ row_count,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: BinaryTableStats::new(Vec::new(), Vec::new(),
Vec::new()),
+ value_stats: BinaryTableStats::new(Vec::new(), Vec::new(),
Vec::new()),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id: 0,
+ level: 1,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count: None,
+ embedded_index: None,
+ first_row_id,
+ write_cols: None,
+ external_path: None,
+ file_source: None,
+ value_stats_cols: None,
+ }
+ }
+
+ fn split(files: Vec<DataFileMeta>, raw_convertible: bool) -> DataSplit {
+ DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("file:/tmp/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(files)
+ .with_raw_convertible(raw_convertible)
+ .build()
+ .unwrap()
+ }
+
+ /// Raw convertible split without deletion files: physical sum is exact.
+ #[test]
+ fn test_merged_row_count_raw_convertible_sums_physical_rows() {
+ let s = split(vec![file("a", 10, None), file("b", 5, None)], true);
+ assert_eq!(s.merged_row_count(), Some(15));
+ }
+
+ /// Merge-needed split (multiple versions of a key may collapse): the
+ /// physical sum is only an upper bound, so the count is unknown.
+ /// Mirrors Java `rawMergedRowCountAvailable`.
+ #[test]
+ fn test_merged_row_count_unknown_for_merge_splits() {
+ let s = split(vec![file("a", 10, None), file("b", 5, None)], false);
+ assert_eq!(s.merged_row_count(), None);
+ }
+
+ /// Non-raw-convertible split where all files carry `first_row_id`: the
+ /// data-evolution branch still applies (overlapping row-id groups count
+ /// the max row count per group).
+ #[test]
+ fn test_merged_row_count_data_evolution_branch() {
+ let mut a = file("a", 10, Some(0));
+ a.row_count = 10;
+ let mut b = file("b", 4, Some(0));
+ b.row_count = 4;
+ let c = file("c", 7, Some(100));
+ let s = split(vec![a, b, c], false);
+ // a and b share row ids [0, ..): max(10, 4) = 10; c adds 7.
+ assert_eq!(s.merged_row_count(), Some(17));
+ }
+
+ /// Raw convertible split with a deletion file of known cardinality:
+ /// deleted rows are subtracted; unknown cardinality makes the raw branch
+ /// unavailable.
+ #[test]
+ fn test_merged_row_count_with_deletion_files() {
+ let s = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("file:/tmp/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![file("a", 10, None)])
+ .with_data_deletion_files(vec![Some(DeletionFile::new(
+ "file:/tmp/a.dv".to_string(),
+ 0,
+ 0,
+ Some(3),
+ ))])
+ .with_raw_convertible(true)
+ .build()
+ .unwrap();
+ assert_eq!(s.merged_row_count(), Some(7));
+
+ let unknown = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("file:/tmp/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![file("a", 10, None)])
+ .with_data_deletion_files(vec![Some(DeletionFile::new(
+ "file:/tmp/a.dv".to_string(),
+ 0,
+ 0,
+ None,
+ ))])
+ .with_raw_convertible(true)
+ .build()
+ .unwrap();
+ assert_eq!(unknown.merged_row_count(), None);
+ }
+}
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 5493938..7fe4156 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -76,9 +76,11 @@ impl<'a> TableRead<'a> {
let core_options = CoreOptions::new(self.table.schema.options());
let merge_engine = core_options.merge_engine()?;
- // PK table with Deduplicate engine: splits containing level-0 files
- // need KeyValueFileReader for sort-merge dedup; splits with only
- // compacted files (level > 0) can use the faster DataFileReader.
+ // PK table with Deduplicate engine: splits that may hold multiple
+ // versions of a key need KeyValueFileReader for sort-merge dedup;
+ // splits marked raw convertible by scan planning — and all compacted
+ // files of deletion-vector tables, where DVs mask stale versions —
+ // use the faster DataFileReader.
if has_primary_keys
&& matches!(
merge_engine,
@@ -95,8 +97,12 @@ impl<'a> TableRead<'a> {
}
}
- /// Read PK table with Deduplicate engine: level-0 splits go through
- /// KeyValueFileReader for sort-merge dedup, compacted splits use
DataFileReader.
+ /// Read PK table with Deduplicate engine: splits marked raw convertible
+ /// by scan planning (mirrors Java `DataSplit#convertToRawFiles`) use the
+ /// faster DataFileReader; the rest go through KeyValueFileReader for
+ /// sort-merge dedup. Deletion-vector tables are exempt: their stale
+ /// versions are masked by DVs, and KeyValueFileReader does not support
+ /// DVs, so they keep the plain level-0 dispatch.
fn read_pk(
&self,
data_splits: &[DataSplit],
@@ -106,10 +112,15 @@ impl<'a> TableRead<'a> {
return self.read_kv(data_splits, core_options);
}
+ // Deletion-vector tables read raw by design: stale versions of a key
+ // are masked by DVs, not merged, and KeyValueFileReader does not
+ // support DVs. Keep the plain level-0 dispatch for them.
+ let dv_enabled = core_options.deletion_vectors_enabled();
+
let mut kv_splits = Vec::new();
let mut raw_splits = Vec::new();
for split in data_splits {
- if split.data_files().iter().any(|f| f.level == 0) {
+ if pk_split_needs_merge(split, dv_enabled) {
kv_splits.push(split.clone());
} else {
raw_splits.push(split.clone());
@@ -192,3 +203,92 @@ impl<'a> TableRead<'a> {
)
}
}
+
+/// Whether a primary-key split must go through the sort-merge reader.
+///
+/// Mirrors Java `PrimaryKeyTableRawFileSplitReadProvider#match`: a raw read
+/// needs the split marked raw convertible AND a known `delete_row_count` on
+/// every file. Legacy files without the stat may hide delete rows — scan
+/// planning treats the missing stat as "no deletes" for compatibility, so the
+/// read side must fall back to the merge reader, which drops them.
+///
+/// Deletion-vector tables keep the plain level-0 dispatch: stale versions are
+/// masked by DVs and KeyValueFileReader does not support DVs.
+fn pk_split_needs_merge(split: &DataSplit, dv_enabled: bool) -> bool {
+ if dv_enabled {
+ return split.data_files().iter().any(|f| f.level == 0);
+ }
+ !split.raw_convertible()
+ || split
+ .data_files()
+ .iter()
+ .any(|f| f.delete_row_count.is_none())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::stats::BinaryTableStats;
+ use crate::spec::{BinaryRow, DataFileMeta};
+ use crate::table::source::DataSplitBuilder;
+
+ fn file(name: &str, level: i32, delete_row_count: Option<i64>) ->
DataFileMeta {
+ DataFileMeta {
+ file_name: name.to_string(),
+ file_size: 128,
+ row_count: 10,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: BinaryTableStats::new(Vec::new(), Vec::new(),
Vec::new()),
+ value_stats: BinaryTableStats::new(Vec::new(), Vec::new(),
Vec::new()),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id: 0,
+ level,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count,
+ embedded_index: None,
+ first_row_id: None,
+ write_cols: None,
+ external_path: None,
+ file_source: None,
+ value_stats_cols: None,
+ }
+ }
+
+ fn split(files: Vec<DataFileMeta>, raw_convertible: bool) -> DataSplit {
+ DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("file:/tmp/bucket-0".to_string())
+ .with_total_buckets(1)
+ .with_data_files(files)
+ .with_raw_convertible(raw_convertible)
+ .build()
+ .unwrap()
+ }
+
+ #[test]
+ fn test_pk_split_needs_merge_routing() {
+ // Raw convertible with known delete counts: raw read.
+ let raw = split(vec![file("a", 5, Some(0))], true);
+ assert!(!pk_split_needs_merge(&raw, false));
+
+ // Not raw convertible: merge read.
+ let merge = split(vec![file("a", 5, Some(0))], false);
+ assert!(pk_split_needs_merge(&merge, false));
+
+ // Raw convertible but a legacy file lacks delete_row_count: the file
+ // may hide delete rows, so it must go through the merge reader.
+ let legacy = split(vec![file("a", 5, None)], true);
+ assert!(pk_split_needs_merge(&legacy, false));
+
+ // Deletion-vector tables dispatch on level 0 only.
+ let dv_l0 = split(vec![file("a", 0, None)], false);
+ assert!(pk_split_needs_merge(&dv_l0, true));
+ let dv_compacted = split(vec![file("a", 5, None)], false);
+ assert!(!pk_split_needs_merge(&dv_compacted, true));
+ }
+}
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 273a5c9..3135e77 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -35,6 +35,9 @@ use crate::spec::{
TimeTravelSelector,
};
use crate::table::bin_pack::split_for_batch;
+use crate::table::merge_tree_split_generator::{
+ merge_tree_split_for_batch, KeyComparator, SplitGroup,
+};
use crate::table::source::{
any_range_overlaps_file, intersect_ranges_with_file, merge_row_ranges,
DataSplit,
DataSplitBuilder, DeletionFile, PartitionBucket, Plan, RowRange,
@@ -413,16 +416,13 @@ impl<'a> TableScan<'a> {
/// Apply a limit-pushdown hint to the generated splits.
///
- /// Iterates through splits and accumulates `merged_row_count()` until the
- /// limit hint is reached. Returns only the splits likely needed to satisfy
- /// that hint.
- ///
- /// This does not guarantee an exact final row count. If a split's
- /// `merged_row_count()` is `None` (for example because of unknown deletion
- /// cardinality), that split is kept even though its contribution to the
- /// limit is unknown. Planning may still stop early later if the
- /// accumulated known `merged_row_count()` reaches the limit, and the
- /// caller or query engine must enforce the final LIMIT.
+ /// Mirrors Java `DataTableBatchScan#applyPushDownLimit`: splits whose
+ /// `merged_row_count()` is unknown (for example merge-needed PK splits or
+ /// unknown deletion cardinality) are skipped — they contribute an unknown
+ /// number of rows, so they cannot help satisfy the limit. Pruning is
+ /// 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.
fn apply_limit_pushdown(&self, splits: Vec<DataSplit>) -> Vec<DataSplit> {
let limit = match self.limit {
Some(l) => l,
@@ -439,22 +439,17 @@ impl<'a> TableScan<'a> {
let mut limited_splits = Vec::new();
let mut scanned_row_count: i64 = 0;
- for split in splits {
- match split.merged_row_count() {
- Some(merged_count) => {
- limited_splits.push(split);
- scanned_row_count += merged_count;
- if scanned_row_count >= limit as i64 {
- return limited_splits;
- }
- }
- None => {
- limited_splits.push(split);
+ 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;
}
}
}
- limited_splits
+ splits
}
/// Read all manifest entries from a snapshot, applying filters and
merging.
@@ -620,6 +615,24 @@ impl<'a> TableScan<'a> {
None
};
+ // Primary-key tables must keep key-overlapping files in one split so
the
+ // sort-merge reader sees every version of a key. The comparator
decodes
+ // the trimmed-PK min/max keys written by the kv writer.
+ //
+ // Deletion-vector and first-row tables read without merging (stale
rows
+ // are masked by DVs / level-0 is skipped), so they keep plain
size-based
+ // packing like Java's MergeTreeSplitGenerator fast path.
+ let read_merges_overlapping_keys =
!core_options.deletion_vectors_enabled()
+ && !matches!(
+ core_options.merge_engine(),
+ Ok(crate::spec::MergeEngine::FirstRow)
+ );
+ let pk_comparator = if read_merges_overlapping_keys {
+ KeyComparator::from_table_schema(self.table.schema())
+ } else {
+ None
+ };
+
// Read deletion vector index manifest once (like Java generateSplits
/ scanDvIndex).
let (deletion_files_map, effective_row_ranges) =
if let Some(index_manifest_name) = snapshot.index_manifest() {
@@ -670,7 +683,7 @@ impl<'a> TableScan<'a> {
// Data-evolution tables merge overlapping row-id groups
column-wise during read.
// Keep that split boundary intact and only bin-pack single-file
groups.
// Apply group-level predicate filtering after grouping by row_id
range.
- let file_groups: Vec<Vec<DataFileMeta>> = if
data_evolution_enabled {
+ let file_groups: Vec<SplitGroup> = if data_evolution_enabled {
let row_id_groups = group_by_overlapping_row_id(data_files);
// Filter groups by merged stats before splitting.
@@ -705,20 +718,60 @@ impl<'a> TableScan<'a> {
let mut result = Vec::new();
for group in multis {
- result.push(group);
+ // Files sharing a row-id range hold column slices of the
+ // same logical rows; physical counts overcount them
+ // (Java DataEvolutionSplitGenerator: not raw convertible).
+ result.push(SplitGroup {
+ files: group,
+ raw_convertible: false,
+ });
}
let single_files: Vec<DataFileMeta> =
singles.into_iter().flatten().collect();
for file_group in split_for_batch(single_files,
target_split_size, open_file_cost) {
- result.push(file_group);
+ result.push(SplitGroup {
+ files: file_group,
+ raw_convertible: true,
+ });
}
result
+ } else if let Some(ref comparator) = pk_comparator {
+ // Merge-tree path: keep key-overlapping files in one split and
+ // mark which splits the sort-merge reader can skip (mirrors
+ // Java MergeTreeSplitGenerator#splitForBatch). Only engines
+ // whose writer deduplicates at flush guarantee a file never
+ // holds two rows of one key, so only they may mark groups raw
+ // convertible; see merge_tree_split_for_batch. (First-row
+ // tables do not take this path today, but its writer dedups
+ // too, so keep the gate accurate.)
+ let file_keys_unique = matches!(
+ core_options.merge_engine(),
+ Ok(crate::spec::MergeEngine::Deduplicate)
+ | Ok(crate::spec::MergeEngine::FirstRow)
+ );
+ merge_tree_split_for_batch(
+ data_files,
+ comparator,
+ target_split_size,
+ open_file_cost,
+ file_keys_unique,
+ )
} else {
split_for_batch(data_files, target_split_size, open_file_cost)
+ .into_iter()
+ .map(|files| SplitGroup {
+ files,
+ raw_convertible: true,
+ })
+ .collect()
};
- for file_group in file_groups {
+ for group in file_groups {
+ let SplitGroup {
+ files: file_group,
+ raw_convertible,
+ } = group;
let data_deletion_files =
per_bucket_deletion_map.map(|per_bucket| {
file_group
.iter()
@@ -748,7 +801,8 @@ impl<'a> TableScan<'a> {
.with_bucket(bucket)
.with_bucket_path(bucket_path.clone())
.with_total_buckets(total_buckets)
- .with_data_files(file_group);
+ .with_data_files(file_group)
+ .with_raw_convertible(raw_convertible);
if let Some(files) = data_deletion_files {
builder = builder.with_data_deletion_files(files);
}
@@ -1008,8 +1062,11 @@ mod tests {
assert!(pruned.is_empty());
}
+ /// Java semantics: unknown-count splits are skipped — they cannot prove
+ /// progress toward the limit — and pruning commits once the counted
+ /// splits alone cover the limit.
#[test]
- fn test_apply_limit_pushdown_keeps_unknown_merged_row_count() {
+ fn test_apply_limit_pushdown_skips_unknown_merged_row_count() {
let table = limit_test_table();
let scan = TableScan::new(&table, None, vec![], None, Some(3), None);
let splits = vec![
@@ -1020,12 +1077,58 @@ mod tests {
let pruned = scan.apply_limit_pushdown(splits);
+ assert_eq!(split_file_names(&pruned), vec!["a.parquet", "c.parquet"]);
+ }
+
+ /// When counted splits never reach the limit, the original split list is
+ /// returned unchanged (mirrors Java `applyPushDownLimit`).
+ #[test]
+ fn test_apply_limit_pushdown_returns_all_when_limit_not_reached() {
+ let table = limit_test_table();
+ let scan = TableScan::new(&table, None, vec![], None, Some(100), None);
+ let splits = vec![
+ limit_test_split("a.parquet", 2),
+ limit_test_split_with_unknown_merged_row_count("b.parquet", 4),
+ limit_test_split("c.parquet", 3),
+ ];
+
+ let pruned = scan.apply_limit_pushdown(splits);
+
assert_eq!(
split_file_names(&pruned),
vec!["a.parquet", "b.parquet", "c.parquet"]
);
}
+ /// A non-raw-convertible split (merge-needed PK split) has an unknown
+ /// merged row count: its physical rows overcount merged versions, so it
+ /// must not satisfy the limit on its own.
+ #[test]
+ fn test_apply_limit_pushdown_treats_merge_splits_as_unknown() {
+ let table = limit_test_table();
+ let scan = TableScan::new(&table, None, vec![], None, Some(3), None);
+
+ let mut merge_file = test_data_file_meta(Vec::new(), Vec::new(),
Vec::new(), 10);
+ merge_file.file_name = "merge.parquet".to_string();
+ let merge_split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path("file:/tmp/merge.parquet".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![merge_file])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap();
+ assert_eq!(merge_split.merged_row_count(), None);
+
+ let splits = vec![merge_split, limit_test_split("a.parquet", 3)];
+ let pruned = scan.apply_limit_pushdown(splits);
+
+ // The merge split is skipped; only the counted split commits pruning.
+ assert_eq!(split_file_names(&pruned), vec!["a.parquet"]);
+ }
+
#[test]
fn test_first_row_skips_level_zero_by_default() {
assert!(should_skip_level_zero_for_scan(
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index fea6ff7..d8add25 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -2567,4 +2567,139 @@ mod tests {
assert_eq!(ids, vec![1]);
assert_eq!(values, vec![20]);
}
+
+ fn tiny_split_pk_table(file_io: &FileIO, table_path: &str) -> Table {
+ Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_tiny_split_pk"),
+ table_path.to_string(),
+ pk_changelog_schema(&[
+ ("source.split.target-size", "1b"),
+ ("source.split.open-file-cost", "1b"),
+ ]),
+ None,
+ )
+ }
+
+ async fn commit_one_batch(table: &Table, ids: Vec<i32>, values: Vec<i32>) {
+ let mut tw = TableWrite::new(table, "test-user".to_string()).unwrap();
+ tw.write_arrow_batch(&make_batch(ids, values))
+ .await
+ .unwrap();
+ let msgs = tw.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(msgs)
+ .await
+ .unwrap();
+ }
+
+ async fn read_id_value_rows(table: &Table) -> Vec<(i32, i32)> {
+ let rb = table.new_read_builder();
+ let plan = rb.new_scan().plan().await.unwrap();
+ let read = rb.new_read().unwrap();
+ let batches: Vec<RecordBatch> =
+
futures::TryStreamExt::try_collect(read.to_arrow(plan.splits()).unwrap())
+ .await
+ .unwrap();
+ let mut rows: Vec<(i32, i32)> = batches
+ .iter()
+ .flat_map(|b| {
+ let ids =
b.column(0).as_any().downcast_ref::<Int32Array>().unwrap();
+ let values =
b.column(1).as_any().downcast_ref::<Int32Array>().unwrap();
+ (0..b.num_rows()).map(|i| (ids.value(i), values.value(i)))
+ })
+ .collect();
+ rows.sort_unstable();
+ rows
+ }
+
+ /// Regression test: three commits of the same primary key produce three
+ /// key-overlapping files. Even with a 1-byte split target they must stay
+ /// in one split so the sort-merge reader merges them, instead of each
+ /// split emitting its own (stale) version of the row.
+ #[tokio::test]
+ async fn
test_pk_plan_keeps_overlapping_files_in_one_split_under_tiny_target() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tiny_split_overlap";
+ setup_dirs(&file_io, table_path).await;
+ let table = tiny_split_pk_table(&file_io, table_path);
+
+ commit_one_batch(&table, vec![1], vec![10]).await;
+ commit_one_batch(&table, vec![1], vec![20]).await;
+ commit_one_batch(&table, vec![1], vec![30]).await;
+
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ assert_eq!(
+ plan.splits().len(),
+ 1,
+ "overlapping files must share a split"
+ );
+ assert_eq!(plan.splits()[0].data_files().len(), 3);
+
+ assert_eq!(read_id_value_rows(&table).await, vec![(1, 30)]);
+ }
+
+ /// Files with disjoint key ranges may still be distributed across splits
+ /// for parallelism when the split target is small.
+ #[tokio::test]
+ async fn test_pk_plan_separates_disjoint_files_under_tiny_target() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tiny_split_disjoint";
+ setup_dirs(&file_io, table_path).await;
+ let table = tiny_split_pk_table(&file_io, table_path);
+
+ commit_one_batch(&table, vec![1], vec![10]).await;
+ commit_one_batch(&table, vec![2], vec![20]).await;
+ commit_one_batch(&table, vec![3], vec![30]).await;
+
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ assert_eq!(
+ plan.splits().len(),
+ 3,
+ "disjoint files keep split parallelism"
+ );
+
+ assert_eq!(
+ read_id_value_rows(&table).await,
+ vec![(1, 10), (2, 20), (3, 30)]
+ );
+ }
+
+ /// Append-only tables have no primary keys (and empty min/max keys); they
+ /// must keep using plain file-level bin packing instead of degrading to a
+ /// single all-files section.
+ #[tokio::test]
+ async fn
test_append_table_plan_uses_file_level_bin_pack_under_tiny_target() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_tiny_split_append";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("source.split.target-size", "1b")
+ .option("source.split.open-file-cost", "1b")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_tiny_split_append"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+
+ commit_one_batch(&table, vec![1], vec![10]).await;
+ commit_one_batch(&table, vec![2], vec![20]).await;
+ commit_one_batch(&table, vec![3], vec![30]).await;
+
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ assert_eq!(
+ plan.splits().len(),
+ 3,
+ "append tables keep file-level bin pack"
+ );
+ }
}
