mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 16:09:05 +00:00
Huaijin
0641e1e486
## Why The kind of merge — plain time-ordered file, trace time-index aggregation, file-list ordering, enterprise metrics downsampling — was re-derived at every layer, each from stream type / stream name / config / batch time range: - compactor scheduler (`merge_by_stream`): `skip_group_files` (downsampling rule lookup #1) → whole-hour batch vs size groups - `merge_files`: `is_match_downsampling_rule` (lookup #2) → early returns, size-limit bypass - `merge_parquet_files`: lookup #3 + SQL selection by stream type, with `is_ingester: bool` standing in for the writer's role `merge_parquet_files` and `merge_parquet_files_with_downsampling` also carried two copies of the DataFusion context / union table / plan / execute-stream / channel / read-task plumbing. Reading the code meant reconciling three derivations of the same fact to know what a batch would produce. ## What - **`search::datafusion::merge::mode::MergeMode { Classic, TraceTimeIndex, FileList, Downsampling(rule) }`** — decided once: `MergeMode::for_compactor(stream_type, stream_name, max_ts)` by the scheduler (carried in `MergeBatch.mode`), `MergeMode::for_ingester(stream_type, stream_name)` by the movers. Every layer only asks it questions: `merges_whole_batch()`, `input_sort_order()`, `sql(schema)`. - **`MergeOutput { file_format, parquet_compression }`** (`for_ingester` / `for_compactor`) replaces the `is_ingester` bool. - **`merge_parquet_files(schema, tables, bloom, meta, &mode, output)`**: one `run_merge_query()` pipeline for every mode, then the mode's writer. Downsampling is a `MergeMode` arm calling `downsampling::write_files` instead of an early-return second implementation (`merge_parquet_files_with_downsampling` removed, −81 lines). - Compactor: `skip_group_files` / `is_match_downsampling_rule` and their `cfg(enterprise)` twin declarations are gone; `compaction` no longer imports `get_largest_downsampling_rule`. Commit 1 (`672ab315a1`) is pure refactor, no behavior change. ## Behavior change (commit 2, `8030eefd04`): whole-hour merges only for a closed hour, over every file of the hour Metrics downsampling used to merge the whole batch whenever the rule matched the batch's newest sample — including incremental rounds on the still-open hour — and it only ever saw the files below the 95%-of-`ZO_COMPACT_MAX_FILE_SIZE` bound of `query_for_merge`: files already above the bound were never downsampled, and small outputs were re-aggregated round after round. - `MergeMode::for_compactor(stream_type, stream_name, range_end_ts, finalize)` is decided **once per hour, before the file_list query**, from the end of the job's date range; a whole-hour mode (downsampling) is only chosen when `finalize` (the hour is closed). Incremental merges of a matching hour use the plain per-stream mode and leave the aggregation to the hour-end pass. - `query_for_merge` takes the `original_size` bound from the caller (`infra::file_list::merge_max_original_size()` = the classic 95% rule); a whole-hour merge asks for **every** file of the hour (`i64::MAX`), the same input range the TSID-major hour-end merge in #13839 uses. - Rule matching uses the hour's end instead of the batch's max `max_ts` (≤ 1h difference; the hour is decided as a unit). ## Guards (commit 3, cherry-picked from #13839) - compactor: fail the merge instead of deleting the sources when `merge_parquet_files` produced no files - search storage `file_list::get_scan_selection` strips the multi-account `{account}/::/` prefix so selections registered by `set` are found again (+ round-trip test) - `deleted.rs`: `delete_from_storage` takes the file slice and a key closure (`Cow`) instead of a pre-collected `Vec` ## Not in this PR The write stage still dispatches to four writers (`write_parquet`, `write_vortex`, `write_downsampled_parquet`, `write_downsampled_vortex`) that share the open / write / rotate / close skeleton and differ only in the rotation policy and how `FileMeta` is derived. Unifying them into one rotating writer is a follow-up (needs a decision on deriving `min_ts/max_ts` from the output batches for the classic path too). ## Validation - `cargo fmt --check`; `cargo clippy -p search -p compaction -p openobserve-jobs --tests` clean - `cargo test -p search --lib -- merge::` 6 passed (new: `mode_by_stream`, `sql_by_mode`, `output_file_format_uses_parquet_for_ingester_metrics`; trace-time-index merge test now goes through `MergeMode::for_ingester` and asserts `TraceTimeIndex`) - `cargo test -p compaction --lib -- merge` 30 passed; `cargo test -p infra --lib -- file_list` 49 passed - `downsampling.rs` is enterprise-only and not compilable locally; `write_files` / `pub(super) generate_downsampling_sql` were edited by hand and need the enterprise CI.