laskoviymishka commented on code in PR #3474: URL: https://github.com/apache/iceberg-rust/pull/3474#discussion_r4341765520
########## crates/iceberg/src/transaction/simple_snapshot_producer.rs: ########## @@ -0,0 +1,630 @@ +// 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. + +//! Persistent producer for append-shaped snapshot actions. + +use std::collections::{HashMap, HashSet}; + +use futures::TryStreamExt; +use futures::stream::FuturesUnordered; +use uuid::Uuid; + +use crate::error::{Result, invalid_data}; +use crate::spec::{ + DataContentType, DataFile, FormatVersion, ManifestContentType, ManifestEntry, ManifestFile, + ManifestStatus, Operation, +}; +use crate::table::Table; +use crate::transaction::ActionCommit; +use crate::transaction::snapshot_helpers::{ + SnapshotCommitInput, current_manifests, generate_snapshot_id, manifest_list_path, + manifest_path, new_manifest_writer, validate_partition_value, write_snapshot_commit, +}; +use crate::{Error, ErrorKind}; + +/// Retry-persistent state for an append-shaped action: added data files go +/// into new manifests, and the current snapshot's manifests are carried +/// forward untouched (no filtering, no merging). +/// +/// One producer lives for one logical execution of an action. It is driven +/// once per attempt via [`SimpleSnapshotProducer::apply`]: logical identity +/// (commit UUID, snapshot id) is fixed on first use, the manifests written for +/// the added files are reused across attempts while their semantic inputs are +/// unchanged, and every attempt writes a fresh manifest list under a distinct +/// path. +pub(crate) struct SimpleSnapshotProducer { + /// Fixed when the producer is created, so every attempt shares one UUID. + commit_uuid: Uuid, + /// Resolved against the table on the first attempt, then stable. + snapshot_id: Option<i64>, + /// Incremented per attempt; distinguishes manifest-list paths. + attempt: u64, + /// Numbers this producer's manifests. Never reset, so reused manifests keep + /// their paths and new ones never collide with earlier attempts. + manifest_counter: u64, + /// Manifests written for the added files, reused across attempts. + added_manifests: Option<AddedManifests>, + /// Metadata files written by this producer (manifests and every attempt's + /// manifest list). + // TODO: cleanup is not implemented yet; terminal cleanup (stateful Review Comment: Could this TODO and the matching one in `cleanup` link an issue, like the #3499 and #3500 TODOs below? Until cleanup lands, `owned_artifacts` is write-only behind `#[allow(dead_code)]`. Java's `SnapshotProducer` deletes the unused manifest lists from failed attempts ([here](https://github.com/apache/iceberg/blob/2114bf631e49af532d66e2ce148ee49dd1dd1f1f/core/src/main/java/org/apache/iceberg/SnapshotProducer.java#L486-L496)), so we should track the gap. ########## crates/iceberg/src/transaction/append.rs: ########## @@ -995,4 +944,135 @@ mod tests { ); assert_eq!(data_file, *manifest.entries()[0].data_file()); } + + /// Two writers append concurrently. Writer A starts from S0, but writer B + /// commits S1 first, so A's first attempt fails the `RefSnapshotIdMatch` + /// requirement and A rebases onto S1 to commit S2. + /// + /// The producer state survives the retry: A's manifest is written once and + /// reused, while parent, sequence number, manifest list and summary are + /// recomputed against S1 (see #3418). + #[tokio::test] + async fn test_fast_append_retry_rebases_onto_concurrent_commit() { + // `TableCommit::apply` derives the next metadata location from the + // current one, so it needs a well-formed name. + let s0 = with_retry_properties(make_v2_minimal_table(), "3").with_metadata_location( Review Comment: Could we add a v3 version of this rebase test? On v3 the retry also has to move the row range. `first_row_id` comes from the rebased table's `next_row_id` ([here](https://github.com/apache/iceberg-rust/blob/f1361c5f1724174c92f0af4ea7850ea338135585/crates/iceberg/src/transaction/snapshot_helpers.rs#L281)) and feeds `with_row_range` ([here](https://github.com/apache/iceberg-rust/blob/f1361c5f1724174c92f0af4ea7850ea338135585/crates/iceberg/src/transaction/snapshot_helpers.rs#L345-L347)). The only row-lineage test commits sequentially ([here](https://github.com/apache/iceberg-rust/blob/f1361c5f1724174c92f0af4ea7850ea338135585/crates/iceberg/src/transaction/mod.rs#L687-L760)), so nothing checks that S2's row range starts at S1's `next_row_id`. ########## crates/iceberg/src/transaction/simple_snapshot_producer.rs: ########## @@ -0,0 +1,630 @@ +// 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. + +//! Persistent producer for append-shaped snapshot actions. + +use std::collections::{HashMap, HashSet}; + +use futures::TryStreamExt; +use futures::stream::FuturesUnordered; +use uuid::Uuid; + +use crate::error::{Result, invalid_data}; +use crate::spec::{ + DataContentType, DataFile, FormatVersion, ManifestContentType, ManifestEntry, ManifestFile, + ManifestStatus, Operation, +}; +use crate::table::Table; +use crate::transaction::ActionCommit; +use crate::transaction::snapshot_helpers::{ + SnapshotCommitInput, current_manifests, generate_snapshot_id, manifest_list_path, + manifest_path, new_manifest_writer, validate_partition_value, write_snapshot_commit, +}; +use crate::{Error, ErrorKind}; + +/// Retry-persistent state for an append-shaped action: added data files go +/// into new manifests, and the current snapshot's manifests are carried +/// forward untouched (no filtering, no merging). +/// +/// One producer lives for one logical execution of an action. It is driven +/// once per attempt via [`SimpleSnapshotProducer::apply`]: logical identity +/// (commit UUID, snapshot id) is fixed on first use, the manifests written for +/// the added files are reused across attempts while their semantic inputs are +/// unchanged, and every attempt writes a fresh manifest list under a distinct +/// path. +pub(crate) struct SimpleSnapshotProducer { + /// Fixed when the producer is created, so every attempt shares one UUID. + commit_uuid: Uuid, + /// Resolved against the table on the first attempt, then stable. + snapshot_id: Option<i64>, + /// Incremented per attempt; distinguishes manifest-list paths. + attempt: u64, + /// Numbers this producer's manifests. Never reset, so reused manifests keep + /// their paths and new ones never collide with earlier attempts. + manifest_counter: u64, + /// Manifests written for the added files, reused across attempts. + added_manifests: Option<AddedManifests>, + /// Metadata files written by this producer (manifests and every attempt's + /// manifest list). + // TODO: cleanup is not implemented yet; terminal cleanup (stateful + // transaction RFC, milestone 3) consumes this set. + #[allow(dead_code)] + owned_artifacts: HashSet<String>, +} + +/// The manifests written for one append's added files, together with the +/// table-level inputs they were written under. +/// +/// A manifest file embeds its schema and partition spec and is encoded for a +/// specific format version, so the set is only reusable while the table it is +/// rebased onto still has the same schema and format version. The partition +/// spec is implied: every file is validated against the default spec, so the +/// spec is fixed for the lifetime of the producer by that check. +/// +/// The guard is table-level because its inputs are table-level: when they +/// change, every manifest in the set is invalid together. +struct AddedManifests { + /// One manifest today. Several once added files are grouped by partition + /// spec (#3500) or rolled over at a target size (#3499). + manifests: Vec<ManifestFile>, + schema_id: i32, + format_version: FormatVersion, +} + +impl SimpleSnapshotProducer { + /// Creates a producer. `commit_uuid` overrides the generated commit UUID. + pub(crate) fn new(commit_uuid: Option<Uuid>) -> Self { + Self { + commit_uuid: commit_uuid.unwrap_or_else(Uuid::now_v7), + snapshot_id: None, + attempt: 0, + manifest_counter: 0, + added_manifests: None, + owned_artifacts: HashSet::new(), + } + } + + /// Checks that every file is a data file written under the table's default + /// partition spec with a compatible partition value. + pub(crate) fn validate_added_data_files( + &self, + table: &Table, + data_files: &[DataFile], + ) -> Result<()> { + let metadata = table.metadata(); + for data_file in data_files { + if data_file.content_type() != DataContentType::Data { + return Err(invalid_data!( + "Only data content type is allowed for fast append" + )); + } + // TODO(#3500): accept any spec known to the table and group the + // files by spec when writing manifests, instead of requiring the + // default spec. + if metadata.default_partition_spec_id() != data_file.partition_spec_id { + return Err(invalid_data!( + "Data file partition spec id does not match table default partition spec id" + )); + } + validate_partition_value(data_file.partition(), metadata.default_partition_type())?; + } + Ok(()) + } + + /// Fails if any of `data_files` is already live in the current snapshot. + /// + /// This must re-run against each attempt's base, since a concurrent commit + /// may have added one of these files. Manifest reads go through the table's + /// object cache, so unchanged manifests are not re-read on retry. + pub(crate) async fn validate_duplicate_files( + &self, + table: &Table, + data_files: &[DataFile], + ) -> Result<()> { + let Some(current_snapshot) = table.metadata().current_snapshot() else { + return Ok(()); + }; + + let new_files: HashSet<&str> = data_files.iter().map(|df| df.file_path()).collect(); + if new_files.is_empty() { + return Ok(()); + } + + let manifest_list = table.manifest_list_reader(current_snapshot).load().await?; + let runtime = table.runtime(); + let object_cache = table.object_cache(); + + let new_files_ref = &new_files; + let referenced_files: Vec<String> = manifest_list + .consume_entries() + .into_iter() + .map(|entry| { + let object_cache = object_cache.clone(); + runtime + .io() + .spawn(async move { object_cache.get_manifest(&entry).await }) + }) + .collect::<FuturesUnordered<_>>() + .try_fold(Vec::new(), |mut acc, manifest| async move { + acc.extend( + manifest? + .entries() + .iter() + .filter(|e| new_files_ref.contains(e.file_path()) && e.is_alive()) + .map(|e| e.file_path().to_string()), + ); + Ok(acc) + }) + .await?; + + if !referenced_files.is_empty() { + return Err(invalid_data!( + "Cannot add files that are already referenced by table, files: {}", + referenced_files.join(", ") + )); + } + + Ok(()) + } + + /// Runs one attempt: produces the resulting manifest set for the current + /// table and writes this attempt's snapshot. + pub(crate) async fn apply( + &mut self, + table: &Table, + properties: HashMap<String, String>, + data_files: &[DataFile], + ) -> Result<ActionCommit> { + // TODO: Allowing snapshot property setup with no added data files is a workaround. + // We should clean it up after all necessary actions are supported. + // For details, please refer to https://github.com/apache/iceberg-rust/issues/1548 + if data_files.is_empty() && properties.is_empty() { + return Err(Error::new( + ErrorKind::PreconditionFailed, + "No added data files or added snapshot properties found when write a manifest file", + )); + } + + let snapshot_id = self.snapshot_id(table); + + let mut manifests = current_manifests(table).await?; + if !data_files.is_empty() { + manifests.extend(self.added_manifests(table, snapshot_id, data_files).await?); + } + + let manifest_list_path = self.next_manifest_list_path(table, snapshot_id)?; + + write_snapshot_commit(table, SnapshotCommitInput { + snapshot_id, + manifest_list_path, + operation: Operation::Append, + properties, + added_files: data_files, + removed_files: &[], + manifests, + truncate_full_table: false, + }) + .await + } + + fn snapshot_id(&mut self, table: &Table) -> i64 { + *self + .snapshot_id + .get_or_insert_with(|| generate_snapshot_id(table)) + } + + fn next_manifest_list_path(&mut self, table: &Table, snapshot_id: i64) -> Result<String> { + let path = manifest_list_path(table, snapshot_id, self.attempt, self.commit_uuid)?; + self.attempt += 1; + self.owned_artifacts.insert(path.clone()); + Ok(path) + } + + /// Returns the manifests for the added files, writing them only if no + /// reusable set exists for the current schema and format version. + async fn added_manifests( + &mut self, + table: &Table, + snapshot_id: i64, + data_files: &[DataFile], + ) -> Result<Vec<ManifestFile>> { + let metadata = table.metadata(); + let schema_id = metadata.current_schema_id(); + let format_version = metadata.format_version(); + + if let Some(cached) = &self.added_manifests Review Comment: Should `AddedManifests` also record the spec id and the added files, and skip reuse when either changes? The check compares only `schema_id` and `format_version`, but the summary is built from whatever `data_files` this call gets ([here](https://github.com/apache/iceberg-rust/blob/f1361c5f1724174c92f0af4ea7850ea338135585/crates/iceberg/src/transaction/simple_snapshot_producer.rs#L216)). Today it's correct for two reasons. `FastAppendAction` passes the same files on every attempt ([here](https://github.com/apache/iceberg-rust/blob/f1361c5f1724174c92f0af4ea7850ea338135585/crates/iceberg/src/transaction/append.rs#L95-L105)), and the default-spec check pins the spec ([here](https://github.com/apache/iceberg-rust/blob/f1361c5f1724174c92f0af4ea7850ea338135585/crates/iceberg/src/transaction/simple_snapshot_producer.rs#L115-L123)), but `TODO(#3500)` removes that check. Comparing them, or moving the files into the producer, both work. We should pick one before #3500 builds on this. ########## crates/iceberg/src/transaction/snapshot_helpers.rs: ########## @@ -0,0 +1,542 @@ +// 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. + +//! Stateless building blocks shared by snapshot producers. +//! +//! Producers own retry-persistent state (identity, reusable artifacts); these +//! helpers only translate explicit inputs into Iceberg metadata and files. +//! They never select operation policy and never cache across calls. + +use std::collections::HashMap; + +use uuid::Uuid; + +use crate::error::{Result, invalid_data}; +use crate::spec::{ + DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType, ManifestFile, + ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpecRef, + SchemaRef, Snapshot, SnapshotReference, SnapshotRetention, SnapshotSummaryCollector, Struct, + StructType, Summary, TableProperties, set_iceberg_version, update_snapshot_summaries, +}; +use crate::table::Table; +use crate::transaction::ActionCommit; +use crate::{Error, ErrorKind, TableRequirement, TableUpdate}; + +/// Generates a positive snapshot id not already present in `table`. +pub(crate) fn generate_snapshot_id(table: &Table) -> i64 { + loop { + let (lhs, rhs) = Uuid::new_v4().as_u64_pair(); + let snapshot_id = (lhs ^ rhs) as i64 & i64::MAX; + if table.metadata().snapshot_by_id(snapshot_id).is_none() { + return snapshot_id; + } + } +} + +/// Returns the metadata path for a producer-owned manifest. +/// +/// `counter` must be unique per `commit_uuid` for the lifetime of the producer, +/// so that no attempt overwrites a manifest written by an earlier one. +pub(crate) fn manifest_path(table: &Table, commit_uuid: Uuid, counter: u64) -> Result<String> { + Ok(format!( + "{}/{}-m{}.{}", + table.metadata().metadata_location()?, + commit_uuid, + counter, + DataFileFormat::Avro + )) +} + +/// Returns the metadata path for one attempt's manifest list. +/// +/// Each attempt gets a distinct `attempt` number so the path is write-once. +pub(crate) fn manifest_list_path( + table: &Table, + snapshot_id: i64, + attempt: u64, + commit_uuid: Uuid, +) -> Result<String> { + Ok(format!( + "{}/snap-{}-{}-{}.{}", + table.metadata().metadata_location()?, + snapshot_id, + attempt, + commit_uuid, + DataFileFormat::Avro + )) +} + +/// Creates a manifest writer at `path`, honoring the table's format version and +/// encryption configuration, under the given schema and partition spec. +pub(crate) fn new_manifest_writer( + table: &Table, + path: String, + snapshot_id: i64, + content: ManifestContentType, + schema: SchemaRef, + partition_spec: PartitionSpecRef, +) -> Result<ManifestWriter> { + let output_file = table.file_io().new_output(path)?; + let partition_spec = partition_spec.as_ref().clone(); + + let builder = if let Some(em) = table.encryption_manager() { + ManifestWriterBuilder::new_from_encrypted( + em.encrypt(output_file), + Some(snapshot_id), + schema, + partition_spec, + )? + } else { + ManifestWriterBuilder::new(output_file, Some(snapshot_id), schema, partition_spec) + }; + + match table.metadata().format_version() { + FormatVersion::V1 => Ok(builder.build_v1()), + FormatVersion::V2 => match content { + ManifestContentType::Data => Ok(builder.build_v2_data()), + ManifestContentType::Deletes => Ok(builder.build_v2_deletes()), + }, + FormatVersion::V3 => match content { + ManifestContentType::Data => Ok(builder.build_v3_data()), + ManifestContentType::Deletes => Ok(builder.build_v3_deletes()), + }, + } +} + +/// Checks that a partition value is compatible with the partition type. +pub(crate) fn validate_partition_value( + partition_value: &Struct, + partition_type: &StructType, +) -> Result<()> { + if partition_value.fields().len() != partition_type.fields().len() { + return Err(invalid_data!( + "Partition value is not compatible with partition type" + )); + } + + for (value, field) in partition_value.fields().iter().zip(partition_type.fields()) { + let field = field.field_type.as_primitive_type().ok_or_else(|| { + Error::new( + ErrorKind::Unexpected, + "Partition field should only be primitive type.", + ) + })?; + if let Some(value) = value + && !field.compatible(&value.as_primitive_literal().unwrap()) + { + return Err(invalid_data!( + "Partition value is not compatible partition type" + )); + } + } + Ok(()) +} + +/// Returns the current snapshot's manifests that a new snapshot carries forward. +/// +/// Delete-only manifests are kept too: they record which files were removed and +/// must persist across snapshots until `expire_snapshots` cleans them up. +/// Dropping them lets the removed files reappear as live data (see #2148). +pub(crate) async fn current_manifests(table: &Table) -> Result<Vec<ManifestFile>> { + let Some(snapshot) = table.metadata().current_snapshot() else { + return Ok(vec![]); + }; + + let manifest_list = table.manifest_list_reader(snapshot).load().await?; + + Ok(manifest_list + .consume_entries() + .into_iter() + .filter(|entry| { + entry.has_added_files() || entry.has_existing_files() || entry.has_deleted_files() + }) + .collect()) +} + +/// Builds a snapshot summary from an explicit set of file changes. +/// +/// User-supplied `properties` are applied first, then the computed metrics +/// overwrite any colliding keys. This matches iceberg-java +/// (`SnapshotProducer.summary`), where computed `added-*`/`total-*` values are +/// written after user properties so a user cannot shadow them with a bad (or +/// merely wrong) value that would corrupt the snapshot summary. +pub(crate) fn build_summary( Review Comment: Can this stay private until `MergingSnapshotProducer` needs it? Its only callers are `write_snapshot_commit` and the test in this module. On `main` it was a private method on `SnapshotProducer` ([here](https://github.com/apache/iceberg-rust/blob/219571a4185df2bc6e144dbbcb72cf016714b61b/crates/iceberg/src/transaction/snapshot.rs#L398-L401)). ```suggestion fn build_summary( ``` -- 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]
