Method assign_row_tracking_meta

crates/paimon/src/table/table_commit.rs:672–705

Assign row tracking metadata: snapshot ID as sequence number, and first_row_id for new APPEND files that don't already have one. Normal files advance the main counter. Blob files (identified by file name) use per-column counters starting from the same base, since each blob column rolls independently.

(
        &self,
        snapshot_id: i64,
        first_row_id_start: i64,
        entries: Vec<ManifestEntry>,
    )

    /// use per-column counters starting from the same base, since each blob column
    /// rolls independently.
    fn assign_row_tracking_meta(
        &self,
        snapshot_id: i64,
        first_row_id_start: i64,
        entries: Vec<ManifestEntry>,
    ) -> (Vec<ManifestEntry>, i64) {
        let mut result = Vec::with_capacity(entries.len());
        let mut start = first_row_id_start;
        // Per blob column (write_cols key) counter, each starts from first_row_id_start.
        let mut blob_starts: HashMap<Vec<String>, i64> = HashMap::new();

        for entry in entries {
            let mut entry = entry.with_sequence_number(snapshot_id, snapshot_id);
            if *entry.kind() == FileKind::Add
                && entry.file().file_source == Some(0) // APPEND
                && entry.file().first_row_id.is_none()
            {
                let is_blob_file =
                    crate::table::blob_file_writer::is_blob_file_name(&entry.file().file_name);
                if is_blob_file {
                    let key = entry.file().write_cols.clone().unwrap_or_default();
                    let blob_start = blob_starts.entry(key).or_insert(first_row_id_start);
                    entry = entry.with_first_row_id(*blob_start);
                    *blob_start += entry.file().row_count;
                } else {
                    entry = entry.with_first_row_id(start);
                    start += entry.file().row_count;
                }
            }
            result.push(entry);
        }

        (result, start)
    }

    /// Validate that files with pre-assigned `first_row_id` (e.g. partial-column
    /// files from MERGE INTO) still match existing files in the current snapshot.