Method AddPuffinDeletionVectorEntries

be/src/runtime/dml-exec-state.cc:713–762

Helper function to add puffin file entries with deletion vectors. Creates one FbIcebergDataFile entry per data file that has deletion vectors. This is called for puffin files only.

// Creates one FbIcebergDataFile entry per data file that has deletion vectors.
// This is called for puffin files only.
void DmlExecState::AddPuffinDeletionVectorEntries(const OutputPartition& partition) {
  // Note: lock_ is already held by AddFileAux

  const string& puffin_path = partition.current_file_final_name.empty()
      ? partition.current_file_name
      : partition.current_file_final_name;

  // Build a unified map of all data files with deletion vectors (old and/or new)
  // Map: data_file_path -> (old_dv_ptr, new_dv_ptr)
  std::map<std::string,
      std::pair<const TIcebergDeletionVector*, const TIcebergDeletionVector*>>
      data_file_to_dvs;

  // Collect old deletion vectors
  for (const auto& entry : partition.puffin_result.old_deletion_vectors) {
    data_file_to_dvs[entry.first].first = &entry.second;
  }

  // Collect new deletion vectors
  for (const auto& entry : partition.puffin_result.new_deletion_vectors) {
    data_file_to_dvs[entry.first].second = &entry.second;
  }

  // Create one FbIcebergDataFile entry per data file with deletion vectors
  PartitionStatusMap::iterator partition_entry =
      per_partition_status_.find(partition.partition_name);
  DCHECK(partition_entry != per_partition_status_.end());

  for (const auto& entry : data_file_to_dvs) {
    const string& referenced_data_file = entry.first;
    const TIcebergDeletionVector* old_dv = entry.second.first;
    const TIcebergDeletionVector* new_dv = entry.second.second;

    DmlFileStatusPb* file = partition_entry->second.add_created_delete_files();

    file->set_final_path(puffin_path);
    if (!partition.current_file_final_name.empty()) {
      file->set_staging_path(partition.current_file_name);
    }

    // For deletion vectors, use the DV record count and the total puffin file size.
    file->set_num_rows(new_dv && new_dv->__isset.record_count ? new_dv->record_count : 0);
    file->set_size(partition.current_file_bytes);

    file->set_iceberg_data_file_fb(
        createIcebergDataFileWithDeletionVectorString(
            partition, puffin_path, partition.current_file_bytes,
            referenced_data_file, old_dv, new_dv));
  }
}

void DmlExecState::AddFileAux(const OutputPartition& partition, bool is_iceberg,
    const IcebergFileStats& insert_stats, bool is_delete) {