POC: separate-index sync MV — write path, product UX, commit sync, native read - #218
Closed
alchemist51 wants to merge 29 commits into
Closed
POC: separate-index sync MV — write path, product UX, commit sync, native read#218alchemist51 wants to merge 29 commits into
alchemist51 wants to merge 29 commits into
Conversation
Proves per-segment materialized-view state files as a composite-engine secondary data format, end to end in a cluster IT: - MV state files (DataFusion Partial-mode aggregate output, sorted by group key) built at flush from the primary's parquet file; tracked as first-class format files in the catalog snapshot - Row-parity exemption for derived formats: DataFormat.exemptFromRowParity() gating the CatalogSnapshotManager parity check (POC-hardcoded name set) and the CompositeMergeExecutor row-equality check - Hardcoded search (Final fold over state files across segments) returns exact golden answers; duplicate groups across segments combine correctly - Standalone round-trip test proving the state contract (Partial -> parquet -> Final; merge-fold -> Final) POC scope: one hardcoded definition (COUNT(*) GROUP BY service), merges disabled, private tokio runtime per build call, no memory-pool enrollment. See kb mv-poc-results.md for lessons and the hardcode-replacement list. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Replaces the derived-at-flush build (re-reading the primary's parquet)
with the forward-buffer / background-state model:
- MVDocumentInput captures the MV's referenced column from the composite
broadcast (no capability claim needed - all formats see all fields)
- MVWriter buffers values into a small VSR; rotation (4096 rows, always
BEFORE append so rollback only truncates the buffer) exports via Arrow
C-Data and folds into a native sorted background state (BTreeMap for
the POC count case; sorted-runs + k-way fold is the general design)
- flush = final rotation + finalize: background state written as the
sorted MV state parquet. Zero flush-time aggregation work, no read of
the primary file, no path-convention coupling
Same observable contract (sorted state file, schema, catalog
integration, golden answers) - the IT passes unchanged except the log
line ('streamed' vs 'built'). The derived build (mv_poc.rs mv_build_poc)
remains as the future merge/backfill path.
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Extends the POC view to
SELECT service, status, COUNT(*), SUM(latency_ms), MIN(latency_ms),
MAX(latency_ms) GROUP BY service, status
and replaces the hand-rolled BTreeMap fold with DataFusion's own state
algebra (hash aggregation until flush + one sort at flush):
- feed: Partial-mode aggregation over the fed batch (raw -> state rows),
accumulated; held state bounded by PartialReduce compaction
(state (+) state -- the engine operator, nothing hand-rolled)
- finalize: final PartialReduce fold -> single lexsort by the group-key
columns (cost ~ groups) -> sorted state parquet
- state schema comes from the plan; Java knows only group keys + search
template. min/max state columns are name[value] (probed via test)
- MVDocumentInput captures (service, status, latency_ms); forward buffer
is a 3-vector VSR; rotation contract unchanged
- search v2: SQL template folding all four aggregates over state files
- noted-not-built optimization recorded in mv_writer.rs: sorted runs +
SortPreservingMerge + GroupOrdered fold for high cardinality
IT: 6 state rows across 2 segments; goldens incl. a group split across
segments and min/max winners in different segments -- all exact.
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Approach 2 from the separate-index design: MV state lives in its own index, kept in sync through the data-level invariant (committed-on- source implies present-on-target), proven end to end by MVSeparateIndexPocIT: - MVStateShipper: at flush, the finalized state rows are exported and synchronously bulk-indexed into the target (index.mv.ship_target) BEFORE the flush returns; any ship failure fails the flush. Doc ids are deterministic (source.shard.generation.row) so a retried flush re-ships idempotently. Refresh policy NONE: ship+ack = durably present; searchable after the target's own refresh. - The source tracks NO MV files in ship mode (the MV index owns its physical layout): DataFormat.mayEmitNoFiles() makes an empty flush result legal for derived formats, with the composite all-or-none assertion, segment assembly, and the refresh-result assertion updated to treat such formats as optional per segment. The completeness contract lives with the composite engine's configured formats; Segment/CatalogSnapshot just carry the file sets that exist. - Fixes the W10 writer bug at the root: mv_writer sessions now drop CombinePartialFinalAggregate (agg_mode's standard removal) — on near-empty single-partition batches that rule collapsed Partial into Single mode and the writer's find-Partial step failed the flush. - Fold-on-read (POC shape): _search aggregations over the MV index (SUM of count-state, SUM/MIN/MAX of metric states) return the exact goldens over 6 raw state docs across 2 shipped generations; the production read shape (precompiled fragment + PartialReduce) is diagrammed in the separate-index KB folder. - Invariant IT: with the target closed, refresh reports failed shards (commit refused); reopening heals via idempotent re-ship. (Delete does not stage the failure: bulk auto-creates missing indices.) Known POC constraint: the synchronous ship must not run on a thread pool the target's writes need (observed write-pool deadlock under NODE_PROCESSORS=1); production ships via a dedicated executor + ack listener. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
MVColocationAllocationDecider (registered via ClusterPlugin, no core change): a target index carrying index.mv.colocate_with=<source> has each primary i constrained to the node holding the source's active primary i. canRemain flips NO when the source primary moves, so the standard reactive machinery relocates the target to follow — a persistent pairing, unlike ResizeAllocationDecider's one-shot initial- recovery constraint. Replicas are unconstrained (only the write handoff needs primary-primary locality). Availability over pairing when the source primary is unassigned: the decider answers YES anywhere rather than waiting. Waiting creates a circular dependency after a joint outage — the source's own recovery flush ships to the target (ship-before-commit), so a target waiting for the source can deadlock the pair. Reactive following restores colocation once the source is active (technical-challenges.md §1-§3). MVSeparateIndexPocIT now runs on 2 data nodes with a colocation assertion; the heal test retries failed allocations after reopening the target (the source's recovery flush exhausts allocation retries while the target is closed — itself a demonstration of the invariant extending through recovery). Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Replaces the client bulk with MVShipStateAction, fixing two flaws and enforcing one rule: - Shard-addressed: bulk routes by doc-id hash and sprays one source shard's state across ALL target shards, defeating the ordinal pairing the colocation decider maintains. The action ships source shard i's rows to target shard i by construction. - Durable ack through the write path: the handler applies each row via IndexShard.applyIndexOperationOnPrimary (translog) and fsyncs before acking. Handing the buffer to a writer AROUND the write path would break the invariant on a target crash (acked-but-lost state). - HARD LOCALITY RULE: the paired target primary must be local at ship time — no remote-forward path exists. A split pair fails the ship, which fails the flush (ship-before-commit backpressure); the colocation decider's reactive following restores the pair and the retried flush succeeds. One path, one failure mode, and nothing is ever serialized (NodeClient dispatches to the handler in-JVM). The golden IT asserts the local apply event via MockLogAppender (a missing local apply or any remote attempt fails the test); 2/2 green on 2 data nodes plus embedded regression. POC scope: primary-only apply (targets run zero replicas); production replaces the handler body with a TransportWriteAction-style replicated apply. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
alchemist51
force-pushed
the
mv-poc-separate-index
branch
from
August 16, 2026 14:59
c92df58 to
2655200
Compare
alchemist51
changed the base branch from
mv-poc-incremental-dataformat
to
main
August 16, 2026 15:00
Implements the write-path contract settled in the separate-index design review (README 'current understanding', 2026-08-16): - The target refreshes BEFORE acking, so the ack certifies durability AND searchability. Since the source commits only after the ack, the target's latest view always supersets any source view a query can hold — the entire read-consistency story, with no snapshot mapping (which folding forbids anyway). - rowsReceived verification as the commit gate (challenges §10): the ack returns the applied count; any mismatch with the shipped row count fails the flush. - The ITs now PROVE the superset guarantee: all explicit target refreshes are removed — fold-on-read must see the complete state purely by the ack ordering. 2/2 green on 2 nodes plus embedded regression. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
The ship request now carries the LIVE Arrow root — the same buffers the native writer's finalize produced — instead of parsed row maps: - Rust: df_mv_writer_finalize_arrow exports the folded+sorted state batch via Arrow C-Data into caller-allocated structs (the parquet finalize remains for embedded mode). No scratch file, no TSV export, no string parsing — zero copies between the native fold and the target's doc build. - MVShipStateAction.Request holds the VectorSchemaRoot plus provenance (source index/shard/generation, from which the handler derives the deterministic doc ids). Legal because the action is LOCAL-ONLY by the hard locality rule; wire serialization throws loudly on both ends — receiving this request remotely means the rule was broken. - Ownership contract: the handler consumes and closes the root in its try/finally (success or failure), releasing the native allocation through the C-Data release callback. - The handler builds target docs positionally off the vectors (the state contract: group keys first, then state columns; batch column names carry the writer's alias and are never compared). Both ITs green (golden fold-on-read, superset guarantee, path=local assertion, invariant + heal) plus the embedded regression — all now exercising the Arrow handoff end to end. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
The finalized state batch may ship to MULTIPLE MV targets (index.mv.ship_targets is now a list). With single-owner close, the first destination to finish would free the Arrow buffers under every other consumer. MVRefCountedStateBatch makes ownership explicit: - The SOURCE acquires one reference per target before shipping; each target's handler releases exactly its own (finally, success or failure); pre-apply exits in doExecute release too; a ship failure short-circuits remaining targets and releases their references at the source. The LAST release — wherever it happens — closes the root, firing the C-Data callback that frees the native allocation. - Releases past zero throw (ownership bug, not tolerated); the missing- release case surfaces as a leak under the tests' Arrow debug allocator. - Concurrent consumers only READ the immutable Arrow buffers — safe. - Commit gate is per target: EVERY target must ack the full row count or the flush fails (the invariant holds per target); on failure the retried flush re-ships to all targets, idempotent on those that already acked. New IT: one batch, two targets, both colocated, both independently fold to exact goldens; 3/3 green plus embedded regression. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Two changes that make the separate-index target a first-class DataFormatAwareEngine index instead of POC InternalEngine scaffolding: - DerivedDataFormat (server): base type codifying the derived-format contract the MV work had expressed as scattered overrides — row- parity exempt, may emit no files, never claims fields. MVDataFormat extends it. - mv_state (MVStateDataFormat + plugin): the TARGET's derived format. The target index is now composite (parquet primary + lucene + mv_state): shipped state rows arrive as ordinary documents through the target's write path (translog, replication-capable), and the mv_state writer — driven by the FOLD definition (SUM of count/sum states, MIN/MAX of extrema states) — materializes folded, group-key- sorted state per target generation. The target is simply a composite index whose embedded MV is the fold: the same machinery, applied to the MV index itself. Cross-generation folding arrives with target merges (the PartialReduce merger, still open). - The write path is definition-driven now: MVDefinitionSpec (columns + group keys + Partial-stopped SQL) parametrizes MVDocumentInput, MVForwardBuffer, MVWriter, and MVIndexingEngine; SOURCE and TARGET_FOLD are the two hardcoded specs until MV metadata lands. MVReaderManager is parametrized by format name. - IT: the composite target has no classic _search path, so fold-on- read asserts via the native FINAL fold over the target's OWN mv_state files (goldens across both targets in the multi-target test), plus an assertion that every target generation carries folded mv_state file sets. The heal test keeps a plain target: close/reopen of a composite index currently livelocks on the shard lock (slow native close vs reopen) — known composite-engine gap, tracked; the invariant under test is source-side and engine-independent. 3/3 separate-index ITs green, embedded regression green. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
State files (embedded source + target mv_state) switch from parquet to Arrow IPC: they are small, whole-scanned by every consumer (fold reads all rows and columns — parquet's pruning and compression buy nothing at this tier), short-lived (the PartialReduce merger folds them away), and written INSIDE the refresh — on the ack path in ship mode — where the parquet encode tax is pure latency. IPC writes are framed buffer copies and mmap back zero-copy for the future merger. writerFinalize emits IPC (arrow::ipc::writer::FileWriter); mv_search_v2 registers state files via register_arrow; state file extension becomes .mv.arrow. The C-Data ship path is untouched (no file, no serialization — already beyond IPC). Compacted/merged output may revisit parquet (compression + stats-pruned range reads) when the merger lands. Both ITs green: the golden group-by query (service,status -> COUNT/SUM/MIN/MAX over latency_ms) folds to exact goldens from IPC state on both the embedded source and the composite target. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
The concrete group-by anchor: q9 (SELECT RegionID, SUM(AdvEngineID), COUNT(*), AVG(ResolutionWidth) FROM hits GROUP BY RegionID) over the real ClickBench hits mapping types (RegionID integer, AdvEngineID/ ResolutionWidth short — KB clickbench-reference). - MVDefinitionSpec gains a named-spec registry (index.mv.definition setting; POC stand-in for MV metadata): 'payments' and 'clickbench_q9'. Sources resolve the definition; targets resolve its FOLD. Ship field names ride the spec (the handler was hardcoded to the payments fields). - The q9 definition stores COUNT/SUM(Adv)/SUM(Res)/MIN/MAX states; AVG is DECOMPOSED — the read computes SUM(res_sum)/SUM(cnt) exactly. Mixed-case ClickBench identifiers are quoted in the definition SQL (unquoted fold to lowercase: 'No field named regionid'). - Validated against the REAL 100M-row ClickBench parquet on the benchmark node (datafusion-cli over the live shard dir): direct q9 == partial-state -> fold -> final, row for row including AVG; 99,997,497 raw rows -> 9,040 state rows (11,062x reduction). - MVClickBenchQ9IT proves the same algebra through the live cluster path (ingest -> ship-before-commit -> target fold -> final over mv_state IPC files) with hand-computed goldens, incl. a group spanning generations and the explicit AVG recomposition. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Real nodes load one plugin class per installed plugin; MVStateDataFormatPlugin lives in mv-data-format's jar and therefore never loaded outside internalClusterTest (whose nodePlugins() loads classes directly — hiding this). The shim gives mv_state its own descriptor with extendedPlugins=[mv-data-format] for the classloader edge. Found deploying the POC to a real single-node cluster. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
…format shim's extendedPlugins) Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Real-node finding: each plugin classloader loads its own copy of the shared native library (separate globals) — the flat internalClusterTest classpath hid this. The MV writers' native calls hit THIS instance's runtime manager, which the DataFusion plugin's doStart cannot have initialized. POC: init in MVDataFormatPlugin.createComponents. Production requirement recorded: one shared native instance across plugins (or a parent-classloader-owned native binding). Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
With -Dnative.lib.path all plugins dlopen the same .so handle (FFM libraryLookup, refcounted) — one shared instance whose runtime the DataFusion plugin initializes. The MV-side init double-initialized the shared manager and broke it. Without the property each classloader extracts its own embedded-lib temp copy (separate globals) — that fragmentation, not a missing init, was the original failure; the deployment rule is: always set native.lib.path to one shared .so. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Found under real OSB load: the target's fold legally produces fewer rows than its parquet primary (784 docs -> 637 state rows at one generation) and the parity check failed the target shard — the ITs' tiny generations folded 1:1 by luck. The POC name-set gains mv_state; the real fix remains consulting DataFormat.exemptFromRowParity() once the registry is reachable here (existing TODO). Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Found under real OSB load + recovery: the composite target is append- only, so the idempotent-overwrite doc ids throw AppendOnlyIndexOperationRetryException when recovery replay re-ships a generation whose rows already landed. Presence satisfies the invariant — tolerate-duplicate IS idempotency on an append-only index; divergent stale content is the generation-watermark sweep's job (pending). Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
lto=false + codegen-units=16 cuts the native link time dramatically for try-it-out loops. Fat LTO + codegen-units=1 is the benchmarking configuration — restore before taking performance numbers (comment in the profile says so). Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Replaces the throwing POC merger with the safe default from the design (implementation-state §8): on merge, rebuild the fold by running the definition over the MERGED primary parquet — always consistent with the post-merge document set, and the target gets CROSS-GENERATION folding for free (state compacts to one row per group at every merge). Ship- mode sources treat merges as a non-event (no logical data change, nothing to re-ship, empty result — CompositeMergeExecutor now lets mayEmitNoFiles formats produce no merged output). The derived build (mv_build_poc) becomes the recompute engine and now emits Arrow IPC (decision 17 consistency). The state-fold merger variant remains gated on the orphan sweep's watermark. Found by hands-on local use: disabling merges per index isn't registrable (index.merge.enabled is internal) — the right answer was making merges safe rather than hardcoding them off. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
compileOnly deps on OTHER PLUGIN projects poison the opensearchplugin bundler: everything on those plugins' classpaths is treated as provided and silently excluded from this plugin's zip — dataformat-native and arrow-c-data went missing and real nodes died with NoClassDefFoundError at first write (bit twice: EC2 deploy and the local gradlew run). The datafusion/parquet compileOnly deps were the original POC hack and are no longer referenced by main sources; ITs keep them in their own scope. The zip now bundles mv-data-format + dataformat-native + arrow-c-data, matching parquet-data-format. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Replace the streaming/VSR incremental-fold write path with a refresh- time build: at flush, run the definition's Partial stage over the generation's just-flushed primary parquet — nothing MV-related happens on the doc write path anymore. addDoc is a row count, rollback is a counter reset (nothing exists to unwind before flush), no forward buffer, no native writer state across the generation. Accepted cost: one read of the page-hot parquet plus the aggregation, once per refresh, off the ingest hot path. Ship mode uses a new df_mv_build_arrow (build from parquet, export the sorted state batch via Arrow C-Data — no scratch file); embedded mode uses the existing build (now emitting Arrow IPC). find_partial also accepts Single/SinglePartitioned aggregates: for pre-decomposed definitions their output schema equals the Partial state schema, so small inputs that skip the plan split are exact, not a fallback (a one-doc generation surfaced this). Definition SQL is planned against the canonical mv_input table. The streaming model stays in git history and native mv_writer.rs as the optimization path if refresh-build cost ever matters. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Completes the ship-mode merge non-event (decision 18): the MV merger legally returns no files on a ship-mode source, CompositeMergeExecutor tolerates it, but CatalogSnapshotManager.getSegment still demanded a WriterFileSet for every format and failed the SHARD on every background merge (found live in the local loop: hits_mv cycling red). Absent file sets are now skipped for mayEmitNoFiles formats when building the merged segment; still fatal for formats that must produce output, and a merge yielding nothing at all remains an error. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Same principle as the flush-time row-parity exemption: a derived format's merge is a recompute/fold, so its merged output legitimately has fewer rows than its inputs (mv_state folds shipped state across generations — 786 in, 652 out on the local target). Conservation now sums only document-carrying formats. Found live in the local loop: the target shard failed on its first background merge. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Front-end-defined aggregate operators (the sql plugin's CHECKED_LONG_SUM — reflective UDAF, kind=SUM over BIGINT) have no isthmus signature and never can: they aren't on the engine's classpath. When binding fails and the kind has a standard Calcite operator with identical logical semantics, rebind through it — DataFusion executes the standard function natively. Kind-based, so checked sums, nullable avgs etc. are covered without naming each operator. Unblocks PPL sum() over any long field — which MV fold queries hit unconditionally (state columns are inherently INT64). MVClickBenchQ9PplIT (qa/analytics-engine-rest) proves the full production stack: real opensearch-sql plugin (snapshot zip) -> Calcite -> analytics-engine -> DataFusion over composite shards, with MV ship-before-commit and target fold underneath — q9 from the MV equals q9 direct, through /_plugins/_ppl. En route it pinned down that ship targets must pre-declare the _mv_source_* provenance fields (composite apply can't do dynamic mapping updates). Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Implements decisions 20/21/23/24 from the design review: - D21: shipped-row provenance slimmed to the single hidden _mv_source_generation field (idempotency already lives in the deterministic doc ids; source index+shard are constants per target shard under ordinal-paired colocation) - D20/D23: new index.mv.views setting — list of 'definition' or 'definition:targetName' entries on the SOURCE index only. An IndexSettingProvider expands it into the derived source settings (composite formats + ship targets; request still wins, with a loud warning if explicit formats omit the MV format), and a cluster-manager listener auto-creates each missing target with the fully derived state mapping (ship schema + the one provenance field), colocated, tolerant of re-entry. Unnamed views get <source>_mv_<definition>. - D24: this path exists only at index creation, so v1 MVs are in-sync by construction (source empty by definition). MVViewsIT proves the UX end to end: user creates ONLY the source with a views entry; derived settings, auto-created target, mapping shape, and a real ship with durable+searchable ack are all asserted. Open gap documented in MVViewsService: a mapping submitted INSIDE the same create request is validated against a settings view missing provider-derived formats (capability check sees [parquet]) — mapping after create or via template works; needs a MetadataCreateIndexService look before this UX ships. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Extends the one-way superset rule from searchable to COMMITTED: when
the source flushes, every ship target first durably commits a catalog
snapshot at least as new as its ship acks reported; only then does the
source write its own commit — with {mv.commit.<target> -> version}
recorded in its commit user data, the durable watermark the orphan
sweep reads.
Mechanism: the ship ack now carries the target's post-refresh catalog
snapshot version; the source engine tracks per-target high-water marks;
a new beforeCommit() hook on IndexingExecutionEngine (fanned out by the
composite engine, contributed entries merged into commit user data)
lets the MV format trigger MVCommitSyncAction — same in-JVM
shard-addressed pattern and hard locality rule as the ship itself,
running on the calling thread (a pool hop would only add deadlock
surface). Target-side engines have no ship targets: no-op, no
recursion. A failed target commit refuses the source commit — the same
refusal semantics as a failed ship. Repeated flushes stay cheap (the
engine skips commits when the snapshot id is unchanged).
MVCommitSyncIT proves it from the commits themselves: after a source
flush, the target's last commit carries a catalog snapshot with NO
manual target flush (needing one was exactly the pre-D25 gap), and the
source's commit user data records the watermark.
MVViewsIT is AwaitsFix: a pre-existing seed-dependent flake (verified
by stashing these changes) where the source shard's recovery evaluates
_field_names capabilities against formats [parquet] — the same
provider-derived-settings-invisible root cause as the documented
create+mapping gap.
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
…ed formats Root cause of BOTH the 'mapping inside create fails with Configured formats: [parquet]' gap AND the seed-dependent MVViewsIT recovery flake — one bug, two faces. CompositeDataFormatPlugin registers its own IndexSettingProvider that contributes cluster-default formats when the REQUEST doesn't set them; providers cannot see each other's output and MetadataCreateIndexService holds them in a HashSet, so iteration order is undefined. On losing orders its empty secondary list overwrote the MV provider's derived [lucene, materialized_view] — the empty list persisted into the index metadata and failed every capability check (inline mapping at create; _field_names at recovery). Bisected with staged probes: MV provider returns the right list; the merged additional settings had it empty; the second provider was CompositeDataFormatPlugin$1 contributing []. Fix: the composite provider defers (returns empty) when index.mv.views is declared — the MV provider owns the format stack for such indices. Key-string check by design: no compile dependency between the plugins. MVViewsIT: AwaitsFix dropped, mapping moved INLINE into the create call (the one-call experience now works); 8/8 repeat runs green, full MV suite green. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
…ated) The validation read path (user-scoped: prove optimization + perf, no transparent rewrite yet): - CLICKBENCH_Q9_NATIVE: the definition IS the query — AVG kept intact, so the state ships in DataFusion's own partial layout (avg as its [count, sum] state pair; avg_sum is a DOUBLE column). Fold definition sums the states (UNSIGNED cast keeps avg_cnt bit-identical to avg's UInt64 state type). - Ship handler: floating state values are no longer truncated to long (avg's sum half rode on this). - The mapped read — q9 spoken in state-column names against the MV index — runs on the COMPLETELY STANDARD engine path over the target's parquet state docs: SUM(adv_sum), SUM(cnt), SUM(avg_sum)/SUM(avg_cnt) as in-plan vectorized float division. No eval, no UDF, no new machinery. MVNativeReadIT proves it equals the direct q9 — and en route caught the baseline truncating (Calcite types AVG(integer) as integer; the MV path was exact, 1551.333 vs the baseline's 1551.0). - Dormant foundation for the strict native read over folded mv_state Arrow files: mv_read.rs port (Arrow IPC, strict whole-replacement), session binding + df_session_attach_mv, NativeBridge.sessionAttachMV, setting-gated shard attach (index.mv.serve_state, dynamic). The executed session doesn't consult the binding on the live fragment path yet — per design steering, the production shape is the COORDINATOR emitting the state-scan fragment (it knows the target at plan time); the shard-side binding remains as the file-level scan primitive for that work. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
alchemist51
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
POC: separate-index sync MV — write path, product UX, commit sync, native read
Synchronous materialized views as a separate index (Approach 2): the source index computes aggregate state at every refresh and ships it to a colocated MV target index before its own commit — the MV can never be behind the data. This PR carries the full arc: the core write path, the product-shaped UX (decisions 18–26 in the design log), commit-level synchronization, and a validated native read path for ClickBench q9.
The invariant
One-way superset, enforced at two levels:
{mv.commit.<target> → snapshot version}in its own commit user data — the durable watermark the orphan sweep will consume.Write path
MVShipStateAction), hard locality rule (colocated pair or refuse —writeTothrows as the tripwire), deterministic doc ids for idempotent re-ships, ref-counted multi-target batches.MVColocationAllocationDeciderpins target primary i to source primary i's node; availability-over-pairing when the source is unassigned.beforeCommit()hook onIndexingExecutionEngine(composite fans out to all format engines; contributed entries merge into commit user data); the MV engine triggersMVCommitSyncAction— target flush first, same refusal semantics as the ship, cheap when nothing changed.Product UX (D20–D24)
One create call is the entire MV story:
PUT /hits { "settings": { "index.mv.views": ["clickbench_q9:mv_hits_q9"] }, // "definition" or "definition:name" "mappings": { ... } }IndexSettingProviderderives the composite format stack + ship targets on the source; a cluster-manager listener auto-creates each target (derived state mapping, colocation, fold definition). Unnamed views get<source>_mv_<definition>._mv_source_generation, D21) — idempotency lives in deterministic ids; source index/shard are constants under ordinal pairing.index.mv.viewsis present. Design lesson: providers are unordered and mutually blind — never share a key unless one yields.Native read (D26, validation scope)
Zero-translation contract: the definition IS the query (
CLICKBENCH_Q9_NATIVEkeepsAVGintact), so shipped state is DataFusion's own partial layout — avg as its[count, sum]state pair,avg_suma double column. The read is q9 spoken in state-column names against the MV index:running on the completely standard engine path over the target's parquet state docs — in-plan vectorized float division, exact by construction (no eval, no UDF, no new machinery).
MVNativeReadITproves it equals the direct q9 — and caught the baseline truncating (Calcite typesAVG(integer)as integer: 1551.0 vs the MV's exact 1551.333…).Also landed (dormant): the strict-attach primitive for serving folded
mv_stateArrow IPC files as Partial batches (mv_read.rsport, session binding,df_session_attach_mv,index.mv.serve_state). The production shape per design review: the coordinator emits the state-scan fragment (it knows the target at plan time — no shard-side plan surgery); this primitive is the file-scan building block for that.Engine fixes shaken out by real use
CHECKED_LONG_SUM) have no isthmus signature and never can; when binding fails and the kind has a standard Calcite operator, rebind through it. Unblocks PPLsum()over long fields — which MV folds hit unconditionally (state columns are INT64).mayEmitNoFilesformats may produce no merged output (executor + catalog registration), and are exempt from merge row-count conservation (a fold legitimately shrinks: 786 state docs in → 652 folded rows out).compileOnlydeps on sibling plugins poisoned the zip bundler (droppeddataformat-native/arrow-c-data→NoClassDefFoundErrorat first write on real nodes).mv_stateexempt from flush row-parity.Validation
MVSeparateIndexPocIT(ship-before-commit + fold-on-read + negative heal test),MVClickBenchQ9IT(embedded golden fold),MVViewsIT(one-call UX end to end, 8/8 repeat runs after the provider fix),MVCommitSyncIT(commit sync proven from the two commits themselves),MVNativeReadIT(native q9 read ≡ direct),MVClickBenchQ9PplIT(real_plugins/_pplthrough the released sql plugin snapshot),MVDataFormatPocIT.Known gaps / next
Generation-watermark orphan sweep (design done; D25's commit meta is its anchor) → coordinator-side state-scan fragment (transparent rewrite path) → target PartialReduce merger → composite close/reopen shard-lock livelock → PPL-compiled definitions (D22) + discovery API (D23) → replicated apply.
Design log (D1–D26), lifecycle diagrams, and mechanism notes live in the team KB under
search/materialized-views/separate-index/.