PHOENIX-7874 Eliminate replicating index mutations; regenerate on standby

phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java

@@ -33,6 +33,7 @@
 import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.ReplayWrite;
 import org.apache.phoenix.hbase.index.covered.IndexMetaData;
 import org.apache.phoenix.hbase.index.covered.data.CachedLocalTable;
+import org.apache.phoenix.hbase.index.covered.data.LocalHBaseState;
 import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
 import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
 import org.apache.phoenix.index.PhoenixIndexMetaData;
@@ -92,14 +93,28 @@ public void getIndexUpdates(
     ListMultimap<HTableInterfaceReference, Pair<Mutation, byte[]>> indexUpdates,
     MiniBatchOperationInProgress<Mutation> miniBatchOp, Collection<? extends Mutation> mutations,
     IndexMetaData indexMetaData) throws Throwable {
-    // notify the delegate that we have started processing a batch
-    this.delegate.batchStarted(miniBatchOp, indexMetaData);
     CachedLocalTable cachedLocalTable = CachedLocalTable.build(mutations,
       (PhoenixIndexMetaData) indexMetaData, this.regionCoprocessorEnvironment.getRegion());
+    getIndexUpdates(indexUpdates, miniBatchOp, mutations, indexMetaData, cachedLocalTable);
+  }
+
+  /**
+   * Variant of
+   * {@link #getIndexUpdates(ListMultimap, MiniBatchOperationInProgress, Collection, IndexMetaData)}
+   * that uses a caller-supplied {@link LocalHBaseState} for prior-row-state lookups instead of
+   * building a region-scanning {@link CachedLocalTable}. Used on the standby replay path, where
+   * prior state comes from the per-batch pre-image rather than the (not-yet-written) region.
+   */
+  public void getIndexUpdates(
+    ListMultimap<HTableInterfaceReference, Pair<Mutation, byte[]>> indexUpdates,
+    MiniBatchOperationInProgress<Mutation> miniBatchOp, Collection<? extends Mutation> mutations,
+    IndexMetaData indexMetaData, LocalHBaseState localHBaseState) throws Throwable {
+    // notify the delegate that we have started processing a batch
+    this.delegate.batchStarted(miniBatchOp, indexMetaData);
     // Avoid the Object overhead of the executor when it's not actually parallelizing anything.
     for (Mutation m : mutations) {
       Collection<Pair<Mutation, byte[]>> updates =
-        delegate.getIndexUpdate(m, indexMetaData, cachedLocalTable);
+        delegate.getIndexUpdate(m, indexMetaData, localHBaseState);
       for (Pair<Mutation, byte[]> update : updates) {
         indexUpdates.put(new HTableInterfaceReference(new ImmutableBytesPtr(update.getSecond())),
           new Pair<>(update.getFirst(), m.getRow()));

phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/covered/data/PreImageLocalTable.java

@@ -0,0 +1,97 @@
+/*
+ * 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.phoenix.hbase.index.covered.data;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.phoenix.hbase.index.IndexRegionObserver.RowTsKey;
+import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.util.IndexUtil;
+
+import org.apache.phoenix.thirdparty.com.google.common.base.Preconditions;
+
+/**
+ * Standby-side {@link LocalHBaseState} that serves the local-index builder's prior-row-state lookup
+ * from the per-batch pre-image the active shipped, instead of scanning the data-table region (which
+ * {@link CachedLocalTable} does on the active).
+ * <p>
+ * On the standby the replication reader can concatenate several active-side batches into one replay
+ * batch, so a row recurs at several active timestamps and the region does not yet hold the
+ * intermediate state each batch saw. The active shipped one pre-image per row per batch, which is
+ * exactly the prior row state that batch built its local index against. We therefore key the lookup
+ * by {@code (row, ts)} — recovered from the mutation's row and max cell timestamp — and return that
+ * group's own pre-image cells. Each group's build is thus an independent reproduction of the
+ * corresponding active {@code preBatchMutate}; no chaining across groups is needed.
+ * <p>
+ * A {@code null} cell list is the "active saw an empty row" sentinel and is the documented
+ * {@link LocalHBaseState#getCurrentRowState} return for "no prior row". A key that is absent
+ * entirely (never populated) is a contract violation and throws, so the sentinel stays unambiguous.
+ * <p>
+ * Not thread-safe, and does not need to be: an instance is built per replay batch and consumed by
+ * that batch's single-threaded local-index build. The backing map is populated once at construction
+ * and never mutated afterward.
+ */
+public class PreImageLocalTable implements LocalHBaseState {
+
+  /** Distinct marker for "key absent", so a stored null (empty-row sentinel) is not mistaken. */
+  private static final List<Cell> ABSENT = Collections.emptyList();
+
+  private final String dataTableName;
+  private final Map<RowTsKey, List<Cell>> preImageCellsByRowTs;
+
+  public PreImageLocalTable(String dataTableName, Map<RowTsKey, List<Cell>> preImageCellsByRowTs) {
+    this.dataTableName = dataTableName;
+    this.preImageCellsByRowTs =
+      Preconditions.checkNotNull(preImageCellsByRowTs, "preImageCellsByRowTs must not be null");
+  }
+
+  /**
+   * {@inheritDoc}
+   * <p>
+   * {@code toCover} and {@code ignoreNewerMutations} are ignored: we already hold the exact
+   * per-group prior-row snapshot the active built against, so there is nothing to narrow by column
+   * or filter by timestamp. The {@code (row, ts)} key is derived from the mutation the same way
+   * {@code IndexRegionObserver.buildReplicatedRowGroups} keys the map that populated this table.
+   */
+  @Override
+  public List<Cell> getCurrentRowState(Mutation mutation,
+    Collection<? extends ColumnReference> toCover, boolean ignoreNewerMutations)
+    throws IOException {
+    RowTsKey key =
+      new RowTsKey(new ImmutableBytesPtr(mutation.getRow()), IndexUtil.getMaxTimestamp(mutation));
+    // A stored null is the "active saw an empty row" sentinel, so a null value cannot be told from
+    // an absent key via get() alone; look up once against a distinct ABSENT marker instead. A true
+    // miss means the (row, ts) derivation drifted from the populating side
+    // (buildReplayLocalIndexInputs); fail loud rather than return null, which the builder would
+    // read
+    // as the empty-row sentinel and silently regenerate the index against an empty prior state.
+    List<Cell> cells = preImageCellsByRowTs.getOrDefault(key, ABSENT);
+    if (cells == ABSENT) {
+      throw new DoNotRetryIOException("No pre-image for replayed local-index row on table "
+        + dataTableName + "; (row, ts) key not populated: " + key);
+    }
+    return cells;
+  }
+}

phoenix-core-server/src/main/java/org/apache/phoenix/replication/MutationCellGrouper.java

@@ -19,12 +19,19 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.phoenix.hbase.index.IndexRegionObserver;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.index.PhoenixIndexCodec;
 
 /**
  * Groups a flat cell stream into Put/Delete mutations, mirroring the algorithm HBase's
@@ -37,6 +44,9 @@
  * recurs non-consecutively yields a separate mutation per run), not correctness -- cell order is
  * preserved, so replaying the resulting mutations in order reproduces the effect of applying the
  * input cells in order.
+ * <p>
+ * Stateless and thread-safe: a non-instantiable holder of static methods that operate solely on
+ * their arguments.
  */
 public final class MutationCellGrouper {
 
@@ -48,6 +58,110 @@ private static boolean isNewRowOrType(Cell previousCell, Cell cell) {
       || !CellUtil.matchingRows(previousCell, cell);
   }
 
+  /**
+   * Flatten a mutation's family cell map into a single ordered cell list, preserving family
+   * iteration order (typically TreeMap-ordered).
+   */
+  public static List<Cell> flattenCells(Mutation mutation) {
+    List<Cell> body = new ArrayList<>();
+    for (List<Cell> familyCells : mutation.getFamilyCellMap().values()) {
+      body.addAll(familyCells);
+    }
+    return body;
+  }
+
+  /**
+   * Extract the well-known replication attributes
+   * ({@link ReplicationLogGroup#REPLICATION_ATTR_KEYS}) from the mutation, copied verbatim. Returns
+   * an empty (mutable) map if the mutation has no attributes or none match.
+   * <p>
+   * {@link PhoenixIndexCodec#INDEX_UUID} is deliberately NOT part of this envelope: whether the
+   * standby regenerates indexes is decided by the active from its resolved index maintainers, not
+   * from the client-set UUID attribute. The active stamps an empty UUID onto the returned map only
+   * for indexed tables (see {@code IndexRegionObserver}).
+   */
+  public static Map<String, byte[]> extractReplicationAttributes(Mutation mutation) {
+    Map<String, byte[]> envelope = new HashMap<>();
+    Map<String, byte[]> mutationAttrs = mutation.getAttributesMap();
+    if (mutationAttrs == null || mutationAttrs.isEmpty()) {
+      return envelope;
+    }
+    for (String key : ReplicationLogGroup.REPLICATION_ATTR_KEYS) {
+      byte[] v = mutationAttrs.get(key);
+      if (v != null) {
+        envelope.put(key, v);
+      }
+    }
+    return envelope;
+  }
+
+  /**
+   * Build the flat cell stream the active ships for a batch of replicated mutations: each
+   * mutation's data cells in family order, followed by one METAFAMILY pre-image cell carrying
+   * {@code preImage}'s row state. This is the inverse of {@link #reconstructMutations}; replaying
+   * its output through that method yields back the mutations with their
+   * {@link IndexRegionObserver#PRE_IMAGE} attribute attached. A {@code null} {@code preImage}
+   * encodes the empty-row sentinel. The pre-image cell is keyed by each mutation's own row,
+   * mirroring the active's per-row pre-image capture.
+   */
+  public static List<Cell> buildReplicatedCells(List<Mutation> mutations, Put preImage)
+    throws IOException {
+    List<Cell> cells = new ArrayList<>();
+    for (Mutation m : mutations) {
+      cells.addAll(flattenCells(m));
+      cells.add(IndexRegionObserver.buildPreImageCell(m.getRow(), preImage));
+    }
+    return cells;
+  }
+
+  /**
+   * Stamp an empty {@link PhoenixIndexCodec#INDEX_UUID} onto a replication attribute envelope. An
+   * empty UUID forces the standby down the server-PTable resolution path, which rebuilds index
+   * maintainers from the schema/table/tenant attributes in the same envelope. Callers apply this
+   * only for indexed tables (a non-indexed table needs no regeneration, and an empty UUID there
+   * would fail on the standby with INDEX_METADATA_NOT_FOUND).
+   */
+  public static void stampIndexAttribute(Map<String, byte[]> attrs) {
+    attrs.put(PhoenixIndexCodec.INDEX_UUID, HConstants.EMPTY_BYTE_ARRAY);
+  }
+
+  /**
+   * Walk the record body, peeling off METAFAMILY pre-image cells (one per row) into a row-keyed
+   * bucket and grouping the remaining data cells into Put/Delete mutations. Each result mutation is
+   * stamped with the replication attributes and the generic
+   * {@link IndexRegionObserver#REPLICATED_MUTATION} marker. When a pre-image entry exists for its
+   * row, the pre-image bytes are also attached as {@link IndexRegionObserver#PRE_IMAGE}.
+   */
+  public static List<Mutation> reconstructMutations(Iterable<Cell> cells,
+    Map<String, byte[]> replicationAttrs) throws IOException {
+    Map<ImmutableBytesPtr, byte[]> preImages = new HashMap<>();
+    List<Cell> dataCells = new ArrayList<>();
+    for (Cell c : cells) {
+      if (
+        CellUtil.matchingFamily(c, WALEdit.METAFAMILY)
+          && CellUtil.matchingQualifier(c, IndexRegionObserver.PRE_IMAGE_WAL_QUALIFIER)
+      ) {
+        preImages.put(new ImmutableBytesPtr(CellUtil.cloneRow(c)), CellUtil.cloneValue(c));
+      } else {
+        dataCells.add(c);
+      }
+    }
+    List<Mutation> mutations = splitCellsIntoMutations(dataCells);
+    for (Mutation m : mutations) {
+      if (replicationAttrs != null) {
+        for (Map.Entry<String, byte[]> e : replicationAttrs.entrySet()) {
+          m.setAttribute(e.getKey(), e.getValue());
+        }
+      }
+      m.setAttribute(IndexRegionObserver.REPLICATED_MUTATION, HConstants.EMPTY_BYTE_ARRAY);
+      byte[] preImageBytes = preImages.get(new ImmutableBytesPtr(m.getRow()));
+      if (preImageBytes != null) {
+        m.setAttribute(IndexRegionObserver.PRE_IMAGE, preImageBytes);
+      }
+    }
+    return mutations;
+  }
+
   /** Group a cell stream into Put/Delete mutations using the row+type boundary algorithm. */
   public static List<Mutation> splitCellsIntoMutations(Iterable<Cell> cells) throws IOException {
     List<Mutation> result = new ArrayList<>();

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java

@@ -310,7 +310,7 @@ private void replayCurrentBatch() throws IOException {
     LOG.info("Replaying {} unsynced records into new writer {}", currentBatch.size(),
       currentWriter);
     for (Record r : currentBatch) {
-      currentWriter.append(r.tableName, r.commitId, r.cells);
+      currentWriter.append(r.tableName, r.commitId, r.cells, r.attributes);
     }
   }
 
@@ -350,7 +350,7 @@ private void apply(Action action) throws IOException {
   protected void append(Record r) throws IOException {
     final boolean[] blockSynced = { false };
     apply(writer -> {
-      blockSynced[0] = writer.append(r.tableName, r.commitId, r.cells);
+      blockSynced[0] = writer.append(r.tableName, r.commitId, r.cells, r.attributes);
     });
     // Add to current batch only after we succeed at appending
     currentBatch.add(r);