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 0132d488 feat(core): align batch changelog planning and row-kind
normalization (#1079)
0132d488 is described below
commit 0132d488756ab2b3a388f3312332096bbe74647b
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 23:09:46 2026 +0800
feat(core): align batch changelog planning and row-kind normalization
(#1079)
---
bindings/python/tests/test_read.py | 34 ++
crates/paimon/src/table/incremental_scan.rs | 101 ++---
crates/paimon/src/table/read_builder.rs | 6 +-
crates/paimon/src/table/scan_trace.rs | 34 --
crates/paimon/src/table/source.rs | 4 -
crates/paimon/src/table/table_scan.rs | 29 +-
crates/paimon/src/table/table_write.rs | 202 ++++++++--
.../tests/incremental_changelog_batch_test.rs | 443 +++++++++++++++++++++
crates/paimon/tests/writer_resources_test.rs | 49 ++-
9 files changed, 755 insertions(+), 147 deletions(-)
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 1a88cbe2..abd0cf16 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -1547,6 +1547,40 @@ def
test_incremental_changelog_and_auto_plan_physical_changelog_files():
assert empty.splits() == []
[email protected]("mode", ["changelog", "auto"])
+def test_batch_incremental_changelog_packing_sharding_and_limits(mode):
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_input_changelog_table(warehouse)
+ builder = table.new_read_builder().with_include_row_kind(True)
+ full = builder.new_incremental_scan(0, 2, mode).plan()
+ # Java packs changelogs over the whole range, rather than once per
+ # snapshot. The two commits in this bucket fit into one split.
+ assert len(full.splits()) == 1
+ assert len(_native_split_file_names(full.splits())) == 2
+ names, rows = [], []
+ for index in range(3):
+ scan = builder.new_incremental_scan(0, 2, mode).with_shard(index,
3)
+ plan = scan.plan()
+ assert plan.snapshot_id() == 2
+ assert all(split.is_streaming() for split in plan.splits())
+ names.extend(_native_split_file_names(plan.splits()))
+ batches = builder.new_read().read(plan.splits())
+ if batches:
+ rows.extend(pa.Table.from_batches(batches).to_pylist())
+ for boundary in (0, 2):
+ empty = builder.new_incremental_scan(boundary, boundary,
mode).with_shard(index, 3).plan()
+ assert empty.snapshot_id() == boundary
+ assert empty.splits() == []
+ assert sorted(names) == sorted(_native_split_file_names(full.splits()))
+ assert sorted((row['rowkind'], row['id'], row['value']) for row in
rows) == [
+ ('+I', 1, 'a'), ('+I', 2, 'b'), ('+I', 3, 'c')]
+ limited =
(table.new_read_builder().with_limit(1).with_include_row_kind(True))
+ plan = limited.new_incremental_scan(0, 2, mode).with_shard(0, 3).plan()
+ result =
pa.Table.from_batches(limited.new_read().read(plan.splits())).to_pylist()
+ assert len(result) == 1
+ assert result[0] in rows
+
+
def test_incremental_changelog_keeps_filter_projection_and_validates_mode():
with tempfile.TemporaryDirectory() as warehouse:
table = _make_input_changelog_table(warehouse)
diff --git a/crates/paimon/src/table/incremental_scan.rs
b/crates/paimon/src/table/incremental_scan.rs
index 7d30fd73..acddf033 100644
--- a/crates/paimon/src/table/incremental_scan.rs
+++ b/crates/paimon/src/table/incremental_scan.rs
@@ -301,6 +301,7 @@ impl<'a> IncrementalScan<'a> {
}
/// Select one worker shard using the same rules as
[`TableScan::with_shard`].
+ /// Use [`Self::plan_combined`] for Delta and Changelog batch sharding.
pub fn with_shard(mut self, index: usize, count: usize) ->
crate::Result<Self> {
self.scan = self.scan.with_shard(index, count)?;
Ok(self)
@@ -314,7 +315,7 @@ impl<'a> IncrementalScan<'a> {
}
match mode {
IncrementalScanMode::Delta => self.plan_delta(mode).await,
- IncrementalScanMode::Changelog => self.plan_changelog(mode,
None).await,
+ IncrementalScanMode::Changelog => self.plan_changelog(mode).await,
IncrementalScanMode::Auto => unreachable!("Auto must resolve
before planning"),
IncrementalScanMode::Diff => self.plan_diff(mode).await,
}
@@ -327,7 +328,7 @@ impl<'a> IncrementalScan<'a> {
|| self.scan.has_shard()
{
return Err(crate::Error::Unsupported {
- message: "Incremental row-position selection, chunk_shuffle
and sharding require combined delta planning"
+ message: "Incremental row-position selection and chunk_shuffle
require combined Delta planning; sharding requires combined Delta or Changelog
planning"
.into(),
});
}
@@ -338,8 +339,8 @@ impl<'a> IncrementalScan<'a> {
///
/// This is the bridge used by readers which already consume
/// [`DataSplit`]s. Delta keeps its cross-snapshot packing semantics;
- /// Changelog preserves physical changelog files and row kinds in snapshot
- /// order. `Auto` resolves from `changelog-producer`. Diff cannot be
+ /// Changelog packs physical changelog files across snapshots while
preserving
+ /// every row kind. `Auto` resolves from `changelog-producer`. Diff cannot
be
/// represented by an ordinary split list because each unit contains a
/// before/after pair.
pub async fn plan_combined(&self) -> crate::Result<Plan> {
@@ -350,25 +351,10 @@ impl<'a> IncrementalScan<'a> {
&self,
trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
- match self.resolve_mode() {
- IncrementalScanMode::Delta =>
self.plan_combined_delta_with_trace(trace).await,
- IncrementalScanMode::Changelog => {
-
self.validate_per_snapshot_plan(IncrementalScanMode::Changelog).await?;
- let incremental =
self.plan_changelog(IncrementalScanMode::Changelog, trace).await?;
- let mut splits =
Vec::with_capacity(incremental.splits().len());
- for split in incremental.splits() {
- match split {
- IncrementalSplit::Data(split) =>
splits.push(split.clone().with_snapshot_id(self.end_inclusive)),
- IncrementalSplit::DiffPair { .. } => {
- return Err(crate::Error::UnexpectedError {
- message: "DiffPair appeared in a Changelog
incremental plan"
- .to_string(),
- source: None,
- });
- }
- }
- }
- Ok(Plan::new(splits).with_snapshot_id(self.end_inclusive))
+ let mode = self.resolve_mode();
+ match mode {
+ IncrementalScanMode::Delta | IncrementalScanMode::Changelog => {
+ self.plan_combined_events(mode, trace).await
}
IncrementalScanMode::Diff => Err(crate::Error::Unsupported {
message: "Combined incremental planning does not support Diff
mode; Diff requires before/after split pairs"
@@ -387,39 +373,73 @@ impl<'a> IncrementalScan<'a> {
/// must exist and supplies snapshot metadata. Snapshot deletion vectors
and
/// automatic global-index pruning do not apply to these historical events.
pub async fn plan_combined_delta(&self) -> crate::Result<Plan> {
- self.plan_combined_delta_with_trace(None).await
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ let mode = self.resolve_mode();
+ if mode != IncrementalScanMode::Delta {
+ return Err(crate::Error::Unsupported {
+ message: "Combined incremental planning only supports Delta
mode".to_string(),
+ });
+ }
+ self.plan_combined_events(mode, None).await
}
- async fn plan_combined_delta_with_trace(
+ async fn plan_combined_events(
&self,
+ mode: IncrementalScanMode,
trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
- let mode = self.resolve_mode();
- if mode != IncrementalScanMode::Delta {
+ // Row positions and shuffled chunks refer to one APPEND-delta row
+ // space. Changelog supports Java's file/bucket shard selection only.
+ if mode == IncrementalScanMode::Changelog
+ && (self.scan.has_row_position_selection() ||
self.scan.has_chunk_shuffle())
+ {
return Err(crate::Error::Unsupported {
- message: "Combined incremental planning only supports Delta
mode".to_string(),
+ message: "Changelog batch planning does not support
row-position selection or chunk_shuffle".into(),
});
}
self.validate_snapshot_range(mode).await?;
+ if mode == IncrementalScanMode::Changelog && self.start_exclusive ==
self.end_inclusive {
+ if let Some(trace) = trace {
+ trace.record_final_plan(0, 0, 0);
+ }
+ // Changelog's empty range may end at earliest-1, which is a
+ // valid boundary even though that snapshot need not exist.
+ return
Ok(Plan::new(Vec::new()).with_snapshot_id(self.end_inclusive));
+ }
let end_snapshot = self
.snapshot_manager
.get_snapshot(self.end_inclusive)
.await?;
let mut snapshots = Vec::new();
+ let selected = |snapshot: &crate::spec::Snapshot| match mode {
+ IncrementalScanMode::Delta => snapshot.commit_kind() ==
&CommitKind::APPEND,
+ IncrementalScanMode::Changelog => {
+ snapshot.commit_kind() != &CommitKind::OVERWRITE
+ && snapshot.changelog_manifest_list().is_some()
+ }
+ _ => unreachable!("combined event planning only accepts Delta or
Changelog"),
+ };
for snapshot_id in (self.start_exclusive + 1)..self.end_inclusive {
let snapshot =
self.snapshot_manager.get_snapshot(snapshot_id).await?;
- if snapshot.commit_kind() == &CommitKind::APPEND {
+ if selected(&snapshot) {
snapshots.push(snapshot);
}
}
- if self.start_exclusive < self.end_inclusive
- && end_snapshot.commit_kind() == &CommitKind::APPEND
- {
+ if self.start_exclusive < self.end_inclusive &&
selected(&end_snapshot) {
snapshots.push(end_snapshot.clone());
}
+ let manifest_lists = snapshots
+ .iter()
+ .filter_map(|snapshot| match mode {
+ IncrementalScanMode::Delta =>
Some(snapshot.delta_manifest_list()),
+ IncrementalScanMode::Changelog =>
snapshot.changelog_manifest_list(),
+ _ => unreachable!("combined event planning only accepts Delta
or Changelog"),
+ })
+ .map(str::to_owned)
+ .collect();
self.scan
- .plan_snapshot_deltas_with_trace(&snapshots, &end_snapshot, trace)
+ .plan_incremental_lists_with_trace(&end_snapshot, manifest_lists,
trace)
.await
}
@@ -488,11 +508,7 @@ impl<'a> IncrementalScan<'a> {
IncrementalPlan::try_new(mode, splits)
}
- async fn plan_changelog(
- &self,
- mode: IncrementalScanMode,
- mut trace: Option<&mut ScanTrace>,
- ) -> crate::Result<IncrementalPlan> {
+ async fn plan_changelog(&self, mode: IncrementalScanMode) ->
crate::Result<IncrementalPlan> {
let mut splits = Vec::new();
for snapshot_id in (self.start_exclusive + 1)..=self.end_inclusive {
let snapshot =
self.snapshot_manager.get_snapshot(snapshot_id).await?;
@@ -504,17 +520,10 @@ impl<'a> IncrementalScan<'a> {
if snapshot.changelog_manifest_list().is_none() {
continue;
}
- let mut snapshot_trace = ScanTrace::default();
let plan = self
.scan
- .plan_snapshot_changelog_with_trace(
- &snapshot,
- trace.as_ref().map(|_| &mut snapshot_trace),
- )
+ .plan_snapshot_changelog_with_trace(&snapshot, None)
.await?;
- if let Some(trace) = trace.as_deref_mut() {
- trace.accumulate(&snapshot_trace);
- }
splits.extend(plan.splits().iter().cloned().map(IncrementalSplit::Data));
}
IncrementalPlan::try_new(mode, splits)
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index ce971a17..ea327b31 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -307,9 +307,9 @@ impl<'a> ReadBuilder<'a> {
/// `(start_exclusive, end_inclusive]`.
///
/// Filters and projection configured on this builder are pushed into the
- /// incremental plan (partition / bucket pruning on the delta path).
- /// Use [`IncrementalScan::plan_combined_delta`] to merge APPEND deltas
into
- /// one ordinary batch plan instead of preserving each snapshot separately.
+ /// incremental plan (including partition / bucket pruning).
+ /// Use [`IncrementalScan::plan_combined`] to pack DELTA or CHANGELOG files
+ /// across the range into one batch plan.
pub fn new_incremental_scan(
&self,
mode: IncrementalScanMode,
diff --git a/crates/paimon/src/table/scan_trace.rs
b/crates/paimon/src/table/scan_trace.rs
index 8cd8aa25..45063800 100644
--- a/crates/paimon/src/table/scan_trace.rs
+++ b/crates/paimon/src/table/scan_trace.rs
@@ -59,40 +59,6 @@ pub struct ScanTrace {
}
impl ScanTrace {
- /// Add counters from another independently planned changelog snapshot.
- pub(crate) fn accumulate(&mut self, other: &Self) {
- self.base_manifest_files += other.base_manifest_files;
- self.delta_manifest_files += other.delta_manifest_files;
- self.manifest_files_before_partition_pruning +=
- other.manifest_files_before_partition_pruning;
- self.manifest_files_after_partition_pruning +=
other.manifest_files_after_partition_pruning;
- self.manifest_files_pruned_by_row_ranges +=
other.manifest_files_pruned_by_row_ranges;
- self.manifest_entries_read += other.manifest_entries_read;
- self.manifest_entries_pruned_by_bucket +=
other.manifest_entries_pruned_by_bucket;
- self.manifest_entries_pruned_by_partition +=
other.manifest_entries_pruned_by_partition;
- self.manifest_entries_after_entry_pruning +=
other.manifest_entries_after_entry_pruning;
- self.manifest_entries_pruned_by_level +=
other.manifest_entries_pruned_by_level;
- self.manifest_entries_pruned_by_row_ranges +=
other.manifest_entries_pruned_by_row_ranges;
- self.manifest_entries_pruned_by_data_stats +=
other.manifest_entries_pruned_by_data_stats;
- self.manifest_entries_after_manifest_filters +=
- other.manifest_entries_after_manifest_filters;
- self.manifest_entries_after_merge +=
other.manifest_entries_after_merge;
- self.manifest_entries_pruned_by_cross_schema_stats +=
- other.manifest_entries_pruned_by_cross_schema_stats;
- self.manifest_entries_after_cross_schema_stats +=
- other.manifest_entries_after_cross_schema_stats;
- self.data_evolution_groups_before_stats +=
other.data_evolution_groups_before_stats;
- self.data_evolution_groups_pruned_by_stats +=
other.data_evolution_groups_pruned_by_stats;
- self.data_evolution_groups_pruned_by_row_ranges +=
- other.data_evolution_groups_pruned_by_row_ranges;
- self.split_candidates_built += other.split_candidates_built;
- self.splits_before_limit += other.splits_before_limit;
- self.splits_after_limit += other.splits_after_limit;
- self.final_splits += other.final_splits;
- self.final_files += other.final_files;
- self.limit_early_stopped |= other.limit_early_stopped;
- }
-
pub(crate) fn record_manifest_lists(&mut self, base_count: usize,
delta_count: usize) {
self.base_manifest_files = base_count;
self.delta_manifest_files = delta_count;
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index fd7a8d13..7c106fc1 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -553,10 +553,6 @@ impl DataSplit {
self.snapshot_id
}
- pub(crate) fn with_snapshot_id(mut self, snapshot_id: i64) -> Self {
- self.snapshot_id = snapshot_id;
- self
- }
pub fn partition(&self) -> &BinaryRow {
&self.partition
}
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 53f8bb63..367e0de1 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -197,7 +197,7 @@ async fn read_manifest_bytes_with_sidecar(
fn validate_incremental_entries(entries: &[ManifestEntry]) ->
crate::Result<()> {
if entries.iter().any(|entry| *entry.kind() != FileKind::Add) {
return Err(crate::Error::DataInvalid {
- message: "Incremental delta manifests must contain only ADD
entries".into(),
+ message: "Incremental manifests must contain only ADD
entries".into(),
source: None,
});
}
@@ -212,7 +212,7 @@ enum IncrementalSplitMode {
enum ManifestListSource<'a> {
Snapshot(&'a Snapshot),
- AppendDeltas(&'a [Snapshot]),
+ IncrementalLists(Vec<String>),
IncrementalList(&'a str),
}
@@ -255,11 +255,7 @@ async fn read_all_manifest_entries(
Vec::new(),
read_manifest_list(file_io, table_path, name).await?,
),
- ManifestListSource::AppendDeltas(snapshots) => {
- let names = snapshots
- .iter()
- .map(|snapshot| snapshot.delta_manifest_list().to_string())
- .collect::<Vec<_>>();
+ ManifestListSource::IncrementalLists(names) => {
let delta = futures::stream::iter(names)
.map(|name| async move { read_manifest_list(file_io,
table_path, &name).await })
.buffered(manifest_parallelism)
@@ -1450,19 +1446,19 @@ impl<'a> TableScan<'a> {
}
}
- pub(crate) async fn plan_snapshot_deltas_with_trace(
+ pub(crate) async fn plan_incremental_lists_with_trace(
&self,
- snapshots: &[Snapshot],
end_snapshot: &Snapshot,
+ manifest_lists: Vec<String>,
trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
match &self.0 {
TableScanKind::Paimon(scan) => {
- scan.plan_snapshot_deltas(snapshots, end_snapshot, trace)
+ scan.plan_incremental_lists(end_snapshot, manifest_lists,
trace)
.await
}
TableScanKind::Format(_) => Err(crate::Error::Unsupported {
- message: "Format tables do not support incremental delta
scan".to_string(),
+ message: "Format tables do not support incremental batch
scans".to_string(),
}),
}
}
@@ -2374,12 +2370,13 @@ impl<'a> PaimonTableScan<'a> {
.await
}
- /// Plan all selected APPEND deltas together. Entries are merged before
- /// splitting so versions of a primary key cannot escape into separate
plans.
- async fn plan_snapshot_deltas(
+ /// Pack all selected delta or changelog files as one batch, as Java's
+ /// IncrementalDeltaStartingScanner does. Every physical event is retained;
+ /// sharding, split costs and LIMIT apply to the complete selected range.
+ async fn plan_incremental_lists(
&self,
- snapshots: &[Snapshot],
end_snapshot: &Snapshot,
+ manifest_lists: Vec<String>,
trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
self.validate_read_options()?;
@@ -2388,7 +2385,7 @@ impl<'a> PaimonTableScan<'a> {
scan.incremental_split_mode = Some(IncrementalSplitMode::Batch);
scan.plan_snapshot_from_lists(
end_snapshot,
- ManifestListSource::AppendDeltas(snapshots),
+ ManifestListSource::IncrementalLists(manifest_lists),
data_evolution_read_field_ids.as_ref(),
trace,
)
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index a03eae62..e2a8e203 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -48,7 +48,7 @@ use crate::table::row_kind_generator::RowKindGenerator;
use crate::table::write_batch_normalize::normalize_write_array;
use crate::table::{Snapshot, Table, TableScan};
use crate::Result;
-use arrow_array::{ArrayRef, Int8Array, RecordBatch};
+use arrow_array::{Array, ArrayRef, Int8Array, RecordBatch};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
@@ -875,10 +875,10 @@ impl TableWrite {
let actual_field_count = actual_schema.fields().len();
// Primary-key data writers consume `_VALUE_KIND` independently of
whether
// separate changelog files are enabled. Row-kind generation and
- // cross-partition routing add the column themselves, so callers must
not.
- let allows_value_kind = !self.primary_key_indices.is_empty()
- && self.row_kind_generator.is_none()
- && !matches!(self.bucket_assigner,
BucketAssignerEnum::CrossPartition(_));
+ // cross-partition routing replaces the column after preserving input
+ // kinds and emitting migration DELETEs. A configured generator owns
it.
+ let allows_value_kind =
+ !self.primary_key_indices.is_empty() &&
self.row_kind_generator.is_none();
let includes_value_kind = allows_value_kind && actual_field_count ==
table_field_count + 1;
if actual_field_count != table_field_count && !includes_value_kind {
@@ -1060,7 +1060,13 @@ impl TableWrite {
);
let (rows, row_kinds) = groups.entry(key).or_default();
rows.push(row_idx);
- row_kinds.push(kinds.map_or(RowKind::Insert.to_value(), |kinds|
kinds.value(row_idx)));
+ row_kinds.push(kinds.map_or(RowKind::Insert.to_value(), |kinds| {
+ if kinds.is_null(row_idx) {
+ RowKind::Insert.to_value()
+ } else {
+ kinds.value(row_idx)
+ }
+ }));
}
// Generated row kinds are replaced below; migration DELETEs must not
// append a second _VALUE_KIND column to a rowkind.field batch.
@@ -1121,6 +1127,17 @@ impl TableWrite {
fn generate_rowkind_batch(&self, batch: &RecordBatch) ->
Result<RecordBatch> {
let Some(generator) = &self.row_kind_generator else {
+ // Every PK batch sent to a buffered writer uses the same schema.
+ // Mixing default INSERT batches with explicit kinds must neither
+ // discard DELETEs nor concatenate columns by the wrong position.
+ if !self.primary_key_indices.is_empty()
+ && batch.column_by_name(VALUE_KIND_FIELD_NAME).is_none()
+ {
+ return Self::add_per_row_value_kind_column(
+ batch,
+ vec![RowKind::Insert.to_value(); batch.num_rows()],
+ );
+ }
return Ok(batch.clone());
};
if batch
@@ -4396,6 +4413,77 @@ pub(in crate::table) mod tests {
);
}
+ #[tokio::test]
+ async fn test_pk_writer_mixes_default_and_explicit_row_kinds() {
+ for producer in ["none", "input"] {
+ for row_first in [false, true] {
+ let file_io = test_file_io();
+ let table_path =
format!("memory:/mixed-kinds-{producer}-{row_first}");
+ setup_dirs(&file_io, &table_path).await;
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "mixed_kinds"),
+ table_path,
+ pk_changelog_schema(&[("changelog-producer", producer)]),
+ None,
+ );
+ let mut writer = TableWrite::new(&table,
"test-user".to_owned()).unwrap();
+ let first = if row_first {
+ make_batch_with_value_kind(vec![1], vec![10], vec![0])
+ } else {
+ make_batch(vec![1], vec![10])
+ };
+ let second = if row_first {
+ make_batch(vec![1], vec![20])
+ } else {
+ make_batch_with_value_kind(vec![1], vec![20], vec![3])
+ };
+ writer.write_arrow_batch(&first).await.unwrap();
+ writer.write_arrow_batch(&second).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_owned())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ read_id_value_rows(&table).await,
+ if row_first { vec![(1, 20)] } else { vec![] }
+ );
+ if producer == "input" {
+ let builder = table.new_read_builder();
+ let plan = builder
+
.new_incremental_scan(super::super::IncrementalScanMode::Changelog, 0, 1)
+ .plan_combined()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> =
futures::TryStreamExt::try_collect(
+ builder
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(plan.splits())
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let kinds: Vec<String> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .iter()
+ .map(|kind| kind.unwrap().to_owned())
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ assert_eq!(kinds, vec!["+I", if row_first { "+I" } else {
"-D" }]);
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn test_default_changelog_producer_accepts_value_kind() {
let file_io = test_file_io();
@@ -5387,8 +5475,10 @@ pub(in crate::table) mod tests {
}
#[tokio::test]
- async fn
test_input_changelog_cross_partition_write_rejects_caller_value_kind() {
+ async fn
test_input_changelog_cross_partition_write_preserves_caller_value_kind() {
let file_io = test_file_io();
+ let table_path = "memory:/test_cross_partition_input_kinds";
+ setup_dirs(&file_io, table_path).await;
let schema = Schema::builder()
.column("pt", DataType::VarChar(VarCharType::string_type()))
.column("id", DataType::Int(IntType::new()))
@@ -5400,32 +5490,92 @@ pub(in crate::table) mod tests {
.unwrap();
let table = Table::new(
file_io,
- Identifier::new("default",
"test_cross_partition_schema_validation"),
- "memory:/test_cross_partition_schema_validation".to_string(),
+ Identifier::new("default", "test_cross_partition_input_kinds"),
+ table_path.to_string(),
TableSchema::new(0, &schema),
None,
);
let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
- let error = table_write
- .write_arrow_batch(&make_partitioned_batch_with_value_kind(
- vec!["a"],
- vec![1],
- vec![10],
- vec![0],
- ))
+ let batch = make_partitioned_batch_with_value_kind(
+ vec!["a", "b", "b", "b"],
+ vec![1, 1, 1, 1],
+ vec![10, 20, 30, 40],
+ vec![0, 2, 1, 3],
+ );
+ let mut fields = batch.schema().fields().to_vec();
+ fields[3] = Arc::new(ArrowField::new(
+ VALUE_KIND_FIELD_NAME,
+ ArrowDataType::Int8,
+ true,
+ ));
+ let mut columns = batch.columns().to_vec();
+ // The null slot deliberately carries a DELETE byte in its underlying
+ // buffer. Cross-partition routing must use INSERT for the logical
null.
+ columns[3] = Arc::new(Int8Array::new(
+ vec![3, 2, 1, 3].into(),
+ Some(arrow_buffer::NullBuffer::from(vec![
+ false, true, true, true,
+ ])),
+ ));
+ let batch = RecordBatch::try_new(Arc::new(ArrowSchema::new(fields)),
columns).unwrap();
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
.await
- .unwrap_err();
+ .unwrap();
- assert_data_invalid_contains(
- error,
- &[
- "field count",
- "expected 3",
- "actual 4",
- "<no field>",
- VALUE_KIND_FIELD_NAME,
- ],
+ let builder = table.new_read_builder();
+ let plan = builder
+
.new_incremental_scan(super::super::IncrementalScanMode::Changelog, 0, 1)
+ .plan_combined()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> = futures::TryStreamExt::try_collect(
+ builder
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(plan.splits())
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let mut events = Vec::new();
+ for batch in batches {
+ let kinds = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let partitions = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let values = batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ events.extend((0..batch.num_rows()).map(|i| {
+ (
+ kinds.value(i).to_owned(),
+ partitions.value(i).to_owned(),
+ values.value(i),
+ )
+ }));
+ }
+ events.sort_unstable();
+ assert_eq!(
+ events,
+ vec![
+ ("+I".to_owned(), "a".to_owned(), 10),
+ ("+U".to_owned(), "b".to_owned(), 20),
+ ("-D".to_owned(), "a".to_owned(), 20),
+ ("-D".to_owned(), "b".to_owned(), 40),
+ ("-U".to_owned(), "b".to_owned(), 30),
+ ]
);
}
diff --git a/crates/paimon/tests/incremental_changelog_batch_test.rs
b/crates/paimon/tests/incremental_changelog_batch_test.rs
new file mode 100644
index 00000000..da652283
--- /dev/null
+++ b/crates/paimon/tests/incremental_changelog_batch_test.rs
@@ -0,0 +1,443 @@
+// 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.
+
+mod common;
+
+use arrow_array::{Int32Array, RecordBatch, StringArray};
+use futures::TryStreamExt;
+use paimon::spec::{Datum, PredicateBuilder};
+use paimon::table::{IncrementalScanMode, Plan, Table};
+
+use common::incremental_helpers::{
+ make_batch, make_batch_with_kinds, memory_table, persist_table_schema,
pk_schema, setup_dirs,
+ write_batch,
+};
+
+async fn table(name: &str, bucket: &str, dv: bool, split_size: &str) -> Table {
+ let path = format!("memory:/changelog-batch/{name}");
+ let (file_io, table) = memory_table(
+ &path,
+ pk_schema(&[
+ ("changelog-producer", "input"),
+ ("bucket", bucket),
+ (
+ "deletion-vectors.enabled",
+ if dv { "true" } else { "false" },
+ ),
+ ("source.split.target-size", split_size),
+ ("source.split.open-file-cost", "1b"),
+ ]),
+ );
+ setup_dirs(&file_io, &path).await;
+ persist_table_schema(&file_io, &path, table.schema()).await;
+ table
+}
+
+async fn events(table: &Table, plan: &Plan) -> Vec<(String, i32, i32)> {
+ let batches: Vec<RecordBatch> = table
+ .new_read_builder()
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in batches {
+ let kinds = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ rows.extend((0..batch.num_rows()).map(|row| {
+ (
+ kinds.value(row).to_owned(),
+ ids.value(row),
+ values.value(row),
+ )
+ }));
+ }
+ rows.sort_unstable();
+ rows
+}
+
+#[tokio::test]
+async fn combined_changelog_packs_overlapping_versions_across_snapshots() {
+ let table = table("overlapping", "1", false, "1mb").await;
+ for (value, kind) in [(10, 0), (11, 2), (12, 1), (13, 3)] {
+ write_batch(
+ &table,
+ &make_batch_with_kinds(vec![1], vec![value], vec![kind]),
+ )
+ .await;
+ }
+ let builder = table.new_read_builder();
+ let per_snapshot = builder
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 4)
+ .plan()
+ .await
+ .unwrap();
+ assert_eq!(per_snapshot.splits().len(), 4);
+ for mode in [IncrementalScanMode::Changelog, IncrementalScanMode::Auto] {
+ let plan = builder
+ .new_incremental_scan(mode, 0, 4)
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(4));
+ assert_eq!(plan.splits().len(), 1, "batch packing must span
snapshots");
+ let split = &plan.splits()[0];
+ assert_eq!(split.snapshot_id(), 4);
+ assert!(split.is_streaming());
+ assert_eq!(split.data_files().len(), 4);
+ assert!(split
+ .data_files()
+ .iter()
+ .all(|file| file.file_name.starts_with("changelog-")));
+ assert_eq!(
+ events(&table, &plan).await,
+ vec![
+ ("+I".into(), 1, 10),
+ ("+U".into(), 1, 11),
+ ("-D".into(), 1, 13),
+ ("-U".into(), 1, 12),
+ ]
+ );
+ }
+}
+
+#[tokio::test]
+async fn combined_changelog_shards_cover_every_physical_event_once() {
+ for dv in [false, true] {
+ let table = table(&format!("shards-{dv}"), "4", dv, "1mb").await;
+ for version in 0..3 {
+ write_batch(&table, &make_batch((0..32).collect(), vec![version;
32])).await;
+ }
+ let builder = table.new_read_builder();
+ let full = builder
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 3)
+ .plan_combined()
+ .await
+ .unwrap();
+ let expected = events(&table, &full).await;
+ assert_eq!(expected.len(), 96);
+ for mode in [IncrementalScanMode::Changelog,
IncrementalScanMode::Auto] {
+ for count in [2, 5] {
+ let mut union = Vec::new();
+ let mut file_names = std::collections::HashSet::new();
+ for index in 0..count {
+ let plan = builder
+ .new_incremental_scan(mode, 0, 3)
+ .with_shard(index, count)
+ .unwrap()
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(3));
+ for split in plan.splits() {
+ assert!(split.is_streaming());
+ assert_eq!(split.snapshot_id(), 3);
+ if !dv {
+ assert_eq!(split.bucket() as usize % count, index);
+ }
+ for file in split.data_files() {
+ assert!(
+ file_names.insert(file.file_name.clone()),
+ "a file belongs to exactly one shard"
+ );
+ }
+ }
+ union.extend(events(&table, &plan).await);
+ }
+ union.sort_unstable();
+ assert_eq!(union, expected, "dv={dv}, mode={mode:?},
count={count}");
+ assert_eq!(
+ file_names.len(),
+ full.splits()
+ .iter()
+ .map(|split| split.data_files().len())
+ .sum::<usize>()
+ );
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn combined_changelog_applies_limit_once_to_the_whole_batch() {
+ let table = table("limit", "1", false, "1b").await;
+ for id in 0..4 {
+ write_batch(&table, &make_batch(vec![id], vec![id * 10])).await;
+ }
+ for limit in [0, 1, 2, 5] {
+ let mut builder = table.new_read_builder();
+ builder.with_limit(limit);
+ let plan = builder
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 4)
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(4));
+ assert_eq!(plan.splits().len(), limit.min(4));
+ assert_eq!(events(&table, &plan).await.len(), limit.min(4));
+ }
+ // Value predicates remain residual filters over physical events; early
+ // LIMIT must not discard a later matching event.
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .greater_than("value", Datum::Int(15))
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_filter(predicate).with_limit(1);
+ let plan = builder
+ .new_incremental_scan(IncrementalScanMode::Auto, 0, 4)
+ .plan_combined()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let value = batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0);
+ assert!(value > 15);
+}
+
+async fn patch_snapshot(table: &Table, id: i64, key: &str, value:
serde_json::Value) {
+ let path = table.snapshot_manager().snapshot_path(id);
+ let bytes = table
+ .file_io()
+ .new_input(&path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let mut snapshot: serde_json::Value =
serde_json::from_slice(&bytes).unwrap();
+ snapshot[key] = value;
+ table
+ .file_io()
+ .new_output(&path)
+ .unwrap()
+ .write(serde_json::to_vec(&snapshot).unwrap().into())
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn
combined_changelog_selects_compact_events_but_skips_overwrite_and_missing_lists()
{
+ let table = table("commit-selection", "1", false, "1mb").await;
+ for id in 1..=4 {
+ write_batch(&table, &make_batch(vec![id], vec![id * 10])).await;
+ }
+ patch_snapshot(&table, 2, "commitKind", "COMPACT".into()).await;
+ patch_snapshot(&table, 3, "commitKind", "OVERWRITE".into()).await;
+ patch_snapshot(&table, 4, "changelogManifestList",
serde_json::Value::Null).await;
+ let builder = table.new_read_builder();
+ for mode in [IncrementalScanMode::Changelog, IncrementalScanMode::Auto] {
+ let plan = builder
+ .new_incremental_scan(mode, 0, 4)
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(4));
+ assert!(plan.splits().iter().all(|split| split.snapshot_id() == 4));
+ assert_eq!(
+ events(&table, &plan).await,
+ vec![("+I".into(), 1, 10), ("+I".into(), 2, 20)]
+ );
+ let last = builder
+ .new_incremental_scan(mode, 2, 4)
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(last.snapshot_id(), Some(4));
+ assert!(last.splits().is_empty());
+ }
+ let per_snapshot = builder
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 4)
+ .plan()
+ .await
+ .unwrap();
+ let ids: Vec<_> = per_snapshot
+ .splits()
+ .iter()
+ .map(|split| match split {
+ paimon::table::IncrementalSplit::Data(split) =>
split.snapshot_id(),
+ _ => panic!("changelog must not contain diff pairs"),
+ })
+ .collect();
+ assert_eq!(ids, vec![1, 2]);
+}
+
+#[tokio::test]
+async fn combined_changelog_preserves_empty_boundaries_with_shards() {
+ let table = table("empty-boundaries", "1", false, "1mb").await;
+ for id in 1..=3 {
+ write_batch(&table, &make_batch(vec![id], vec![id * 10])).await;
+ }
+ let builder = table.new_read_builder();
+ for mode in [IncrementalScanMode::Changelog, IncrementalScanMode::Auto] {
+ for id in [0, 1, 3] {
+ let plan = builder
+ .new_incremental_scan(mode, id, id)
+ .with_shard(1, 3)
+ .unwrap()
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(id));
+ assert!(plan.splits().is_empty());
+ }
+ for (start, end) in [(-1, 0), (4, 4), (3, 2)] {
+ assert!(matches!(
+ builder
+ .new_incremental_scan(mode, start, end)
+ .with_shard(1, 3)
+ .unwrap()
+ .plan_combined()
+ .await,
+ Err(paimon::Error::DataInvalid { .. })
+ ));
+ }
+ }
+ // Delta retains its existing requirement for an actual ending snapshot.
+ assert!(builder
+ .new_incremental_scan(IncrementalScanMode::Delta, 0, 0)
+ .plan_combined_delta()
+ .await
+ .is_err());
+}
+
+#[tokio::test]
+async fn
combined_changelog_sharding_keeps_residual_filters_before_reader_limit() {
+ let table = table("sharded-residual", "4", false, "1b").await;
+ for value in 0..3 {
+ write_batch(&table, &make_batch((0..32).collect(), vec![value;
32])).await;
+ }
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .greater_than("value", Datum::Int(0))
+ .unwrap();
+ let mut read = table.new_read_builder();
+ read.with_filter(predicate);
+ read.with_projection(&["value", "id"]).unwrap();
+ let mut all_candidates = 0;
+ for index in 0..5 {
+ let plan = read
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 3)
+ .with_shard(index, 5)
+ .unwrap()
+ .plan_combined()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let candidates =
batches.iter().map(RecordBatch::num_rows).sum::<usize>();
+ all_candidates += candidates;
+ let mut limited = read.clone();
+ limited.with_limit(1);
+ let plan = limited
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 3)
+ .with_shard(index, 5)
+ .unwrap()
+ .plan_combined()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> = limited
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ candidates.min(1)
+ );
+ for batch in batches {
+ assert_eq!(batch.schema().fields()[0].name(), "value");
+ assert_eq!(batch.schema().fields()[1].name(), "id");
+ let values = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert!((0..batch.num_rows()).all(|row| values.value(row) > 0));
+ }
+ }
+ assert_eq!(all_candidates, 64);
+}
+
+#[tokio::test]
+async fn timestamp_changelog_trace_describes_one_packed_batch() {
+ let table = table("trace", "1", false, "1mb").await;
+ for id in 1..=3 {
+ write_batch(&table, &make_batch(vec![1], vec![id])).await;
+ patch_snapshot(&table, i64::from(id), "timeMillis", (id *
100).into()).await;
+ }
+ let table = table.copy_with_options(std::collections::HashMap::from([
+ ("scan.mode".into(), "incremental".into()),
+ ("incremental-between-timestamp".into(), "0,300".into()),
+ ]));
+ let (plan, trace) = table
+ .new_read_builder()
+ .new_scan()
+ .plan_with_trace()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(3));
+ assert_eq!(trace.snapshot_id, Some(3));
+ assert_eq!(trace.base_manifest_files, 0);
+ assert_eq!(trace.delta_manifest_files, 3);
+ assert_eq!(trace.manifest_entries_read, 3);
+ assert_eq!(trace.final_splits, 1);
+ assert_eq!(trace.final_files, 3);
+ assert_eq!(
+ trace.planned_data_file_bytes,
+ plan.splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .map(|file| file.file_size as u64)
+ .sum::<u64>()
+ );
+ assert_eq!(events(&table, &plan).await.len(), 3);
+}
diff --git a/crates/paimon/tests/writer_resources_test.rs
b/crates/paimon/tests/writer_resources_test.rs
index 02bd6711..264ece12 100644
--- a/crates/paimon/tests/writer_resources_test.rs
+++ b/crates/paimon/tests/writer_resources_test.rs
@@ -17,11 +17,13 @@
mod common;
+use arrow_array::{Array, Int8Array, RecordBatch};
use common::incremental_helpers::{
- make_batch, make_partitioned_batch, memory_table, partitioned_pk_schema,
pk_schema, setup_dirs,
+ make_batch, make_batch_with_kinds, make_partitioned_batch, memory_table,
partitioned_pk_schema,
+ pk_schema, setup_dirs,
};
use paimon::resource::{MemoryPool, ResourceContext};
-use paimon::spec::{DataType, IntType, Schema, TableSchema};
+use paimon::spec::{DataType, IntType, RowKind, Schema, TableSchema,
VALUE_KIND_FIELD_NAME};
use paimon::Error;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
@@ -51,17 +53,34 @@ async fn assert_no_data_files(io: &paimon::io::FileIO,
path: &str) {
);
}
+fn buffered_pk_bytes(batch: &RecordBatch) -> usize {
+ let bytes = batch
+ .columns()
+ .iter()
+ .map(|column| column.get_buffer_memory_size())
+ .sum::<usize>();
+ // PK input normalization adds INSERT kinds when callers omit the column.
+ if batch.column_by_name(VALUE_KIND_FIELD_NAME).is_none() {
+ bytes
+ + Int8Array::from(vec![RowKind::Insert.to_value();
batch.num_rows()])
+ .get_buffer_memory_size()
+ } else {
+ bytes
+ }
+}
+
#[tokio::test]
async fn key_value_buffer_shares_limit_and_releases_on_failure() {
+ assert_key_value_buffer_limit(make_batch(vec![1, 2], vec![10, 20])).await;
+ assert_key_value_buffer_limit(make_batch_with_kinds(vec![1, 2], vec![10,
20], vec![0, 2]))
+ .await;
+}
+
+async fn assert_key_value_buffer_limit(batch: RecordBatch) {
let path = "memory:/writer_resources_pk";
let (io, table) = memory_table(path, pk_schema(&[("file.format",
"parquet")]));
setup_dirs(&io, path).await;
- let batch = make_batch(vec![1, 2], vec![10, 20]);
- let bytes: usize = batch
- .columns()
- .iter()
- .map(|column| column.get_buffer_memory_size())
- .sum();
+ let bytes = buffered_pk_bytes(&batch);
let resources = ResourceContext::builder()
.memory_limit(bytes)
.build()
@@ -273,11 +292,7 @@ async fn
key_value_flush_rejection_prevents_partial_commit() {
);
setup_dirs(&io, path).await;
let batch = make_batch(vec![1, 2], vec![10, 20]);
- let bytes: usize = batch
- .columns()
- .iter()
- .map(|column| column.get_buffer_memory_size())
- .sum();
+ let bytes = buffered_pk_bytes(&batch);
let resources = ResourceContext::builder()
.memory_limit(bytes)
.build()
@@ -291,6 +306,8 @@ async fn
key_value_flush_rejection_prevents_partial_commit() {
write.write_arrow_batch(&batch).await,
Err(Error::ResourceExhausted { .. })
));
+ // The batch must be admitted before its triggered flush exhausts the pool.
+ assert_eq!(resources.metrics().peak_reserved_memory_bytes, bytes);
assert!(write.prepare_commit().await.is_err());
assert_eq!(resources.metrics().reserved_memory_bytes, 0);
assert_no_data_files(&io, path).await;
@@ -302,11 +319,7 @@ async fn
key_value_prepare_rejection_cleans_uncommitted_file() {
let (io, table) = memory_table(path, pk_schema(&[("file.format",
"parquet")]));
setup_dirs(&io, path).await;
let batch = make_batch(vec![1, 2], vec![10, 20]);
- let bytes: usize = batch
- .columns()
- .iter()
- .map(|column| column.get_buffer_memory_size())
- .sum();
+ let bytes = buffered_pk_bytes(&batch);
let resources = ResourceContext::builder()
.memory_limit(bytes)
.build()
