diff --git a/src/mito2/src/compaction/compactor.rs b/src/mito2/src/compaction/compactor.rs index 5c15da731c..17e5d83bf9 100644 --- a/src/mito2/src/compaction/compactor.rs +++ b/src/mito2/src/compaction/compactor.rs @@ -267,16 +267,18 @@ impl CompactionRegion { return; }; - // Local compaction fills `sst_infos` 1:1 with `files_to_add`. Remote - // compaction deserializes `MergeOutput` over gRPC, where `sst_infos` - // is `#[serde(skip)]`; substitute empty defaults — the hook still reads - // `record_count` from [`FileMeta::num_rows`]. - let defaults: Vec; + // Remote compaction deserializes `MergeOutput` over gRPC, where + // `sst_infos` is `#[serde(skip)]`. Rebuild each `SstInfo` from the + // matching `FileMeta` so the hook observes real ids/rows/sizes rather + // than zeros — see [`sst_info_from_file_meta`] for what stays default. + let synthesized: Vec; let infos: &[SstInfo] = if merge_output.sst_infos.is_empty() { - defaults = (0..merge_output.files_to_add.len()) - .map(|_| SstInfo::default()) + synthesized = merge_output + .files_to_add + .iter() + .map(sst_info_from_file_meta) .collect(); - &defaults + &synthesized } else { // `sst_infos` and `files_to_add` are documented as 1:1. If they // ever diverge, `zip` below would silently truncate; warn so the @@ -306,6 +308,41 @@ impl CompactionRegion { } } +/// Builds an [`SstInfo`] for the remote-compaction path by copying the scalar +/// fields that [`FileMeta`] also carries. +/// +/// Remote compaction runs off-process and ships `MergeOutput` over the wire; +/// [`MergeOutput::sst_infos`] is `#[serde(skip)]` because the parquet footer +/// (`ParquetMetaData`) is not serde-serializable. To avoid feeding the hook +/// default (zero) values for real SSTs, this rebuilds the seven fields `FileMeta` +/// and `SstInfo` share: `file_id`, `time_range`, `file_size`, +/// `max_row_group_uncompressed_size`, `num_rows`, `num_row_groups`, `num_series`. +/// +/// Two fields stay default and **cannot be recovered on the datanode**: +/// - `file_metadata` — the parquet footer, whose column min/max/null-count +/// statistics are required by hooks building richer artifacts (e.g. an Iceberg +/// manifest). That data is produced by the compactor's writer and exists only +/// in the freshly-written SST on object storage. +/// - `index_metadata` — `FileMeta` tracks indexes via `available_indexes` / +/// `indexes`, not as an [`IndexOutput`]. +/// +/// A hook that needs column stats must fetch the footer from object storage +/// (the [`CompactionRegion`]'s `access_layer` reaches the store), or the footer +/// must be shipped over the wire (revisit the `#[serde(skip)]` on `sst_infos`). +/// `num_rows` may read `0` for legacy `FileMeta`s where the count is unknown. +fn sst_info_from_file_meta(meta: &FileMeta) -> SstInfo { + SstInfo { + file_id: meta.file_id, + time_range: meta.time_range, + file_size: meta.file_size, + max_row_group_uncompressed_size: meta.max_row_group_uncompressed_size, + num_rows: meta.num_rows as usize, + num_row_groups: meta.num_row_groups, + num_series: meta.num_series, + ..Default::default() + } +} + /// `[MergeOutput]` represents the output of merging SST files. #[derive(Default, Debug, Serialize, Deserialize)] pub struct MergeOutput {