POC: incremental materialized view — derived data format + transparent read path

[alchemist51]

What this proves

The incremental-MV design (per-segment aggregate state files as a composite-engine secondary data format) works end to end in a live cluster — both halves:

1. Write path: state files built as segments flush, tracked as first-class catalog files (MVDataFormatPocIT)
2. Read path (new): an aggregate query on the source index is transparently served from the state files, with per-segment fallback to raw (MVReadPathIT)

Write path (commits 1–3)

For the hardcoded view SELECT service, status, COUNT(*), SUM(latency_ms), MIN(latency_ms), MAX(latency_ms) FROM payments GROUP BY service, status:

• Ingest: composite index (parquet primary + materialized_view secondary), golden docs in 2 segments
• Streaming build: MVDocumentInput captures the referenced columns from the composite broadcast → forward VSR buffer (4096-row rotation, rotate-before-append for rollback safety) → Arrow C-Data → DataFusion-maintained native state (Partial-mode accumulate + PartialReduce compaction) → flush finalizes into a group-key-sorted state parquet. No primary re-read, no flush-time aggregation spike.
• Catalog integration: state files ride Segment's format map — pinning, snapshot membership, per-format cleanup all free. Row-parity exemption (DataFormat.exemptFromRowParity()) lets a segment carry 8 raw rows next to 3 state rows.
• Standalone Rust round-trip test proves the state contract (Partial → parquet → Final; merge-fold → Final).

Read path (commits 4–7)

The full transparent-rewrite chain, star-tree style (gate once per request, substitute per segment, zero new downstream types):

planner: MVRewritePhase (post-decomposeAggregates, pre-mark)
  matches the canonicalized Aggregate+Scan against MV definitions
  → records an annotation in a PlannerContext side-channel (tree untouched)
     ↓
FragmentConversionDriver: binding rides the shard-scan instruction
  (only when the fragment has a partial aggregate; never with QTF row-ids
   or delegated predicates)
     ↓
data node: ShardScanInstructionHandler computes the coverage split from
  ONE catalog snapshot (MV generations ∩ parquet generations — atomic)
  → df_session_attach_mv(state file paths, covered raw file names)
     ↓
Rust mv_read: at prepare_partial_plan time the stripped Partial plan becomes
  UNION( scan(state files, positionally aliased to the partial output names),
         Partial over the raw scan narrowed to uncovered files )
     ↓
coordinator FINAL: unchanged — state files ARE Partial-mode output
  (zero-translation contract)

Fallback-first everywhere: any mismatch (plan shape, state schema, scan leaf shape, attach failure) keeps today's raw-only plan. The only possible failure mode is "no speedup", never a wrong answer.

MV-only proof (strict mode): a POC verification switch (mv.poc.strict_read) inverts the fallback contract to fail-closed — every fallback becomes a hard error, every raw file must be covered, and the native plan is the state-file scan alone (no raw scan node exists, raw parquet physically cannot be read). testStrictModeProvesQueryServedFromMVOnly runs the aggregate through the live cluster in this mode and gets the golden answers — constructive proof the query was served exclusively from the MV.

Proof (assertions layered so a silent fallback FAILS the test):

• MVReadPathIT (2 shards): MockLogAppender expects the shard handler's mv-binding … attached event during the MV-enabled query; differential — MV-enabled results ≡ raw-path results ≡ goldens; negative control — a non-matching group-by doesn't bind and still answers correctly. Observed chain in one run: annotation → bind=true → attached 2 state files, 2 covered raw files → mv_read: bound 2 state files; raw branch narrowed by 2 covered files.
• tests/mv_read_union_test.rs (6 tests): mixed coverage equals the full-raw answer (a narrowing failure would double-count and fail), all-covered serves entirely from state files, mismatched state schema returns the untouched plan; strict variants — the strict plan's display contains no raw-dir path, an uncovered file errors, a mismatched schema errors.

Reading order

• server/…/DataFormat.java + CatalogSnapshotManager.java — parity-exemption core change (small)
• sandbox/plugins/mv-data-format/ — the format plugin (MVWriter is the write-path heart; MVReaderManager now exposes per-generation coverage)
• analytics-engine …/planner/mv/ — MVRewritePhase + MVRegistry (the planner gate)
• analytics-backend-datafusion …/ShardScanInstructionHandler.java + rust/src/mv_read.rs — the shard-side split + native union surgery
• MVDataFormatPocIT (write path) and MVReadPathIT (read path) — the demos:

  ./gradlew :sandbox:plugins:mv-data-format:internalClusterTest -Dsandbox.enabled=true --tests 'org.opensearch.mv.MVDataFormatPocIT'
  ./gradlew :sandbox:qa:analytics-engine-coordinator:internalClusterTest -Dsandbox.enabled=true --tests 'org.opensearch.mv.MVReadPathIT'
  

POC hardcodes / known v0 gaps (not production)

• Definition compiled in; MVRegistryHolder static wiring (replaced by cluster-state MV metadata at M0/M1); state-fingerprint carried on the wire but enforced only via the positional schema check
• Single-shard queries don't bind: 1-shard fragments run SINGLE-mode aggregation (no Partial node = no surgery point). Tracked as W9 in the plan doc
• POC writer parity on near-empty shard batches fails flush (pre-existing; the read-path IT routes docs to one shard). Tracked as W10
• TopK fragments excluded from binding in v0; merges disabled; per-call tokio runtime on the write path; parity exemption as a hardcoded name set

Design decisions (D1–D8: annotation transport, surgery ordering, wire transport, coverage split, schema alignment, registry wiring) + full work log live in the KB: mv-search-side-integration-plan.md.

Not for merge — design-validation artifact for the incremental-MV track, now covering the complete write→read loop.