write documentaion

Author: Jan Kaul

iceberg-rust/src/table/manifest.rs

@@ -365,6 +365,41 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> {
         })
     }
 
+    /// Creates a ManifestWriter from an existing manifest file with selective filtering of entries.
+    ///
+    /// This method reads an existing manifest file and creates a new writer that includes
+    /// only the entries whose file paths are NOT in the provided filter set. Entries that
+    /// pass the filter have their status updated to "Existing" and their sequence numbers
+    /// and snapshot IDs updated as needed.
+    ///
+    /// This is particularly useful for overwrite operations where specific files need to be
+    /// excluded from the new manifest while preserving other existing entries.
+    ///
+    /// # Arguments
+    /// * `bytes` - The raw bytes of the existing manifest file
+    /// * `manifest` - The manifest list entry describing the existing manifest
+    /// * `filter` - A set of file paths to exclude from the new manifest
+    /// * `schema` - The Avro schema used for serializing manifest entries
+    /// * `table_metadata` - The table metadata containing schema and partition information
+    /// * `branch` - Optional branch name to get the current schema from
+    ///
+    /// # Returns
+    /// * `Result<Self, Error>` - A new ManifestWriter instance or an error if initialization fails
+    ///
+    /// # Errors
+    /// Returns an error if:
+    /// * The existing manifest cannot be read
+    /// * The Avro writer cannot be created
+    /// * Required metadata fields cannot be serialized
+    /// * The partition spec ID is not found in table metadata
+    ///
+    /// # Behavior
+    /// - Entries whose file paths are in the `filter` set are excluded from the new manifest
+    /// - Remaining entries have their status set to `Status::Existing`
+    /// - Sequence numbers are updated for entries that don't have them
+    /// - Snapshot IDs are updated for entries that don't have them
+    /// - The manifest's sequence number is incremented
+    /// - File counts are updated to reflect the filtered entries
     pub(crate) fn from_existing_with_filter(
         bytes: &[u8],
         mut manifest: ManifestListEntry,

iceberg-rust/src/table/manifest_list.rs

@@ -334,6 +334,53 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
         })
     }
 
+    /// Creates a ManifestListWriter from an existing manifest list, excluding manifests scheduled for overwriting.
+    ///
+    /// This constructor is specifically designed for overwrite operations where certain manifests
+    /// need to be replaced while preserving others. It analyzes an existing manifest list and:
+    /// 1. Identifies manifests that should be overwritten (excluded from the new manifest list)
+    /// 2. Selects compatible manifests that can be reused for appending new data
+    /// 3. Copies non-selected, non-overwritten manifests to the new manifest list
+    /// 4. Returns both the writer and the list of manifests that will be overwritten
+    ///
+    /// This approach optimizes overwrite operations by:
+    /// - Avoiding unnecessary rewrites of unaffected manifests
+    /// - Providing efficient append capabilities for new data
+    /// - Returning metadata about what will be overwritten for cleanup operations
+    ///
+    /// # Arguments
+    /// * `bytes` - The raw bytes of the existing manifest list file
+    /// * `data_files` - Iterator over new data files to be appended
+    /// * `manifests_to_overwrite` - Set of manifest paths that should be excluded/overwritten
+    /// * `schema` - The Avro schema to use for manifest list serialization
+    /// * `table_metadata` - Reference to the table metadata for partition field information
+    /// * `branch` - Optional branch name for multi-branch table operations
+    ///
+    /// # Returns
+    /// * `Result<(Self, Vec<ManifestListEntry>), Error>` - A tuple containing:
+    ///   - A new ManifestListWriter instance with selected manifest for appends
+    ///   - A vector of ManifestListEntry objects that will be overwritten
+    ///
+    /// # Errors
+    /// Returns an error if:
+    /// * The existing manifest list cannot be parsed
+    /// * Partition fields cannot be retrieved from table metadata
+    /// * Partition boundary computation fails
+    /// * Manifest selection logic fails
+    /// * The Avro writer cannot be initialized
+    ///
+    /// # Example Usage
+    /// ```ignore
+    /// let manifests_to_overwrite = HashSet::from(["manifest1.avro", "manifest2.avro"]);
+    /// let (writer, overwritten_manifests) = ManifestListWriter::from_existing_without_overwrites(
+    ///     &existing_manifest_list_bytes,
+    ///     new_data_files.iter(),
+    ///     &manifests_to_overwrite,
+    ///     &manifest_list_schema,
+    ///     &table_metadata,
+    ///     Some("main"),
+    /// )?;
+    /// ```
     pub(crate) fn from_existing_without_overwrites<'datafiles>(
         bytes: &[u8],
         data_files: impl Iterator<Item = &'datafiles DataFile>,
@@ -494,6 +541,60 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
         .await
     }
 
+    /// Appends data files to a single manifest with optional filtering and finalizes the manifest list.
+    ///
+    /// This method extends the basic `append_and_finish` functionality by providing the ability to
+    /// filter data files during the append process. It creates a single manifest file containing
+    /// the provided data files (after filtering), either by appending to an existing reusable
+    /// manifest or creating a new one.
+    ///
+    /// The filtering capability is particularly useful for:
+    /// - Excluding certain files from being included in the manifest
+    /// - Conditional processing based on file properties or metadata
+    /// - Implementing custom business logic during manifest creation
+    /// - Selective processing of existing manifest entries when reusing manifests
+    ///
+    /// This approach is optimal for:
+    /// - Small to medium append operations with conditional logic
+    /// - Cases where certain files need to be excluded or processed differently
+    /// - Operations requiring custom filtering logic during manifest creation
+    ///
+    /// The process:
+    /// 1. Determines whether to reuse an existing manifest or create a new one
+    /// 2. If reusing, applies the filter when reading existing manifest entries
+    /// 3. Creates/updates a manifest writer with the selected manifest
+    /// 4. Appends all provided data files to the manifest
+    /// 5. Finalizes the manifest and writes it to storage
+    /// 6. Adds the manifest entry to the manifest list
+    /// 7. Writes the complete manifest list to storage
+    ///
+    /// # Arguments
+    /// * `data_files` - Iterator over manifest entries to append
+    /// * `snapshot_id` - The snapshot ID for the new manifest
+    /// * `filter` - Optional filter function to apply to existing manifest entries when reusing
+    /// * `object_store` - The object store for writing files
+    ///
+    /// # Returns
+    /// * `Result<String, Error>` - The location of the new manifest list file or an error
+    ///
+    /// # Errors
+    /// Returns an error if:
+    /// * Partition field retrieval fails
+    /// * Manifest schema creation fails
+    /// * Manifest writer creation or operation fails
+    /// * Object storage operations fail
+    /// * Avro serialization fails
+    /// * Filter function encounters an error
+    ///
+    /// # Example Usage
+    /// ```ignore
+    /// let manifest_list_location = writer.append_filtered_and_finish(
+    ///     data_files_iter,
+    ///     snapshot_id,
+    ///     Some(|entry| entry.as_ref().map(|e| e.status() == &Status::Added).unwrap_or(false)),
+    ///     object_store,
+    /// ).await?;
+    /// ```
     pub(crate) async fn append_filtered_and_finish(
         mut self,
         data_files: impl Iterator<Item = Result<ManifestEntry, Error>>,
@@ -644,6 +745,67 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
         .await
     }
 
+    /// Appends data files across multiple manifests with optional filtering and finalizes the manifest list.
+    ///
+    /// This method extends the `append_multiple_and_finish` functionality by providing the ability to
+    /// filter data files during the append and splitting process. It distributes the data files
+    /// (after filtering) across the specified number of splits based on partition boundaries,
+    /// optimizing for large operations that require conditional processing.
+    ///
+    /// The filtering capability is particularly useful for:
+    /// - Excluding certain files from being included in any manifest
+    /// - Conditional processing based on file properties, status, or metadata
+    /// - Implementing custom business logic during large-scale manifest operations
+    /// - Selective processing of existing manifest entries when reusing manifests
+    /// - Complex overwrite scenarios where certain entries need special handling
+    ///
+    /// This approach is optimal for:
+    /// - Large append operations with hundreds or thousands of files requiring filtering
+    /// - Partitioned tables where files need both splitting and filtering
+    /// - Complex operations combining append, overwrite, and conditional logic
+    /// - Cases requiring high query parallelism with selective data inclusion
+    ///
+    /// The process:
+    /// 1. Computes optimal partition boundaries for splitting
+    /// 2. If reusing an existing manifest, applies filter when reading existing entries
+    /// 3. Merges new data files with filtered existing files from selected manifest
+    /// 4. Splits all files across the specified number of manifest files
+    /// 5. Creates and writes multiple manifest files concurrently
+    /// 6. Adds all manifest entries to the manifest list
+    /// 7. Writes the complete manifest list to storage
+    ///
+    /// # Arguments
+    /// * `data_files` - Iterator over manifest entries to append and split
+    /// * `snapshot_id` - The snapshot ID for the new manifests
+    /// * `n_splits` - The number of manifest files to create (should match `n_splits()` result)
+    /// * `filter` - Optional filter function to apply to existing manifest entries when reusing
+    /// * `object_store` - The object store for writing files
+    ///
+    /// # Returns
+    /// * `Result<String, Error>` - The location of the new manifest list file or an error
+    ///
+    /// # Errors
+    /// Returns an error if:
+    /// * Partition field retrieval fails
+    /// * Manifest schema creation fails
+    /// * File splitting logic fails
+    /// * Manifest writer creation or operation fails
+    /// * Concurrent manifest writing fails
+    /// * Object storage operations fail
+    /// * Avro serialization fails
+    /// * Filter function encounters an error
+    ///
+    /// # Example Usage
+    /// ```ignore
+    /// let n_splits = writer.n_splits(data_files.len());
+    /// let manifest_list_location = writer.append_multiple_filtered_and_finish(
+    ///     data_files_iter,
+    ///     snapshot_id,
+    ///     n_splits,
+    ///     Some(|entry| entry.as_ref().map(|e| e.status() != &Status::Deleted).unwrap_or(false)),
+    ///     object_store,
+    /// ).await?;
+    /// ```
     pub(crate) async fn append_multiple_filtered_and_finish(
         mut self,
         data_files: impl Iterator<Item = Result<ManifestEntry, Error>>,
@@ -764,6 +926,67 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> {
         Ok(new_manifest_list_location)
     }
 
+    /// Processes manifests for overwrite operations by filtering out specific data files.
+    ///
+    /// This method is specifically designed for complex overwrite scenarios where certain data files
+    /// within existing manifests need to be removed while preserving others. It processes a list of
+    /// manifests, filters out specified data files from each one, and adds the filtered manifests
+    /// to the manifest list being constructed.
+    ///
+    /// This operation is essential for:
+    /// - **Overwrite operations**: Removing specific files that are being replaced by new data
+    /// - **Partial table updates**: Selectively removing files while keeping others
+    /// - **Data deduplication**: Filtering out duplicate or obsolete data files
+    /// - **Complex merge operations**: Managing file-level changes during table merges
+    ///
+    /// The method operates at the manifest level rather than the manifest list level, providing
+    /// fine-grained control over which data files are included in the final table state.
+    ///
+    /// The process:
+    /// 1. Processes each manifest in the provided list concurrently
+    /// 2. For each manifest, retrieves the list of data files to filter out
+    /// 3. Loads the manifest content from object storage
+    /// 4. Creates a new manifest location and updates the manifest path
+    /// 5. Uses `ManifestWriter::from_existing_with_filter` to exclude specified files
+    /// 6. Writes the filtered manifest to storage with a new location
+    /// 7. Adds the new manifest entry to the manifest list being constructed
+    ///
+    /// # Arguments
+    /// * `manifests_to_overwrite` - Vector of manifest list entries to process and filter
+    /// * `data_files_to_filter` - Map from manifest path to list of data file paths to exclude
+    /// * `object_store` - The object store for reading existing and writing new manifest files
+    ///
+    /// # Returns
+    /// * `Result<(), Error>` - Ok if all manifests were successfully processed and filtered
+    ///
+    /// # Errors
+    /// Returns an error if:
+    /// * A manifest path is not found in the `data_files_to_filter` map
+    /// * Object storage operations fail (reading existing or writing new manifests)
+    /// * Manifest parsing or writing operations fail
+    /// * Avro serialization fails
+    /// * Concurrent processing encounters errors
+    ///
+    /// # Example Usage
+    /// ```ignore
+    /// let mut manifest_list_writer = ManifestListWriter::new(...)?;
+    /// let data_files_to_filter = HashMap::from([
+    ///     ("manifest1.avro".to_string(), vec!["file1.parquet".to_string(), "file2.parquet".to_string()]),
+    ///     ("manifest2.avro".to_string(), vec!["file3.parquet".to_string()]),
+    /// ]);
+    /// 
+    /// manifest_list_writer.append_and_filter(
+    ///     manifests_to_overwrite,
+    ///     &data_files_to_filter,
+    ///     object_store,
+    /// ).await?;
+    /// ```
+    ///
+    /// # Implementation Notes
+    /// - Manifests are processed concurrently for optimal performance
+    /// - Each filtered manifest gets a new location to avoid conflicts
+    /// - The method modifies the manifest list writer's internal state by adding filtered manifests
+    /// - This method is typically called as part of a larger overwrite operation workflow
     pub(crate) async fn append_and_filter(
         &mut self,
         manifests_to_overwrite: Vec<ManifestListEntry>,