laskoviymishka commented on code in PR #2543:
URL: https://github.com/apache/iceberg-rust/pull/2543#discussion_r3625967513
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -36,6 +36,28 @@ use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
const META_ROOT_PATH: &str = "metadata";
+pub(crate) fn generate_unique_snapshot_id(table: &Table) -> i64 {
+ let generate_random_id = || -> i64 {
+ let (lhs, rhs) = Uuid::new_v4().as_u64_pair();
+ let snapshot_id = (lhs ^ rhs) as i64;
+ if snapshot_id < 0 {
+ -snapshot_id
Review Comment:
Now that this is `pub(crate)` and shared, I'd fix the negation here.
`-snapshot_id` on `i64::MIN` panics in debug and wraps back to `i64::MIN`
(still negative) in release, so the sign-flip silently fails for that one
value. Masking the sign bit is always non-negative and drops the branch: `let
snapshot_id = ((lhs ^ rhs) & i64::MAX as u64) as i64;`. It's pre-existing, but
promoting it to a shared helper is a good moment to get it right in one place.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
Review Comment:
Not a blocker, but `bytes_per_entry` divides on-disk `manifest_length` by
the live-entry count, so deleted/dropped entries still count toward the byte
numerator and inflate the estimate — Java sizes against actual bytes written.
Fine as a heuristic, but worth a comment noting it can over-split.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
+ Some(&i) => i,
+ None => {
+ let i = grouped.len();
+ group_index.insert(data_file.partition.clone(), i);
+ grouped.push(Vec::new());
+ i
+ }
+ };
+ grouped[idx].push((data_file, snap_id, seq,
entry.file_sequence_number));
+ entries_processed += 1;
+ }
+ }
+
+ let mut counter: u64 = 0;
+ let mut new_manifests: Vec<ManifestFile> =
Vec::with_capacity(grouped.len());
+
+ for group in grouped {
+ let mut writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ let mut accumulated: u64 = 0;
+ let mut min_first_row_id: Option<u64> = None;
+
+ for (data_file, snap_id, seq, file_seq) in group {
+ if accumulated > 0
+ && accumulated.saturating_add(bytes_per_entry) >
target_size_bytes
+ {
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ accumulated = 0;
+ min_first_row_id = None;
+ }
+ if let Some(frid_u) = data_file.first_row_id.and_then(|f|
u64::try_from(f).ok()) {
+ min_first_row_id = Some(min_first_row_id.map_or(frid_u,
|m| m.min(frid_u)));
+ }
+ writer.add_existing_file(data_file, snap_id, seq, file_seq)?;
+ accumulated = accumulated.saturating_add(bytes_per_entry);
+ }
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ }
+
+ let manifest_list_path = format!(
+ "{}/metadata/snap-{}-0-{}.{}",
+ metadata.location(),
+ snapshot_id,
+ commit_uuid,
+ DataFileFormat::Avro,
+ );
+ let next_seq_num = metadata.next_sequence_number();
+ let next_row_id = metadata.next_row_id();
+ let writer = table
+ .file_io()
+ .new_output(manifest_list_path.clone())?
+ .writer()
+ .await?;
+ let mut list_writer = match format_version {
+ FormatVersion::V1 => {
+ ManifestListWriter::v1(writer, snapshot_id,
metadata.current_snapshot_id())
+ }
+ FormatVersion::V2 => ManifestListWriter::v2(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ ),
+ FormatVersion::V3 => ManifestListWriter::v3(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ Some(next_row_id),
+ ),
+ };
+ let manifests_created = new_manifests.len();
+ let manifests_kept = kept.len();
+ list_writer.add_manifests(new_manifests.into_iter().chain(kept))?;
+ list_writer.close().await?;
+
+ let mut additional_properties: HashMap<String, String> =
self.snapshot_properties.clone();
+ for k in [
+ "total-data-files",
+ "total-delete-files",
+ "total-records",
+ "total-files-size",
+ "total-position-deletes",
+ "total-equality-deletes",
+ ] {
+ if let Some(v) =
current_snapshot.summary().additional_properties.get(k) {
+ additional_properties.insert(k.to_string(), v.clone());
+ }
+ }
+ additional_properties.insert(
+ "manifests-created".to_string(),
+ manifests_created.to_string(),
+ );
+ additional_properties.insert(
+ "manifests-replaced".to_string(),
+ manifests_replaced.to_string(),
+ );
+ additional_properties.insert("manifests-kept".to_string(),
manifests_kept.to_string());
+ additional_properties.insert(
+ "entries-processed".to_string(),
+ entries_processed.to_string(),
+ );
+ let summary = Summary {
+ operation: Operation::Replace,
+ additional_properties,
+ };
+
+ let commit_ts = chrono::Utc::now().timestamp_millis();
+ let snapshot_builder = Snapshot::builder()
+ .with_manifest_list(manifest_list_path)
+ .with_snapshot_id(snapshot_id)
+ .with_parent_snapshot_id(metadata.current_snapshot_id())
+ .with_sequence_number(next_seq_num)
+ .with_summary(summary)
+ .with_schema_id(metadata.current_schema_id())
+ .with_timestamp_ms(commit_ts);
+ let new_snapshot = match format_version {
+ FormatVersion::V3 => snapshot_builder.with_row_range(next_row_id,
0).build(),
+ _ => snapshot_builder.build(),
+ };
+
+ let updates = vec![
+ TableUpdate::AddSnapshot {
+ snapshot: new_snapshot,
+ },
+ TableUpdate::SetSnapshotRef {
+ ref_name: MAIN_BRANCH.to_string(),
+ reference: SnapshotReference::new(
+ snapshot_id,
+ SnapshotRetention::branch(None, None, None),
+ ),
+ },
+ ];
+ let requirements = vec![
+ TableRequirement::UuidMatch {
+ uuid: metadata.uuid(),
+ },
+ TableRequirement::RefSnapshotIdMatch {
+ r#ref: MAIN_BRANCH.to_string(),
+ snapshot_id: metadata.current_snapshot_id(),
+ },
+ ];
+ Ok(ActionCommit::new(updates, requirements))
+ }
+}
+
+fn new_manifest_writer(
Review Comment:
This is a near-copy of `SnapshotProducer::new_manifest_writer`, and the path
format differs slightly (this one skips the `META_ROOT_PATH` const). The
row-lineage bug above is partly a consequence of the two paths drifting. Either
share the helper or add a "keep in sync with SnapshotProducer" comment so the
next change touches both.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
+ Some(&i) => i,
+ None => {
+ let i = grouped.len();
+ group_index.insert(data_file.partition.clone(), i);
+ grouped.push(Vec::new());
+ i
+ }
+ };
+ grouped[idx].push((data_file, snap_id, seq,
entry.file_sequence_number));
+ entries_processed += 1;
+ }
+ }
+
+ let mut counter: u64 = 0;
+ let mut new_manifests: Vec<ManifestFile> =
Vec::with_capacity(grouped.len());
+
+ for group in grouped {
+ let mut writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ let mut accumulated: u64 = 0;
+ let mut min_first_row_id: Option<u64> = None;
+
+ for (data_file, snap_id, seq, file_seq) in group {
+ if accumulated > 0
+ && accumulated.saturating_add(bytes_per_entry) >
target_size_bytes
+ {
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
Review Comment:
Building on @dhruvarya-db's thread on lines 187/224 — agreed the read path
never materializes per-file `first_row_id`, so `min_first_row_id` stays None
and the `(Some, None)` arm reassigns fresh ranges to existing rows.
The part I'd add is the downstream blast radius. Because we also commit
`with_row_range(next_row_id, 0)`, `metadata.next_row_id` never advances — so
the next FastAppend reuses that same range and the two collide. And
Java/PyIceberg compute `_row_id = manifest.first_row_id + pos`, so they'd read
remapped ids for rows that were originally in a lower range. That makes it
silent cross-engine corruption, not just a local mislabel — which is what I'd
block on.
A couple of ways to handle, wdyt:
- Mirror what Java's `BaseRewriteManifests` does — drop this
`written.first_row_id = ...` assignment, pass `None` to
`ManifestListWriter::v3` for these existing-only manifests, and follow
`SnapshotProducer::commit` (snapshot.rs:485–499): read
`list_writer.next_row_id()` after close and set `with_row_range(table_next,
assigned - table_next)`.
- If materializing per-entry first_row_id in `inherit_data` is out of scope
here, I'd rather return an error on V3 than ship silent corruption behind a
passing test.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
Review Comment:
Not a blocker, but taking a full `HashMap` and replacing the whole map reads
oddly next to the additive `set_target_size_bytes`. A
`with_snapshot_property(k, v)` that inserts one entry would be more consistent
with the builder style — optional.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
Review Comment:
`buffer_unordered(16)` makes the rewritten manifest layout non-deterministic
— manifests come back in completion order, so entry grouping and the resulting
manifest set aren't reproducible run to run, which also shifts
partition-pruning order downstream. I'd use `buffered(16)` to preserve input
order, or sort `loaded` by manifest path before grouping.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
+ Some(&i) => i,
+ None => {
+ let i = grouped.len();
+ group_index.insert(data_file.partition.clone(), i);
+ grouped.push(Vec::new());
+ i
+ }
+ };
+ grouped[idx].push((data_file, snap_id, seq,
entry.file_sequence_number));
+ entries_processed += 1;
+ }
+ }
+
+ let mut counter: u64 = 0;
+ let mut new_manifests: Vec<ManifestFile> =
Vec::with_capacity(grouped.len());
+
+ for group in grouped {
+ let mut writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ let mut accumulated: u64 = 0;
+ let mut min_first_row_id: Option<u64> = None;
+
+ for (data_file, snap_id, seq, file_seq) in group {
+ if accumulated > 0
+ && accumulated.saturating_add(bytes_per_entry) >
target_size_bytes
+ {
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ accumulated = 0;
+ min_first_row_id = None;
+ }
+ if let Some(frid_u) = data_file.first_row_id.and_then(|f|
u64::try_from(f).ok()) {
+ min_first_row_id = Some(min_first_row_id.map_or(frid_u,
|m| m.min(frid_u)));
+ }
+ writer.add_existing_file(data_file, snap_id, seq, file_seq)?;
+ accumulated = accumulated.saturating_add(bytes_per_entry);
+ }
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ }
+
+ let manifest_list_path = format!(
+ "{}/metadata/snap-{}-0-{}.{}",
+ metadata.location(),
+ snapshot_id,
+ commit_uuid,
+ DataFileFormat::Avro,
+ );
+ let next_seq_num = metadata.next_sequence_number();
+ let next_row_id = metadata.next_row_id();
+ let writer = table
+ .file_io()
+ .new_output(manifest_list_path.clone())?
+ .writer()
+ .await?;
+ let mut list_writer = match format_version {
+ FormatVersion::V1 => {
+ ManifestListWriter::v1(writer, snapshot_id,
metadata.current_snapshot_id())
+ }
+ FormatVersion::V2 => ManifestListWriter::v2(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ ),
+ FormatVersion::V3 => ManifestListWriter::v3(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ Some(next_row_id),
+ ),
+ };
+ let manifests_created = new_manifests.len();
+ let manifests_kept = kept.len();
+ list_writer.add_manifests(new_manifests.into_iter().chain(kept))?;
Review Comment:
Not a blocker, but Java's `BaseRewriteManifests` validates
created-file-count == replaced-file-count before committing
(`validateFilesCounts`). I'd sum added+existing entries across `to_rewrite` vs
the new manifests and error on mismatch — cheap insurance against a grouping
bug silently dropping a data file.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
+ Some(&i) => i,
+ None => {
+ let i = grouped.len();
+ group_index.insert(data_file.partition.clone(), i);
+ grouped.push(Vec::new());
+ i
+ }
+ };
+ grouped[idx].push((data_file, snap_id, seq,
entry.file_sequence_number));
+ entries_processed += 1;
+ }
+ }
+
+ let mut counter: u64 = 0;
+ let mut new_manifests: Vec<ManifestFile> =
Vec::with_capacity(grouped.len());
+
+ for group in grouped {
+ let mut writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ let mut accumulated: u64 = 0;
+ let mut min_first_row_id: Option<u64> = None;
+
+ for (data_file, snap_id, seq, file_seq) in group {
+ if accumulated > 0
+ && accumulated.saturating_add(bytes_per_entry) >
target_size_bytes
+ {
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ accumulated = 0;
+ min_first_row_id = None;
+ }
+ if let Some(frid_u) = data_file.first_row_id.and_then(|f|
u64::try_from(f).ok()) {
+ min_first_row_id = Some(min_first_row_id.map_or(frid_u,
|m| m.min(frid_u)));
+ }
+ writer.add_existing_file(data_file, snap_id, seq, file_seq)?;
+ accumulated = accumulated.saturating_add(bytes_per_entry);
+ }
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ }
+
+ let manifest_list_path = format!(
+ "{}/metadata/snap-{}-0-{}.{}",
+ metadata.location(),
+ snapshot_id,
+ commit_uuid,
+ DataFileFormat::Avro,
+ );
+ let next_seq_num = metadata.next_sequence_number();
+ let next_row_id = metadata.next_row_id();
+ let writer = table
+ .file_io()
+ .new_output(manifest_list_path.clone())?
+ .writer()
+ .await?;
+ let mut list_writer = match format_version {
+ FormatVersion::V1 => {
+ ManifestListWriter::v1(writer, snapshot_id,
metadata.current_snapshot_id())
+ }
+ FormatVersion::V2 => ManifestListWriter::v2(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ ),
+ FormatVersion::V3 => ManifestListWriter::v3(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ Some(next_row_id),
+ ),
+ };
+ let manifests_created = new_manifests.len();
+ let manifests_kept = kept.len();
+ list_writer.add_manifests(new_manifests.into_iter().chain(kept))?;
+ list_writer.close().await?;
+
+ let mut additional_properties: HashMap<String, String> =
self.snapshot_properties.clone();
+ for k in [
+ "total-data-files",
Review Comment:
These keys already exist as consts in `spec/snapshot_summary.rs` — I'd
reference those instead of retyping the strings. The `manifests-*` /
`entries-processed` ones aren't const anywhere yet, so worth promoting them to
`pub(crate)` consts while we're here so the two paths can't drift.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
+ Some(&i) => i,
+ None => {
+ let i = grouped.len();
+ group_index.insert(data_file.partition.clone(), i);
+ grouped.push(Vec::new());
+ i
+ }
+ };
+ grouped[idx].push((data_file, snap_id, seq,
entry.file_sequence_number));
+ entries_processed += 1;
+ }
+ }
+
+ let mut counter: u64 = 0;
+ let mut new_manifests: Vec<ManifestFile> =
Vec::with_capacity(grouped.len());
+
+ for group in grouped {
+ let mut writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ let mut accumulated: u64 = 0;
+ let mut min_first_row_id: Option<u64> = None;
+
+ for (data_file, snap_id, seq, file_seq) in group {
+ if accumulated > 0
+ && accumulated.saturating_add(bytes_per_entry) >
target_size_bytes
+ {
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ accumulated = 0;
+ min_first_row_id = None;
+ }
+ if let Some(frid_u) = data_file.first_row_id.and_then(|f|
u64::try_from(f).ok()) {
+ min_first_row_id = Some(min_first_row_id.map_or(frid_u,
|m| m.min(frid_u)));
+ }
+ writer.add_existing_file(data_file, snap_id, seq, file_seq)?;
+ accumulated = accumulated.saturating_add(bytes_per_entry);
+ }
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ }
+
+ let manifest_list_path = format!(
+ "{}/metadata/snap-{}-0-{}.{}",
+ metadata.location(),
+ snapshot_id,
+ commit_uuid,
+ DataFileFormat::Avro,
+ );
+ let next_seq_num = metadata.next_sequence_number();
+ let next_row_id = metadata.next_row_id();
+ let writer = table
+ .file_io()
+ .new_output(manifest_list_path.clone())?
+ .writer()
+ .await?;
+ let mut list_writer = match format_version {
+ FormatVersion::V1 => {
+ ManifestListWriter::v1(writer, snapshot_id,
metadata.current_snapshot_id())
+ }
+ FormatVersion::V2 => ManifestListWriter::v2(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ ),
+ FormatVersion::V3 => ManifestListWriter::v3(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ Some(next_row_id),
+ ),
+ };
+ let manifests_created = new_manifests.len();
+ let manifests_kept = kept.len();
+ list_writer.add_manifests(new_manifests.into_iter().chain(kept))?;
+ list_writer.close().await?;
+
+ let mut additional_properties: HashMap<String, String> =
self.snapshot_properties.clone();
+ for k in [
+ "total-data-files",
+ "total-delete-files",
+ "total-records",
+ "total-files-size",
+ "total-position-deletes",
+ "total-equality-deletes",
+ ] {
+ if let Some(v) =
current_snapshot.summary().additional_properties.get(k) {
+ additional_properties.insert(k.to_string(), v.clone());
+ }
+ }
+ additional_properties.insert(
+ "manifests-created".to_string(),
+ manifests_created.to_string(),
+ );
+ additional_properties.insert(
+ "manifests-replaced".to_string(),
+ manifests_replaced.to_string(),
+ );
+ additional_properties.insert("manifests-kept".to_string(),
manifests_kept.to_string());
+ additional_properties.insert(
+ "entries-processed".to_string(),
+ entries_processed.to_string(),
+ );
+ let summary = Summary {
+ operation: Operation::Replace,
+ additional_properties,
+ };
+
+ let commit_ts = chrono::Utc::now().timestamp_millis();
+ let snapshot_builder = Snapshot::builder()
+ .with_manifest_list(manifest_list_path)
+ .with_snapshot_id(snapshot_id)
+ .with_parent_snapshot_id(metadata.current_snapshot_id())
+ .with_sequence_number(next_seq_num)
+ .with_summary(summary)
+ .with_schema_id(metadata.current_schema_id())
+ .with_timestamp_ms(commit_ts);
+ let new_snapshot = match format_version {
+ FormatVersion::V3 => snapshot_builder.with_row_range(next_row_id,
0).build(),
+ _ => snapshot_builder.build(),
+ };
+
+ let updates = vec![
+ TableUpdate::AddSnapshot {
+ snapshot: new_snapshot,
+ },
+ TableUpdate::SetSnapshotRef {
+ ref_name: MAIN_BRANCH.to_string(),
+ reference: SnapshotReference::new(
+ snapshot_id,
+ SnapshotRetention::branch(None, None, None),
+ ),
+ },
+ ];
+ let requirements = vec![
+ TableRequirement::UuidMatch {
+ uuid: metadata.uuid(),
+ },
+ TableRequirement::RefSnapshotIdMatch {
+ r#ref: MAIN_BRANCH.to_string(),
+ snapshot_id: metadata.current_snapshot_id(),
+ },
+ ];
+ Ok(ActionCommit::new(updates, requirements))
+ }
+}
+
+fn new_manifest_writer(
+ table: &Table,
+ commit_uuid: &Uuid,
+ n: u64,
+ snapshot_id: i64,
+) -> Result<ManifestWriter> {
+ let metadata = table.metadata();
+ let path = format!(
+ "{}/metadata/{}-m{}.{}",
+ metadata.location(),
+ commit_uuid,
+ n,
+ DataFileFormat::Avro,
+ );
+ let output = table.file_io().new_output(path)?;
+ let builder = ManifestWriterBuilder::new(
+ output,
+ Some(snapshot_id),
+ metadata.current_schema().clone(),
+ metadata.default_partition_spec().as_ref().clone(),
+ );
+ Ok(match metadata.format_version() {
+ FormatVersion::V1 => builder.build_v1(),
+ FormatVersion::V2 => builder.build_v2_data(),
+ FormatVersion::V3 => builder.build_v3_data(),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ use crate::memory::tests::new_memory_catalog;
+ use crate::spec::{
+ DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal,
Operation, Struct,
+ };
+ use crate::table::Table;
+ use crate::transaction::tests::{make_v2_minimal_table,
make_v3_minimal_table_in_catalog};
+ use crate::transaction::{ApplyTransactionAction, Transaction,
TransactionAction};
+ use crate::{Catalog, TableUpdate};
+
+ fn data_file(name: &str, partition: i64, size: u64, records: u64) ->
DataFile {
+ DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path(format!("test/{name}.parquet"))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(size)
+ .record_count(records)
+ .partition_spec_id(0)
+ .partition(Struct::from_iter([Some(Literal::long(partition))]))
+ .build()
+ .unwrap()
+ }
+
+ async fn append_one(catalog: &impl Catalog, table: Table, file: DataFile)
-> Table {
+ let tx = Transaction::new(&table);
+ tx.fast_append()
+ .add_data_files(vec![file])
+ .apply(tx)
+ .unwrap()
+ .commit(catalog)
+ .await
+ .unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_no_current_snapshot_errors() {
+ let table = make_v2_minimal_table();
+ let tx = Transaction::new(&table);
+ let action = tx.rewrite_manifests();
+ match Arc::new(action).commit(&table).await {
+ Ok(_) => panic!("expected error"),
+ Err(e) => assert_eq!(e.kind(),
crate::ErrorKind::PreconditionFailed),
+ }
+ }
+
+ #[tokio::test]
+ async fn test_single_small_manifest_is_noop() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let table = append_one(&catalog, table, data_file("a", 1, 100,
1)).await;
+ let original_snapshot_id = table.metadata().current_snapshot_id();
+
+ let tx = Transaction::new(&table);
+ let action = tx.rewrite_manifests();
+ let mut commit = Arc::new(action).commit(&table).await.unwrap();
+ let updates = commit.take_updates();
+ assert!(updates.is_empty());
+
+ let table = tx
+ .rewrite_manifests()
+ .apply(Transaction::new(&table))
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+ assert_eq!(
+ table.metadata().current_snapshot_id(),
+ original_snapshot_id,
+ "no-op should not change the current snapshot"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_multi_manifest_merge_preserves_sequence_numbers() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+
+ let table = append_one(&catalog, table, data_file("a", 1, 1_000,
10)).await;
+ let seq_a = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ let table = append_one(&catalog, table, data_file("b", 1, 2_000,
20)).await;
+ let seq_b = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ let table = append_one(&catalog, table, data_file("c", 2, 3_000,
30)).await;
+ let seq_c = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ assert!(seq_a < seq_b && seq_b < seq_c);
+
+ let pre_manifest_count = table
+ .manifest_list_reader(table.metadata().current_snapshot().unwrap())
+ .load()
+ .await
+ .unwrap()
+ .entries()
+ .len();
+ assert_eq!(pre_manifest_count, 3);
+
+ let tx = Transaction::new(&table);
+ let table = tx
+ .rewrite_manifests()
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let snapshot = table.metadata().current_snapshot().unwrap();
+ assert_eq!(snapshot.summary().operation, Operation::Replace);
+
+ let post_list =
table.manifest_list_reader(snapshot).load().await.unwrap();
+ let total_entries: usize = {
+ let mut n = 0;
+ for m in post_list.entries() {
+ let manifest = m.load_manifest(table.file_io()).await.unwrap();
+ n += manifest.entries().len();
+ }
+ n
+ };
+ assert_eq!(total_entries, 3, "all entries preserved across rewrite");
+
+ let mut seen_seqs: Vec<i64> = Vec::new();
+ for m in post_list.entries() {
+ let manifest = m.load_manifest(table.file_io()).await.unwrap();
+ for entry in manifest.entries() {
+ seen_seqs.push(entry.sequence_number().unwrap());
+ }
+ }
+ seen_seqs.sort();
+ assert_eq!(seen_seqs, vec![seq_a, seq_b, seq_c]);
+
+ assert!(post_list.entries().len() < pre_manifest_count);
+
+ let summary = &snapshot.summary().additional_properties;
+ assert_eq!(summary.get("total-records").unwrap(), "60");
+ assert_eq!(summary.get("total-data-files").unwrap(), "3");
+ assert_eq!(summary.get("entries-processed").unwrap(), "3");
+ assert_eq!(summary.get("manifests-replaced").unwrap(), "3");
+ }
+
+ #[tokio::test]
+ async fn test_target_size_from_table_property() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let mut t = table;
+ for i in 0..6 {
+ t = append_one(&catalog, t, data_file(&format!("f{i}"), 1, 10_000,
1)).await;
+ }
+
+ let tx = Transaction::new(&t);
+ let t = tx
+ .update_table_properties()
+ .set(
+ "commit.manifest.target-size-bytes".to_string(),
+ "400".to_string(),
+ )
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let tx = Transaction::new(&t);
+ let t = tx
+ .rewrite_manifests()
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let post_list = t
+ .manifest_list_reader(t.metadata().current_snapshot().unwrap())
+ .load()
+ .await
+ .unwrap();
+ assert!(post_list.entries().len() > 1);
+ }
+
+ #[tokio::test]
+ async fn test_target_size_rolls_multiple_manifests() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let mut t = table;
+ for i in 0..6 {
+ t = append_one(&catalog, t, data_file(&format!("f{i}"), 1, 10_000,
1)).await;
+ }
+
+ let tx = Transaction::new(&t);
+ let t = tx
+ .rewrite_manifests()
+ .set_target_size_bytes(400)
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let post_list = t
+ .manifest_list_reader(t.metadata().current_snapshot().unwrap())
+ .load()
+ .await
+ .unwrap();
+ assert!(post_list.entries().len() > 1);
+
+ let mut total = 0;
+ for m in post_list.entries() {
+ let manifest = m.load_manifest(t.file_io()).await.unwrap();
+ let n = manifest.entries().len();
+ assert!(
+ n <= 2,
+ "each rolled manifest should hold at most ~2 entries when
target is just above bytes_per_entry"
+ );
+ total += n;
+ }
+ assert_eq!(total, 6);
+ }
+
+ #[tokio::test]
+ async fn test_oversized_single_manifest_is_split() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let files: Vec<_> = (0..6)
+ .map(|i| data_file(&format!("f{i}"), 1, 10_000, 1))
+ .collect();
+
+ let tx = Transaction::new(&table);
+ let table = tx
+ .fast_append()
+ .add_data_files(files)
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let pre_list = table
+ .manifest_list_reader(table.metadata().current_snapshot().unwrap())
+ .load()
+ .await
+ .unwrap();
+ assert_eq!(pre_list.entries().len(), 1);
+
+ let tx = Transaction::new(&table);
+ let table = tx
+ .rewrite_manifests()
+ .set_target_size_bytes(400)
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let snap = table.metadata().current_snapshot().unwrap();
+ let post_list = table.manifest_list_reader(snap).load().await.unwrap();
+ assert!(post_list.entries().len() > 1);
+ assert_eq!(
+ snap.summary()
+ .additional_properties
+ .get("manifests-replaced")
+ .unwrap(),
+ "1"
+ );
+
+ let mut total = 0;
+ for m in post_list.entries() {
+ total += m
+ .load_manifest(table.file_io())
+ .await
+ .unwrap()
+ .entries()
+ .len();
+ }
+ assert_eq!(total, 6);
+ }
+
+ #[tokio::test]
+ async fn test_v3_row_lineage_preserved() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let table = append_one(&catalog, table, data_file("a", 1, 100,
30)).await;
+ let table = append_one(&catalog, table, data_file("b", 1, 100,
17)).await;
+ let table = append_one(&catalog, table, data_file("c", 1, 100,
11)).await;
+
+ async fn collect(t: &Table) -> Vec<(String, Option<i64>, Option<i64>,
Option<i64>)> {
+ let list = t
+ .manifest_list_reader(t.metadata().current_snapshot().unwrap())
+ .load()
+ .await
+ .unwrap();
+ let mut v = Vec::new();
+ for m in list.entries() {
+ let manifest = m.load_manifest(t.file_io()).await.unwrap();
+ for entry in manifest.entries() {
+ v.push((
+ entry.data_file().file_path().to_string(),
+ entry.data_file().first_row_id,
+ entry.sequence_number(),
+ entry.file_sequence_number,
+ ));
+ }
+ }
+ v.sort();
+ v
+ }
+
+ let pre = collect(&table).await;
+ let next_row_id_before = table.metadata().next_row_id();
+
+ let tx = Transaction::new(&table);
+ let table = tx
+ .rewrite_manifests()
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ assert_eq!(
+ table.metadata().next_row_id(),
+ next_row_id_before,
+ "rewrite must not consume new row ids"
+ );
+ let snap = table.metadata().current_snapshot().unwrap();
+ assert_eq!(snap.row_range(), Some((next_row_id_before, 0)));
+
+ let post = collect(&table).await;
+ assert_eq!(
Review Comment:
This test doesn't actually exercise row-lineage preservation, which is why
the bug @dhruvarya-db flagged on lines 187/224 slips through green.
`collect` only reads `entry.data_file().first_row_id` (field 142), which is
None for every file FastAppend writes, so `pre` and `post` are both all-None
and `pre == post` passes no matter what the rewrite does to the manifest
ranges. It also asserts `added_rows == 0` using the same code path it's meant
to verify. What I'd want: fixtures with explicit per-file first_row_id, an
assertion on the manifest-level `ManifestFile.first_row_id` in the post list, a
check that no two manifests share a range, and an append-after-rewrite
assertion that the new data's range doesn't overlap the rewritten manifests'.
That last one is the regression test that locks the contract against the bug
above.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
Review Comment:
Worth documenting on the action: grouping strictly by partition means at
least one manifest per partition, so on high-cardinality partitioning this
won't actually compact. Java exposes a `clusterBy` knob for exactly this. I'm
not asking for the knob in this PR, just a doc note so callers aren't surprised.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
+ Some(&i) => i,
+ None => {
+ let i = grouped.len();
+ group_index.insert(data_file.partition.clone(), i);
+ grouped.push(Vec::new());
+ i
+ }
+ };
+ grouped[idx].push((data_file, snap_id, seq,
entry.file_sequence_number));
+ entries_processed += 1;
+ }
+ }
+
+ let mut counter: u64 = 0;
+ let mut new_manifests: Vec<ManifestFile> =
Vec::with_capacity(grouped.len());
+
+ for group in grouped {
+ let mut writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ let mut accumulated: u64 = 0;
+ let mut min_first_row_id: Option<u64> = None;
+
+ for (data_file, snap_id, seq, file_seq) in group {
+ if accumulated > 0
+ && accumulated.saturating_add(bytes_per_entry) >
target_size_bytes
+ {
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ accumulated = 0;
+ min_first_row_id = None;
+ }
+ if let Some(frid_u) = data_file.first_row_id.and_then(|f|
u64::try_from(f).ok()) {
+ min_first_row_id = Some(min_first_row_id.map_or(frid_u,
|m| m.min(frid_u)));
+ }
+ writer.add_existing_file(data_file, snap_id, seq, file_seq)?;
+ accumulated = accumulated.saturating_add(bytes_per_entry);
+ }
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ }
+
+ let manifest_list_path = format!(
+ "{}/metadata/snap-{}-0-{}.{}",
+ metadata.location(),
+ snapshot_id,
+ commit_uuid,
+ DataFileFormat::Avro,
+ );
+ let next_seq_num = metadata.next_sequence_number();
+ let next_row_id = metadata.next_row_id();
+ let writer = table
+ .file_io()
+ .new_output(manifest_list_path.clone())?
+ .writer()
+ .await?;
+ let mut list_writer = match format_version {
+ FormatVersion::V1 => {
+ ManifestListWriter::v1(writer, snapshot_id,
metadata.current_snapshot_id())
+ }
+ FormatVersion::V2 => ManifestListWriter::v2(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ ),
+ FormatVersion::V3 => ManifestListWriter::v3(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ Some(next_row_id),
+ ),
+ };
+ let manifests_created = new_manifests.len();
+ let manifests_kept = kept.len();
+ list_writer.add_manifests(new_manifests.into_iter().chain(kept))?;
+ list_writer.close().await?;
+
+ let mut additional_properties: HashMap<String, String> =
self.snapshot_properties.clone();
+ for k in [
+ "total-data-files",
+ "total-delete-files",
+ "total-records",
+ "total-files-size",
+ "total-position-deletes",
+ "total-equality-deletes",
+ ] {
+ if let Some(v) =
current_snapshot.summary().additional_properties.get(k) {
+ additional_properties.insert(k.to_string(), v.clone());
+ }
+ }
+ additional_properties.insert(
+ "manifests-created".to_string(),
+ manifests_created.to_string(),
+ );
+ additional_properties.insert(
+ "manifests-replaced".to_string(),
+ manifests_replaced.to_string(),
+ );
+ additional_properties.insert("manifests-kept".to_string(),
manifests_kept.to_string());
+ additional_properties.insert(
+ "entries-processed".to_string(),
+ entries_processed.to_string(),
+ );
+ let summary = Summary {
+ operation: Operation::Replace,
+ additional_properties,
+ };
+
+ let commit_ts = chrono::Utc::now().timestamp_millis();
+ let snapshot_builder = Snapshot::builder()
+ .with_manifest_list(manifest_list_path)
+ .with_snapshot_id(snapshot_id)
+ .with_parent_snapshot_id(metadata.current_snapshot_id())
+ .with_sequence_number(next_seq_num)
+ .with_summary(summary)
+ .with_schema_id(metadata.current_schema_id())
+ .with_timestamp_ms(commit_ts);
+ let new_snapshot = match format_version {
+ FormatVersion::V3 => snapshot_builder.with_row_range(next_row_id,
0).build(),
+ _ => snapshot_builder.build(),
+ };
+
+ let updates = vec![
+ TableUpdate::AddSnapshot {
+ snapshot: new_snapshot,
+ },
+ TableUpdate::SetSnapshotRef {
+ ref_name: MAIN_BRANCH.to_string(),
+ reference: SnapshotReference::new(
+ snapshot_id,
+ SnapshotRetention::branch(None, None, None),
+ ),
+ },
+ ];
+ let requirements = vec![
+ TableRequirement::UuidMatch {
+ uuid: metadata.uuid(),
+ },
+ TableRequirement::RefSnapshotIdMatch {
+ r#ref: MAIN_BRANCH.to_string(),
+ snapshot_id: metadata.current_snapshot_id(),
+ },
+ ];
+ Ok(ActionCommit::new(updates, requirements))
+ }
+}
+
+fn new_manifest_writer(
+ table: &Table,
+ commit_uuid: &Uuid,
+ n: u64,
+ snapshot_id: i64,
+) -> Result<ManifestWriter> {
+ let metadata = table.metadata();
+ let path = format!(
+ "{}/metadata/{}-m{}.{}",
+ metadata.location(),
+ commit_uuid,
+ n,
+ DataFileFormat::Avro,
+ );
+ let output = table.file_io().new_output(path)?;
+ let builder = ManifestWriterBuilder::new(
+ output,
+ Some(snapshot_id),
+ metadata.current_schema().clone(),
+ metadata.default_partition_spec().as_ref().clone(),
+ );
+ Ok(match metadata.format_version() {
+ FormatVersion::V1 => builder.build_v1(),
+ FormatVersion::V2 => builder.build_v2_data(),
+ FormatVersion::V3 => builder.build_v3_data(),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ use crate::memory::tests::new_memory_catalog;
+ use crate::spec::{
+ DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal,
Operation, Struct,
+ };
+ use crate::table::Table;
+ use crate::transaction::tests::{make_v2_minimal_table,
make_v3_minimal_table_in_catalog};
+ use crate::transaction::{ApplyTransactionAction, Transaction,
TransactionAction};
+ use crate::{Catalog, TableUpdate};
+
+ fn data_file(name: &str, partition: i64, size: u64, records: u64) ->
DataFile {
+ DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path(format!("test/{name}.parquet"))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(size)
+ .record_count(records)
+ .partition_spec_id(0)
+ .partition(Struct::from_iter([Some(Literal::long(partition))]))
+ .build()
+ .unwrap()
+ }
+
+ async fn append_one(catalog: &impl Catalog, table: Table, file: DataFile)
-> Table {
+ let tx = Transaction::new(&table);
+ tx.fast_append()
+ .add_data_files(vec![file])
+ .apply(tx)
+ .unwrap()
+ .commit(catalog)
+ .await
+ .unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_no_current_snapshot_errors() {
+ let table = make_v2_minimal_table();
+ let tx = Transaction::new(&table);
+ let action = tx.rewrite_manifests();
+ match Arc::new(action).commit(&table).await {
+ Ok(_) => panic!("expected error"),
+ Err(e) => assert_eq!(e.kind(),
crate::ErrorKind::PreconditionFailed),
+ }
+ }
+
+ #[tokio::test]
+ async fn test_single_small_manifest_is_noop() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let table = append_one(&catalog, table, data_file("a", 1, 100,
1)).await;
+ let original_snapshot_id = table.metadata().current_snapshot_id();
+
+ let tx = Transaction::new(&table);
+ let action = tx.rewrite_manifests();
+ let mut commit = Arc::new(action).commit(&table).await.unwrap();
+ let updates = commit.take_updates();
+ assert!(updates.is_empty());
+
+ let table = tx
+ .rewrite_manifests()
+ .apply(Transaction::new(&table))
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+ assert_eq!(
+ table.metadata().current_snapshot_id(),
+ original_snapshot_id,
+ "no-op should not change the current snapshot"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_multi_manifest_merge_preserves_sequence_numbers() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+
+ let table = append_one(&catalog, table, data_file("a", 1, 1_000,
10)).await;
+ let seq_a = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ let table = append_one(&catalog, table, data_file("b", 1, 2_000,
20)).await;
+ let seq_b = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ let table = append_one(&catalog, table, data_file("c", 2, 3_000,
30)).await;
+ let seq_c = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ assert!(seq_a < seq_b && seq_b < seq_c);
+
+ let pre_manifest_count = table
+ .manifest_list_reader(table.metadata().current_snapshot().unwrap())
+ .load()
+ .await
+ .unwrap()
+ .entries()
+ .len();
+ assert_eq!(pre_manifest_count, 3);
+
+ let tx = Transaction::new(&table);
+ let table = tx
+ .rewrite_manifests()
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let snapshot = table.metadata().current_snapshot().unwrap();
+ assert_eq!(snapshot.summary().operation, Operation::Replace);
+
+ let post_list =
table.manifest_list_reader(snapshot).load().await.unwrap();
+ let total_entries: usize = {
+ let mut n = 0;
+ for m in post_list.entries() {
+ let manifest = m.load_manifest(table.file_io()).await.unwrap();
+ n += manifest.entries().len();
+ }
+ n
+ };
+ assert_eq!(total_entries, 3, "all entries preserved across rewrite");
+
+ let mut seen_seqs: Vec<i64> = Vec::new();
+ for m in post_list.entries() {
+ let manifest = m.load_manifest(table.file_io()).await.unwrap();
+ for entry in manifest.entries() {
+ seen_seqs.push(entry.sequence_number().unwrap());
+ }
+ }
+ seen_seqs.sort();
+ assert_eq!(seen_seqs, vec![seq_a, seq_b, seq_c]);
+
+ assert!(post_list.entries().len() < pre_manifest_count);
+
+ let summary = &snapshot.summary().additional_properties;
+ assert_eq!(summary.get("total-records").unwrap(), "60");
+ assert_eq!(summary.get("total-data-files").unwrap(), "3");
+ assert_eq!(summary.get("entries-processed").unwrap(), "3");
+ assert_eq!(summary.get("manifests-replaced").unwrap(), "3");
+ }
+
+ #[tokio::test]
+ async fn test_target_size_from_table_property() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let mut t = table;
+ for i in 0..6 {
+ t = append_one(&catalog, t, data_file(&format!("f{i}"), 1, 10_000,
1)).await;
+ }
+
+ let tx = Transaction::new(&t);
+ let t = tx
+ .update_table_properties()
+ .set(
+ "commit.manifest.target-size-bytes".to_string(),
Review Comment:
Small thing — since you just added
`TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES`, use the const
here so a rename can't leave the test asserting a stale key.
##########
crates/iceberg/src/transaction/rewrite_manifests.rs:
##########
@@ -0,0 +1,804 @@
+// 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.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::TryStreamExt;
+use futures::stream::{self, StreamExt};
+use uuid::Uuid;
+
+use crate::error::Result;
+use crate::spec::{
+ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType,
ManifestFile,
+ ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
Snapshot,
+ SnapshotReference, SnapshotRetention, Struct, Summary, TableProperties,
+};
+use crate::table::Table;
+use crate::transaction::snapshot::generate_unique_snapshot_id;
+use crate::transaction::{ActionCommit, TransactionAction};
+use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+
+const FALLBACK_BYTES_PER_ENTRY: u64 = 256;
+
+pub struct RewriteManifestsAction {
+ target_size_bytes: Option<u64>,
+ snapshot_properties: HashMap<String, String>,
+}
+
+impl RewriteManifestsAction {
+ pub(crate) fn new() -> Self {
+ Self {
+ target_size_bytes: None,
+ snapshot_properties: HashMap::new(),
+ }
+ }
+
+ pub fn set_target_size_bytes(mut self, target_size_bytes: u64) -> Self {
+ self.target_size_bytes = Some(target_size_bytes);
+ self
+ }
+
+ pub fn set_snapshot_properties(mut self, snapshot_properties:
HashMap<String, String>) -> Self {
+ self.snapshot_properties = snapshot_properties;
+ self
+ }
+}
+
+#[async_trait]
+impl TransactionAction for RewriteManifestsAction {
+ async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ let metadata = table.metadata();
+ let Some(current_snapshot) = metadata.current_snapshot() else {
+ return Err(Error::new(
+ ErrorKind::PreconditionFailed,
+ "RewriteManifests requires the table to have a current
snapshot",
+ ));
+ };
+
+ let target_size_bytes = self.target_size_bytes.unwrap_or_else(|| {
+ metadata
+ .properties()
+
.get(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES)
+ .and_then(|s| s.parse::<u64>().ok())
+
.unwrap_or(TableProperties::PROPERTY_COMMIT_MANIFEST_TARGET_SIZE_BYTES_DEFAULT)
+ });
+ let default_spec_id = metadata.default_partition_spec_id();
+ let format_version = metadata.format_version();
+
+ let manifest_list =
table.manifest_list_reader(current_snapshot).load().await?;
+
+ let (to_rewrite, kept): (Vec<ManifestFile>, Vec<ManifestFile>) =
+ manifest_list.entries().iter().cloned().partition(|m| {
+ m.content == ManifestContentType::Data
+ && m.partition_spec_id == default_spec_id
+ && (m.has_added_files() || m.has_existing_files())
+ });
+
+ if to_rewrite.is_empty() {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+
+ let total_size: u64 = to_rewrite
+ .iter()
+ .map(|m| u64::try_from(m.manifest_length).unwrap_or(0))
+ .sum();
+ if to_rewrite.len() == 1 && total_size <= target_size_bytes {
+ return Ok(ActionCommit::new(vec![], vec![]));
+ }
+ let total_input_entries: u64 = to_rewrite
+ .iter()
+ .map(|m| {
+ u64::from(m.added_files_count.unwrap_or(0))
+ + u64::from(m.existing_files_count.unwrap_or(0))
+ })
+ .sum();
+ let bytes_per_entry = total_size
+ .checked_div(total_input_entries)
+ .map_or(FALLBACK_BYTES_PER_ENTRY, |b| b.max(1));
+ let manifests_replaced = to_rewrite.len();
+
+ let commit_uuid = Uuid::now_v7();
+ let snapshot_id = generate_unique_snapshot_id(table);
+
+ let loaded: Vec<_> = stream::iter(to_rewrite)
+ .map(|m| {
+ let file_io = table.file_io().clone();
+ async move { m.load_manifest(&file_io).await }
+ })
+ .buffer_unordered(16)
+ .try_collect()
+ .await?;
+
+ let mut grouped: Vec<Vec<(DataFile, i64, i64, Option<i64>)>> =
Vec::new();
+ let mut group_index: HashMap<Struct, usize> = HashMap::new();
+ let mut entries_processed: u64 = 0;
+
+ for manifest in loaded {
+ for entry in manifest.entries() {
+ if !entry.is_alive() {
+ continue;
+ }
+ let snap_id = entry.snapshot_id().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing snapshot_id",
+ )
+ })?;
+ let seq = entry.sequence_number().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Live manifest entry is missing sequence_number",
+ )
+ })?;
+ let data_file = entry.data_file().clone();
+ let idx = match group_index.get(&data_file.partition) {
+ Some(&i) => i,
+ None => {
+ let i = grouped.len();
+ group_index.insert(data_file.partition.clone(), i);
+ grouped.push(Vec::new());
+ i
+ }
+ };
+ grouped[idx].push((data_file, snap_id, seq,
entry.file_sequence_number));
+ entries_processed += 1;
+ }
+ }
+
+ let mut counter: u64 = 0;
+ let mut new_manifests: Vec<ManifestFile> =
Vec::with_capacity(grouped.len());
+
+ for group in grouped {
+ let mut writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ let mut accumulated: u64 = 0;
+ let mut min_first_row_id: Option<u64> = None;
+
+ for (data_file, snap_id, seq, file_seq) in group {
+ if accumulated > 0
+ && accumulated.saturating_add(bytes_per_entry) >
target_size_bytes
+ {
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ writer = new_manifest_writer(table, &commit_uuid, counter,
snapshot_id)?;
+ counter += 1;
+ accumulated = 0;
+ min_first_row_id = None;
+ }
+ if let Some(frid_u) = data_file.first_row_id.and_then(|f|
u64::try_from(f).ok()) {
+ min_first_row_id = Some(min_first_row_id.map_or(frid_u,
|m| m.min(frid_u)));
+ }
+ writer.add_existing_file(data_file, snap_id, seq, file_seq)?;
+ accumulated = accumulated.saturating_add(bytes_per_entry);
+ }
+ let mut written = writer.write_manifest_file().await?;
+ if format_version == FormatVersion::V3 {
+ written.first_row_id = min_first_row_id;
+ }
+ new_manifests.push(written);
+ }
+
+ let manifest_list_path = format!(
+ "{}/metadata/snap-{}-0-{}.{}",
+ metadata.location(),
+ snapshot_id,
+ commit_uuid,
+ DataFileFormat::Avro,
+ );
+ let next_seq_num = metadata.next_sequence_number();
+ let next_row_id = metadata.next_row_id();
+ let writer = table
+ .file_io()
+ .new_output(manifest_list_path.clone())?
+ .writer()
+ .await?;
+ let mut list_writer = match format_version {
+ FormatVersion::V1 => {
+ ManifestListWriter::v1(writer, snapshot_id,
metadata.current_snapshot_id())
+ }
+ FormatVersion::V2 => ManifestListWriter::v2(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ ),
+ FormatVersion::V3 => ManifestListWriter::v3(
+ writer,
+ snapshot_id,
+ metadata.current_snapshot_id(),
+ next_seq_num,
+ Some(next_row_id),
+ ),
+ };
+ let manifests_created = new_manifests.len();
+ let manifests_kept = kept.len();
+ list_writer.add_manifests(new_manifests.into_iter().chain(kept))?;
+ list_writer.close().await?;
+
+ let mut additional_properties: HashMap<String, String> =
self.snapshot_properties.clone();
+ for k in [
+ "total-data-files",
+ "total-delete-files",
+ "total-records",
+ "total-files-size",
+ "total-position-deletes",
+ "total-equality-deletes",
+ ] {
+ if let Some(v) =
current_snapshot.summary().additional_properties.get(k) {
+ additional_properties.insert(k.to_string(), v.clone());
+ }
+ }
+ additional_properties.insert(
+ "manifests-created".to_string(),
+ manifests_created.to_string(),
+ );
+ additional_properties.insert(
+ "manifests-replaced".to_string(),
+ manifests_replaced.to_string(),
+ );
+ additional_properties.insert("manifests-kept".to_string(),
manifests_kept.to_string());
+ additional_properties.insert(
+ "entries-processed".to_string(),
+ entries_processed.to_string(),
+ );
+ let summary = Summary {
+ operation: Operation::Replace,
+ additional_properties,
+ };
+
+ let commit_ts = chrono::Utc::now().timestamp_millis();
+ let snapshot_builder = Snapshot::builder()
+ .with_manifest_list(manifest_list_path)
+ .with_snapshot_id(snapshot_id)
+ .with_parent_snapshot_id(metadata.current_snapshot_id())
+ .with_sequence_number(next_seq_num)
+ .with_summary(summary)
+ .with_schema_id(metadata.current_schema_id())
+ .with_timestamp_ms(commit_ts);
+ let new_snapshot = match format_version {
+ FormatVersion::V3 => snapshot_builder.with_row_range(next_row_id,
0).build(),
+ _ => snapshot_builder.build(),
+ };
+
+ let updates = vec![
+ TableUpdate::AddSnapshot {
+ snapshot: new_snapshot,
+ },
+ TableUpdate::SetSnapshotRef {
+ ref_name: MAIN_BRANCH.to_string(),
+ reference: SnapshotReference::new(
+ snapshot_id,
+ SnapshotRetention::branch(None, None, None),
+ ),
+ },
+ ];
+ let requirements = vec![
+ TableRequirement::UuidMatch {
+ uuid: metadata.uuid(),
+ },
+ TableRequirement::RefSnapshotIdMatch {
+ r#ref: MAIN_BRANCH.to_string(),
+ snapshot_id: metadata.current_snapshot_id(),
+ },
+ ];
+ Ok(ActionCommit::new(updates, requirements))
+ }
+}
+
+fn new_manifest_writer(
+ table: &Table,
+ commit_uuid: &Uuid,
+ n: u64,
+ snapshot_id: i64,
+) -> Result<ManifestWriter> {
+ let metadata = table.metadata();
+ let path = format!(
+ "{}/metadata/{}-m{}.{}",
+ metadata.location(),
+ commit_uuid,
+ n,
+ DataFileFormat::Avro,
+ );
+ let output = table.file_io().new_output(path)?;
+ let builder = ManifestWriterBuilder::new(
+ output,
+ Some(snapshot_id),
+ metadata.current_schema().clone(),
+ metadata.default_partition_spec().as_ref().clone(),
+ );
+ Ok(match metadata.format_version() {
+ FormatVersion::V1 => builder.build_v1(),
+ FormatVersion::V2 => builder.build_v2_data(),
+ FormatVersion::V3 => builder.build_v3_data(),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ use crate::memory::tests::new_memory_catalog;
+ use crate::spec::{
+ DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal,
Operation, Struct,
+ };
+ use crate::table::Table;
+ use crate::transaction::tests::{make_v2_minimal_table,
make_v3_minimal_table_in_catalog};
+ use crate::transaction::{ApplyTransactionAction, Transaction,
TransactionAction};
+ use crate::{Catalog, TableUpdate};
+
+ fn data_file(name: &str, partition: i64, size: u64, records: u64) ->
DataFile {
+ DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path(format!("test/{name}.parquet"))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(size)
+ .record_count(records)
+ .partition_spec_id(0)
+ .partition(Struct::from_iter([Some(Literal::long(partition))]))
+ .build()
+ .unwrap()
+ }
+
+ async fn append_one(catalog: &impl Catalog, table: Table, file: DataFile)
-> Table {
+ let tx = Transaction::new(&table);
+ tx.fast_append()
+ .add_data_files(vec![file])
+ .apply(tx)
+ .unwrap()
+ .commit(catalog)
+ .await
+ .unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_no_current_snapshot_errors() {
+ let table = make_v2_minimal_table();
+ let tx = Transaction::new(&table);
+ let action = tx.rewrite_manifests();
+ match Arc::new(action).commit(&table).await {
+ Ok(_) => panic!("expected error"),
+ Err(e) => assert_eq!(e.kind(),
crate::ErrorKind::PreconditionFailed),
+ }
+ }
+
+ #[tokio::test]
+ async fn test_single_small_manifest_is_noop() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+ let table = append_one(&catalog, table, data_file("a", 1, 100,
1)).await;
+ let original_snapshot_id = table.metadata().current_snapshot_id();
+
+ let tx = Transaction::new(&table);
+ let action = tx.rewrite_manifests();
+ let mut commit = Arc::new(action).commit(&table).await.unwrap();
+ let updates = commit.take_updates();
+ assert!(updates.is_empty());
+
+ let table = tx
+ .rewrite_manifests()
+ .apply(Transaction::new(&table))
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+ assert_eq!(
+ table.metadata().current_snapshot_id(),
+ original_snapshot_id,
+ "no-op should not change the current snapshot"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_multi_manifest_merge_preserves_sequence_numbers() {
+ let catalog = new_memory_catalog().await;
+ let table = make_v3_minimal_table_in_catalog(&catalog).await;
+
+ let table = append_one(&catalog, table, data_file("a", 1, 1_000,
10)).await;
+ let seq_a = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ let table = append_one(&catalog, table, data_file("b", 1, 2_000,
20)).await;
+ let seq_b = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ let table = append_one(&catalog, table, data_file("c", 2, 3_000,
30)).await;
+ let seq_c = table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+ assert!(seq_a < seq_b && seq_b < seq_c);
+
+ let pre_manifest_count = table
+ .manifest_list_reader(table.metadata().current_snapshot().unwrap())
+ .load()
+ .await
+ .unwrap()
+ .entries()
+ .len();
+ assert_eq!(pre_manifest_count, 3);
+
+ let tx = Transaction::new(&table);
+ let table = tx
+ .rewrite_manifests()
+ .apply(tx)
+ .unwrap()
+ .commit(&catalog)
+ .await
+ .unwrap();
+
+ let snapshot = table.metadata().current_snapshot().unwrap();
+ assert_eq!(snapshot.summary().operation, Operation::Replace);
+
+ let post_list =
table.manifest_list_reader(snapshot).load().await.unwrap();
+ let total_entries: usize = {
+ let mut n = 0;
+ for m in post_list.entries() {
+ let manifest = m.load_manifest(table.file_io()).await.unwrap();
+ n += manifest.entries().len();
+ }
+ n
+ };
+ assert_eq!(total_entries, 3, "all entries preserved across rewrite");
+
+ let mut seen_seqs: Vec<i64> = Vec::new();
+ for m in post_list.entries() {
+ let manifest = m.load_manifest(table.file_io()).await.unwrap();
+ for entry in manifest.entries() {
+ seen_seqs.push(entry.sequence_number().unwrap());
+ }
+ }
+ seen_seqs.sort();
+ assert_eq!(seen_seqs, vec![seq_a, seq_b, seq_c]);
+
+ assert!(post_list.entries().len() < pre_manifest_count);
Review Comment:
`< pre_manifest_count` is looser than the actual invariant — files in
partitions 1 and 2 should consolidate to exactly 2 manifests, so I'd assert `==
2`. `<` would still pass if the rewrite silently dropped a partition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
