feat(merge-executor): YAML flag to route regular merges through streaming engine (default off, no-op rollout) (#6441)

* test(merge): engine-parity tests + share MS-7 serial lock

Adds `merge::tests::parity` with two tests that run the same realistic
input fixture through both `merge_sorted_parquet_files` (in-memory
engine) and `execute_merge_operation` (streaming engine over the same
`LocalFileByteSource` the executor uses in production), then asserts
row-by-row equivalence on every visible column. These gate the upcoming
YAML flag that flips regular merges to the streaming engine: parity
must hold before the default is flipped in production.

The streaming engine writes a process-global atomic
(`PEAK_BODY_COL_PAGE_CACHE_LEN`) that the MS-7 tests reset-then-read.
Any test that runs a streaming merge must serialise against MS-7 or
inflate its readings. Move `ms7_serial_lock` from the streaming-tests
submodule to module scope (still `#[cfg(test)] pub(crate)`) so the new
parity tests acquire the same lock.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(merge-executor): YAML flag to route regular merges through the streaming engine

Wires the streaming Parquet merge engine into the regular (non-promotion)
merge path behind a node-level YAML flag,
`indexer.parquet_merge_use_streaming_engine`, defaulted to false. When
true, `ParquetMergeExecutor::handle` runs every merge through
`execute_merge_operation` (the column-major, page-bounded streaming
engine) instead of the in-memory `merge_sorted_parquet_files`.
Promotion merges (`target_prefix_len_override.is_some()`) continue to
take the streaming path unconditionally — the in-memory engine can't
handle mixed `rg_partition_prefix_len` inputs.

The in-memory engine stays in place as the runtime fallback. If the
streaming engine hits a bug in production, an operator can flip the
flag back to `false` via YAML without redeploying. Once the streaming
path has soaked, the fallback branch and `merge_sorted_parquet_files`
itself can be removed.

The flag is plumbed `IndexerConfig` → `IndexingService` →
`ParquetMergePipelineParams` → `ParquetMergeExecutor::new`, and exercised
end-to-end by the engine parity tests in the previous commit.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(config): exercise parquet_merge_use_streaming_engine in YAML/JSON/TOML fixtures

Adds the new flag to the YAML, JSON, and TOML node-config test fixtures
and bumps the expected `IndexerConfig` in `node_config_parse_*` to
`parquet_merge_use_streaming_engine: true`. Catches parse / serde
regressions on the field — e.g., a rename or a default-fn typo would
fail the test instead of silently parsing as `false`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(merge-pipeline): end-to-end streaming-engine flag verification

Adds `test_merge_pipeline_end_to_end_with_streaming_engine_flag`, an
integration test that runs the full actor chain (planner → downloader
→ executor → uploader → publisher) with
`ParquetMergePipelineParams::use_streaming_engine = true`. Asserts:

1. Publish fired with the right replaced_split_ids (merge ran
   end-to-end through the executor).
2. `PEAK_BODY_COL_PAGE_CACHE_LEN > 0` after the merge. The streaming
   engine increments this on every body-col page assembly; the
   in-memory engine never touches it. Non-zero is direct evidence
   the streaming path executed — not a silent fallback to in-memory.
3. The merge output row count and metric names are correct.

To make assertion (2) work cross-crate, exposes
`PEAK_BODY_COL_PAGE_CACHE_LEN` as `pub` under
`#[cfg(any(test, feature = "testsuite"))]`. The visibility widening is
test-only — production builds never see the symbol.

This is the closest analog to the sesh-mode "production-path" rule
that is feasible today: the metrics pipeline's OTLP gRPC ingest path
is not yet wired into `quickwit-serve`, so the closest end-to-end
test is the actor-chain integration test that this PR adds.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* docs(adr): record dual Parquet merge engines as deviation #1

Captures the intentional, time-bounded divergence from ADR-003 §4
introduced by the streaming-engine wire-in: two engines coexist in
production behind `IndexerConfig::parquet_merge_use_streaming_engine`,
with the in-memory engine retained as the runtime fallback.

Documents:
- The ADR-003 §4 quote the deviation diverges from (page-granular
  streaming, bounded memory).
- The current dual-engine implementation and routing logic.
- Why this exists (production safety, staged rollout, parity is
  strong-but-not-total).
- Explicit exit criteria: default flipped to `true`, ≥ 2-week
  production soak with no merge-correctness incidents, no rollback.
  When met, a follow-up PR deletes the in-memory branch and engine,
  the flag, and the parity tests.

This is the first deviation recorded under the EVOLUTION.md framework.
Indexes the doc in `deviations/README.md`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(merge-pipeline): share full correctness contract between both engine tests

Extracts the steps-5-through-8 assertions (replaced_split_ids, staged
metadata, Parquet file content, Parquet KV headers) into
`assert_cpu_mem_merge_outputs_correct` and calls it from both
`test_merge_pipeline_end_to_end` (in-memory engine) and
`test_merge_pipeline_end_to_end_with_streaming_engine_flag` (streaming
engine). The streaming-engine test had been doing only a small subset
of the checks — row count and metric names. It now runs the full
contract: time_range, num_merge_ops, sort_fields, row_keys_proto,
zonemap_regexes, low_cardinality_tags, all 100 timestamps,
sorted_series monotonicity, cpu/mem sort-order semantics, and every
`qh.*` Parquet KV header.

By construction both engines must produce a file that satisfies the
same contract — the helper is the executable parity between engines
at the pipeline-integration level, complementing the column-level
parity tests in `quickwit-parquet-engine::merge::tests::parity`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(merge-pipeline,merge-engine): multi-metric + multi-RG + m:n disjointness

Expands test coverage along three axes the existing helpers didn't hit:

1. **Multi-input, multi-metric pipeline tests** (new file
   `parquet_merge_pipeline_multi_metric_test.rs`). Three inputs, each
   carrying three metrics with overlapping per-metric timeseries IDs
   and overlapping-but-distinct timestamps — the merge must
   row-by-row interleave across all three inputs. Output writer uses
   `row_group_size = 50` so the 180-row merge output breaks into
   four row groups, exercising the writer's multi-RG path in both
   engines. Both engine variants (in-memory + streaming) covered.
   Streaming-engine test asserts `PEAK_BODY_COL_PAGE_CACHE_LEN > 0`
   to confirm the flag routed through the streaming path.

2. **Engine-level multi-output contract** in
   `merge::tests::parity::assert_engine_parity`. Beyond the existing
   engine-vs-engine column equivalence, every parity test now also
   verifies on the in-memory engine's outputs (equivalent to the
   streaming engine's): sum of per-output row counts equals total
   input rows, each output internally monotonic on `sorted_series`,
   and across outputs the partition is disjoint (no two outputs
   share any `sorted_series` value). This is the m:n
   non-overlap contract.

3. **Multi-metric overlapping-input m:n** test
   `parity_multi_metric_overlapping_inputs_multi_output` exercises
   the strengthened contract with three inputs × three metrics where
   per-metric keyspaces overlap across inputs. n = 3 outputs target.

Honest scope note in the new pipeline test module's doc: the actor
pipeline today hardcodes `num_outputs = 1` in `ParquetMergeExecutor`,
so n > 1 is not reachable end-to-end through the actor system. The
new engine-level test covers the n > 1 correctness contract for now;
when the executor is taught to accept `num_outputs > 1` from the
merge policy, the pipeline tests can grow an n > 1 variant.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(merge-executor): compute num_outputs from target_split_size_bytes

Replaces the hardcoded `MergeConfig { num_outputs: 1, ... }` in
`ParquetMergeExecutor::handle` with a per-merge computation:

    num_outputs = max(1, ceil(total_input_bytes / target_split_size_bytes))

So a merge that ingests more than one target's worth of data spreads
across multiple output files; merges that fit in one target keep
producing a single output (preserving today's behavior for the
common case). The engine clamps the request to the number of
`sorted_series` boundaries actually available, so the value is an
upper bound, not an exact count.

Plumbing: `IndexerConfig` already carries `target_split_size_bytes`
in `ParquetMergePolicyConfig`. Pass that through
`ParquetMergePipelineParams.target_split_size_bytes` →
`ParquetMergeExecutor::new`. Default for tests:
`256 * 1024 * 1024` (matches the production default).

Latent multi-output bug fixed at the same time: with n>1, the
executor used to assign the planner-supplied `merge_split_id` to
**every** output split, which would have collided on the rename to
`{split_id}.parquet`. First output keeps the planner ID for
observability continuity; subsequent outputs use the fresh IDs
generated by `merge_parquet_split_metadata`.

Also exposes `quickwit_parquet_engine::merge::streaming::ms7_serial_lock`
as `pub` under the `testsuite` feature so cross-crate streaming tests
(in `quickwit-indexing`) can serialise against the same global lock
the in-crate MS-7 tests use. The streaming engine writes to a
process-global atomic on every merge — without shared locking, the
existing pipeline streaming-engine test races `store(0)` against
other tests' merges. Adds the appropriate
`#[allow(clippy::await_holding_lock)]` to the in-crate
`test_merge_pipeline_end_to_end_with_streaming_engine_flag` to
match.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(merge-pipeline): bonus — prefix_len=1 multi-RG inputs + m:n outputs

Adds the bonus scenario: three multi-metric inputs each written with
`rg_partition_prefix_len = 1` and one row group per distinct
metric_name (via `row_group_size = ROWS_PER_METRIC_PER_INPUT` so the
writer flushes at every metric boundary after sorting). Merged with a
small `ParquetMergePipelineParams::target_split_size_bytes = 500`
that forces the executor's `num_outputs` calculation to ask the
engine for multiple outputs — exercising the m:n merge path now
reachable through the actor pipeline (PR's earlier commit removed
the `num_outputs = 1` hardcode).

Both engines covered:

- `test_prefix_aligned_multi_metric_three_input_multi_output_in_memory_engine`
- `test_prefix_aligned_multi_metric_three_input_multi_output_streaming_engine`

The streaming-engine variant also asserts
`PEAK_BODY_COL_PAGE_CACHE_LEN > 0` (under `ms7_serial_lock`) so a
silent fallback to the in-memory path would fail.

The shared assertion helper
`assert_three_input_three_metric_multi_output_correct` checks the
m:n contract end-to-end at the pipeline level:

- All three input splits replaced.
- ≥ 2 output splits staged (proves splitting happened).
- Sum of per-output row counts = total input rows.
- Each output internally monotonic on `sorted_series`.
- Across outputs, the `sorted_series` partition is disjoint — no
  two outputs share any key, which is the "non-overlapping output"
  contract the engine promises.
- Union of metric_names / services across outputs = full input set.
- Every output has `num_merge_ops = 1`, `row_keys_proto`, and a
  `metric_name` zonemap regex.

To pin the test to exactly one merge (not a cascade of merges over
the now-multiple staged outputs), `make_pipeline_params` now takes
`max_merge_ops` and the bonus tests set it to `1`: outputs land at
`num_merge_ops = 1`, equal to the policy ceiling, and the planner
refuses to merge them again. The existing n=1 tests stay at 5
(headroom — they produce a single output that can't trigger another
merge anyway, since `merge_factor = 3`).

Updates the module doc to drop the now-stale scope note about m:n
not being reachable through the pipeline.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* style: nightly rustfmt + drop useless borrows in assert!

Reformats doc comments / format strings under nightly rustfmt
(`wrap_comments`, `format_strings`), and removes two redundant `&` in
`assert!` arguments flagged by clippy.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: g-talbot

docs/internals/adr/deviations/001-dual-parquet-merge-engines.md

@@ -0,0 +1,178 @@
+# Deviation 001: Dual Parquet merge engines during streaming-engine rollout
+
+## Summary
+
+Two Parquet merge engines coexist in production behind a runtime YAML flag.
+The streaming engine (`execute_merge_operation`) matches the intent of
+[ADR-003 §4](../003-time-windowed-sorted-compaction.md) (page-granular
+streaming, bounded memory). The in-memory engine
+(`merge_sorted_parquet_files`) is retained as the runtime fallback so an
+operator can flip back via configuration if the streaming engine hits a
+production bug. The dual-engine state is intentional and time-bounded —
+it ends when the streaming engine has soaked at the new default in
+production.
+
+## Related ADR
+
+- **ADR**: [ADR-003 Time-Windowed Sorted Compaction](../003-time-windowed-sorted-compaction.md)
+- **Section**: §4 Sorted Merge, Phase 2 (column streaming)
+
+## ADR States
+
+> Phase 2: Stream columns through the merge.
+>
+> Once the global sort order is determined, each column is read from the
+> input splits and written to the output in sorted order. Columns are
+> processed one at a time (or in small groups) for memory efficiency.
+>
+> For large columns, it may be advantageous to operate at **page
+> granularity** rather than loading an entire column from each input:
+> read individual Parquet pages from inputs as needed and write
+> individual pages to the output. This bounds memory usage for columns
+> with large values (e.g., high-cardinality string tags, large attribute
+> maps) and avoids materializing an entire column across all inputs
+> simultaneously.
+
+ADR-003 §4 describes the merge as a streaming operation that bounds
+memory by reading and writing pages incrementally. The in-memory
+`merge_sorted_parquet_files` engine pre-materializes whole columns from
+all inputs simultaneously — directly contrary to the ADR's stated
+memory model.
+
+## Current Implementation
+
+Both engines live in `quickwit-parquet-engine/src/merge/`:
+
+- **Streaming engine** (`execute_merge_operation`, in `merge/mod.rs`,
+  backed by `merge/streaming.rs`). Column-major, page-bounded body cache,
+  reads inputs through `RemoteByteSource`. This is the
+  ADR-003-compliant implementation. It is the unconditional path for
+  promotion merges (the in-memory path can't handle mixed
+  `rg_partition_prefix_len`) and the opt-in path for regular merges.
+- **In-memory engine** (`merge_sorted_parquet_files`, in `merge/mod.rs`).
+  Buffers all inputs through arrow-rs into memory, runs the merge under
+  `run_cpu_intensive`. This is the original bootstrap implementation
+  retained as the runtime fallback.
+
+`ParquetMergeExecutor::handle` routes between them:
+
+```rust
+let is_promotion = scratch.merge_operation.target_prefix_len_override.is_some();
+if is_promotion || self.use_streaming_engine {
+    execute_merge_operation(&op, sources, &output_dir, &config).await
+} else {
+    run_cpu_intensive(move || {
+        merge_sorted_parquet_files(&input_paths, &output_dir_clone, &config)
+    }).await
+}
+```
+
+The `use_streaming_engine` boolean is sourced from the node-level
+`IndexerConfig::parquet_merge_use_streaming_engine` YAML field, default
+`false`.
+
+Row-content equivalence between the two engines is enforced by parity
+tests in `quickwit-parquet-engine/src/merge/tests.rs::parity`. These
+must keep passing as long as both engines coexist.
+
+## Signal Impact
+
+Applies to **metrics** (the only signal currently using the Parquet
+pipeline). Will apply to **traces** and **logs** when those signals
+adopt Parquet splits. The deviation does not affect Tantivy-backed
+pipelines.
+
+## Impact
+
+| Aspect | ADR Target | Current Reality |
+|--------|------------|-----------------|
+| Engines in production | One streaming engine | Two (streaming + in-memory) |
+| Memory model | Page-bounded; ~constant per column | In-memory engine: O(total input column size) per merge |
+| Configuration surface | None — engine choice is internal | One YAML flag (`parquet_merge_use_streaming_engine`) |
+| Code surface to maintain | One engine | Two engines + parity tests + routing branch |
+| Operator rollback | Not applicable — only one path | Flip flag to `false`, no redeploy needed |
+
+## Why This Exists
+
+The streaming engine is new code. ADR-003 describes the target memory
+model but does not guarantee bug-free first-deployment behavior. Three
+forces produced the dual-engine state:
+
+1. **Production safety.** The in-memory engine has been the live merge
+   path during the metrics pipeline's bring-up. Replacing it wholesale
+   on a single PR, without an in-place fallback, would mean any bug in
+   the streaming engine requires a redeploy to recover. With the flag,
+   recovery is `config edit + restart`.
+2. **Staged rollout.** Production confidence is built by enabling the
+   streaming engine on a soak fleet, observing for some time, then
+   flipping the default. The dual-engine state is the necessary
+   infrastructure for that rollout.
+3. **Parity verifiable, not certain.** The parity tests in
+   `merge::tests::parity` cover representative synthetic fixtures.
+   Production data has shapes those fixtures don't cover. The fallback
+   exists because parity is a strong-but-not-total guarantee.
+
+## Priority Assessment
+
+- **PoC / MVP**: acceptable — dual-engine is in fact the deliberate
+  state during MVP rollout.
+- **Production (current)**: acceptable — flag defaults to `false`,
+  rollout has not begun. The streaming engine is exercised only by
+  promotion merges (whose execution will start once GAP-011 is closed).
+- **Production (post-soak)**: not acceptable. Once the streaming engine
+  has soaked at default-`true` in production, the in-memory engine
+  becomes dead code that complicates the merge-executor and obscures
+  the ADR-003 memory contract. Resolve before merging additional
+  significant work into `parquet_merge_executor.rs`.
+
+## Exit Criteria
+
+The deviation resolves when **all** of the following hold:
+
+1. `IndexerConfig::default_parquet_merge_use_streaming_engine` defaults
+   to `true` in `quickwit-config`.
+2. At least one production fleet has run with the flag set to `true` for
+   a soak window of ≥ 2 weeks with no merge-correctness incidents (no
+   data loss, no schema mismatch, no merge-output-rows-≠-input-rows
+   alerts).
+3. No deviation-resolving rollback has been issued during the soak.
+
+When those are met, the follow-up PR:
+
+- Deletes `merge_sorted_parquet_files` from `quickwit-parquet-engine`.
+- Deletes the in-memory branch in `ParquetMergeExecutor::handle`.
+- Deletes the `use_streaming_engine` field on `ParquetMergeExecutor` and
+  `ParquetMergePipelineParams`.
+- Deletes `IndexerConfig::parquet_merge_use_streaming_engine`.
+- Deletes `merge::tests::parity` (both engines no longer exist to
+  compare).
+- Closes this deviation.
+
+## Work Required to Match ADR
+
+| Change | Difficulty | Description |
+|--------|------------|-------------|
+| Flip default to `true` | Trivial | One-line change in `IndexerConfig::default_parquet_merge_use_streaming_engine`. Lands after soak. |
+| Production soak | Operational | Run with `true` on at least one fleet for ≥ 2 weeks, monitor merge-correctness signals. |
+| Delete in-memory engine | Moderate | Remove `merge_sorted_parquet_files`, the fallback branch, the flag, and the parity tests. Mechanically straightforward but touches several call sites. |
+
+## Recommendation
+
+**Accept for now.** The dual-engine state is the deliberate output of a
+flag-with-fallback rollout pattern (see commit history of #6441 and
+related PRs). Resolution is a known follow-up, not technical debt that
+needs to be paid down ahead of schedule.
+
+Track the exit criteria in this doc. When all three conditions hold,
+open the deletion PR and close this deviation.
+
+## References
+
+- [ADR-003 Time-Windowed Sorted Compaction](../003-time-windowed-sorted-compaction.md) §4
+- [GAP-011 No legacy promotion planner](../gaps/011-no-legacy-promotion-planner.md)
+- [GAP-012 Merge downloads instead of streaming](../gaps/012-merge-downloads-instead-of-streaming.md)
+- PR #6441 (wire-in of the YAML flag)
+
+## Date
+
+2026-05-18

docs/internals/adr/deviations/README.md

@@ -94,8 +94,7 @@ Deviation files use sequential numbering: `001-short-description.md`
 
 | Deviation | Title | Related ADR | Priority |
 |-----------|-------|-------------|----------|
-
-*No deviations recorded yet.*
+| [001](./001-dual-parquet-merge-engines.md) | Dual Parquet merge engines during streaming-engine rollout | [ADR-003](../003-time-windowed-sorted-compaction.md) | Accept until post-soak |
 
 ## Lifecycle
 

quickwit/quickwit-config/resources/tests/node_config/quickwit.json

@@ -54,7 +54,8 @@
         "split_store_max_num_splits": 10000,
         "max_concurrent_split_uploads": 8,
         "max_merge_write_throughput": "100mb",
-        "merge_concurrency": 2
+        "merge_concurrency": 2,
+        "parquet_merge_use_streaming_engine": true
     },
     "ingest_api": {
         "replication_factor": 2

quickwit/quickwit-config/resources/tests/node_config/quickwit.toml

@@ -45,6 +45,7 @@ split_store_max_num_splits = 10_000
 max_concurrent_split_uploads = 8
 max_merge_write_throughput = "100mb"
 merge_concurrency = 2
+parquet_merge_use_streaming_engine = true
 
 [ingest_api]
 replication_factor = 2

quickwit/quickwit-config/resources/tests/node_config/quickwit.yaml

@@ -49,6 +49,7 @@ indexer:
   max_concurrent_split_uploads: 8
   max_merge_write_throughput: 100mb
   merge_concurrency: 2
+  parquet_merge_use_streaming_engine: true
 
 ingest_api:
   replication_factor: 2

quickwit/quickwit-config/src/node_config/mod.rs

@@ -163,6 +163,16 @@ pub struct IndexerConfig {
     pub enable_cooperative_indexing: bool,
     #[serde(default = "IndexerConfig::default_cpu_capacity")]
     pub cpu_capacity: CpuCapacity,
+    /// If true, run Parquet merges through the streaming column-major engine
+    /// (`execute_merge_operation`). If false (default), use the in-memory
+    /// `merge_sorted_parquet_files` engine. The legacy in-memory engine is
+    /// kept as the runtime fallback so production can flip back to it
+    /// without redeploying if the streaming engine hits a bug. Promotion
+    /// merges (those with `target_prefix_len_override`) always go through
+    /// the streaming engine regardless of this flag — the in-memory path
+    /// can't handle mixed prefix lengths.
+    #[serde(default = "IndexerConfig::default_parquet_merge_use_streaming_engine")]
+    pub parquet_merge_use_streaming_engine: bool,
 }
 
 impl IndexerConfig {
@@ -201,6 +211,10 @@ impl IndexerConfig {
         CpuCapacity::one_cpu_thread() * (quickwit_common::num_cpus() as u32)
     }
 
+    fn default_parquet_merge_use_streaming_engine() -> bool {
+        false
+    }
+
     #[cfg(any(test, feature = "testsuite"))]
     pub fn for_test() -> anyhow::Result<Self> {
         use quickwit_proto::indexing::PIPELINE_FULL_CAPACITY;
@@ -213,6 +227,7 @@ impl IndexerConfig {
             cpu_capacity: PIPELINE_FULL_CAPACITY * 4u32,
             max_merge_write_throughput: None,
             merge_concurrency: NonZeroUsize::new(3).unwrap(),
+            parquet_merge_use_streaming_engine: Self::default_parquet_merge_use_streaming_engine(),
         };
         Ok(indexer_config)
     }
@@ -229,6 +244,7 @@ impl Default for IndexerConfig {
             cpu_capacity: Self::default_cpu_capacity(),
             merge_concurrency: Self::default_merge_concurrency(),
             max_merge_write_throughput: None,
+            parquet_merge_use_streaming_engine: Self::default_parquet_merge_use_streaming_engine(),
         }
     }
 }

quickwit/quickwit-config/src/node_config/serialize.rs

@@ -657,6 +657,7 @@ mod tests {
                 cpu_capacity: IndexerConfig::default_cpu_capacity(),
                 enable_cooperative_indexing: false,
                 max_merge_write_throughput: Some(ByteSize::mb(100)),
+                parquet_merge_use_streaming_engine: true,
             }
         );
         assert_eq!(

quickwit/quickwit-indexing/src/actors/indexing_service.rs

@@ -113,6 +113,12 @@ pub struct IndexingService {
     counters: IndexingServiceCounters,
     local_split_store: Arc<IndexingSplitCache>,
     pub(crate) max_concurrent_split_uploads: usize,
+    /// Cached from `IndexerConfig`. Selects whether new Parquet merge
+    /// pipelines route regular merges through the streaming engine or
+    /// the in-memory fallback. Promotion merges always use the
+    /// streaming engine regardless of this flag.
+    #[cfg(feature = "metrics")]
+    pub(crate) parquet_merge_use_streaming_engine: bool,
     merge_pipeline_handles: HashMap<MergePipelineId, MergePipelineHandle>,
     #[cfg(feature = "metrics")]
     parquet_merge_pipeline_handles: HashMap<IndexUid, ParquetMergePipelineHandle>,
@@ -178,6 +184,8 @@ impl IndexingService {
             indexing_pipelines: Default::default(),
             counters: Default::default(),
             max_concurrent_split_uploads: indexer_config.max_concurrent_split_uploads,
+            #[cfg(feature = "metrics")]
+            parquet_merge_use_streaming_engine: indexer_config.parquet_merge_use_streaming_engine,
             merge_pipeline_handles: HashMap::new(),
             #[cfg(feature = "metrics")]
             parquet_merge_pipeline_handles: HashMap::new(),
@@ -723,6 +731,8 @@ impl IndexingService {
             max_concurrent_split_uploads: self.max_concurrent_split_uploads,
             event_broker: self.event_broker.clone(),
             writer_config,
+            use_streaming_engine: self.parquet_merge_use_streaming_engine,
+            target_split_size_bytes: cfg.target_split_size_bytes,
         };
 
         let pipeline = super::parquet_pipeline::ParquetMergePipeline::new(

quickwit/quickwit-indexing/src/actors/parquet_pipeline/mod.rs

@@ -61,6 +61,10 @@ mod parquet_merge_pipeline_trace_conformance_test;
 #[allow(clippy::disallowed_methods)]
 mod parquet_merge_pipeline_sketch_test;
 
+#[cfg(test)]
+#[allow(clippy::disallowed_methods)]
+mod parquet_merge_pipeline_multi_metric_test;
+
 pub use parquet_doc_processor::{
     ParquetDocProcessor, ParquetDocProcessorCounters, ParquetDocProcessorError, is_arrow_ipc,
 };