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 8c3527be Align row-tracking commit metadata and conflicts with Java
(#1029)
8c3527be is described below
commit 8c3527be61884d0ebee35a7491c020d25f796d0b
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 4 22:26:45 2026 +0800
Align row-tracking commit metadata and conflicts with Java (#1029)
---
crates/paimon/src/spec/core_options.rs | 15 +
crates/paimon/src/table/table_commit.rs | 171 +++----
.../paimon/src/table/table_commit/row_tracking.rs | 141 ++++++
.../src/table/table_commit/row_tracking_tests.rs | 564 +++++++++++++++++++++
4 files changed, 783 insertions(+), 108 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 8aabb1d3..1775b5b7 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -85,6 +85,8 @@ const DEFAULT_METADATA_STATS_KEEP_FIRST_N_COLUMNS: i32 = -1;
const FIELDS_PREFIX: &str = "fields";
const STATS_MODE_SUFFIX: &str = "stats-mode";
const ROW_TRACKING_ENABLED_OPTION: &str = "row-tracking.enabled";
+const ROW_TRACKING_PARTITION_GROUP_ON_COMMIT_OPTION: &str =
+ "row-tracking.partition-group-on-commit";
const CLUSTERING_INCREMENTAL_OPTION: &str = "clustering.incremental";
pub(crate) const TABLE_TYPE_OPTION: &str = "type";
@@ -218,6 +220,9 @@ pub enum ChangelogProducer {
pub enum GlobalIndexColumnUpdateAction {
ThrowError,
DropPartitionIndex,
+ /// Preserve existing indexes. As in Java, the caller must refresh affected
+ /// index ranges separately before relying on their updated values.
+ Ignore,
}
/// Search mode for global index queries.
@@ -1130,6 +1135,7 @@ impl<'a> CoreOptions<'a> {
{
"THROW_ERROR" => Ok(GlobalIndexColumnUpdateAction::ThrowError),
"DROP_PARTITION_INDEX" =>
Ok(GlobalIndexColumnUpdateAction::DropPartitionIndex),
+ "IGNORE" => Ok(GlobalIndexColumnUpdateAction::Ignore),
other => Err(crate::Error::ConfigInvalid {
message: format!("Unsupported
global-index.column-update-action: {other}"),
}),
@@ -1353,6 +1359,15 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(false)
}
+ /// Group new file metadata by partition before assigning row IDs. Defaults
+ /// to true, matching Java, so each partition receives contiguous row IDs.
+ pub fn row_tracking_partition_group_on_commit(&self) -> bool {
+ self.options
+ .get(ROW_TRACKING_PARTITION_GROUP_ON_COMMIT_OPTION)
+ .map(|v| v.eq_ignore_ascii_case("true"))
+ .unwrap_or(true)
+ }
+
/// Whether incremental clustering is enabled. Default is false.
pub fn clustering_incremental_enabled(&self) -> bool {
self.options
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 9b2f48e9..78f17710 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -47,6 +47,9 @@ use std::time::{SystemTime, UNIX_EPOCH};
const BATCH_COMMIT_IDENTIFIER: i64 = i64::MAX;
/// Java RollingFileWriter.CHECK_ROLLING_RECORD_CNT.
const CHECK_ROLLING_RECORD_COUNT: usize = 1000;
+
+mod row_tracking;
+
const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
type PartitionBucketKey = (Vec<u8>, i32);
@@ -1208,10 +1211,16 @@ impl TableCommit {
.and_then(|s| s.next_row_id())
.or(Some(first_row_id_start));
} else {
- let (assigned, nrid) = self.assign_row_tracking_meta(
+ let mut entries = std::mem::take(&mut resolved.entries);
+ if CoreOptions::new(self.table.schema().options())
+ .row_tracking_partition_group_on_commit()
+ {
+ entries = row_tracking::group_by_partition(entries);
+ }
+ let (assigned, nrid) = row_tracking::assign_row_tracking(
new_snapshot_id,
first_row_id_start,
- std::mem::take(&mut resolved.entries),
+ entries,
)?;
resolved.entries = assigned;
next_row_id = Some(nrid);
@@ -1998,6 +2007,7 @@ impl TableCommit {
}
match
CoreOptions::new(self.table.schema().options()).global_index_column_update_action()?
{
+ GlobalIndexColumnUpdateAction::Ignore => Ok(vec![]),
GlobalIndexColumnUpdateAction::DropPartitionIndex => Ok(affected
.into_iter()
.map(|entry| IndexManifestEntry {
@@ -2648,7 +2658,10 @@ impl TableCommit {
let mut existing_index: HashSet<(Vec<u8>, i32, i64, i64)> =
HashSet::new();
let mut existing_ranges: ExistingRowIdRanges = HashMap::new();
- for base in base_entries {
+ for base in base_entries
+ .iter()
+ .filter(|entry| !is_dedicated_storage_file(entry.file()))
+ {
if let Some(first_row_id) = base.file().first_row_id {
existing_index.insert((
base.partition().to_vec(),
@@ -2656,12 +2669,10 @@ impl TableCommit {
first_row_id,
base.file().row_count,
));
- if !is_dedicated_storage_file(base.file()) {
- existing_ranges
- .entry((base.partition().to_vec(), base.bucket()))
- .or_default()
- .push((first_row_id, first_row_id +
base.file().row_count - 1));
- }
+ existing_ranges
+ .entry((base.partition().to_vec(), base.bucket()))
+ .or_default()
+ .push((first_row_id, first_row_id + base.file().row_count
- 1));
}
}
@@ -2669,14 +2680,14 @@ impl TableCommit {
let first_row_id = entry.file().first_row_id.unwrap();
if is_dedicated_storage_file(entry.file()) {
if let Some((start, end)) = entry.file().row_id_range() {
- let overlaps_existing = existing_ranges
+ let covered_by_existing = existing_ranges
.get(&(entry.partition().to_vec(), entry.bucket()))
.is_some_and(|ranges| {
ranges.iter().any(|&(base_start, base_end)| {
- ranges_overlap(start, end, base_start,
base_end)
+ base_start <= start && end <= base_end
})
});
- if overlaps_existing {
+ if covered_by_existing {
continue;
}
}
@@ -2746,6 +2757,43 @@ impl TableCommit {
}
}
}
+ Self::check_dedicated_row_id_ranges(&entries, commit_entries)
+ }
+
+ /// Dedicated files can roll independently, but each must fit in one normal
+ /// file range. Adjacent ranges must not be merged: the DE reader joins one
+ /// normal group at a time, matching Java's RowRangeIndex with
mergeAdjacent=false.
+ fn check_dedicated_row_id_ranges(
+ normal_entries: &[&ManifestEntry],
+ entries: &[ManifestEntry],
+ ) -> Result<()> {
+ for entry in entries
+ .iter()
+ .filter(|entry| is_dedicated_storage_file(entry.file()))
+ {
+ let Some((start, end)) = entry.file().row_id_range() else {
+ continue;
+ };
+ let covered = normal_entries.iter().any(|normal| {
+ normal.partition() == entry.partition()
+ && normal.bucket() == entry.bucket()
+ && normal
+ .file()
+ .row_id_range()
+ .is_some_and(|(base_start, base_end)| {
+ base_start <= start && end <= base_end
+ })
+ });
+ if !covered {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "For Data Evolution table, dedicated file '{}'
[{start}, {end}] is not covered by one data file range.",
+ entry.file().file_name,
+ ),
+ source: None,
+ });
+ }
+ }
Ok(())
}
@@ -2797,9 +2845,8 @@ impl TableCommit {
for entry in self
.read_delta_entries(partition_filter.as_ref(), &snapshot)
.await?
- .into_iter()
- .filter(|entry| *entry.kind() == FileKind::Add)
{
+ // Java's DML checker treats deleted column ranges as writes
too.
let Some((start, end)) = entry.file().row_id_range() else {
continue;
};
@@ -2829,10 +2876,7 @@ impl TableCommit {
delta_entries: &[ManifestEntry],
) -> Result<Vec<RowIdWriteRange>> {
let mut ranges = Vec::new();
- for entry in delta_entries
- .iter()
- .filter(|entry| *entry.kind() == FileKind::Add)
- {
+ for entry in delta_entries {
let Some((start, end)) = entry.file().row_id_range() else {
continue;
};
@@ -2901,96 +2945,6 @@ impl TableCommit {
Ok(ids)
}
- /// Assign row tracking metadata: snapshot ID as sequence number, and
- /// first_row_id for new APPEND files that don't already have one.
- /// Normal files advance the main counter. Blob files (identified by file
name)
- /// use per-column counters starting from the same base, since each blob
column
- /// rolls independently.
- fn assign_row_tracking_meta(
- &self,
- snapshot_id: i64,
- first_row_id_start: i64,
- entries: Vec<ManifestEntry>,
- ) -> Result<(Vec<ManifestEntry>, i64)> {
- let mut result = Vec::with_capacity(entries.len());
- let mut start = first_row_id_start;
- let mut blob_start_default = first_row_id_start;
- let mut blob_starts: HashMap<String, i64> = HashMap::new();
- let mut vector_store_start = first_row_id_start;
-
- for entry in entries {
- let mut entry = entry.with_sequence_number(snapshot_id,
snapshot_id);
- if entry.file().file_source.is_none() {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "file_source must be present for row-tracking table,
file={}",
- entry.file().file_name
- ),
- source: None,
- });
- }
- let contains_row_id =
- entry.file().write_cols.as_ref().is_some_and(|cols| {
- cols.iter().any(|col| col ==
crate::spec::ROW_ID_FIELD_NAME)
- });
- if *entry.kind() == FileKind::Add
- && entry.file().file_source == Some(0) // APPEND
- && entry.file().first_row_id.is_none()
- && !contains_row_id
- {
- if is_blob_data_file(entry.file()) {
- let blob_field_name = entry
- .file()
- .write_cols
- .as_ref()
- .and_then(|cols| cols.first())
- .cloned()
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!(
- "Blob file '{}' must have write_cols for
row-tracking assignment.",
- entry.file().file_name
- ),
- source: None,
- })?;
- let blob_start = blob_starts
- .entry(blob_field_name)
- .or_insert(blob_start_default);
- if *blob_start >= start {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "This is a bug, blobStart {} should be less
than start {} when assigning a blob entry file.",
- *blob_start, start
- ),
- source: None,
- });
- }
- entry = entry.with_first_row_id(*blob_start);
- *blob_start += entry.file().row_count;
- } else if is_vector_store_file(entry.file()) {
- if vector_store_start >= start {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "This is a bug, vectorStoreStart {} should be
less than start {} when assigning a vector-store entry file.",
- vector_store_start, start
- ),
- source: None,
- });
- }
- entry = entry.with_first_row_id(vector_store_start);
- vector_store_start += entry.file().row_count;
- } else {
- entry = entry.with_first_row_id(start);
- blob_start_default = start;
- blob_starts.clear();
- start += entry.file().row_count;
- }
- }
- result.push(entry);
- }
-
- Ok((result, start))
- }
-
/// Validate that files with pre-assigned `first_row_id` (e.g.
partial-column
/// files from MERGE INTO) still match existing files in the current
snapshot.
///
@@ -3651,6 +3605,7 @@ mod tests {
use super::*;
include!("table_commit/parity_tests.rs");
include!("table_commit/recovery_tests.rs");
+ include!("table_commit/row_tracking_tests.rs");
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_commit/row_tracking.rs
b/crates/paimon/src/table/table_commit/row_tracking.rs
new file mode 100644
index 00000000..45dacf99
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/row_tracking.rs
@@ -0,0 +1,141 @@
+// 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.
+
+//! Row tracking metadata assignment, following Java RowTrackingCommitUtils.
+
+use std::collections::HashMap;
+
+use super::{is_blob_data_file, is_vector_store_file};
+use crate::spec::{FileKind, ManifestEntry};
+use crate::Result;
+
+/// Group partitions without changing the file order within a partition. This
+/// keeps each partition's new row IDs contiguous and preserves Blob alignment.
+pub(super) fn group_by_partition(entries: Vec<ManifestEntry>) ->
Vec<ManifestEntry> {
+ let mut partitions = indexmap::IndexMap::<Vec<u8>,
Vec<ManifestEntry>>::new();
+ for entry in entries {
+ partitions
+ .entry(entry.partition().to_vec())
+ .or_default()
+ .push(entry);
+ }
+ partitions.into_values().flatten().collect()
+}
+
+/// Zero is the pending-snapshot sentinel. A rewritten file can contain both
+/// unmodified and modified records, so its minimum must survive the commit.
+fn assign_snapshot_id(snapshot_id: i64, entry: ManifestEntry) -> ManifestEntry
{
+ let min = entry.file().min_sequence_number;
+ let max = entry.file().max_sequence_number;
+ if min == 0 {
+ entry.with_sequence_number(snapshot_id, snapshot_id)
+ } else if max == 0 {
+ entry.with_sequence_number(min, snapshot_id)
+ } else {
+ entry
+ }
+}
+
+/// Assign row tracking metadata: snapshot ID as sequence number, and
+/// first_row_id for new APPEND files that don't already have one.
+/// Normal files advance the main counter. Blob files (identified by file name)
+/// use per-column counters starting from the same base, since each blob column
+/// rolls independently.
+pub(super) fn assign_row_tracking(
+ snapshot_id: i64,
+ first_row_id_start: i64,
+ entries: Vec<ManifestEntry>,
+) -> Result<(Vec<ManifestEntry>, i64)> {
+ let mut result = Vec::with_capacity(entries.len());
+ let mut start = first_row_id_start;
+ let mut blob_start_default = first_row_id_start;
+ let mut blob_starts: HashMap<String, i64> = HashMap::new();
+ let mut vector_store_start = first_row_id_start;
+
+ for entry in entries {
+ let mut entry = assign_snapshot_id(snapshot_id, entry);
+ if entry.file().file_source.is_none() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "file_source must be present for row-tracking table,
file={}",
+ entry.file().file_name
+ ),
+ source: None,
+ });
+ }
+ let contains_row_id = entry
+ .file()
+ .write_cols
+ .as_ref()
+ .is_some_and(|cols| cols.iter().any(|col| col ==
crate::spec::ROW_ID_FIELD_NAME));
+ if *entry.kind() == FileKind::Add
+ && entry.file().file_source == Some(0) // APPEND
+ && entry.file().first_row_id.is_none()
+ && !contains_row_id
+ {
+ if is_blob_data_file(entry.file()) {
+ let blob_field_name = entry
+ .file()
+ .write_cols
+ .as_ref()
+ .and_then(|cols| cols.first())
+ .cloned()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' must have write_cols for
row-tracking assignment.",
+ entry.file().file_name
+ ),
+ source: None,
+ })?;
+ let blob_start = blob_starts
+ .entry(blob_field_name)
+ .or_insert(blob_start_default);
+ if *blob_start >= start {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "This is a bug, blobStart {} should be less than
start {} when assigning a blob entry file.",
+ *blob_start, start
+ ),
+ source: None,
+ });
+ }
+ entry = entry.with_first_row_id(*blob_start);
+ *blob_start += entry.file().row_count;
+ } else if is_vector_store_file(entry.file()) {
+ if vector_store_start >= start {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "This is a bug, vectorStoreStart {} should be less
than start {} when assigning a vector-store entry file.",
+ vector_store_start, start
+ ),
+ source: None,
+ });
+ }
+ entry = entry.with_first_row_id(vector_store_start);
+ vector_store_start += entry.file().row_count;
+ } else {
+ entry = entry.with_first_row_id(start);
+ blob_start_default = start;
+ blob_starts.clear();
+ start += entry.file().row_count;
+ }
+ }
+ result.push(entry);
+ }
+
+ Ok((result, start))
+}
diff --git a/crates/paimon/src/table/table_commit/row_tracking_tests.rs
b/crates/paimon/src/table/table_commit/row_tracking_tests.rs
new file mode 100644
index 00000000..611d023d
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/row_tracking_tests.rs
@@ -0,0 +1,564 @@
+// 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.
+
+// Persisted metadata must follow Java RowTrackingCommitUtils, including
entries
+// written by other engines with sequence numbers that are already assigned.
+#[tokio::test]
+async fn row_tracking_sequence_assignment_preserves_existing_versions() {
+ let io = test_file_io();
+ let path = "memory:/row-tracking-sequences";
+ setup_dirs(&io, path).await;
+ let commit = setup_data_evolution_commit(&io, path);
+ let mut initial = test_data_file("initial.parquet", 10);
+ initial.file_source = Some(0);
+ commit
+ .commit(vec![CommitMessage::new(
+ EMPTY_SERIALIZED_ROW.clone(),
+ 0,
+ vec![initial],
+ )])
+ .await
+ .unwrap();
+
+ for (index, min, max, expected_min, expected_max) in [
+ (2, 0, 0, 2, 2),
+ (3, 1, 0, 1, 3),
+ (4, 1, 2, 1, 2),
+ (5, 0, 2, 5, 5),
+ ] {
+ let mut file = test_data_file(&format!("partial-{index}.parquet"), 10);
+ file.file_source = Some(0);
+ file.first_row_id = Some(0);
+ file.write_cols = Some(vec!["name".into()]);
+ file.min_sequence_number = min;
+ file.max_sequence_number = max;
+ file.column_max_sequence_numbers = Some(vec![max]);
+ let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![file]);
+ message.check_from_snapshot = Some(index - 1);
+ commit.commit(vec![message]).await.unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.id(), index);
+ assert_eq!(snapshot.next_row_id(), Some(10));
+ let entries = commit.read_delta_entries(None,
&snapshot).await.unwrap();
+ let file = entries[0].file();
+ assert_eq!(
+ (file.min_sequence_number, file.max_sequence_number),
+ (expected_min, expected_max)
+ );
+ assert_eq!(file.first_row_id, Some(0));
+ assert_eq!(file.column_max_sequence_numbers, Some(vec![max]));
+ }
+}
+
+#[tokio::test]
+async fn row_tracking_overwrite_retains_deleted_file_sequences() {
+ let io = test_file_io();
+ let path = "memory:/row-tracking-overwrite-sequences";
+ setup_dirs(&io, path).await;
+ let commit = setup_row_tracking_commit(&io, path);
+ let mut file = test_data_file("initial.parquet", 10);
+ file.file_source = Some(0);
+ commit
+ .commit(vec![CommitMessage::new(
+ EMPTY_SERIALIZED_ROW.clone(),
+ 0,
+ vec![file],
+ )])
+ .await
+ .unwrap();
+ let mut replacement = test_data_file("replacement.parquet", 4);
+ replacement.file_source = Some(0);
+ commit
+ .overwrite(
+ vec![CommitMessage::new(
+ EMPTY_SERIALIZED_ROW.clone(),
+ 0,
+ vec![replacement],
+ )],
+ None,
+ )
+ .await
+ .unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.next_row_id(), Some(14));
+ let entries = commit.read_delta_entries(None, &snapshot).await.unwrap();
+ assert_eq!(entries.len(), 2);
+ for entry in entries {
+ let expected = if *entry.kind() == FileKind::Delete {
+ (1, 0)
+ } else {
+ (2, 10)
+ };
+ assert_eq!(entry.file().min_sequence_number, expected.0);
+ assert_eq!(entry.file().max_sequence_number, expected.0);
+ assert_eq!(entry.file().first_row_id, Some(expected.1));
+ }
+}
+
+#[tokio::test]
+async fn row_tracking_partition_grouping_follows_java_option() {
+ for enabled in [None, Some(true), Some(false)] {
+ let io = test_file_io();
+ let path = format!("memory:/row-tracking-partition-{enabled:?}");
+ setup_dirs(&io, &path).await;
+ let mut options = HashMap::from([("row-tracking.enabled".to_string(),
"true".to_string())]);
+ if let Some(enabled) = enabled {
+ options.insert(
+ "row-tracking.partition-group-on-commit".into(),
+ enabled.to_string(),
+ );
+ }
+ let table = test_partitioned_table(&io,
&path).copy_with_options(options);
+ let commit = TableCommit::new(table, "test-user".into());
+ let messages = [("a", "a-1", 2), ("b", "b-1", 3), ("a", "a-2", 4)]
+ .into_iter()
+ .map(|(partition, name, count)| {
+ let mut file = test_data_file(&format!("{name}.parquet"),
count);
+ file.file_source = Some(0);
+ CommitMessage::new(partition_bytes(partition), 0, vec![file])
+ })
+ .collect();
+ commit.commit(messages).await.unwrap();
+ let snapshot = latest_snapshot(&io, &path).await.unwrap();
+ assert_eq!(snapshot.next_row_id(), Some(9));
+ let entries = commit.read_delta_entries(None,
&snapshot).await.unwrap();
+ let actual: Vec<_> = entries
+ .iter()
+ .map(|e| (e.file().file_name.as_str(), e.file().first_row_id))
+ .collect();
+ let expected = if enabled == Some(false) {
+ vec![
+ ("a-1.parquet", Some(0)),
+ ("b-1.parquet", Some(2)),
+ ("a-2.parquet", Some(5)),
+ ]
+ } else {
+ vec![
+ ("a-1.parquet", Some(0)),
+ ("a-2.parquet", Some(2)),
+ ("b-1.parquet", Some(6)),
+ ]
+ };
+ assert_eq!(actual, expected, "grouping={enabled:?}");
+ }
+}
+
+#[test]
+fn row_tracking_groups_preserve_independently_rolled_blob_columns() {
+ fn entry(partition: &str, name: &str, rows: i64, column: Option<&str>) ->
ManifestEntry {
+ let mut file = test_data_file(name, rows);
+ file.file_source = Some(0);
+ file.write_cols = column.map(|name| vec![name.to_string()]);
+ ManifestEntry::new(FileKind::Add, partition_bytes(partition), 0, -1,
file, 2)
+ }
+ // Writers can interleave partitions while each Blob column rolls at a
+ // different size. Grouping must retain normal-file / Blob-file order.
+ let entries = vec![
+ entry("a", "a-0.parquet", 5, None),
+ entry("a", "a-left-0.blob", 2, Some("left")),
+ entry("a", "a-left-1.blob", 3, Some("left")),
+ entry("a", "a-right-0.blob", 5, Some("right")),
+ entry("b", "b-0.parquet", 2, None),
+ entry("b", "b-left-0.blob", 2, Some("left")),
+ entry("a", "a-1.parquet", 3, None),
+ entry("a", "a-left-2.blob", 3, Some("left")),
+ entry("a", "a-right-1.blob", 1, Some("right")),
+ entry("a", "a-right-2.blob", 2, Some("right")),
+ ];
+ let (entries, next) =
+ row_tracking::assign_row_tracking(7, 10,
row_tracking::group_by_partition(entries))
+ .unwrap();
+ assert_eq!(next, 20);
+ let row_ids: Vec<_> = entries
+ .iter()
+ .map(|entry| (entry.file().file_name.as_str(),
entry.file().first_row_id))
+ .collect();
+ assert_eq!(
+ row_ids,
+ vec![
+ ("a-0.parquet", Some(10)),
+ ("a-left-0.blob", Some(10)),
+ ("a-left-1.blob", Some(12)),
+ ("a-right-0.blob", Some(10)),
+ ("a-1.parquet", Some(15)),
+ ("a-left-2.blob", Some(15)),
+ ("a-right-1.blob", Some(15)),
+ ("a-right-2.blob", Some(16)),
+ ("b-0.parquet", Some(18)),
+ ("b-left-0.blob", Some(18)),
+ ]
+ );
+ assert!(entries.iter().all(
+ |entry| entry.file().min_sequence_number == 7 &&
entry.file().max_sequence_number == 7
+ ));
+}
+
+#[test]
+fn row_tracking_keeps_explicit_row_ids_and_completed_rewrite_metadata() {
+ let mut embedded = test_data_file("embedded.parquet", 3);
+ embedded.file_source = Some(0);
+ embedded.write_cols = Some(vec!["id".into(),
crate::spec::ROW_ID_FIELD_NAME.into()]);
+ embedded.min_sequence_number = 2;
+ embedded.max_sequence_number = 0;
+ let mut rewritten = test_data_file("rewritten.parquet", 6);
+ rewritten.file_source = Some(1);
+ rewritten.first_row_id = Some(20);
+ rewritten.min_sequence_number = 2;
+ rewritten.max_sequence_number = 4;
+ let entries = [embedded, rewritten]
+ .into_iter()
+ .map(|file| ManifestEntry::new(FileKind::Add,
EMPTY_SERIALIZED_ROW.clone(), 0, -1, file, 2))
+ .collect();
+ let (entries, next) = row_tracking::assign_row_tracking(9, 100,
entries).unwrap();
+ assert_eq!(next, 100);
+ assert_eq!(entries[0].file().first_row_id, None);
+ assert_eq!(entries[0].file().min_sequence_number, 2);
+ assert_eq!(entries[0].file().max_sequence_number, 9);
+ assert_eq!(entries[1].file().first_row_id, Some(20));
+ assert_eq!(entries[1].file().min_sequence_number, 2);
+ assert_eq!(entries[1].file().max_sequence_number, 4);
+}
+
+struct ConcurrentRowTrackingAppend {
+ table: Table,
+ attempts: std::sync::Mutex<Vec<Snapshot>>,
+}
+
+#[async_trait::async_trait]
+impl SnapshotCommit for ConcurrentRowTrackingAppend {
+ async fn commit(
+ &self,
+ _: Option<&str>,
+ snapshot: &Snapshot,
+ _: &[PartitionStatistics],
+ ) -> Result<bool> {
+ let first = {
+ let mut attempts = self.attempts.lock().unwrap();
+ attempts.push(snapshot.clone());
+ attempts.len() == 1
+ };
+ if first {
+ let mut file = test_data_file("concurrent.parquet", 3);
+ file.file_source = Some(0);
+ TableCommit::new(self.table.clone(), "other".into())
+ .commit(vec![CommitMessage::new(
+ EMPTY_SERIALIZED_ROW.clone(),
+ 0,
+ vec![file],
+ )])
+ .await?;
+ return Ok(false);
+ }
+ self.table
+ .snapshot_manager()
+ .commit_snapshot(snapshot)
+ .await
+ }
+}
+
+#[tokio::test]
+async fn row_tracking_retry_reassigns_unpublished_ids_and_is_idempotent() {
+ let io = test_file_io();
+ let path = "memory:/row-tracking-retry";
+ setup_dirs(&io, path).await;
+ let table = test_data_evolution_table(&io,
path).copy_with_options(HashMap::from([
+ ("manifest.sidecar.enabled".into(), "true".into()),
+ ("manifest.target-file-size".into(), "1 B".into()),
+ ]));
+ let mut commit = TableCommit::new(table.clone(), "job".into());
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let publisher = Arc::new(ConcurrentRowTrackingAppend {
+ table,
+ attempts: std::sync::Mutex::new(Vec::new()),
+ });
+ commit.snapshot_commit = publisher.clone();
+ let mut file = test_data_file("pending.parquet", 4);
+ file.file_source = Some(0);
+ let message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![file]);
+ commit
+ .commit_with_identifier(vec![message.clone()], 42)
+ .await
+ .unwrap();
+ let attempts = publisher.attempts.lock().unwrap().clone();
+ assert_eq!(
+ attempts
+ .iter()
+ .map(|s| (s.id(), s.next_row_id()))
+ .collect::<Vec<_>>(),
+ vec![(1, Some(4)), (2, Some(7))]
+ );
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.id(), 2);
+ assert_eq!(snapshot.next_row_id(), Some(7));
+ let entries = commit.read_delta_entries(None, &snapshot).await.unwrap();
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].file().first_row_id, Some(3));
+ assert_eq!(entries[0].file().min_sequence_number, 2);
+ assert_eq!(entries[0].file().max_sequence_number, 2);
+ let before = manifest_paths(&io, path).await;
+ commit
+ .filter_and_commit_with_identifier(vec![message], 42)
+ .await
+ .unwrap();
+ assert_eq!(publisher.attempts.lock().unwrap().len(), 2);
+ assert_eq!(manifest_paths(&io, path).await, before);
+ assert_eq!(
+ latest_snapshot(&io, path).await.unwrap().next_row_id(),
+ Some(7)
+ );
+}
+
+#[tokio::test]
+async fn row_tracking_dedicated_updates_require_one_complete_normal_range() {
+ // The latest snapshot may contain two adjacent normal ranges after another
+ // engine rewrites a former [0, 9] file. A stale Blob must not span both.
+ for extension in ["blob", "vector.vortex"] {
+ for (start, count, valid) in [
+ (1, 3, true),
+ (0, 5, true),
+ (0, 10, false),
+ (4, 2, false),
+ (9, 2, false),
+ ] {
+ let io = test_file_io();
+ let path =
format!("memory:/dedicated-range-{extension}-{start}-{count}");
+ setup_dirs(&io, &path).await;
+ let commit = setup_data_evolution_commit(&io, &path);
+ let files = ["first.parquet", "second.parquet"]
+ .into_iter()
+ .map(|name| {
+ let mut file = test_data_file(name, 5);
+ file.file_source = Some(0);
+ file
+ })
+ .collect();
+ commit
+ .commit(vec![CommitMessage::new(
+ EMPTY_SERIALIZED_ROW.clone(),
+ 0,
+ files,
+ )])
+ .await
+ .unwrap();
+ let mut file = test_data_file(&format!("update.{extension}"),
count);
+ file.file_source = Some(0);
+ file.first_row_id = Some(start);
+ file.write_cols = Some(vec!["name".into()]);
+ let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(),
0, vec![file]);
+ message.check_from_snapshot = Some(1);
+ let result = commit.commit(vec![message]).await;
+ if valid {
+ result.unwrap();
+ assert_eq!(latest_snapshot(&io, &path).await.unwrap().id(), 2);
+ } else {
+ let error = result.expect_err("dedicated range must fit in one
normal file");
+ assert!(
+ error.to_string().contains("Row ID existence conflict"),
+ "{error}"
+ );
+ assert_eq!(latest_snapshot(&io, &path).await.unwrap().id(), 1);
+ }
+ }
+ }
+}
+
+#[test]
+fn row_tracking_dedicated_file_cannot_supply_normal_range_existence() {
+ let io = test_file_io();
+ let commit = setup_data_evolution_commit(&io,
"memory:/dedicated-existence");
+ let entry = |name: &str, start: i64, rows: i64| {
+ let mut file = test_data_file(name, rows);
+ file.file_source = Some(0);
+ file.first_row_id = Some(start);
+ file.write_cols = Some(vec!["name".into()]);
+ ManifestEntry::new(FileKind::Add, EMPTY_SERIALIZED_ROW.clone(), 0, -1,
file, 2)
+ };
+ let base = vec![entry("first.parquet", 0, 5), entry("stale.blob", 0, 10)];
+ for name in ["update.blob", "update.parquet"] {
+ let delta = vec![entry(name, 0, 10)];
+ assert!(commit
+ .check_row_id_existence(&base, &delta, Some(10))
+ .is_err());
+ }
+ // The merged-layout check must also catch an older dedicated file that
+ // remains after a change to the normal file boundaries.
+ let merged = vec![
+ entry("first.parquet", 0, 5),
+ entry("second.parquet", 5, 5),
+ entry("stale.blob", 0, 10),
+ ];
+ assert!(commit
+ .check_row_id_range_conflicts(&CommitKind::APPEND, Some(1), &merged)
+ .is_err());
+ assert!(commit
+ .check_row_id_range_conflicts(&CommitKind::APPEND, None, &merged)
+ .is_ok());
+}
+
+#[tokio::test]
+async fn row_tracking_ignore_index_update_preserves_index_metadata() {
+ let io = test_file_io();
+ let path = "memory:/ignore-index-update";
+ setup_dirs(&io, path).await;
+ let table = test_data_evolution_table(&io,
path).copy_with_options(HashMap::from([(
+ "global-index.column-update-action".into(),
+ "IGNORE".into(),
+ )]));
+ let commit = TableCommit::new(table, "test-user".into());
+ let mut file = test_data_file("initial.parquet", 10);
+ file.file_source = Some(0);
+ let mut seed = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![file]);
+ let index = test_global_index_file("global-name.index", 1, 0, 9);
+ seed.new_index_files.push(index.clone());
+ commit.commit(vec![seed]).await.unwrap();
+ let mut partial = test_data_file("updated.parquet", 10);
+ partial.file_source = Some(0);
+ partial.first_row_id = Some(0);
+ partial.write_cols = Some(vec!["name".into()]);
+ let mut update = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![partial]);
+ update.check_from_snapshot = Some(1);
+ commit.commit(vec![update]).await.unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.id(), 2);
+ assert_eq!(snapshot.next_row_id(), Some(10));
+ let index_entries = IndexManifest::read(
+ &io,
+ &format!("{path}/manifest/{}", snapshot.index_manifest().unwrap()),
+ )
+ .await
+ .unwrap();
+ assert_eq!(index_entries.len(), 1);
+ assert_eq!(index_entries[0].index_file, index);
+ assert_eq!(index_entries[0].kind, FileKind::Add);
+}
+
+async fn row_tracking_partial_delete_setup(io: &FileIO, path: &str) ->
(TableCommit, DataFileMeta) {
+ setup_dirs(io, path).await;
+ let commit = setup_data_evolution_commit(io, path);
+ let mut initial = test_data_file("initial.parquet", 10);
+ initial.file_source = Some(0);
+ commit
+ .commit(vec![CommitMessage::new(
+ EMPTY_SERIALIZED_ROW.clone(),
+ 0,
+ vec![initial],
+ )])
+ .await
+ .unwrap();
+ let mut partial = test_data_file("partial-name.parquet", 10);
+ partial.file_source = Some(0);
+ partial.first_row_id = Some(0);
+ partial.write_cols = Some(vec!["name".into()]);
+ let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![partial]);
+ message.check_from_snapshot = Some(1);
+ commit.commit(vec![message]).await.unwrap();
+ let snapshot = latest_snapshot(io, path).await.unwrap();
+ let entries = active_entries(io, path, &snapshot).await;
+ let partial = entries
+ .iter()
+ .find(|entry| entry.file().file_name == "partial-name.parquet")
+ .unwrap()
+ .file()
+ .clone();
+ (commit, partial)
+}
+
+#[tokio::test]
+async fn row_tracking_historical_partial_delete_checks_updated_columns() {
+ for column in ["name", "id"] {
+ let io = test_file_io();
+ let path = format!("memory:/historical-partial-delete-{column}");
+ let (commit, partial) = row_tracking_partial_delete_setup(&io,
&path).await;
+ let mut deletion = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![]);
+ deletion.deleted_files.push(partial);
+ deletion.check_from_snapshot = Some(2);
+ commit.commit(vec![deletion]).await.unwrap();
+ let snapshot = latest_snapshot(&io, &path).await.unwrap();
+ assert_eq!(snapshot.id(), 3);
+ assert_eq!(snapshot.commit_kind(), &CommitKind::OVERWRITE);
+ let changes = commit.read_delta_entries(None,
&snapshot).await.unwrap();
+ assert_eq!(changes.len(), 1);
+ assert_eq!(changes[0].kind(), &FileKind::Delete);
+
+ let mut stale = test_data_file("stale-update.parquet", 10);
+ stale.file_source = Some(0);
+ stale.first_row_id = Some(0);
+ stale.write_cols = Some(vec![column.into()]);
+ let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![stale]);
+ message.check_from_snapshot = Some(2);
+ let result = commit.commit(vec![message]).await;
+ let expected_snapshot = if column == "name" {
+ let error = result.expect_err("a historical DELETE must conflict
on the same column");
+ assert!(
+ error
+ .to_string()
+ .contains("multiple MERGE INTO operations have encountered
conflicts"),
+ "{error}"
+ );
+ 3
+ } else {
+ result.unwrap();
+ 4
+ };
+ let latest = latest_snapshot(&io, &path).await.unwrap();
+ assert_eq!(latest.id(), expected_snapshot);
+ assert_eq!(latest.next_row_id(), Some(10));
+ }
+}
+
+#[tokio::test]
+async fn row_tracking_partial_delete_checks_concurrent_updated_columns() {
+ for column in ["name", "id"] {
+ let io = test_file_io();
+ let path = format!("memory:/current-partial-delete-{column}");
+ let (commit, partial) = row_tracking_partial_delete_setup(&io,
&path).await;
+ let mut concurrent = test_data_file("concurrent-update.parquet", 10);
+ concurrent.file_source = Some(0);
+ concurrent.first_row_id = Some(0);
+ concurrent.write_cols = Some(vec![column.into()]);
+ let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![concurrent]);
+ message.check_from_snapshot = Some(2);
+ commit.commit(vec![message]).await.unwrap();
+
+ let mut deletion = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![]);
+ deletion.deleted_files.push(partial);
+ deletion.check_from_snapshot = Some(2);
+ let result = commit.commit(vec![deletion]).await;
+ let expected_snapshot = if column == "name" {
+ let error = result.expect_err("a current DELETE must conflict on
the same column");
+ assert!(
+ error
+ .to_string()
+ .contains("multiple MERGE INTO operations have encountered
conflicts"),
+ "{error}"
+ );
+ 3
+ } else {
+ result.unwrap();
+ 4
+ };
+ let latest = latest_snapshot(&io, &path).await.unwrap();
+ assert_eq!(latest.id(), expected_snapshot);
+ assert_eq!(latest.next_row_id(), Some(10));
+ let partial_still_exists = active_entries(&io, &path, &latest)
+ .await
+ .iter()
+ .any(|entry| entry.file().file_name == "partial-name.parquet");
+ assert_eq!(partial_still_exists, column == "name");
+ }
+}
