[PR] [MaterializedView] Harden partition-state engine [pinot]

["hongkunxu (via GitHub)" <gi...@apache.org>]

hongkunxu opened a new pull request, #18642:
URL: https://github.com/apache/pinot/pull/18642

   ## Title
   
   [MV] Harden materialized-view partition-state engine: watermark cap, DELETE backfill, and centralized mutation entry point
   
   ## Description
   
   This PR hardens the materialized-view partition-state engine in three areas: it
   fixes two correctness bugs that let stale data look fresh and let backfilled
   buckets silently fail to re-sync; it consolidates every per-partition mutation
   behind a single CAS engine so future state transitions cannot drift across the
   executor, scheduler, and consistency manager; and it centralizes the
   window-fingerprint algorithm that those three components use to agree on
   "what the source looked like at task time".
   
   ### Design reference
   
   The full state machine — partition states, operations, allowed transitions,
   and how the consistency manager keeps the partition map in sync with
   base-table segment changes — is documented here:
   
   > [MV partition state engine — design notes (Google Doc)](<link>)
   
   It contains four sections that together describe the contract this PR enforces:
   
   1. **Partition state definition** — `VACANT` / `VALID` / `STALE` and what each
      one means physically (entry presence, fingerprint, last-refresh time).
   2. **Operation definition** — the seven operations
      (`MinionTaskCycle_APPEND` / `_OVERWRITE` / `_CLEAR`,
      `ConsistencyMgr_MARK_STALE`, `Scheduler_REVERT_VALID`,
      `Manual_REFRESH_MV`, `Manual_DROP_MV_PARTITION`),
      their triggers (Auto vs Manual), and which actor invokes each.
   3. **State transition table** — the 3×3 from-state × to-state matrix listing
      every operation that drives each transition.
   4. **Sync with base-table change** — what the consistency manager does when it
      observes a base segment add/update/delete event, and which transitions get
      produced as a result.
   
   This PR is the first implementation pass that fully matches that document; the
   post-merge state machine has a single owner (`MaterializedViewPartitionManager`)
   and one test surface that exercises every documented transition.
   
   ### Why introduce `MaterializedViewPartitionManager`
   
   Before this PR the per-partition mutation logic was spread across three
   sites — the minion executor, the controller-side task scheduler, and the
   consistency manager — each shipping its **own** ~50-line CAS-write retry loop
   with its own retry budget, its own jittered backoff, and its own
   hand-rolled `Map<Long, PartitionInfo>` copy/edit code:
   
   | Site | Trigger | Bespoke CAS loop |
   |------|---------|------------------|
   | Executor | `APPEND` / `OVERWRITE` / `DELETE` task commit | `DEFAULT_MAX_RUNTIME_UPDATE_ATTEMPTS` |
   | Scheduler | False-positive `STALE` revert | `MAX_PARTITION_STATE_PERSIST_RETRIES` (8, fixed) |
   | Consistency mgr | Segment-change flush | `MAX_MARK_RETRIES` + `ThreadLocalRandom` backoff |
   
   Concrete problems this caused:
   
   - Three retry/backoff policies that could (and did) drift apart whenever any
     one was tuned. Operators had no single knob to control "MV runtime znode
     contention pressure".
   - Watermark advancement on `APPEND` was a separate ZK write from the bucket
     insert in some refactor proposals; under concurrent writers this opens a
     window where the partition map and `watermarkMs` disagree, which in turn
     poisons the broker's `now - watermarkMs <= stalenessThresholdMs` check.
   - Adding a new state to the state machine required touching ~3 files with
     ~3 different idioms. Every new transition was an opportunity to forget a
     precondition check (e.g. `existing.getState() == STALE` before `OVERWRITE`).
   - Cross-package callers could synthesize arbitrary `PartitionInfo` objects
     directly, bypassing every invariant the manager would otherwise enforce.
   
   The new design replaces all three loops with one CAS engine
   (`MaterializedViewPartitionManager#applyMutation`), exposes one public method
   per documented operation (`appendValid` / `refreshValid` / `clearValid` /
   `revertValid` / `markStale` / `deletePartition`), and bundles watermark
   advancement into the same atomic write that mutates the bucket. The
   `PartitionInfo` constructor is locked to package access so production code
   outside the metadata package physically cannot bypass the manager — tests
   opt in via a single named factory `PartitionInfo.forTesting(...)` annotated
   `@VisibleForTesting`.
   
   Net effect: one retry budget, one backoff implementation, one set of
   preconditions, one place to add the next state. Future transitions
   (`VACANT → STALE` synthesize for in-coverage backfill, an explicit
   `VALID-empty` state, etc.) become a one-method addition with predictable
   behavior, not a three-site coordination exercise.
   
   ### Bug fixes that landed alongside the refactor
   
   Two separate correctness bugs were fixed as part of this pass — both touch
   the same partition-state machinery the manager now owns, and both now have
   regression tests in the manager's unit-test surface:
   
   1. **Watermark over-advance past latest source data.** The MV scheduler kept
      advancing `watermarkMs` toward `now - bufferMs` even when the base table
      had stopped receiving new data, which made the broker's freshness check
      `now - watermarkMs <= stalenessThresholdMs` return `true` for stale MV
      data. `watermarkMs` is now capped at the latest `endTime` present in the
      source segments, so the staleness check reflects real data freshness
      instead of wall-clock drift.
   
   2. **Backfill into a previously-deleted bucket silently dropped.** When a
      `STALE` bucket's source data was retention-deleted, the executor's
      `DELETE` branch removed the runtime `PartitionInfo` entry entirely. That
      left the bucket in the `absent` state, which the consistency manager's
      `markStale` pass deliberately skips, so any subsequent base-table
      backfill into that window never propagated to the MV. The `DELETE`
      branch now writes `VALID + PartitionFingerprint.EMPTY` instead, so the
      bucket follows the standard `VALID → STALE → OVERWRITE` cycle on
      backfill.
   
      `PartitionFingerprint.EMPTY` is byte-identical to what
      `computeWindowFingerprint` already produces when the overlapping segment
      list is empty (both feed an empty input through `farmHashFingerprint64`),
      so existing ZK records remain comparable across rolling upgrades — no
      migration required.
   
   ### Other clean-ups bundled in
   
   - Two verbatim copies of the
     filter + sort + `farmHashFingerprint64` window-fingerprint algorithm
     (one in the scheduler, one in the executor) are collapsed into
     `MaterializedViewTaskUtils#computeWindowFingerprint`. Drift between the
     two copies would silently break the executor's commit-time fingerprint
     validation and produce tasks that retry forever without advancing the
     watermark — one of the worst classes of MV bugs to triage.
   
   ### Backward compatibility / rolling upgrade
   
   - ZK format unchanged. The DELETE-branch fix relies on
     `PartitionFingerprint.EMPTY` being byte-identical to the
     empty-overlap fingerprint, so znodes written by the old executor remain
     comparable to znodes written by the new one.
   - No public API or REST surface changes.
   - `PartitionInfo` constructor is now package-private; the only production
     callers were already inside the `metadata` package. Tests use the new
     `PartitionInfo.forTesting(...)` factory.
   
   ### Tests
   
   - New: `MaterializedViewPartitionManagerTest` exercises every public method,
     including CAS retry / version-conflict paths and precondition violations.
   - Updated: scheduler / executor / consistency-manager / rewrite-engine tests
     rebased onto the manager and the `PartitionInfo.forTesting` factory.
   - Manual end-to-end validation against a local Pinot cluster: created
     `airlineStats_mv_hardening_smoke` with `1m` refresh and exercised the
     segment-delete → STALE → executor-DELETE → backfill path; verified that
     the `VALID-empty` write keeps the bucket re-syncing on backfill, and that
     the watermark stops advancing once base-table ingestion stalls.
   
   ### Suggested labels
   
   `bugfix`, `refactor`


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on code in PR #18642:
URL: https://github.com/apache/pinot/pull/18642#discussion_r3391749530


##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -602,19 +560,44 @@ private PartitionFingerprint getTaskFingerprint(Map<String, String> configs, Str
     return fingerprint;
   }
 
-  private void validateSourceFingerprintAtCommit(Map<String, String> configs, String tableName, long windowStartMs,
-      long windowEndMs, PartitionFingerprint taskFingerprint) {
+  /// Computes the current source-table [PartitionFingerprint] for `[windowStartMs, windowEndMs)`
+  /// from live ZK segment metadata.  Shared by the APPEND / OVERWRITE commit-time fingerprint
+  /// validation and the DELETE commit-time emptiness re-check.  Requires
+  /// [MaterializedViewTask#SOURCE_TABLE_NAME_KEY] in the task config (the scheduler sets it for
+  /// every task mode); fails loud if absent so a malformed / pre-upgrade task cannot silently
+  /// skip the validation.
+  private PartitionFingerprint computeSourceWindowFingerprint(Map<String, String> configs, String tableName,
+      long windowStartMs, long windowEndMs) {
     String sourceTableName = configs.get(MaterializedViewTask.SOURCE_TABLE_NAME_KEY);

Review Comment:
   Addressed in c81bd840dd — `computeSourceWindowFingerprint` now falls back to the authoritative source-table reference in `MaterializedViewDefinitionMetadata` when the task config lacks `sourceTableName` (the legacy DELETE shape from a pre-upgrade controller), failing loud only when neither the config key nor the definition znode is available. Covered by `testComputeSourceWindowFingerprintFallsBackToDefinitionForLegacyTask` (positive) and `testComputeSourceWindowFingerprintFailsLoudWithoutConfigOrDefinition` (negative); a sunset TODO marks the shim for removal once pre-`sourceTableName` controllers are out of supported rolling-upgrade paths.



##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -602,19 +560,44 @@ private PartitionFingerprint getTaskFingerprint(Map<String, String> configs, Str
     return fingerprint;
   }
 
-  private void validateSourceFingerprintAtCommit(Map<String, String> configs, String tableName, long windowStartMs,
-      long windowEndMs, PartitionFingerprint taskFingerprint) {
+  /// Computes the current source-table [PartitionFingerprint] for `[windowStartMs, windowEndMs)`
+  /// from live ZK segment metadata.  Shared by the APPEND / OVERWRITE commit-time fingerprint
+  /// validation and the DELETE commit-time emptiness re-check.  Requires
+  /// [MaterializedViewTask#SOURCE_TABLE_NAME_KEY] in the task config (the scheduler sets it for
+  /// every task mode); fails loud if absent so a malformed / pre-upgrade task cannot silently
+  /// skip the validation.
+  private PartitionFingerprint computeSourceWindowFingerprint(Map<String, String> configs, String tableName,
+      long windowStartMs, long windowEndMs) {
     String sourceTableName = configs.get(MaterializedViewTask.SOURCE_TABLE_NAME_KEY);

Review Comment:
   Addressed in c81bd840dd — `computeSourceWindowFingerprint` now falls back to the authoritative source-table reference in `MaterializedViewDefinitionMetadata` when the task config lacks `sourceTableName` (the legacy DELETE shape from a pre-upgrade controller), failing loud only when neither the config key nor the definition znode is available. Covered by `testComputeSourceWindowFingerprintFallsBackToDefinitionForLegacyTask` (positive) and `testComputeSourceWindowFingerprintFailsLoudWithoutConfigOrDefinition` (negative); a sunset TODO marks the shim for removal once pre-`sourceTableName` controllers are out of supported rolling-upgrade paths.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #18642:
URL: https://github.com/apache/pinot/pull/18642#discussion_r3365786701


##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -173,7 +171,8 @@ public SegmentConversionResult executeTask(PinotTaskConfig pinotTaskConfig)
     String taskMode = configs.getOrDefault(MaterializedViewTask.TASK_MODE_KEY,
         MaterializedViewTask.TASK_MODE_APPEND);
 
-    // DELETE mode: skip query execution, only remove existing MV segments
+    // DELETE mode: skip query execution; remove existing MV segments and rewrite the
+    // runtime PartitionInfo to VALID-empty (see updateMaterializedViewRuntime DELETE branch).
     if (MaterializedViewTask.TASK_MODE_DELETE.equals(taskMode)) {

Review Comment:
   The DELETE-mode comment still references the removed updateMaterializedViewRuntime() method. This makes the code harder to follow and points readers to a symbol that no longer exists; the comment should reference the new MaterializedViewPartitionManager-based postProcess path instead.



##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/consistency/MaterializedViewConsistencyManager.java:
##########
@@ -410,55 +405,40 @@ private boolean markPartitionsDirty(String viewTableName, long affectedStartMs,
             + "mark partitions without a bucket size.  Repair the MV table config.",
         viewTableName, bucketMs);
 
-    // Iterate every bucket [partStart, partStart+bucketMs) that overlaps the affected range
-    // [affectedStartMs, affectedEndMs]. Two ranges overlap when start1 <= end2 AND end1 >= start2.
+    // Cap affectedEndMs at watermarkMs.  No partition with partStart > watermarkMs can exist
+    // (the writer invariant), so any bucket beyond that cannot be marked STALE anyway.  This
+    // protects the bucket iteration below from a caller passing Long.MAX_VALUE (full-range
+    // invalidation from notifyMaterializedViewConsistencyManager paths when segment
+    // startTime/endTime is unknown), which would otherwise enumerate ~watermarkMs/bucketMs
+    // candidates — orders of magnitude more than the number of real partitions.
+    long cappedEnd = Math.min(affectedEndMs, runtime.getWatermarkMs());
+    if (cappedEnd < affectedStartMs) {
+      LOGGER.debug("Affected range [{}, {}] is past watermarkMs ({}) for MV table: {}; nothing to mark",
+          affectedStartMs, affectedEndMs, runtime.getWatermarkMs(), viewTableName);
+      return Collections.emptyList();
+    }
+
+    // Enumerate every bucket [partStart, partStart+bucketMs) that overlaps the affected range
+    // [affectedStartMs, cappedEnd]. Two ranges overlap when start1 <= end2 AND end1 >= start2.
     // The first overlapping bucket has partStart = floorDiv(affectedStartMs, bucketMs) * bucketMs;
-    // the last has partStart <= affectedEndMs (any bucket whose start is past affectedEndMs
-    // cannot overlap because partStart > affectedEndMs >= affectedStartMs implies the bucket
-    // starts after the affected range ends). floorDiv (instead of /) is used defensively so a
-    // future caller passing negative affectedStartMs would still produce the correct floor.
-    // Only flip existing VALID entries to STALE.  Absent buckets are NOT synthesized — under
-    // Design C, a bucket's absence from the partition map already means "MV does not cover this
-    // range", so the broker rewrite (PR 2) routes those queries to the base table.  Synthesizing
-    // STALE entries for every uncovered bucket below `watermarkMs` would explode the znode size
-    // on a full-range invalidation (~watermarkMs / bucketMs entries) without affecting routing
-    // correctness — the bucket-iteration loop below stays O(affectedRange / bucketMs) but the
-    // persisted map grows only with real partitions.
+    // the last has partStart <= cappedEnd (any bucket whose start is past cappedEnd cannot
+    // overlap because partStart > cappedEnd >= affectedStartMs implies the bucket starts after
+    // the affected range ends). floorDiv (instead of /) is used defensively so a future caller
+    // passing negative affectedStartMs would still produce the correct floor.
+    //
+    // Absent buckets are still enumerated: the manager's per-bucket lenient filter no-ops them
+    // under the CAS lock.  Synthesizing STALE entries for absent buckets is intentionally NOT
+    // done here — under Design C, a bucket's absence already means "MV does not cover this
+    // range" and the broker routes those queries to the base.  Synthesizing on every
+    // full-range invalidation would explode the znode size by ~watermarkMs/bucketMs entries
+    // without affecting routing correctness.
+    List<Long> buckets = new ArrayList<>();
     long partStart = Math.floorDiv(affectedStartMs, bucketMs) * bucketMs;
-    while (partStart <= affectedEndMs) {
-      PartitionInfo info = updatedInfos.get(partStart);
-      if (info != null && info.getState() == PartitionState.VALID) {
-        updatedInfos.put(partStart, info.withState(PartitionState.STALE));
-        anyChanged = true;
-        markedCount++;
-      }
+    while (partStart <= cappedEnd) {
+      buckets.add(partStart);
       partStart += bucketMs;
     }

Review Comment:
   enumerateCandidateBuckets() materializes every bucket start between affectedStartMs and watermark (cappedEnd) into a List. For full invalidation (affectedEndMs=Long.MAX_VALUE) with a small bucketMs and a large watermark horizon, this can allocate an extremely large list and risk high GC pressure or OOM. Consider chunking the range (e.g., call _partitionManager.markStale() in bounded-size batches) or moving the range-iteration into the manager so you don't need to hold all candidate buckets in memory at once.



##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -392,8 +391,10 @@ private void revertWindowSegmentReplace(String tableNameWithType, String lineage
   }
 
   /// Handles DELETE mode: removes all existing MV segments for the given time window
-  /// via segment lineage replace (segmentsFrom=[old segments], segmentsTo=[]).
-  /// No query is executed and no new segments are created.
+  /// via segment lineage replace (segmentsFrom=[old segments], segmentsTo=[]).  No query
+  /// is executed and no new MV segments are created; the runtime PartitionInfo is rewritten
+  /// to `VALID + PartitionFingerprint.EMPTY` by [#updateMaterializedViewRuntime] in
+  /// [#postProcess], so the partition stays tracked-but-empty rather than disappearing.

Review Comment:
   This Javadoc still says the runtime PartitionInfo is rewritten by updateMaterializedViewRuntime(), but that helper was removed in favor of MaterializedViewPartitionManager. Updating this reference avoids confusion and keeps the documentation consistent with the current implementation.



##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/metadata/MaterializedViewPartitionManager.java:
##########
@@ -0,0 +1,499 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.materializedview.metadata;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+import org.apache.helix.store.HelixPropertyStore;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.zkclient.exception.ZkException;
+import org.apache.pinot.materializedview.scheduler.MaterializedViewTaskUtils;
+import org.apache.pinot.spi.utils.CommonConstants.MaterializedViewTask;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Single source of truth for every per-partition state transition on the
+/// [MaterializedViewRuntimeMetadata] runtime znode.
+///
+/// <h3>Why this class exists</h3>
+///
+/// Before this manager landed, the runtime znode was mutated by three separate sites — the
+/// minion executor (APPEND / OVERWRITE / DELETE task commits), the scheduler (false-positive
+/// STALE reverts), and the consistency manager (VALID → STALE markings).  Each site shipped
+/// its own version-checked CAS loop, with subtly different retry budgets, backoff jitter,
+/// and exception-classification rules.  Subsequent state-machine changes (e.g. extending the
+/// VACANT → STALE synthesize for in-coverage backfill, or revoking the VALID-empty state)
+/// would have required coordinated edits at all three sites — exactly the failure mode that
+/// caused the earlier `computeWindowFingerprint` duplication scare.
+///
+/// This class consolidates all per-partition mutations behind a state-change DSL.  Each
+/// public method maps to exactly one operation in the per-partition state machine; the
+/// ZK CAS retry loop, the writer-side `validateForPersist` invariant gate, and the watermark
+/// recompute on APPEND all live behind the public API.  Callers express intent
+/// (`appendValid`, `markStale`, ...) and never see the persistence machinery.
+///
+/// <h3>Architecture</h3>
+///
+/// Three layers, deliberately separated:
+///
+///   - **Public state-change DSL** — one method per state-machine op.  Each method has a
+///     well-defined precondition and postcondition (see per-method javadoc) and selects the
+///     appropriate retry profile.  Callers never construct `PartitionInfo` directly.
+///   - **Private CRUD primitives** — `addPartitionInfo` / `updatePartitionInfo` /
+///     `removePartitionInfo`.  Each enforces a structural invariant on the in-memory map
+///     (e.g. "cannot add an already-existing entry") that fail-loud on violation, surfacing
+///     races as observable exceptions instead of silent overwrites.
+///   - **Single CAS engine** — `applyMutation` runs a caller-supplied mutator under the
+///     fetch / mutate / version-checked-write loop, with retry classification:
+///       - [MaterializedViewRuntimeMetadataUtils.CasConflictException] → re-fetch + retry
+///       - other [ZkException] (transport / session) → log warn + retry
+///       - [IllegalStateException] / [IllegalArgumentException] (validateForPersist) →
+///         propagate (fail-fast, no retry)
+///       - any other exception → propagate
+///
+/// <h3>Retry profile</h3>
+///
+/// Two profiles, calibrated to op semantics:
+///
+///   - **Critical** ([#DEFAULT_CRITICAL_MAX_ATTEMPTS] = 128, cluster-tunable via
+///     `pinot.materialized.view.executor.runtime.update.max.attempts`) — used by every
+///     state-changing op except `revertValid`.  Failure forces a minion task retry at the
+///     Helix level, which is far more expensive than a CAS retry, so the budget is sized
+///     to absorb realistic contention with `maxTasksPerBatch` parallel writers.
+///   - **Revert** ([#REVERT_MAX_ATTEMPTS] = 8) — used by `revertValid` only.  Failure
+///     means we did not avoid one OVERWRITE task, which the next scheduling cycle will
+///     either run for real or revert again.  Spending 128 retries to save one task is
+///     wasteful.
+///
+/// <h3>Concurrency</h3>
+///
+/// The manager is thread-safe and stateless above the ZK property store handle.  Concurrent
+/// callers contending on the same MV runtime znode are serialized by ZK CAS; one thread's
+/// commit invalidates the other's snapshot, the loser re-fetches and re-runs its mutator.
+/// Different MVs map to different znodes — no cross-MV contention.
+///
+/// <h3>Scope discipline</h3>
+///
+/// This class manages exactly the partition map and `watermarkMs` field of
+/// [MaterializedViewRuntimeMetadata].  It MUST NOT acquire other znodes
+/// (e.g. [MaterializedViewDefinitionMetadata]), call source-table or table-config services,
+/// or grow a method count beyond the public DSL listed below.  See `red lines` in the PR
+/// description.
+public final class MaterializedViewPartitionManager {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MaterializedViewPartitionManager.class);
+
+  /// Compile-time default for the critical retry budget.  Overridable per cluster (no
+  /// restart) via `pinot.materialized.view.executor.runtime.update.max.attempts`.  Sized so
+  /// `maxTasksPerBatch` parallel APPEND completions cannot exhaust the budget under realistic
+  /// contention.  Reused by ConsistencyMgr-style mark ops so a STALE marking is never
+  /// silently dropped.
+  static final int DEFAULT_CRITICAL_MAX_ATTEMPTS = 128;
+
+  /// Compile-time retry budget for `revertValid` only.  Failure is recovered by the next
+  /// scheduling cycle (the partition stays STALE and is retried), so a small budget is
+  /// appropriate.  Not cluster-tunable — operationally, raising this would not buy
+  /// correctness, only delay.
+  static final int REVERT_MAX_ATTEMPTS = 8;
+
+  /// Backoff envelope for the critical profile.  Total wait at the cap with worst-case
+  /// jitter is about `128 * (50 + 150) = 25.6 s`, well within the implicit per-task timeout.
+  private static final long CRITICAL_BACKOFF_BASE_MS = 50L;
+  private static final int CRITICAL_BACKOFF_JITTER_MS = 150;
+
+  /// Backoff envelope for the revert profile.  Total wait at the cap is about
+  /// `8 * (5 + 20) = 200 ms`.  Tight CAS-loop livelock is the failure to defend against;
+  /// scheduler thread blocking is the failure to avoid.
+  private static final long REVERT_BACKOFF_BASE_MS = 5L;
+  private static final int REVERT_BACKOFF_JITTER_MS = 20;
+
+  private final HelixPropertyStore<ZNRecord> _propertyStore;
+
+  /// Optional live cluster-config reader used to override [#DEFAULT_CRITICAL_MAX_ATTEMPTS]
+  /// at runtime.  Null in unit tests; null reader falls back to the compile-time default.
+  @Nullable
+  private final Function<String, String> _clusterConfigReader;
+
+  public MaterializedViewPartitionManager(HelixPropertyStore<ZNRecord> propertyStore,
+      @Nullable Function<String, String> clusterConfigReader) {
+    Preconditions.checkArgument(propertyStore != null, "propertyStore must not be null");
+    _propertyStore = propertyStore;
+    _clusterConfigReader = clusterConfigReader;
+  }
+
+  // ─────────────────────────────────────────────────────────────────────────────────────
+  //  Public state-change DSL
+  // ─────────────────────────────────────────────────────────────────────────────────────
+
+  /// APPEND op: VACANT → VALID(fp).  Invoked by the minion executor at task-commit time
+  /// after a successful materialization of a window that was previously absent from the
+  /// partition map.
+  ///
+  /// Side effects (all in a single CAS write):
+  ///
+  ///   - Inserts a new `PartitionInfo(VALID, fp, now)` at `windowStartMs`.
+  ///   - Advances `watermarkMs` to the highest contiguous VALID upper bound starting at
+  ///     the prior watermark, derived via
+  ///     [MaterializedViewTaskUtils#computeContiguousUpperMs] over the resulting map.
+  ///     The bucket size used for the walk is `windowEndMs - windowStartMs`.
+  ///
+  /// Watermark advancement is bundled with the bucket transition because watermark is a
+  /// derived field of the map state and must stay consistent under concurrent writers.
+  /// Splitting them across two CAS writes would create a transient window where the map
+  /// has the new bucket but the watermark is stale.
+  ///
+  /// Strict precondition: `windowStartMs` must NOT be present in the map.  A violation
+  /// (raced double-dispatch from the scheduler, mid-flight task replay) throws
+  /// [IllegalStateException] inside the CAS loop, which propagates without retry — the
+  /// scheduler dispatch invariant guarantees one APPEND per bucket per cycle, so this
+  /// firing indicates a real bug, not transient contention.
+  ///
+  /// Retry profile: critical (cluster-tunable budget).
+  public void appendValid(String tableNameWithType, long windowStartMs, long windowEndMs,
+      PartitionFingerprint fingerprint) {
+    Preconditions.checkArgument(fingerprint != null, "fingerprint must not be null");
+    Preconditions.checkArgument(windowEndMs > windowStartMs,
+        "Invalid window: windowEndMs (%s) <= windowStartMs (%s) for table %s",
+        windowEndMs, windowStartMs, tableNameWithType);
+    long bucketMs = windowEndMs - windowStartMs;
+    applyMutation(tableNameWithType, current -> {
+      Preconditions.checkState(current != null,
+          "appendValid called before MV runtime znode was initialized for table: %s "
+              + "(cold-start path skipped?)", tableNameWithType);
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      addPartitionInfo(updated, windowStartMs,
+          new PartitionInfo(PartitionState.VALID, fingerprint, System.currentTimeMillis()));
+      long newWatermarkMs = MaterializedViewTaskUtils.computeContiguousUpperMs(
+          current.getWatermarkMs(), updated, bucketMs);
+      LOGGER.info("appendValid: table={} bucket={} watermarkMs {}->{} (partitions={})",
+          tableNameWithType, windowStartMs, current.getWatermarkMs(), newWatermarkMs, updated.size());
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), newWatermarkMs, updated);
+    }, criticalMaxAttempts(), CRITICAL_BACKOFF_BASE_MS, CRITICAL_BACKOFF_JITTER_MS);
+  }
+
+  /// OVERWRITE op: STALE → VALID(fp).  Invoked by the minion executor at task-commit time
+  /// after re-materializing a STALE bucket whose source-side fingerprint has actually
+  /// changed.  Updates the in-place fingerprint and refreshes `lastRefreshTime`; watermark
+  /// is unchanged because OVERWRITE only refreshes existing coverage.
+  ///
+  /// Strict precondition: `windowStartMs` must be present and STALE.  A non-STALE entry
+  /// indicates the scheduler dispatch invariant was violated and throws
+  /// [IllegalStateException] without retry.
+  ///
+  /// Retry profile: critical.
+  public void refreshValid(String tableNameWithType, long windowStartMs,
+      PartitionFingerprint fingerprint) {
+    Preconditions.checkArgument(fingerprint != null, "fingerprint must not be null");
+    applyMutation(tableNameWithType, current -> {
+      Preconditions.checkState(current != null,
+          "refreshValid called before MV runtime znode was initialized for table: %s",
+          tableNameWithType);
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      PartitionInfo existing = updated.get(windowStartMs);
+      Preconditions.checkState(existing != null && existing.getState() == PartitionState.STALE,
+          "refreshValid expects bucket %s to be STALE for table %s, got: %s",
+          windowStartMs, tableNameWithType, existing);
+      updatePartitionInfo(updated, windowStartMs,
+          new PartitionInfo(PartitionState.VALID, fingerprint, System.currentTimeMillis()));
+      LOGGER.info("refreshValid: table={} bucket={} STALE->VALID (fp={})",
+          tableNameWithType, windowStartMs, fingerprint);
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), current.getWatermarkMs(), updated);
+    }, criticalMaxAttempts(), CRITICAL_BACKOFF_BASE_MS, CRITICAL_BACKOFF_JITTER_MS);
+  }
+
+  /// DELETE op: STALE → VALID(EMPTY).  Invoked by the minion executor at task-commit time
+  /// after retention-deleting MV segments for a window whose source data is now empty.
+  /// Persists the `VALID + PartitionFingerprint.EMPTY` shape rather than removing the
+  /// entry — the coverage model treats every processed bucket as "tracked", so a later
+  /// base-table backfill flows through the standard VALID → STALE → OVERWRITE cycle.
+  ///
+  /// <p><b>Design context (open question):</b> The continued existence of the `VALID-empty`
+  /// shape is under review.  An alternative model removes the entry entirely on DELETE
+  /// (treating empty buckets as VACANT), which requires extending
+  /// [MaterializedViewTaskUtils#computeContiguousUpperMs] to walk past VACANT buckets and
+  /// extending the consistency manager to synthesize STALE entries for in-coverage VACANT
+  /// buckets on backfill.  Until that design lands, this method preserves today's
+  /// `VALID-empty` semantics so the scheduler's contiguous-VALID watermark walk continues
+  /// to work without modification.
+  ///
+  /// Strict precondition: `windowStartMs` must be present and STALE.  Non-STALE entries
+  /// indicate the scheduler dispatch invariant was violated; fail-loud surfaces the bug.
+  ///
+  /// Retry profile: critical.
+  public void clearValid(String tableNameWithType, long windowStartMs) {
+    applyMutation(tableNameWithType, current -> {
+      Preconditions.checkState(current != null,
+          "clearValid called before MV runtime znode was initialized for table: %s",
+          tableNameWithType);
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      PartitionInfo existing = updated.get(windowStartMs);
+      Preconditions.checkState(existing != null && existing.getState() == PartitionState.STALE,
+          "clearValid expects bucket %s to be STALE for table %s, got: %s",
+          windowStartMs, tableNameWithType, existing);
+      updatePartitionInfo(updated, windowStartMs,
+          new PartitionInfo(PartitionState.VALID, PartitionFingerprint.EMPTY,
+              System.currentTimeMillis()));
+      LOGGER.info("clearValid: table={} bucket={} STALE->VALID-empty",
+          tableNameWithType, windowStartMs);
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), current.getWatermarkMs(), updated);
+    }, criticalMaxAttempts(), CRITICAL_BACKOFF_BASE_MS, CRITICAL_BACKOFF_JITTER_MS);
+  }
+
+  /// STALE_REVERT op: STALE → VALID, fingerprint and `lastRefreshTime` preserved.  Invoked
+  /// by the scheduler after a precise fingerprint comparison concludes that a STALE marking
+  /// was a false positive (the source did not actually change in the affected window).  The
+  /// underlying data was not re-materialized, so neither fingerprint nor `lastRefreshTime`
+  /// move.
+  ///
+  /// Lenient precondition: if the bucket is no longer STALE on retry (either already
+  /// reverted by a concurrent write or the consistency manager re-marked it for a new
+  /// reason), the call exits successfully without writing — the desired transition is
+  /// already done or no longer applicable.
+  ///
+  /// Retry profile: revert (small budget; failure is recovered by the next scheduling
+  /// cycle).
+  public void revertValid(String tableNameWithType, long windowStartMs) {
+    applyMutation(tableNameWithType, current -> {
+      if (current == null) {
+        LOGGER.debug("revertValid: runtime znode missing for table: {}; skipping",
+            tableNameWithType);
+        return null;
+      }
+      PartitionInfo existing = current.getPartitions().get(windowStartMs);
+      if (existing == null || existing.getState() != PartitionState.STALE) {
+        LOGGER.info("revertValid: bucket {} for table {} is not STALE (got: {}); skipping",
+            windowStartMs, tableNameWithType, existing);
+        return null;
+      }
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      updatePartitionInfo(updated, windowStartMs, existing.withState(PartitionState.VALID));
+      LOGGER.info("revertValid: table={} bucket={} STALE->VALID (false-positive, fp={})",
+          tableNameWithType, windowStartMs, existing.getFingerprint());
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), current.getWatermarkMs(), updated);
+    }, REVERT_MAX_ATTEMPTS, REVERT_BACKOFF_BASE_MS, REVERT_BACKOFF_JITTER_MS);
+  }
+
+  /// MARK_STALE op (single-bucket): VALID → STALE.  Invoked by the consistency manager (and
+  /// future manual REFRESH commands) when base-table data has changed in a way that affects
+  /// `windowStartMs`.  Fingerprint and `lastRefreshTime` are preserved so the scheduler's
+  /// false-positive revert can compare against the prior snapshot.
+  ///
+  /// Lenient: if the bucket is absent or already STALE, the call is a no-op.  Today the
+  /// in-coverage VACANT case is a deliberate skip (matching the prior consistency manager
+  /// behavior); a future change will synthesize a STALE entry for in-coverage VACANT
+  /// buckets to fix the backfill silent-data-loss case.
+  ///

Review Comment:
   PR description states that ConsistencyMgr now covers in-coverage "VACANT → STALE" by synthesizing a STALE entry for absent buckets, but the current manager implementation explicitly treats absent buckets as a no-op (and even calls out synthesis as a future change). If synthesis is required for correctness per the design notes, it likely needs to be implemented here (or the PR description/design docs updated to reflect the current behavior).



##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/scheduler/MaterializedViewTaskUtils.java:
##########
@@ -172,4 +180,51 @@ public static String buildSegmentName(String tableName, long windowStartMs, long
       String attemptId, int segIdx) {
     return tableName + "_" + windowStartMs + "_" + windowEndMs + "_" + attemptId + "_" + segIdx;
   }
+
+  /// Computes a [PartitionFingerprint] over the source segments overlapping
+  /// `[windowStartMs, windowEndMs)`.  Single source of truth — both the scheduler (when
+  /// generating tasks) and the minion executor (when validating fingerprints at commit time)
+  /// must call this method, never re-implement the algorithm.
+  ///
+  /// Algorithm (each step is part of the byte-equality contract — changing any one of them
+  /// breaks the fingerprint and forces an MV-wide recompute):
+  ///
+  ///   1. **Filter** to segments overlapping the half-open window: `start < windowEndMs` AND
+  ///      `end >= windowStartMs`.  Asymmetric `>=` on the lower bound is intentional —
+  ///      `SegmentZKMetadata#getEndTimeMs` is inclusive.
+  ///   2. **Sort** the surviving list by segment name.  Listing order from
+  ///      `getSegmentsZKMetadata` is not guaranteed to be stable across calls, so the sort
+  ///      makes the hash deterministic.  The comparator is part of the algorithm: changing
+  ///      it (e.g. to sort by CRC) is byte-equivalent to changing the hash encoding.
+  ///   3. **Hash** the sorted list with `farmHashFingerprint64`, feeding each segment as
+  ///      `<segmentName>\0<crc>\n`.  FarmHash64 is non-cryptographic but collision-resistant
+  ///      for non-adversarial inputs; it replaces an earlier XOR-CRC scheme that exhibited
+  ///      cancellation collisions (swap two segments with equal combined contribution →
+  ///      identical fingerprint).
+  ///
+  /// Empty overlap returns [PartitionFingerprint#EMPTY] — by construction, since
+  /// `farmHashFingerprint64` over zero input bytes is the constant baked into [#EMPTY].
+  /// Callers MUST NOT pre-filter or pre-sort `allSegments`; the helper is responsible for
+  /// both.  Pre-sorting with a different comparator would silently break byte-equality
+  /// against the executor's commit-time recomputation.
+  public static PartitionFingerprint computeWindowFingerprint(
+      List<SegmentZKMetadata> allSegments, long windowStartMs, long windowEndMs) {
+    List<SegmentZKMetadata> overlapping = new ArrayList<>();
+    for (SegmentZKMetadata seg : allSegments) {
+      long segStartMs = seg.getStartTimeMs();
+      long segEndMs = seg.getEndTimeMs();
+      if (segStartMs < windowEndMs && segEndMs >= windowStartMs) {
+        overlapping.add(seg);
+      }
+    }
+    overlapping.sort(Comparator.comparing(SegmentZKMetadata::getSegmentName));
+    Hasher hasher = Hashing.farmHashFingerprint64().newHasher();
+    for (SegmentZKMetadata seg : overlapping) {
+      hasher.putString(seg.getSegmentName(), StandardCharsets.UTF_8);
+      hasher.putByte((byte) 0);
+      hasher.putLong(seg.getCrc());
+      hasher.putByte((byte) '\n');
+    }
+    return new PartitionFingerprint(overlapping.size(), hasher.hash().asLong());

Review Comment:
   computeWindowFingerprint()'s Javadoc says empty overlap returns PartitionFingerprint.EMPTY, but the implementation always allocates a new PartitionFingerprint (even for empty overlap). Returning the constant in the empty case would make the implementation match the documented contract and avoid per-call allocation on empty windows.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on code in PR #18642:
URL: https://github.com/apache/pinot/pull/18642#discussion_r3367288865


##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -602,19 +560,44 @@ private PartitionFingerprint getTaskFingerprint(Map<String, String> configs, Str
     return fingerprint;
   }
 
-  private void validateSourceFingerprintAtCommit(Map<String, String> configs, String tableName, long windowStartMs,
-      long windowEndMs, PartitionFingerprint taskFingerprint) {
+  /// Computes the current source-table [PartitionFingerprint] for `[windowStartMs, windowEndMs)`
+  /// from live ZK segment metadata.  Shared by the APPEND / OVERWRITE commit-time fingerprint
+  /// validation and the DELETE commit-time emptiness re-check.  Requires
+  /// [MaterializedViewTask#SOURCE_TABLE_NAME_KEY] in the task config (the scheduler sets it for
+  /// every task mode); fails loud if absent so a malformed / pre-upgrade task cannot silently
+  /// skip the validation.
+  private PartitionFingerprint computeSourceWindowFingerprint(Map<String, String> configs, String tableName,
+      long windowStartMs, long windowEndMs) {
     String sourceTableName = configs.get(MaterializedViewTask.SOURCE_TABLE_NAME_KEY);

Review Comment:
   This new hard requirement is not rolling-upgrade-safe. Older controllers already emit DELETE tasks without `SOURCE_TABLE_NAME_KEY`; if a minion is upgraded first, those legacy tasks now fail here instead of cleaning up the bucket. Because V1 routing still relies on `watermarkMs` rather than per-bucket state, that can strand below-watermark STALE partitions during a mixed-version rollout. This needs a backward-compatible fallback for legacy DELETE tasks, or an explicit controller-first rollout gate.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on code in PR #18642:
URL: https://github.com/apache/pinot/pull/18642#discussion_r3366657158


##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -449,22 +450,40 @@ public void postProcess(PinotTaskConfig pinotTaskConfig) {
     long windowStartMs = Long.parseLong(configs.get(MaterializedViewTask.WINDOW_START_MS_KEY));
     long windowEndMs = Long.parseLong(configs.get(MaterializedViewTask.WINDOW_END_MS_KEY));
 
-    updateMaterializedViewRuntime(configs, tableName, taskMode, windowStartMs, windowEndMs);
-  }
+    MaterializedViewPartitionManager partitionManager = new MaterializedViewPartitionManager(
+        MINION_CONTEXT.getHelixPropertyStore(),
+        MaterializedViewTaskExecutor::readMinionClusterConfig);
 
-  /// Updates [MaterializedViewRuntimeMetadata] in a single CAS write, combining:
-  ///
-  ///   - partitions: set VALID with new fingerprint (APPEND/OVERWRITE) or remove (DELETE)
-  ///   - watermarkMs: advance on APPEND only (drives both scheduler dispatch and the
-  ///       broker's SPLIT_REWRITE boundary)
-  ///
-  // Compile-time default for the CAS retry budget when racing to update MaterializedViewRuntimeMetadata.
-  // Up to maxTasksPerBatch executors can contend per batch completion; each retry re-fetches the
-  // latest version with jittered backoff (Thread.sleep below). 128 is well above any realistic
-  // maxTasksPerBatch and stays low enough that genuinely pathological contention still surfaces as
-  // a task failure (caught by Helix and retried at the task level). Overridable per cluster via
-  // `MaterializedViewTask.CLUSTER_CONFIG_KEY_MAX_RUNTIME_UPDATE_ATTEMPTS` (no minion restart).
-  private static final int DEFAULT_MAX_RUNTIME_UPDATE_ATTEMPTS = 128;
+    if (MaterializedViewTask.TASK_MODE_DELETE.equals(taskMode)) {
+      // DELETE writes `VALID + PartitionFingerprint.EMPTY` rather than removing the entry —
+      // see `MaterializedViewPartitionManager#clearValid` for the design rationale (no
+      // fingerprint validation needed; the source contained zero overlapping segments by
+      // construction at scheduler dispatch time).
+      partitionManager.clearValid(tableName, windowStartMs);

Review Comment:
   Addressed in a71f721743: DELETE now re-checks source emptiness inside `clearValid`'s CAS loop (a `Supplier` re-reads the source on every attempt) and leaves the bucket STALE for OVERWRITE when a backfill is detected, instead of writing `VALID+EMPTY`. The narrow residual cross-ZK window is backstopped by a periodic `VALID-empty` re-evaluation sweep in `MaterializedViewConsistencyManager` (8e1295a357).



##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -173,7 +171,8 @@ public SegmentConversionResult executeTask(PinotTaskConfig pinotTaskConfig)
     String taskMode = configs.getOrDefault(MaterializedViewTask.TASK_MODE_KEY,
         MaterializedViewTask.TASK_MODE_APPEND);
 
-    // DELETE mode: skip query execution, only remove existing MV segments
+    // DELETE mode: skip query execution; remove existing MV segments and rewrite the
+    // runtime PartitionInfo to VALID-empty (see updateMaterializedViewRuntime DELETE branch).
     if (MaterializedViewTask.TASK_MODE_DELETE.equals(taskMode)) {

Review Comment:
   Fixed in a71f721743 — the DELETE-mode comments now reference `executeDeleteTask` / `postProcess` (the `MaterializedViewPartitionManager`-based path); `updateMaterializedViewRuntime` is no longer referenced in the production code.



##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -392,8 +391,10 @@ private void revertWindowSegmentReplace(String tableNameWithType, String lineage
   }
 
   /// Handles DELETE mode: removes all existing MV segments for the given time window
-  /// via segment lineage replace (segmentsFrom=[old segments], segmentsTo=[]).
-  /// No query is executed and no new segments are created.
+  /// via segment lineage replace (segmentsFrom=[old segments], segmentsTo=[]).  No query
+  /// is executed and no new MV segments are created; the runtime PartitionInfo is rewritten
+  /// to `VALID + PartitionFingerprint.EMPTY` by [#updateMaterializedViewRuntime] in
+  /// [#postProcess], so the partition stays tracked-but-empty rather than disappearing.

Review Comment:
   Fixed in a71f721743 — `executeDeleteTask`'s javadoc now points to `postProcess` instead of the removed `updateMaterializedViewRuntime`.



##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/scheduler/MaterializedViewTaskUtils.java:
##########
@@ -172,4 +180,51 @@ public static String buildSegmentName(String tableName, long windowStartMs, long
       String attemptId, int segIdx) {
     return tableName + "_" + windowStartMs + "_" + windowEndMs + "_" + attemptId + "_" + segIdx;
   }
+
+  /// Computes a [PartitionFingerprint] over the source segments overlapping
+  /// `[windowStartMs, windowEndMs)`.  Single source of truth — both the scheduler (when
+  /// generating tasks) and the minion executor (when validating fingerprints at commit time)
+  /// must call this method, never re-implement the algorithm.
+  ///
+  /// Algorithm (each step is part of the byte-equality contract — changing any one of them
+  /// breaks the fingerprint and forces an MV-wide recompute):
+  ///
+  ///   1. **Filter** to segments overlapping the half-open window: `start < windowEndMs` AND
+  ///      `end >= windowStartMs`.  Asymmetric `>=` on the lower bound is intentional —
+  ///      `SegmentZKMetadata#getEndTimeMs` is inclusive.
+  ///   2. **Sort** the surviving list by segment name.  Listing order from
+  ///      `getSegmentsZKMetadata` is not guaranteed to be stable across calls, so the sort
+  ///      makes the hash deterministic.  The comparator is part of the algorithm: changing
+  ///      it (e.g. to sort by CRC) is byte-equivalent to changing the hash encoding.
+  ///   3. **Hash** the sorted list with `farmHashFingerprint64`, feeding each segment as
+  ///      `<segmentName>\0<crc>\n`.  FarmHash64 is non-cryptographic but collision-resistant
+  ///      for non-adversarial inputs; it replaces an earlier XOR-CRC scheme that exhibited
+  ///      cancellation collisions (swap two segments with equal combined contribution →
+  ///      identical fingerprint).
+  ///
+  /// Empty overlap returns [PartitionFingerprint#EMPTY] — by construction, since
+  /// `farmHashFingerprint64` over zero input bytes is the constant baked into [#EMPTY].
+  /// Callers MUST NOT pre-filter or pre-sort `allSegments`; the helper is responsible for
+  /// both.  Pre-sorting with a different comparator would silently break byte-equality
+  /// against the executor's commit-time recomputation.
+  public static PartitionFingerprint computeWindowFingerprint(
+      List<SegmentZKMetadata> allSegments, long windowStartMs, long windowEndMs) {
+    List<SegmentZKMetadata> overlapping = new ArrayList<>();
+    for (SegmentZKMetadata seg : allSegments) {
+      long segStartMs = seg.getStartTimeMs();
+      long segEndMs = seg.getEndTimeMs();
+      if (segStartMs < windowEndMs && segEndMs >= windowStartMs) {
+        overlapping.add(seg);
+      }
+    }
+    overlapping.sort(Comparator.comparing(SegmentZKMetadata::getSegmentName));
+    Hasher hasher = Hashing.farmHashFingerprint64().newHasher();
+    for (SegmentZKMetadata seg : overlapping) {
+      hasher.putString(seg.getSegmentName(), StandardCharsets.UTF_8);
+      hasher.putByte((byte) 0);
+      hasher.putLong(seg.getCrc());
+      hasher.putByte((byte) '\n');
+    }
+    return new PartitionFingerprint(overlapping.size(), hasher.hash().asLong());

Review Comment:
   Fixed in 8e1295a357 — `computeWindowFingerprint` now returns `PartitionFingerprint.EMPTY` for empty overlap, matching the documented contract and avoiding the per-call allocation. The value is byte-identical to the prior hash-of-empty-input result, and `testWindowFingerprintForEmptyOverlapMatchesEmptyConstant` pins the contract.



##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/consistency/MaterializedViewConsistencyManager.java:
##########
@@ -410,55 +405,40 @@ private boolean markPartitionsDirty(String viewTableName, long affectedStartMs,
             + "mark partitions without a bucket size.  Repair the MV table config.",
         viewTableName, bucketMs);
 
-    // Iterate every bucket [partStart, partStart+bucketMs) that overlaps the affected range
-    // [affectedStartMs, affectedEndMs]. Two ranges overlap when start1 <= end2 AND end1 >= start2.
+    // Cap affectedEndMs at watermarkMs.  No partition with partStart > watermarkMs can exist
+    // (the writer invariant), so any bucket beyond that cannot be marked STALE anyway.  This
+    // protects the bucket iteration below from a caller passing Long.MAX_VALUE (full-range
+    // invalidation from notifyMaterializedViewConsistencyManager paths when segment
+    // startTime/endTime is unknown), which would otherwise enumerate ~watermarkMs/bucketMs
+    // candidates — orders of magnitude more than the number of real partitions.
+    long cappedEnd = Math.min(affectedEndMs, runtime.getWatermarkMs());
+    if (cappedEnd < affectedStartMs) {
+      LOGGER.debug("Affected range [{}, {}] is past watermarkMs ({}) for MV table: {}; nothing to mark",
+          affectedStartMs, affectedEndMs, runtime.getWatermarkMs(), viewTableName);
+      return Collections.emptyList();
+    }
+
+    // Enumerate every bucket [partStart, partStart+bucketMs) that overlaps the affected range
+    // [affectedStartMs, cappedEnd]. Two ranges overlap when start1 <= end2 AND end1 >= start2.
     // The first overlapping bucket has partStart = floorDiv(affectedStartMs, bucketMs) * bucketMs;
-    // the last has partStart <= affectedEndMs (any bucket whose start is past affectedEndMs
-    // cannot overlap because partStart > affectedEndMs >= affectedStartMs implies the bucket
-    // starts after the affected range ends). floorDiv (instead of /) is used defensively so a
-    // future caller passing negative affectedStartMs would still produce the correct floor.
-    // Only flip existing VALID entries to STALE.  Absent buckets are NOT synthesized — under
-    // Design C, a bucket's absence from the partition map already means "MV does not cover this
-    // range", so the broker rewrite (PR 2) routes those queries to the base table.  Synthesizing
-    // STALE entries for every uncovered bucket below `watermarkMs` would explode the znode size
-    // on a full-range invalidation (~watermarkMs / bucketMs entries) without affecting routing
-    // correctness — the bucket-iteration loop below stays O(affectedRange / bucketMs) but the
-    // persisted map grows only with real partitions.
+    // the last has partStart <= cappedEnd (any bucket whose start is past cappedEnd cannot
+    // overlap because partStart > cappedEnd >= affectedStartMs implies the bucket starts after
+    // the affected range ends). floorDiv (instead of /) is used defensively so a future caller
+    // passing negative affectedStartMs would still produce the correct floor.
+    //
+    // Absent buckets are still enumerated: the manager's per-bucket lenient filter no-ops them
+    // under the CAS lock.  Synthesizing STALE entries for absent buckets is intentionally NOT
+    // done here — under Design C, a bucket's absence already means "MV does not cover this
+    // range" and the broker routes those queries to the base.  Synthesizing on every
+    // full-range invalidation would explode the znode size by ~watermarkMs/bucketMs entries
+    // without affecting routing correctness.
+    List<Long> buckets = new ArrayList<>();
     long partStart = Math.floorDiv(affectedStartMs, bucketMs) * bucketMs;
-    while (partStart <= affectedEndMs) {
-      PartitionInfo info = updatedInfos.get(partStart);
-      if (info != null && info.getState() == PartitionState.VALID) {
-        updatedInfos.put(partStart, info.withState(PartitionState.STALE));
-        anyChanged = true;
-        markedCount++;
-      }
+    while (partStart <= cappedEnd) {
+      buckets.add(partStart);
       partStart += bucketMs;
     }

Review Comment:
   Fixed in 8e1295a357 — `enumerateCandidateBuckets` now iterates the existing partition keys in range instead of every bucket slot, bounding the candidate list to the partition count (a full-range invalidation with a small `bucketMs` and a large watermark no longer allocates ~`watermarkMs/bucketMs` entries). Functionally identical, since `markStale` already no-ops absent buckets under its CAS lock.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on code in PR #18642:
URL: https://github.com/apache/pinot/pull/18642#discussion_r3367288355


##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -602,19 +560,44 @@ private PartitionFingerprint getTaskFingerprint(Map<String, String> configs, Str
     return fingerprint;
   }
 
-  private void validateSourceFingerprintAtCommit(Map<String, String> configs, String tableName, long windowStartMs,
-      long windowEndMs, PartitionFingerprint taskFingerprint) {
+  /// Computes the current source-table [PartitionFingerprint] for `[windowStartMs, windowEndMs)`
+  /// from live ZK segment metadata.  Shared by the APPEND / OVERWRITE commit-time fingerprint
+  /// validation and the DELETE commit-time emptiness re-check.  Requires
+  /// [MaterializedViewTask#SOURCE_TABLE_NAME_KEY] in the task config (the scheduler sets it for
+  /// every task mode); fails loud if absent so a malformed / pre-upgrade task cannot silently
+  /// skip the validation.
+  private PartitionFingerprint computeSourceWindowFingerprint(Map<String, String> configs, String tableName,
+      long windowStartMs, long windowEndMs) {
     String sourceTableName = configs.get(MaterializedViewTask.SOURCE_TABLE_NAME_KEY);

Review Comment:
   This new hard requirement is not rolling-upgrade-safe. Older controllers already emit DELETE tasks without ; if a minion is upgraded first, those legacy tasks now fail here instead of cleaning up the bucket. Because V1 routing still relies on  rather than per-bucket state, that can strand below-watermark STALE partitions during a mixed-version rollout. This needs a backward-compatible fallback for legacy DELETE tasks, or an explicit controller-first rollout gate.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 merged PR #18642:
URL: https://github.com/apache/pinot/pull/18642


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on PR #18642:
URL: https://github.com/apache/pinot/pull/18642#issuecomment-4692964854

   Opened the companion docs update: pinot-contrib/pinot-docs#870 https://github.com/pinot-contrib/pinot-docs/pull/870


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on code in PR #18642:
URL: https://github.com/apache/pinot/pull/18642#discussion_r3333999423


##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/materializedview/MaterializedViewTaskExecutor.java:
##########
@@ -449,22 +450,40 @@ public void postProcess(PinotTaskConfig pinotTaskConfig) {
     long windowStartMs = Long.parseLong(configs.get(MaterializedViewTask.WINDOW_START_MS_KEY));
     long windowEndMs = Long.parseLong(configs.get(MaterializedViewTask.WINDOW_END_MS_KEY));
 
-    updateMaterializedViewRuntime(configs, tableName, taskMode, windowStartMs, windowEndMs);
-  }
+    MaterializedViewPartitionManager partitionManager = new MaterializedViewPartitionManager(
+        MINION_CONTEXT.getHelixPropertyStore(),
+        MaterializedViewTaskExecutor::readMinionClusterConfig);
 
-  /// Updates [MaterializedViewRuntimeMetadata] in a single CAS write, combining:
-  ///
-  ///   - partitions: set VALID with new fingerprint (APPEND/OVERWRITE) or remove (DELETE)
-  ///   - watermarkMs: advance on APPEND only (drives both scheduler dispatch and the
-  ///       broker's SPLIT_REWRITE boundary)
-  ///
-  // Compile-time default for the CAS retry budget when racing to update MaterializedViewRuntimeMetadata.
-  // Up to maxTasksPerBatch executors can contend per batch completion; each retry re-fetches the
-  // latest version with jittered backoff (Thread.sleep below). 128 is well above any realistic
-  // maxTasksPerBatch and stays low enough that genuinely pathological contention still surfaces as
-  // a task failure (caught by Helix and retried at the task level). Overridable per cluster via
-  // `MaterializedViewTask.CLUSTER_CONFIG_KEY_MAX_RUNTIME_UPDATE_ATTEMPTS` (no minion restart).
-  private static final int DEFAULT_MAX_RUNTIME_UPDATE_ATTEMPTS = 128;
+    if (MaterializedViewTask.TASK_MODE_DELETE.equals(taskMode)) {
+      // DELETE writes `VALID + PartitionFingerprint.EMPTY` rather than removing the entry —
+      // see `MaterializedViewPartitionManager#clearValid` for the design rationale (no
+      // fingerprint validation needed; the source contained zero overlapping segments by
+      // construction at scheduler dispatch time).
+      partitionManager.clearValid(tableName, windowStartMs);

Review Comment:
   This removes the commit-time source check from the DELETE path. If a backfill lands after the scheduler emitted DELETE but before `postProcess()` runs, the create-segment event is ignored because the partition is already STALE; then `clearValid()` flips it to `VALID + EMPTY` and the broker will rewrite queries to an empty MV bucket even though source rows now exist. The old behavior of removing the entry still fell back to base-table routing, so this is a silent wrong-results regression. DELETE needs a commit-time fingerprint/emptiness validation (or another way to preserve STALE on mid-flight backfills) before writing `VALID + EMPTY`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["codecov-commenter (via GitHub)" <gi...@apache.org>]

codecov-commenter commented on PR #18642:
URL: https://github.com/apache/pinot/pull/18642#issuecomment-4589638828

   ## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18642?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :white_check_mark: All modified and coverable lines are covered by tests.
   :white_check_mark: Project coverage is 56.80%. Comparing base ([`eddf4c0`](https://app.codecov.io/gh/apache/pinot/commit/eddf4c0deb3f8c0b710e27b4d946333ac5fa9311?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`af88fd6`](https://app.codecov.io/gh/apache/pinot/commit/af88fd66aacc0d6f45b38cdb8cdfd6c5beb6ea87?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 2 commits behind head on master.
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@            Coverage Diff            @@
   ##             master   #18642   +/-   ##
   =========================================
     Coverage     56.79%   56.80%           
   + Complexity        7        1    -6     
   =========================================
     Files          2579     2576    -3     
     Lines        149557   149551    -6     
     Branches      24165    24165           
   =========================================
   + Hits          84939    84945    +6     
   + Misses        57424    57417    -7     
   + Partials       7194     7189    -5     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration1](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration2](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [java-21](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.80% <ø> (+<0.01%)` | :arrow_up: |
   | [temurin](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.80% <ø> (+<0.01%)` | :arrow_up: |
   | [unittests](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.80% <ø> (+<0.01%)` | :arrow_up: |
   | [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18642/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.80% <ø> (+<0.01%)` | :arrow_up: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   </details>
   
   [:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/18642?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on code in PR #18642:
URL: https://github.com/apache/pinot/pull/18642#discussion_r3366657509


##########
pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/metadata/MaterializedViewPartitionManager.java:
##########
@@ -0,0 +1,499 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.materializedview.metadata;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+import org.apache.helix.store.HelixPropertyStore;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.zkclient.exception.ZkException;
+import org.apache.pinot.materializedview.scheduler.MaterializedViewTaskUtils;
+import org.apache.pinot.spi.utils.CommonConstants.MaterializedViewTask;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Single source of truth for every per-partition state transition on the
+/// [MaterializedViewRuntimeMetadata] runtime znode.
+///
+/// <h3>Why this class exists</h3>
+///
+/// Before this manager landed, the runtime znode was mutated by three separate sites — the
+/// minion executor (APPEND / OVERWRITE / DELETE task commits), the scheduler (false-positive
+/// STALE reverts), and the consistency manager (VALID → STALE markings).  Each site shipped
+/// its own version-checked CAS loop, with subtly different retry budgets, backoff jitter,
+/// and exception-classification rules.  Subsequent state-machine changes (e.g. extending the
+/// VACANT → STALE synthesize for in-coverage backfill, or revoking the VALID-empty state)
+/// would have required coordinated edits at all three sites — exactly the failure mode that
+/// caused the earlier `computeWindowFingerprint` duplication scare.
+///
+/// This class consolidates all per-partition mutations behind a state-change DSL.  Each
+/// public method maps to exactly one operation in the per-partition state machine; the
+/// ZK CAS retry loop, the writer-side `validateForPersist` invariant gate, and the watermark
+/// recompute on APPEND all live behind the public API.  Callers express intent
+/// (`appendValid`, `markStale`, ...) and never see the persistence machinery.
+///
+/// <h3>Architecture</h3>
+///
+/// Three layers, deliberately separated:
+///
+///   - **Public state-change DSL** — one method per state-machine op.  Each method has a
+///     well-defined precondition and postcondition (see per-method javadoc) and selects the
+///     appropriate retry profile.  Callers never construct `PartitionInfo` directly.
+///   - **Private CRUD primitives** — `addPartitionInfo` / `updatePartitionInfo` /
+///     `removePartitionInfo`.  Each enforces a structural invariant on the in-memory map
+///     (e.g. "cannot add an already-existing entry") that fail-loud on violation, surfacing
+///     races as observable exceptions instead of silent overwrites.
+///   - **Single CAS engine** — `applyMutation` runs a caller-supplied mutator under the
+///     fetch / mutate / version-checked-write loop, with retry classification:
+///       - [MaterializedViewRuntimeMetadataUtils.CasConflictException] → re-fetch + retry
+///       - other [ZkException] (transport / session) → log warn + retry
+///       - [IllegalStateException] / [IllegalArgumentException] (validateForPersist) →
+///         propagate (fail-fast, no retry)
+///       - any other exception → propagate
+///
+/// <h3>Retry profile</h3>
+///
+/// Two profiles, calibrated to op semantics:
+///
+///   - **Critical** ([#DEFAULT_CRITICAL_MAX_ATTEMPTS] = 128, cluster-tunable via
+///     `pinot.materialized.view.executor.runtime.update.max.attempts`) — used by every
+///     state-changing op except `revertValid`.  Failure forces a minion task retry at the
+///     Helix level, which is far more expensive than a CAS retry, so the budget is sized
+///     to absorb realistic contention with `maxTasksPerBatch` parallel writers.
+///   - **Revert** ([#REVERT_MAX_ATTEMPTS] = 8) — used by `revertValid` only.  Failure
+///     means we did not avoid one OVERWRITE task, which the next scheduling cycle will
+///     either run for real or revert again.  Spending 128 retries to save one task is
+///     wasteful.
+///
+/// <h3>Concurrency</h3>
+///
+/// The manager is thread-safe and stateless above the ZK property store handle.  Concurrent
+/// callers contending on the same MV runtime znode are serialized by ZK CAS; one thread's
+/// commit invalidates the other's snapshot, the loser re-fetches and re-runs its mutator.
+/// Different MVs map to different znodes — no cross-MV contention.
+///
+/// <h3>Scope discipline</h3>
+///
+/// This class manages exactly the partition map and `watermarkMs` field of
+/// [MaterializedViewRuntimeMetadata].  It MUST NOT acquire other znodes
+/// (e.g. [MaterializedViewDefinitionMetadata]), call source-table or table-config services,
+/// or grow a method count beyond the public DSL listed below.  See `red lines` in the PR
+/// description.
+public final class MaterializedViewPartitionManager {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MaterializedViewPartitionManager.class);
+
+  /// Compile-time default for the critical retry budget.  Overridable per cluster (no
+  /// restart) via `pinot.materialized.view.executor.runtime.update.max.attempts`.  Sized so
+  /// `maxTasksPerBatch` parallel APPEND completions cannot exhaust the budget under realistic
+  /// contention.  Reused by ConsistencyMgr-style mark ops so a STALE marking is never
+  /// silently dropped.
+  static final int DEFAULT_CRITICAL_MAX_ATTEMPTS = 128;
+
+  /// Compile-time retry budget for `revertValid` only.  Failure is recovered by the next
+  /// scheduling cycle (the partition stays STALE and is retried), so a small budget is
+  /// appropriate.  Not cluster-tunable — operationally, raising this would not buy
+  /// correctness, only delay.
+  static final int REVERT_MAX_ATTEMPTS = 8;
+
+  /// Backoff envelope for the critical profile.  Total wait at the cap with worst-case
+  /// jitter is about `128 * (50 + 150) = 25.6 s`, well within the implicit per-task timeout.
+  private static final long CRITICAL_BACKOFF_BASE_MS = 50L;
+  private static final int CRITICAL_BACKOFF_JITTER_MS = 150;
+
+  /// Backoff envelope for the revert profile.  Total wait at the cap is about
+  /// `8 * (5 + 20) = 200 ms`.  Tight CAS-loop livelock is the failure to defend against;
+  /// scheduler thread blocking is the failure to avoid.
+  private static final long REVERT_BACKOFF_BASE_MS = 5L;
+  private static final int REVERT_BACKOFF_JITTER_MS = 20;
+
+  private final HelixPropertyStore<ZNRecord> _propertyStore;
+
+  /// Optional live cluster-config reader used to override [#DEFAULT_CRITICAL_MAX_ATTEMPTS]
+  /// at runtime.  Null in unit tests; null reader falls back to the compile-time default.
+  @Nullable
+  private final Function<String, String> _clusterConfigReader;
+
+  public MaterializedViewPartitionManager(HelixPropertyStore<ZNRecord> propertyStore,
+      @Nullable Function<String, String> clusterConfigReader) {
+    Preconditions.checkArgument(propertyStore != null, "propertyStore must not be null");
+    _propertyStore = propertyStore;
+    _clusterConfigReader = clusterConfigReader;
+  }
+
+  // ─────────────────────────────────────────────────────────────────────────────────────
+  //  Public state-change DSL
+  // ─────────────────────────────────────────────────────────────────────────────────────
+
+  /// APPEND op: VACANT → VALID(fp).  Invoked by the minion executor at task-commit time
+  /// after a successful materialization of a window that was previously absent from the
+  /// partition map.
+  ///
+  /// Side effects (all in a single CAS write):
+  ///
+  ///   - Inserts a new `PartitionInfo(VALID, fp, now)` at `windowStartMs`.
+  ///   - Advances `watermarkMs` to the highest contiguous VALID upper bound starting at
+  ///     the prior watermark, derived via
+  ///     [MaterializedViewTaskUtils#computeContiguousUpperMs] over the resulting map.
+  ///     The bucket size used for the walk is `windowEndMs - windowStartMs`.
+  ///
+  /// Watermark advancement is bundled with the bucket transition because watermark is a
+  /// derived field of the map state and must stay consistent under concurrent writers.
+  /// Splitting them across two CAS writes would create a transient window where the map
+  /// has the new bucket but the watermark is stale.
+  ///
+  /// Strict precondition: `windowStartMs` must NOT be present in the map.  A violation
+  /// (raced double-dispatch from the scheduler, mid-flight task replay) throws
+  /// [IllegalStateException] inside the CAS loop, which propagates without retry — the
+  /// scheduler dispatch invariant guarantees one APPEND per bucket per cycle, so this
+  /// firing indicates a real bug, not transient contention.
+  ///
+  /// Retry profile: critical (cluster-tunable budget).
+  public void appendValid(String tableNameWithType, long windowStartMs, long windowEndMs,
+      PartitionFingerprint fingerprint) {
+    Preconditions.checkArgument(fingerprint != null, "fingerprint must not be null");
+    Preconditions.checkArgument(windowEndMs > windowStartMs,
+        "Invalid window: windowEndMs (%s) <= windowStartMs (%s) for table %s",
+        windowEndMs, windowStartMs, tableNameWithType);
+    long bucketMs = windowEndMs - windowStartMs;
+    applyMutation(tableNameWithType, current -> {
+      Preconditions.checkState(current != null,
+          "appendValid called before MV runtime znode was initialized for table: %s "
+              + "(cold-start path skipped?)", tableNameWithType);
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      addPartitionInfo(updated, windowStartMs,
+          new PartitionInfo(PartitionState.VALID, fingerprint, System.currentTimeMillis()));
+      long newWatermarkMs = MaterializedViewTaskUtils.computeContiguousUpperMs(
+          current.getWatermarkMs(), updated, bucketMs);
+      LOGGER.info("appendValid: table={} bucket={} watermarkMs {}->{} (partitions={})",
+          tableNameWithType, windowStartMs, current.getWatermarkMs(), newWatermarkMs, updated.size());
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), newWatermarkMs, updated);
+    }, criticalMaxAttempts(), CRITICAL_BACKOFF_BASE_MS, CRITICAL_BACKOFF_JITTER_MS);
+  }
+
+  /// OVERWRITE op: STALE → VALID(fp).  Invoked by the minion executor at task-commit time
+  /// after re-materializing a STALE bucket whose source-side fingerprint has actually
+  /// changed.  Updates the in-place fingerprint and refreshes `lastRefreshTime`; watermark
+  /// is unchanged because OVERWRITE only refreshes existing coverage.
+  ///
+  /// Strict precondition: `windowStartMs` must be present and STALE.  A non-STALE entry
+  /// indicates the scheduler dispatch invariant was violated and throws
+  /// [IllegalStateException] without retry.
+  ///
+  /// Retry profile: critical.
+  public void refreshValid(String tableNameWithType, long windowStartMs,
+      PartitionFingerprint fingerprint) {
+    Preconditions.checkArgument(fingerprint != null, "fingerprint must not be null");
+    applyMutation(tableNameWithType, current -> {
+      Preconditions.checkState(current != null,
+          "refreshValid called before MV runtime znode was initialized for table: %s",
+          tableNameWithType);
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      PartitionInfo existing = updated.get(windowStartMs);
+      Preconditions.checkState(existing != null && existing.getState() == PartitionState.STALE,
+          "refreshValid expects bucket %s to be STALE for table %s, got: %s",
+          windowStartMs, tableNameWithType, existing);
+      updatePartitionInfo(updated, windowStartMs,
+          new PartitionInfo(PartitionState.VALID, fingerprint, System.currentTimeMillis()));
+      LOGGER.info("refreshValid: table={} bucket={} STALE->VALID (fp={})",
+          tableNameWithType, windowStartMs, fingerprint);
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), current.getWatermarkMs(), updated);
+    }, criticalMaxAttempts(), CRITICAL_BACKOFF_BASE_MS, CRITICAL_BACKOFF_JITTER_MS);
+  }
+
+  /// DELETE op: STALE → VALID(EMPTY).  Invoked by the minion executor at task-commit time
+  /// after retention-deleting MV segments for a window whose source data is now empty.
+  /// Persists the `VALID + PartitionFingerprint.EMPTY` shape rather than removing the
+  /// entry — the coverage model treats every processed bucket as "tracked", so a later
+  /// base-table backfill flows through the standard VALID → STALE → OVERWRITE cycle.
+  ///
+  /// <p><b>Design context (open question):</b> The continued existence of the `VALID-empty`
+  /// shape is under review.  An alternative model removes the entry entirely on DELETE
+  /// (treating empty buckets as VACANT), which requires extending
+  /// [MaterializedViewTaskUtils#computeContiguousUpperMs] to walk past VACANT buckets and
+  /// extending the consistency manager to synthesize STALE entries for in-coverage VACANT
+  /// buckets on backfill.  Until that design lands, this method preserves today's
+  /// `VALID-empty` semantics so the scheduler's contiguous-VALID watermark walk continues
+  /// to work without modification.
+  ///
+  /// Strict precondition: `windowStartMs` must be present and STALE.  Non-STALE entries
+  /// indicate the scheduler dispatch invariant was violated; fail-loud surfaces the bug.
+  ///
+  /// Retry profile: critical.
+  public void clearValid(String tableNameWithType, long windowStartMs) {
+    applyMutation(tableNameWithType, current -> {
+      Preconditions.checkState(current != null,
+          "clearValid called before MV runtime znode was initialized for table: %s",
+          tableNameWithType);
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      PartitionInfo existing = updated.get(windowStartMs);
+      Preconditions.checkState(existing != null && existing.getState() == PartitionState.STALE,
+          "clearValid expects bucket %s to be STALE for table %s, got: %s",
+          windowStartMs, tableNameWithType, existing);
+      updatePartitionInfo(updated, windowStartMs,
+          new PartitionInfo(PartitionState.VALID, PartitionFingerprint.EMPTY,
+              System.currentTimeMillis()));
+      LOGGER.info("clearValid: table={} bucket={} STALE->VALID-empty",
+          tableNameWithType, windowStartMs);
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), current.getWatermarkMs(), updated);
+    }, criticalMaxAttempts(), CRITICAL_BACKOFF_BASE_MS, CRITICAL_BACKOFF_JITTER_MS);
+  }
+
+  /// STALE_REVERT op: STALE → VALID, fingerprint and `lastRefreshTime` preserved.  Invoked
+  /// by the scheduler after a precise fingerprint comparison concludes that a STALE marking
+  /// was a false positive (the source did not actually change in the affected window).  The
+  /// underlying data was not re-materialized, so neither fingerprint nor `lastRefreshTime`
+  /// move.
+  ///
+  /// Lenient precondition: if the bucket is no longer STALE on retry (either already
+  /// reverted by a concurrent write or the consistency manager re-marked it for a new
+  /// reason), the call exits successfully without writing — the desired transition is
+  /// already done or no longer applicable.
+  ///
+  /// Retry profile: revert (small budget; failure is recovered by the next scheduling
+  /// cycle).
+  public void revertValid(String tableNameWithType, long windowStartMs) {
+    applyMutation(tableNameWithType, current -> {
+      if (current == null) {
+        LOGGER.debug("revertValid: runtime znode missing for table: {}; skipping",
+            tableNameWithType);
+        return null;
+      }
+      PartitionInfo existing = current.getPartitions().get(windowStartMs);
+      if (existing == null || existing.getState() != PartitionState.STALE) {
+        LOGGER.info("revertValid: bucket {} for table {} is not STALE (got: {}); skipping",
+            windowStartMs, tableNameWithType, existing);
+        return null;
+      }
+      Map<Long, PartitionInfo> updated = new HashMap<>(current.getPartitions());
+      updatePartitionInfo(updated, windowStartMs, existing.withState(PartitionState.VALID));
+      LOGGER.info("revertValid: table={} bucket={} STALE->VALID (false-positive, fp={})",
+          tableNameWithType, windowStartMs, existing.getFingerprint());
+      return new MaterializedViewRuntimeMetadata(
+          current.getMaterializedViewTableNameWithType(), current.getWatermarkMs(), updated);
+    }, REVERT_MAX_ATTEMPTS, REVERT_BACKOFF_BASE_MS, REVERT_BACKOFF_JITTER_MS);
+  }
+
+  /// MARK_STALE op (single-bucket): VALID → STALE.  Invoked by the consistency manager (and
+  /// future manual REFRESH commands) when base-table data has changed in a way that affects
+  /// `windowStartMs`.  Fingerprint and `lastRefreshTime` are preserved so the scheduler's
+  /// false-positive revert can compare against the prior snapshot.
+  ///
+  /// Lenient: if the bucket is absent or already STALE, the call is a no-op.  Today the
+  /// in-coverage VACANT case is a deliberate skip (matching the prior consistency manager
+  /// behavior); a future change will synthesize a STALE entry for in-coverage VACANT
+  /// buckets to fix the backfill silent-data-loss case.
+  ///

Review Comment:
   The implemented design uses the `VALID-empty` approach rather than `VACANT -> STALE` synthesis: `DELETE` leaves the bucket `VALID-empty` (not `VACANT`), so a backfill flows through the standard `VALID -> STALE -> OVERWRITE` cycle, and the narrow commit-vs-backfill residual is closed by a new periodic `VALID-empty` re-evaluation sweep (8e1295a357). No `VACANT -> STALE` synthesis is required for correctness; absent buckets remain a no-op and the broker routes those queries to the base table. The code matches the design-notes line that frames `VACANT -> STALE` synthesis as future work; the PR description's earlier 'can synthesize VACANT -> STALE' wording (Bugfix summary + operation/transition tables) is now outdated and should be updated to match.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MaterializedView] Harden partition-state engine [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on PR #18642:
URL: https://github.com/apache/pinot/pull/18642#issuecomment-4636262554

   # Review: PR #18642 — [MaterializedView] Harden partition-state engine
   
   **Repo:** apache/pinot
   **Branch:** feat/sse_mv_hardening
   **Author:** hongkunxu (Hongkun Xu)
   **Date:** 2026-06-02
   
   **Repo context loaded:**
   - [x] CLAUDE.md read (root `CLAUDE.md` defining build commands, conventions, mandatory code review)
   - [x] CODEOWNERS read (no CODEOWNERS file in repo)
   - [x] Module-level docs for touched paths read (`pinot-materialized-view/DESIGN.md`)
   
   ---
   
   ## Stage 1: Triage & Evidence Collection
   
   ### PR description
   
   - [x] Description present
   - [x] States what problem is being solved
   - [x] States why this approach
   - [x] States risky areas
   
   **Findings:**
   - Severity: observation. Confidence: high. Evidence: PR body. The description is exceptionally thorough — fixes two correctness bugs (watermark over-advance, backfill-after-delete), refactors three retry sites into one CAS engine, centralizes fingerprinting. The "Why introduce `MaterializedViewPartitionManager`" and "Backward compatibility / rolling upgrade" sections both state the risk surface explicitly. ZK format is unchanged, public REST surface unchanged. No regression-test gaps are claimed.
   
   ### Change summary
   
   - [x] Full diff read
   - [x] Summary produced
   - [x] Files listed by module
   
   **Summary:**
   The PR introduces `MaterializedViewPartitionManager` as the single source of truth for all per-partition state transitions on the MV runtime ZNode. Before this PR, three separate sites (executor, scheduler, consistency manager) each shipped their own CAS retry loop with subtly different budgets, jitter, and exception classification. The new manager exposes a typed DSL (`appendValid` / `refreshValid` / `clearValid` / `revertValid` / `markStale` / `deletePartition`) and routes every mutation through one `applyMutation` CAS engine with two retry profiles (critical / revert). Two correctness bugs are fixed alongside: (1) the APPEND watermark is now capped at `max(source segment endTimeMs)` so it cannot drift past real data when ingestion stalls, and (2) the DELETE task now writes `VALID + PartitionFingerprint.EMPTY` instead of removing the entry so backfill follows the standard `VALID → STALE → OVERWRITE` cycle. Two duplicate `computeWindowFingerprint` copies are collapsed into `M
 aterializedViewTaskUtils#computeWindowFingerprint`. `PartitionInfo`'s constructor becomes package-private with a named `forTesting` factory.
   
   **Files:**
   - `pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/metadata/`
     - `MaterializedViewPartitionManager.java` (new, 499 lines)
     - `PartitionFingerprint.java` (+24 — adds `EMPTY` constant)
     - `PartitionInfo.java` (+17 / -1 — package-private constructor + `forTesting` factory)
     - `PartitionState.java` (+14 / -7 — javadoc updates clarifying `absent` semantics)
   - `pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/scheduler/`
     - `MaterializedViewTaskScheduler.java` (+62 / -121 — removes inline `computeWindowFingerprint` and `persistPartitionStateChangeWithRetry`; adds `computeMaxSourceEndTimeMs`)
     - `MaterializedViewTaskUtils.java` (+55 — centralizes `computeWindowFingerprint`)
   - `pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/consistency/`
     - `MaterializedViewConsistencyManager.java` (+97 / -117 — replaces bespoke retry loop with manager routing)
   - `pinot-materialized-view/src/test/java/...`
     - `metadata/MaterializedViewPartitionManagerTest.java` (new, 477 lines)
     - `metadata/PartitionFingerprintTest.java` (+39 — EMPTY contract tests)
     - `scheduler/MaterializedViewTaskSchedulerTest.java` (+57 — `computeMaxSourceEndTimeMs` tests)
     - `consistency/MaterializedViewConsistencyManagerTest.java`, `rewrite/MaterializedViewQueryRewriteEngineTest.java` — `PartitionInfo` → `PartitionInfo.forTesting` switches
   - `pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/...`
     - `materializedview/MaterializedViewTaskExecutor.java` (+50 / -178 — replaces bespoke CAS loop with manager calls; removes inline `computeWindowFingerprint`)
   - Same test class: `MaterializedViewTaskExecutorTest.java` (+31 / -12 — references the moved helpers + adds an EMPTY-overlap byte-equality test)
   
   ### Risk classification
   
   - [x] Classified as low / medium / high
   - [x] Reason stated
   
   **Classification:** Medium
   
   **Reason:** The PR touches concurrency-sensitive ZK CAS code paths shared by three subsystems (executor, scheduler, consistency manager) and changes a public API (`PartitionInfo` constructor) in a way that's source-incompatible across module boundaries. It is contained to the `pinot-materialized-view` module + the MV minion-task plugin; ZK serialization is unchanged. The materialized-view feature itself is relatively new and gated by table config, so blast radius is bounded to MV-enabled tables. CI is green across binary compatibility, multi-stage query engine, integration, and unit test sets.
   
   ### Test gaps
   
   - [x] Changed behaviors/contracts checked for corresponding tests
   - [x] Test coverage of new/changed behavior verified
   
   **Gaps:**
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewPartitionManagerTest.java` (new). Every public method of the manager is tested with strict/lenient precondition variants, CAS retry classification (`testCasConflictRetriesUntilSuccess`), and budget exhaustion (`testRevertBudgetExhaustedThrowsRuntimeException`). Strong coverage.
   - Severity: observation. Confidence: medium. Evidence: `MaterializedViewTaskSchedulerTest` adds 5 `computeMaxSourceEndTimeMs` cases covering empty-list, mixed-negative, all-negative, and zero-endTime boundaries — directly exercising the watermark-cap bug fix.
   - Severity: concern. Confidence: medium. Evidence: `MaterializedViewTaskExecutorTest.testWindowFingerprintForEmptyOverlapMatchesEmptyConstant` pins the byte-equality contract for the DELETE backfill bug fix, but there is no end-to-end test exercising the full `STALE → DELETE → VALID-empty → backfill → STALE → OVERWRITE` lifecycle. The PR body cites "manual end-to-end validation against a local Pinot cluster", but no automated regression test covers the backfill-after-delete scenario specifically. A future integration test would close this gap. CI's integration test sets pass but do not appear to exercise this specific MV path.
   - Severity: observation. Confidence: medium. Evidence: scheduler.tryHandleStalePartition. There is no unit test that asserts the scheduler invokes `revertValid` on a false-positive STALE marking via the manager (rather than the old `persistPartitionStateChangeWithRetry` it replaced) — coverage of the revertValid wiring relies on the manager test alone. Minor gap; integration coverage is presumed.
   
   ---
   
   ## Stage 2: Deep Review
   
   ### Does the implementation address the stated problem?
   
   - [x] Problem statement identified from the PR description
   - [x] Implementation traced through the code to check whether it appears to address the stated problem
   - [x] No obvious gaps between what the PR claims to fix and what the code actually does
   
   **Findings:**
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewTaskScheduler.java:274-285`, `computeMaxSourceEndTimeMs:962-971`. The watermark over-advance fix is implemented as advertised — the scheduler now resolves the cutoff in two stages and clamps it at `max(source segment endTimeMs)`. The two-stage shape avoids a ZK list call when the rough cutoff already excludes any candidate bucket.
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewTaskExecutor.java:457-464` and `MaterializedViewPartitionManager.java:248-266`. The DELETE backfill fix routes through `clearValid`, which writes `VALID + PartitionFingerprint.EMPTY` instead of removing the entry. `PartitionFingerprint.EMPTY` (`PartitionFingerprint.java:61-62`) is constructed to be byte-identical to what `computeWindowFingerprint` produces on an empty overlap (verified by `testWindowFingerprintForEmptyOverlapMatchesEmptyConstant`).
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewPartitionManager.java`. The CAS engine consolidation is exhaustive — all six documented state-machine ops have a corresponding public method, every transition flows through `applyMutation`, and the executor / scheduler / consistency manager construct or hold a manager rather than rolling their own loops.
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewTaskUtils.java:210-229`. `computeWindowFingerprint` is the single source of truth — the scheduler and executor both invoke `MaterializedViewTaskUtils.computeWindowFingerprint` (no remaining duplicate copies). Javadoc captures the byte-equality contract.
   
   ### High-risk files _(medium)_
   
   - [x] Files ranked by logical risk
   - [x] Risk reasons stated per file
   
   **Findings:**
   1. `MaterializedViewPartitionManager.java` (new, 499 lines). Severity: observation. Confidence: high. Evidence: file. Center of the change. New CAS engine; if its retry classification or precondition checks are wrong, every state transition is affected. Tests are comprehensive; the architecture has clear separation between public DSL, CRUD primitives, and CAS engine; the `Mutator` interface's null-return-means-no-op contract is documented and exercised. Risk is contained by the test surface but the class is now the bottleneck for MV correctness.
   2. `MaterializedViewTaskExecutor.java`. Severity: observation. Confidence: medium. Evidence: file. 178 lines of bespoke CAS loop removed, replaced by 50 lines of manager calls. The simplification is large enough that subtle pre-PR behavior could have shifted — but `validateSourceFingerprintAtCommit` is still called before `appendValid`/`refreshValid` (line 472), and the DELETE branch's lack of fingerprint validation is documented (no source-side state to validate, since `clearValid` writes a fixed EMPTY constant).
   3. `MaterializedViewTaskScheduler.java`. Severity: observation. Confidence: medium. Evidence: file. The watermark-cap logic is new and has direct impact on the broker's `now - watermarkMs <= stalenessThresholdMs` freshness check. The two-stage cutoff resolution adds a code path; the unit tests cover the helper function but not the integrated `generateTasks` path.
   4. `MaterializedViewConsistencyManager.java`. Severity: observation. Confidence: medium. Evidence: file. The bespoke retry loop is replaced with the manager. Two ZK reads per flush instead of one — explicitly justified in the new javadoc. The `_partitionManager` is wired via a `volatile` field with a deref-on-call lambda so a later `setClusterConfigReader` propagates correctly (`init:139-147`). Sequencing is correct: `setClusterConfigReader` is called AFTER `init()` in `BaseControllerStarter.java:621-636`.
   
   ### Center of gravity _(medium)_
   
   - [x] 1-3 key files or decisions identified
   
   **Files:**
   1. `pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/metadata/MaterializedViewPartitionManager.java` — the new manager that owns the state machine.
   2. `pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/metadata/PartitionFingerprint.java` — defines `EMPTY`; its byte-equality with `computeWindowFingerprint`-of-empty is the contract that the DELETE backfill bug fix rests on.
   3. `pinot-materialized-view/src/main/java/org/apache/pinot/materializedview/scheduler/MaterializedViewTaskScheduler.java` — hosts the watermark-cap fix and the shared `computeWindowFingerprint` (via `MaterializedViewTaskUtils`).
   
   **Why these are central:**
   Every consumer (executor postProcess, scheduler revert, consistency manager flush) routes mutations through the manager; every fingerprint computation routes through `MaterializedViewTaskUtils`; the broker's freshness contract depends on the scheduler's watermark advancement.
   
   ### Invariants — targeted _(medium)_
   
   - [x] Authorization — N/A. No new endpoints or permission checks introduced; the PR is internal task plumbing.
   - [x] Idempotency — `revertValid`, `markStale`, `deletePartition` are lenient (no-op if precondition violated). `appendValid` / `refreshValid` / `clearValid` are strict (throw on precondition violation). Helix will retry strict-fail tasks; that retry contract is documented and matches prior behavior.
   - [x] Backward compatibility — APIs — Public REST surface unchanged.
   - [x] Backward compatibility — Schemas — ZNRecord format unchanged. `PartitionFingerprint.EMPTY` is byte-identical to the existing APPEND-empty fingerprint (`(0, farmHash64(""))`), explicitly verified in `testWindowFingerprintForEmptyOverlapMatchesEmptyConstant`. Existing ZNodes are read-compatible.
   - [x] Backward compatibility — SPI — `PartitionInfo` constructor went from `public` to package-private. Cross-package production callers would break. Audited — no production caller of `new PartitionInfo(...)` exists outside `pinot.materializedview.metadata` (verified via repo grep). All cross-package callers are tests, which now use `PartitionInfo.forTesting`.
   - [x] Backward compatibility — Configuration — Existing cluster config key `pinot.materialized.view.executor.runtime.update.max.attempts` (`CLUSTER_CONFIG_KEY_MAX_RUNTIME_UPDATE_ATTEMPTS`) is now also consulted by the consistency manager and scheduler revert paths (via the manager's `criticalMaxAttempts()`). The default value is unchanged (128). Operators who tuned this for the executor only will see the new value applied to all three sites — this is the intended unification per the PR design.
   - [x] Ordering guarantees — Watermark advancement is now bundled with the bucket insert in a single CAS write (`appendValid`). Prior code paths could split these into two ZK writes in some refactor scenarios; the manager closes that window.
   - [x] Null/empty handling — Empty-window fingerprint canonicalized via `PartitionFingerprint.EMPTY`. `markStale` no-ops on empty collection (`MaterializedViewPartitionManager.java:329-331`). Empty-source-segments list returns `Long.MIN_VALUE` from `computeMaxSourceEndTimeMs` and caller falls back to `roughCutoffMs`.
   - [x] Retry safety — CAS retry classification is documented (`CasConflictException` and other `ZkException` retry; `IllegalStateException`/`IllegalArgumentException` propagate). `Mutator` may return null to signal no-op without ZK write, used by lenient ops.
   - [x] Observability — All ops emit INFO-level log lines including table, bucket, and transition. Replaces and consolidates ad-hoc log lines that lived in three sites. No metric changes.
   - [x] Resource cleanup — N/A. No new resources (files / connections / locks) introduced.
   - [x] Migration reversibility — No migration. The PR is wire-compatible. A rollback would re-introduce the bespoke loops; partitions persisted as `VALID + EMPTY` would deserialize without issue under the old code (the old code reads `VALID + arbitraryFP`).
   - [x] Performance in hot paths — The consistency manager's flush now does two ZK reads (one for `enumerateCandidateBuckets`, one inside `applyMutation`). The added cost is one ZK GET per flush per affected MV; debounce window (5 s default) bounds the rate. The scheduler's APPEND path adds one `getSegmentsZKMetadata` call only when at least one bucket can fit before the rough cutoff (Stage 1 short-circuit). Both costs are bounded and explicitly justified in code comments.
   - [x] API design — The manager DSL has one method per state-machine op, with documented strict / lenient preconditions and a per-method retry profile. Error responses (RuntimeException after budget exhaust, IllegalStateException on precondition violation) carry the offending table name and bucket. Names (`appendValid`, `refreshValid`, `clearValid`, `revertValid`, `markStale`, `deletePartition`) are reasonable. `deletePartition` is documented as "reserved for future manual admin commands; not currently invoked from any automatic path" — good defensive surface for upcoming REFRESH MV work.
   
   **Findings:**
   - Severity: concern. Confidence: medium. Evidence: `MaterializedViewTaskExecutor.java:457-464` and `MaterializedViewTaskExecutor.java:469-484`. The DELETE branch in `postProcess` skips `validateSourceFingerprintAtCommit` — by design, since the source was empty at scheduler dispatch time. A race window exists: between scheduler dispatch (`segmentCount == 0`) and executor commit (`clearValid`), a backfill could land in the same bucket. The MV is then persisted as `VALID + EMPTY` while the source has data. The consistency manager will mark the bucket STALE when the segment-add event flows through, so the system self-heals on the next debounce cycle. The window is bounded by debounce + scheduler periodicity but is real. A possible enhancement: validate source fingerprint inside `clearValid` (treat a non-empty source as a precondition violation that propagates back as IllegalStateException, which forces a Helix retry). The PR description acknowledges this design choice indirectly via
  the `VALID-empty` semantics. Leaving as concern, not blocker — self-healing path exists.
   - Severity: observation. Confidence: medium. Evidence: `MaterializedViewPartitionManager.java:281-301` (`revertValid`). The revert profile uses `REVERT_MAX_ATTEMPTS = 8` and is invoked via `new MaterializedViewPartitionManager(...)` for each revert (`MaterializedViewTaskScheduler.java:390-391`). The per-call allocation is cheap; the consistency manager and executor reuse a single instance. The asymmetry is acceptable but worth flagging in case a future refactor wants to hold a single scheduler-side manager too.
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewPartitionManager.java:316`. The `markStale` single-bucket overload uses `java.util.Collections.singletonList(...)` (fully qualified). This works but stylistically inconsistent with the file's other unqualified `Collections.emptyList()` usages in callers. Minor.
   
   ### Test adequacy _(medium)_
   
   - [x] Tests prove intended behavior (not just that code runs)
   - [x] Failure modes covered
   - [x] Boundary conditions covered
   - [x] Not over-coupled to implementation details
   
   **Findings:**
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewPartitionManagerTest.java`. 24 tests cover happy path, strict-precondition-throws, lenient-no-op, CAS retry-until-success, CAS budget exhaustion, and null-arg rejection. The retry test simulates a concurrent writer by bumping the stored version inside an `Answer` lambda — realistic.
   - Severity: observation. Confidence: medium. Evidence: `MaterializedViewConsistencyManagerTest.testCasConflictTriggersRetry`. Verifies the new manager routing still retries on CAS conflict. The test re-uses default Stat versions (0) for all reads, which works because `set` is stubbed to `(false, true)` regardless of version — slightly loose, but functional.
   - Severity: observation. Confidence: medium. Evidence: `MaterializedViewTaskExecutorTest.testWindowFingerprintForEmptyOverlapMatchesEmptyConstant`. Pins the byte-equality contract that the DELETE bug fix relies on. Good regression coverage.
   - Severity: observation. Confidence: low. The new tests don't exercise the concurrent interleaving of `markStale` and `appendValid` against the same map (e.g. `markStale` enumerating bucket X while `appendValid` adds bucket X+1). The manager's CAS engine should serialize these by version, but no test asserts the interleaving. Coverage gap that an integration test would close.
   
   ### Collateral damage _(medium)_
   
   - [x] No copy-paste duplication introduced
   - [x] No partial renames (all references updated)
   - [x] No dead code left behind
   - [x] Docs and config match code changes
   - [x] No one-off exceptions added to shared abstractions
   
   **Findings:**
   - Severity: observation. Confidence: high. Evidence: diff. The two inline `computeWindowFingerprint` copies (one in scheduler, one in executor) are both fully removed and replaced with calls to `MaterializedViewTaskUtils.computeWindowFingerprint`. No leftover dead code.
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewTaskExecutorTest.java` and `MaterializedViewConsistencyManagerTest.java`. Cross-package test usages of `new PartitionInfo(...)` correctly switched to `PartitionInfo.forTesting(...)`. Same-package test (`MaterializedViewPartitionManagerTest.java`) keeps direct constructor usage — fine, since the constructor remains package-accessible.
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewTaskExecutor.java`. The old `computeContiguousUpperMs` and `computeWindowFingerprint` @VisibleForTesting stubs in the executor are removed; the test file now references `MaterializedViewTaskUtils.computeContiguousUpperMs` / `MaterializedViewTaskUtils.computeWindowFingerprint` directly. No partial rename.
   - Severity: observation. Confidence: high. Evidence: `PartitionState.java`. Javadoc updated to reflect the new VALID-empty shape and the explicit "absent is cold-start only" semantic. Matches the code change.
   - Severity: observation. Confidence: medium. Evidence: `MaterializedViewPartitionManager.java:236-242` ("Design context (open question)" javadoc on `clearValid`). The PR author flags the `VALID-empty` design as "under review" with an alternative model documented. This is honest — but it means a future PR may revise this decision. Reasonable for now; worth tracking.
   
   ### Path-triggered review _(medium)_
   
   For each touched path, the inferred failure mode and what was checked.
   
   - [x] Touched paths listed
   - [x] Failure mode inferred per path (what can break when this area changes?)
   - [x] Reviewed for inferred risks
   - [x] Boundary areas checked first (auth, APIs, persistence, queues, shared core, infra, multi-tenancy, manifests)
   
   **Touched paths and inferred risks:**
   
   | Path | Inferred failure mode | What was checked | Findings |
   |---|---|---|---|
   | `pinot-materialized-view/.../metadata/MaterializedViewPartitionManager.java` (new) | Concurrency: CAS race, lost updates, retry-budget thrashing under contention | Reviewed retry classification (`CasConflictException` → retry; `ZkException` → retry; `IllegalStateException`/`IllegalArgumentException` → propagate). Reviewed backoff envelopes (critical: 128 × 50–200 ms ≈ 25 s; revert: 8 × 5–25 ms ≈ 200 ms). Mutator returning null short-circuits without CAS write — lenient ops. | No issues |
   | `pinot-materialized-view/.../metadata/PartitionInfo.java` (constructor package-private) | API/SPI: source-incompat for cross-package callers | Searched repo for `new PartitionInfo(` outside the `metadata` package. Only same-package test (`MaterializedViewPartitionManagerTest`) uses constructor; all cross-package tests switched to `forTesting`. No production caller affected. | No issues |
   | `pinot-materialized-view/.../metadata/PartitionFingerprint.java` (`EMPTY` constant) | ZK byte-equality drift across rolling upgrades | Verified `EMPTY = new PartitionFingerprint(0, farmHash64(""))` exactly matches what `computeWindowFingerprint` produces for an empty segment list. Test `testWindowFingerprintForEmptyOverlapMatchesEmptyConstant` pins the contract. | No issues |
   | `pinot-materialized-view/.../scheduler/MaterializedViewTaskScheduler.java` (watermark cap) | Broker freshness check correctness; APPEND scheduling drift | `computeMaxSourceEndTimeMs` returns `Long.MIN_VALUE` for empty list (caller falls back to `roughCutoffMs`). `endMs >= 0` filter mirrors the cold-start scan. Two-stage cutoff short-circuits the ZK list when no bucket can fit. | No issues |
   | `pinot-materialized-view/.../consistency/MaterializedViewConsistencyManager.java` (manager routing) | Cluster-config reader timing; CAS retry semantic preservation | Verified the `volatile MaterializedViewPartitionManager _partitionManager` is constructed at `init()` time with a closed-over `_clusterConfigReader` field deref (so a later `setClusterConfigReader` propagates). Verified `BaseControllerStarter` calls `setClusterConfigReader` AFTER `init()`. Verified `markStaleThroughManager` swallows manager `RuntimeException` with ERROR log — same behavior as the prior loop's exhausted-retry path. | No issues |
   | `pinot-plugins/.../materializedview/MaterializedViewTaskExecutor.java` (postProcess via manager) | Task-commit correctness; fingerprint validation timing | `validateSourceFingerprintAtCommit` is still called before `appendValid` and `refreshValid` (line 472). The DELETE branch deliberately skips fingerprint validation — see Concern 1 above. The per-call `new MaterializedViewPartitionManager(...)` allocation is acceptable. | One concern flagged on DELETE branch race (see "Invariants — targeted") |
   
   ---
   
   ## Verdict
   
   **Risk:** medium
   **Verdict:** approved
   **Verdict confidence:** medium
   
   ### Confidence reasoning
   
   The PR is well-scoped, internally consistent, and addresses the two stated correctness bugs with documented byte-equality guarantees for rolling upgrades. The new manager has comprehensive unit-test coverage (24 tests across happy/strict/lenient/CAS-retry/budget-exhaust paths) and the centralization of three retry loops into one is a genuine maintainability win. CI is green across all check categories including binary compatibility and integration tests.
   
   Confidence is medium (not high) because:
   - The DELETE branch's lack of commit-time fingerprint validation creates a race window where a `VALID-empty` partition can be persisted while the source has backfilled data. Self-healing exists (consistency manager will re-mark STALE), but the window deserves a regression test or an explicit precondition check in `clearValid`.
   - No integration-level test exercises the full `STALE → DELETE → VALID-empty → backfill → STALE → OVERWRITE` lifecycle. The PR cites manual validation against a local cluster.
   - The cluster-config key `pinot.materialized.view.executor.runtime.update.max.attempts` semantics quietly broaden from "executor only" to "executor + consistency manager + scheduler revert". Operators who tuned this key may see different behavior. The default is unchanged so this is a soft regression only for those who explicitly overrode the value.
   
   These are concerns, not blockers — none would justify holding the merge given the centralization win and the self-healing fallback behavior. The PR meaningfully reduces future surface area for state-machine drift and removes a class of real correctness bugs.
   
   ### Blockers
   None.
   
   ### Concerns
   - Severity: concern. Confidence: medium. Evidence: `MaterializedViewTaskExecutor.java:457-464`. DELETE branch skips `validateSourceFingerprintAtCommit`. Backfill landing between scheduler dispatch and executor commit results in a transient `VALID-empty` MV partition over a non-empty source. Self-heals on next consistency-manager cycle; bounded by debounce window. Recommend either (a) adding a regression test that simulates the race, or (b) adding a source-fingerprint precondition inside `clearValid` so a non-empty source forces Helix to retry the DELETE.
   - Severity: concern. Confidence: low. Evidence: PR scope. No integration test covers the full STALE → DELETE → VALID-empty → backfill → STALE → OVERWRITE lifecycle. PR cites manual validation. A future integration test would close the gap.
   - Severity: concern. Confidence: medium. Evidence: `CommonConstants.MaterializedViewTask.CLUSTER_CONFIG_KEY_MAX_RUNTIME_UPDATE_ATTEMPTS` semantics. Operators who previously tuned this key for the executor only will now see it applied to consistency-manager and scheduler-revert paths too. Default unchanged, so impact is limited to operators who overrode the value. Worth noting in release notes / upgrade guide.
   
   ### Observations
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewPartitionManager.java:39-105`. The class-level javadoc is exceptionally thorough — documents the why, the architecture (three layers), the retry profiles with calibration math, the concurrency model, and the scope discipline. This is the level of documentation other shared-state subsystems should aspire to.
   - Severity: observation. Confidence: medium. Evidence: `MaterializedViewPartitionManager.java:236-242`. `clearValid` javadoc flags "Design context (open question)" — the `VALID-empty` shape is "under review" with an alternative model documented. Acknowledging the design tradeoff in code is honest and helps future contributors. Worth tracking through whatever issue replaces this.
   - Severity: observation. Confidence: high. Evidence: `MaterializedViewPartitionManager.java:316`. `markStale(table, bucket)` single-bucket overload uses `java.util.Collections.singletonList(...)` (fully qualified) inconsistently with other unqualified `Collections` usages. Cosmetic.
   - Severity: observation. Confidence: medium. Evidence: scheduler invokes `revertValid` via `new MaterializedViewPartitionManager(...)` per call. Cheap per-call allocation, but inconsistent with the consistency manager and executor that hold or reuse instances. A future refactor could hold a scheduler-side manager too.
   - Severity: observation. Confidence: medium. Evidence: PR description. The two bug fixes (watermark over-advance, backfill-after-delete) and the refactor (single CAS engine + central fingerprint) are bundled in one PR. Each is individually mergeable; bundling is reasonable here because the bug fixes naturally live in the same code paths the refactor centralizes, but a reviewer auditing post-merge would benefit from the per-fix breakdown the PR body provides.
   
   ### Areas recommended for human focus
   
   1. **DELETE-branch race window** (`MaterializedViewTaskExecutor.executeDeleteTask` + `MaterializedViewPartitionManager.clearValid`). Reviewer should confirm the self-healing behavior via consistency-manager re-mark is acceptable, or push back on adding source-fingerprint validation inside `clearValid`.
   2. **Cluster-config key semantic broadening** (`CLUSTER_CONFIG_KEY_MAX_RUNTIME_UPDATE_ATTEMPTS`). Reviewer should confirm operators won't be surprised; consider release-note language.
   3. **Test gap for full backfill lifecycle**. Reviewer should confirm the manual-cluster validation is sufficient for the initial release of this hardening pass, or request a regression integration test for the next iteration.
   4. **`MaterializedViewPartitionManager.clearValid` "open question" javadoc**. Reviewer should confirm the `VALID-empty` semantic is acceptable as a temporary contract or request a follow-up issue tracking the alternative design.
   5. **Concurrency interleaving coverage**. Manager tests cover single-thread retry classification but not the concurrent `markStale` + `appendValid` race. Reviewer should confirm CAS-by-version is sufficient or ask for an interleaving test.
   
   ### Status
   
   - [x] All applicable Stage 1 boxes checked
   - [x] All applicable Stage 2 boxes checked (or N/A for skipped tiers)
   - [x] Verdict committed
   - [x] Verdict confidence assigned with reasoning
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org