You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by tk...@apache.org on 2026/09/16 23:30:26 UTC
(phoenix) 02/02: PHOENIX-7874 Addendum Skip redundant getMaxTimestamp scans on the active index write path
This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
commit 94fed195f56f30abe076a42b5b773d3eecdb5988
Author: tkhurana <kh...@gmail.com>
AuthorDate: Tue Sep 15 14:20:36 2026 -0700
PHOENIX-7874 Addendum Skip redundant getMaxTimestamp scans on the active index write path
On the active path every generated index-mutation cell carries the batch-wide
timestamp (set by setTimestamps), so re-scanning each mutation's cells with the
O(cells) IndexUtil.getMaxTimestamp on every index-prepare step was wasted work.
Introduce indexMutationTimestamp(context, m): use context.batchTimestamp on the
active path, fall back to getMaxTimestamp only on the standby replay path (which
regenerates index updates per (row, ts) group, where the mutation's own timestamp
is authoritative).
Make context.batchTimestamp the single source of truth for the index-prepare
chain: set it once where it is computed and drop the now-redundant batchTimestamp
parameter threaded through prepareAndCommitGlobalIndexUpdates,
preparePreIndexMutations, and prepareIndexMutations. Also drop the never-used
timestamp parameter from prepareDataRowStates/applyPendingPutMutations.
Standby path is unchanged (byte-for-byte). Validated by GlobalIndexCheckerIT,
GlobalIndexCheckerEventualGenerateIT, UncoveredGlobalMutableNonTxIndexIT, and
ReplicationLogGroupIT.
---
.../phoenix/hbase/index/IndexRegionObserver.java | 162 ++++++++++-----------
.../phoenix/replication/MutationCellGrouper.java | 30 +++-
.../phoenix/replication/ReplicationLogGroupIT.java | 20 ++-
.../reader/ReplicationLogProcessorTestIT.java | 12 +-
4 files changed, 122 insertions(+), 102 deletions(-)
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
index dd887cc73f..3cc6310196 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -134,7 +134,6 @@ import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.PhoenixIndexBuilderHelper;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.index.PhoenixIndexMetaData;
-import org.apache.phoenix.index.PhoenixIndexMetaDataBuilder;
import org.apache.phoenix.jdbc.HAGroupStoreManager;
import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
@@ -442,6 +441,11 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
// The collection of candidate index mutations that will be applied after the data table
// mutations.
private ListMultimap<HTableInterfaceReference, Pair<Mutation, byte[]>> indexUpdates;
+ // The batch-wide timestamp stamped on every data/index mutation on the active path (see
+ // setTimestamps). Lets the index-prepare steps timestamp index mutations without re-scanning
+ // each mutation's cells. Unused on the standby replay path, which derives per-(row, ts)
+ // timestamps from each mutation directly.
+ private long batchTimestamp;
// Map of (data table row key, group ts) to IndexMutations bytes containing pre-index mutations
// for eventually consistent indexes (UNVERIFIED Puts only, no Deletes). Keyed by (row, ts) so
// multiple per-row entries from the standby's per-(row, ts) grouping don't collide. On the
@@ -1053,20 +1057,12 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
if (replicable.isEmpty()) {
return;
}
- Map<String, byte[]> replicationAttrs = new HashMap<>();
- for (String attrKey : ReplicationLogGroup.REPLICATION_ATTR_KEYS) {
- byte[] val = walKeyAttrs.get(attrKey);
- if (val != null) {
- replicationAttrs.put(attrKey, val);
- }
- }
// INDEX_UUID rides the WAL key only when appendReplicationAttributesToWALKey stamped it (i.e.
- // the batch's table was indexed, gated on hasIndex()); copy it through verbatim so this path
- // follows the same server-PTable resolution the synchronous path triggers.
- byte[] indexUuid = walKeyAttrs.get(PhoenixIndexCodec.INDEX_UUID);
- if (indexUuid != null) {
- replicationAttrs.put(PhoenixIndexCodec.INDEX_UUID, indexUuid);
- }
+ // the batch's table was indexed, gated on hasIndex()); its presence drives the same empty-UUID
+ // stamp here, so this path follows the same server-PTable resolution the synchronous path does.
+ // The stamped value is always empty, so re-stamping empty is equivalent to copying it through.
+ Map<String, byte[]> replicationAttrs = MutationCellGrouper.buildReplicationAttributes(
+ walKeyAttrs, walKeyAttrs.containsKey(PhoenixIndexCodec.INDEX_UUID));
logGroup.append(tableName, -1, replicable, replicationAttrs);
logGroup.sync();
}
@@ -1428,7 +1424,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
* called, the next row states is set to current row states.
*/
private void applyPendingPutMutations(MiniBatchOperationInProgress<Mutation> miniBatchOp,
- BatchMutateContext context, long now) throws IOException {
+ BatchMutateContext context) throws IOException {
for (Integer i = 0; i < miniBatchOp.size(); i++) {
if (isAtomicOperationComplete(miniBatchOp.getOperationStatus(i))) {
continue;
@@ -1467,12 +1463,12 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
* * Prepares next data row state
*/
private void prepareDataRowStates(ObserverContext<RegionCoprocessorEnvironment> c,
- MiniBatchOperationInProgress<Mutation> miniBatchOp, BatchMutateContext context, long now)
+ MiniBatchOperationInProgress<Mutation> miniBatchOp, BatchMutateContext context)
throws IOException {
if (context.rowsToLock.size() == 0) {
return;
}
- applyPendingPutMutations(miniBatchOp, context, now);
+ applyPendingPutMutations(miniBatchOp, context);
applyPendingDeleteMutations(miniBatchOp, context);
}
@@ -1733,13 +1729,17 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
}
/**
- * True when this batch will be shipped to the replication log: replication is on, not disabled
- * for testing, and an HA log group is present. Gates the active-side pre-image capture — the
- * pre-image exists only so the standby can regenerate its index, so capturing it on a
- * non-replicated batch would be wasted work (and an unnecessary region scan on the local path).
+ * True when this batch is replicated on an active, replication-configured cluster: replication is
+ * on and an HA log group is present. Gates the active-side pre-image capture — the pre-image
+ * exists only so the standby can regenerate its index, so capturing it on a non-replicated batch
+ * would be wasted work (and an unnecessary region scan on the local path) — and the WAL-key
+ * replication stamp. It intentionally does NOT gate on {@code ignoreSyncReplicationForTesting}:
+ * that flag suppresses only the synchronous ship (checked explicitly at the ship site) to
+ * simulate an RS crash after the WAL write but before the ship, leaving pre-image capture, the
+ * WAL stamp, and the replay path intact so crash-recovery re-ship is faithfully exercised.
*/
private boolean isReplicatedBatch(BatchMutateContext context) {
- return shouldReplicate && !ignoreSyncReplicationForTesting && context.logGroup.isPresent();
+ return shouldReplicate && context.logGroup.isPresent();
}
/**
@@ -2122,8 +2122,8 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
* Generate the index update for a data row from the mutation that are obtained by merging the
* previous data row state with the pending row mutation.
*/
- private void prepareIndexMutations(BatchMutateContext context, List<IndexMaintainer> maintainers,
- long batchTimestamp) throws IOException {
+ private void prepareIndexMutations(BatchMutateContext context, List<IndexMaintainer> maintainers)
+ throws IOException {
List<Pair<IndexMaintainer, HTableInterfaceReference>> indexTables =
buildIndexTablesList(maintainers);
for (Map.Entry<ImmutableBytesPtr, Pair<Put, Put>> entry : context.dataRowStates.entrySet()) {
@@ -2135,8 +2135,9 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
continue;
}
ListMultimap<HTableInterfaceReference, Mutation> idxUpdates = ArrayListMultimap.create();
- generateIndexMutationsForRow(rowKeyPtr, currentDataRowState, nextDataRowState, batchTimestamp,
- encodedRegionName, QueryConstants.UNVERIFIED_BYTES, indexTables, idxUpdates);
+ generateIndexMutationsForRow(rowKeyPtr, currentDataRowState, nextDataRowState,
+ context.batchTimestamp, encodedRegionName, QueryConstants.UNVERIFIED_BYTES, indexTables,
+ idxUpdates);
for (Map.Entry<HTableInterfaceReference, Mutation> idxUpdate : idxUpdates.entries()) {
context.indexUpdates.put(idxUpdate.getKey(),
new Pair<>(idxUpdate.getValue(), rowKeyPtr.get()));
@@ -2174,8 +2175,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
* row is deleted.
*/
private void preparePreIndexMutations(MiniBatchOperationInProgress<Mutation> miniBatchOp,
- BatchMutateContext context, long batchTimestamp, PhoenixIndexMetaData indexMetaData)
- throws Throwable {
+ BatchMutateContext context, PhoenixIndexMetaData indexMetaData) throws Throwable {
List<IndexMaintainer> maintainers = indexMetaData.getIndexMaintainers();
// get the current span, or just use a null-span to avoid a bunch of if statements
try (TraceScope scope = Trace.startSpan("Starting to build index updates")) {
@@ -2193,7 +2193,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
// pre-image — recovers the active-batch boundary the reader's coalescing can erase.
prepareReplicatedIndexMutations(miniBatchOp, context, maintainers);
} else {
- prepareIndexMutations(context, maintainers, batchTimestamp);
+ prepareIndexMutations(context, maintainers);
}
if (serializeCDCMutations) {
@@ -2217,7 +2217,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
List<Pair<Mutation, byte[]>> updates = context.indexUpdates.get(hTableInterfaceReference);
for (Pair<Mutation, byte[]> update : updates) {
Mutation m = update.getFirst();
- long ts = IndexUtil.getMaxTimestamp(m);
+ long ts = indexMutationTimestamp(context, m);
RowTsKey cdcKey = new RowTsKey(new ImmutableBytesPtr(update.getSecond()), ts);
if (m instanceof Put) {
if (indexMaintainer.isCDCIndex() && context.cdcPreMutationsBytes != null) {
@@ -2285,7 +2285,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
for (Pair<Mutation, byte[]> update : updates) {
Mutation m = update.getFirst();
byte[] dataRowKey = update.getSecond();
- long ts = IndexUtil.getMaxTimestamp(m);
+ long ts = indexMutationTimestamp(context, m);
RowTsKey key = new RowTsKey(new ImmutableBytesPtr(dataRowKey), ts);
IndexMutationsProtos.IndexMutations.Builder preBuilder =
preBuilderMap.computeIfAbsent(key, k -> IndexMutationsProtos.IndexMutations.newBuilder());
@@ -2373,6 +2373,17 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
return (PhoenixIndexMetaData) indexMetaData;
}
+ /**
+ * Timestamp to stamp/key a generated index mutation by. On the active path every index-mutation
+ * cell carries {@code batchTimestamp} (see {@link #setTimestamps}), so we use that directly and
+ * skip the O(cells) {@link IndexUtil#getMaxTimestamp} scan. The standby regenerates index updates
+ * per {@code (row, ts)} group, where the mutation's own timestamp is authoritative, so it must
+ * read it from the mutation.
+ */
+ private static long indexMutationTimestamp(BatchMutateContext context, Mutation m) {
+ return context.isReplication ? IndexUtil.getMaxTimestamp(m) : context.batchTimestamp;
+ }
+
private void preparePostIndexMutations(BatchMutateContext context,
PhoenixIndexMetaData indexMetaData) {
context.postIndexUpdates = ArrayListMultimap.<HTableInterfaceReference, Mutation> create();
@@ -2395,7 +2406,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
if (!indexMaintainer.isUncovered()) {
Put verifiedPut = new Put(m.getRow());
// Set the status of the index row to "verified"
- verifiedPut.addColumn(emptyCF, emptyCQ, IndexUtil.getMaxTimestamp(m),
+ verifiedPut.addColumn(emptyCF, emptyCQ, indexMutationTimestamp(context, m),
QueryConstants.VERIFIED_BYTES);
context.postIndexUpdates.put(hTableInterfaceReference, verifiedPut);
}
@@ -2416,7 +2427,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
for (Pair<Mutation, byte[]> update : updates) {
Mutation m = update.getFirst();
if (m instanceof Put) {
- long ts = IndexUtil.getMaxTimestamp(m);
+ long ts = indexMutationTimestamp(context, m);
RowTsKey cdcKey = new RowTsKey(new ImmutableBytesPtr(update.getSecond()), ts);
byte[] cdcMutationsBytes = context.cdcPostMutationsBytes.get(cdcKey);
if (cdcMutationsBytes != null) {
@@ -2707,21 +2718,20 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
}
TableName table = c.getEnvironment().getRegion().getRegionInfo().getTable();
- long batchTimestamp = getBatchTimestamp(context, table);
+ context.batchTimestamp = getBatchTimestamp(context, table);
// Update the timestamps of the data table mutations to prevent overlapping timestamps
// (which prevents index inconsistencies as this case is not handled).
- setTimestamps(miniBatchOp, builder, batchTimestamp, isStrictTTLEnabled(miniBatchOp));
+ setTimestamps(miniBatchOp, builder, context.batchTimestamp, isStrictTTLEnabled(miniBatchOp));
if (context.hasGlobalIndex || context.hasUncoveredIndex || context.hasTransform) {
// Prepare next data rows states for pending mutations (for global indexes).
- prepareDataRowStates(c, miniBatchOp, context, batchTimestamp);
+ prepareDataRowStates(c, miniBatchOp, context);
// dataRowStates is now populated; on a replicated batch write per-row pre-image cells to the
// WAL edit so both replication paths (replicateMutations and replicateEditOnWALRestore) ship
// them. Skip on a non-replicated batch — the pre-image would be unused work.
if (isReplicatedBatch(context)) {
capturePreImageCells(miniBatchOp, context);
}
- prepareAndCommitGlobalIndexUpdates(table, miniBatchOp, context, batchTimestamp,
- indexMetaData);
+ prepareAndCommitGlobalIndexUpdates(table, miniBatchOp, context, indexMetaData);
}
if (context.hasLocalIndex) {
// Group all the updates for a single row into a single update to be processed (for local
@@ -2763,9 +2773,7 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
PhoenixIndexMetaData indexMetaData) throws Throwable {
TableName table = c.getEnvironment().getRegion().getRegionInfo().getTable();
if (context.hasGlobalIndex || context.hasUncoveredIndex || context.hasTransform) {
- // batchTimestamp is unused on this path: prepareReplicatedIndexMutations derives each index
- // update's timestamp from its (row, ts) group, not from a batch-wide timestamp.
- prepareAndCommitGlobalIndexUpdates(table, miniBatchOp, context, 0, indexMetaData);
+ prepareAndCommitGlobalIndexUpdates(table, miniBatchOp, context, indexMetaData);
}
if (context.hasLocalIndex) {
// Group by (row, ts) so each replayed active-side batch's cells stay in their own uniform-ts
@@ -2792,10 +2800,10 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
*/
private void prepareAndCommitGlobalIndexUpdates(TableName table,
MiniBatchOperationInProgress<Mutation> miniBatchOp, BatchMutateContext context,
- long batchTimestamp, PhoenixIndexMetaData indexMetaData) throws Throwable {
+ PhoenixIndexMetaData indexMetaData) throws Throwable {
// early exit if it turns out we don't have any edits
long start = EnvironmentEdgeManager.currentTimeMillis();
- preparePreIndexMutations(miniBatchOp, context, batchTimestamp, indexMetaData);
+ preparePreIndexMutations(miniBatchOp, context, indexMetaData);
metricSource.updateIndexPrepareTime(dataTableName,
EnvironmentEdgeManager.currentTimeMillis() - start);
// Release the locks before making RPC calls for index updates
@@ -2862,16 +2870,11 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
@Override
public void preWALAppend(ObserverContext<RegionCoprocessorEnvironment> c, WALKey key,
WALEdit edit) {
+ BatchMutateContext context = getBatchMutateContext(c);
if (shouldWALAppend) {
- BatchMutateContext context = getBatchMutateContext(c);
appendMutationAttributesToWALKey(key, context);
}
-
- if (shouldReplicate) {
- BatchMutateContext context = getBatchMutateContext(c);
- appendHAGroupAttributeToWALKey(key, context);
- appendReplicationAttributesToWALKey(key, context);
- }
+ appendReplicationAttributesToWALKey(key, context);
}
private void appendMutationAttributesToWALKey(WALKey key,
@@ -2890,29 +2893,37 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
}
/**
- * Save the HA group name if present in the WAL key so that we can use it when restoring from the
- * WAL
+ * Stamp the WAL key with everything the WAL-restore re-ship path consumes: the HA group name it
+ * keys on ({@link #getHAGroupFromWALKey}) plus the replication attribute envelope
+ * (schema/table/tenant, and an empty INDEX_UUID for indexed tables) it reads back in
+ * {@link #replicateEditOnWALRestore}. Gated on {@link #isReplicatedBatch} — the same predicate
+ * that gates pre-image capture — so the WAL carries this stamp exactly when it carries the
+ * pre-images the re-ship needs. On a standby replay the log group is absent, so the standby's WAL
+ * never carries replication metadata it could never re-ship.
*/
- private void appendHAGroupAttributeToWALKey(WALKey key,
- IndexRegionObserver.BatchMutateContext context) {
- if (context != null && context.logGroup.isPresent()) {
- String haGroupName = context.logGroup.get().getHAGroupName();
- IndexRegionObserver.appendToWALKey(key,
- BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB, Bytes.toBytes(haGroupName));
- }
- }
-
private void appendReplicationAttributesToWALKey(WALKey key,
IndexRegionObserver.BatchMutateContext context) {
- if (context == null || context.getOriginalMutations().isEmpty()) {
+ if (context == null || !isReplicatedBatch(context)) {
return;
}
- Map<String, byte[]> replicationAttributes = buildReplicationAttributes(context);
- for (Map.Entry<String, byte[]> e : replicationAttributes.entrySet()) {
+ IndexRegionObserver.appendToWALKey(key, BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB,
+ Bytes.toBytes(context.logGroup.get().getHAGroupName()));
+ for (Map.Entry<String, byte[]> e : buildReplicationAttributes(context).entrySet()) {
IndexRegionObserver.appendToWALKey(key, e.getKey(), e.getValue());
}
}
+ /**
+ * Active-path replication envelope for a batch: the well-known metadata keys carried on the
+ * batch's first mutation, plus an empty INDEX_UUID when the table is indexed
+ * ({@link BatchMutateContext#hasIndex()}). Shared by the synchronous ship and the WAL-append
+ * stamp; the WAL-restore path builds the same envelope from the WAL key attributes instead.
+ */
+ private static Map<String, byte[]> buildReplicationAttributes(BatchMutateContext context) {
+ return MutationCellGrouper.buildReplicationAttributes(
+ context.getOriginalMutations().get(0).getAttributesMap(), context.hasIndex());
+ }
+
/**
* When this hook is called, all the rows in the batch context are locked if the batch of
* mutations is successful. Because the rows are locked, we can safely make updates to pending row
@@ -2975,7 +2986,9 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
// updates
CompletableFuture<Void> postIndexFuture =
CompletableFuture.runAsync(() -> doPost(c, context));
- if (isReplicatedBatch(context)) {
+ // ignoreSyncReplicationForTesting suppresses only this synchronous ship, simulating an RS
+ // crash after the WAL write but before the ship; the WAL-restore path re-ships on replay.
+ if (isReplicatedBatch(context) && !ignoreSyncReplicationForTesting) {
replicateMutations(context.logGroup.get(), miniBatchOp, context);
}
FutureUtils.get(postIndexFuture);
@@ -3612,23 +3625,4 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
}
}
- /**
- * Build the replication attribute envelope shipped with a batch: the well-known metadata keys
- * carried on the batch's mutations, plus an empty {@link PhoenixIndexCodec#INDEX_UUID} when (and
- * only when) the table carries an index. An empty UUID forces the standby down the server-PTable
- * resolution path (see {@link PhoenixIndexMetaDataBuilder}), which rebuilds index maintainers
- * from the schema/table/tenant attributes in this same envelope. It is stamped only for indexed
- * tables: a non-indexed table needs no regeneration, and an empty UUID there would push the
- * standby into the server-cache branch and fail with INDEX_METADATA_NOT_FOUND. The active's own
- * resolved index maintainers ({@link BatchMutateContext#hasIndex()}) are the source of truth, not
- * the client-set UUID attribute.
- */
- private static Map<String, byte[]> buildReplicationAttributes(BatchMutateContext context) {
- Map<String, byte[]> replicationAttributes =
- MutationCellGrouper.extractReplicationAttributes(context.getOriginalMutations().get(0));
- if (context.hasIndex()) {
- MutationCellGrouper.stampIndexAttribute(replicationAttributes);
- }
- return replicationAttributes;
- }
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/MutationCellGrouper.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/MutationCellGrouper.java
index 210e32e63d..08413811e3 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/MutationCellGrouper.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/MutationCellGrouper.java
@@ -81,17 +81,31 @@ public final class MutationCellGrouper {
* for indexed tables (see {@code IndexRegionObserver}).
*/
public static Map<String, byte[]> extractReplicationAttributes(Mutation mutation) {
+ return buildReplicationAttributes(mutation.getAttributesMap(), false);
+ }
+
+ /**
+ * Build the replication attribute envelope from a raw attribute source: the
+ * {@link ReplicationLogGroup#REPLICATION_ATTR_KEYS} present in {@code source}, plus an empty
+ * {@link PhoenixIndexCodec#INDEX_UUID} when {@code indexed}. Shared by the active write path
+ * (source = the batch's first mutation attributes, {@code indexed} = context.hasIndex()) and the
+ * WAL-restore path (source = the WAL key attributes, {@code indexed} = INDEX_UUID already rode
+ * the key). Returns a mutable, possibly empty map.
+ */
+ public static Map<String, byte[]> buildReplicationAttributes(Map<String, byte[]> source,
+ boolean indexed) {
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);
+ if (source != null) {
+ for (String key : ReplicationLogGroup.REPLICATION_ATTR_KEYS) {
+ byte[] v = source.get(key);
+ if (v != null) {
+ envelope.put(key, v);
+ }
}
}
+ if (indexed) {
+ stampIndexAttribute(envelope);
+ }
return envelope;
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
index f926fd70c6..4b5bd6c1d3 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
@@ -873,14 +873,16 @@ public class ReplicationLogGroupIT extends ReplicationLogGroupBaseIT {
final String tableName = "T_" + generateUniqueName();
final String indexName = "I_" + generateUniqueName();
TableName table = TableName.valueOf(tableName);
- try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection) DriverManager
- .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
- String ddl = String.format("create table %s (id1 integer not null, "
+ final String createTableDdl =
+ String.format("create table if not exists %s (id1 integer not null, "
+ "id2 integer not null, val1 varchar, val2 varchar "
+ "constraint pk primary key (id1, id2))", tableName);
- conn.createStatement().execute(ddl);
- ddl = String.format("create index %s on %s (val1) include (val2)", indexName, tableName);
- conn.createStatement().execute(ddl);
+ final String createIndexDdl = String
+ .format("create index if not exists %s on %s (val1) include (val2)", indexName, tableName);
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection) DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(createTableDdl);
+ conn.createStatement().execute(createIndexDdl);
conn.commit();
}
// Mini cluster by default comes with only 1 RS. Starting a second RS so that
@@ -933,6 +935,12 @@ public class ReplicationLogGroupIT extends ReplicationLogGroupBaseIT {
// expected.put(SYSTEM_CATALOG_NAME, 1);
verifyReplication(expected);
}
+ // Beyond the log-entry count, replay the re-shipped log onto cluster 2 and assert the data
+ // table and the standby-regenerated index are byte-for-byte equal across clusters. This
+ // exercises index regeneration from the pre-images the WAL-restore re-ship carries -- which
+ // ignoreSyncReplicationForTesting must leave intact (it suppresses only the synchronous ship).
+ replayAndVerifyAcrossClusters(Arrays.asList(createTableDdl, createIndexDdl), tableName,
+ indexName);
}
@Ignore("Mutations on SYSTEM.CATALOG and SYSTEM.CHILD_LINK are generated on the server side and don't have the HAGroup attribute set")
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
index 6168d2692b..184717d961 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
@@ -631,11 +631,14 @@ public class ReplicationLogProcessorTestIT extends ParallelStatsDisabledIT {
// Recording-site guard (PHOENIX-7992): incrementSuccessfulFileMutationsReplayedCount is
// written inside processLogFile before the catch, not by its callers, so a real replay is the
// only thing that exercises it. The exact replayed-mutation count is intentionally not
- // asserted: generateHBaseMutations builds its Puts from a Phoenix mutation state that is never
- // committed or cleared between rows, so the replayed total is an artifact of that helper rather
+ // asserted: generateHBaseMutations builds its Puts from a Phoenix mutation state that is
+ // never
+ // committed or cleared between rows, so the replayed total is an artifact of that helper
+ // rather
// than the "2 + 5" rows requested (testProcessLogFileBatchSizeLogic pins an exact count on a
// 1:1 record model). We also do not assert the mutationsPerFile histogram here:
- // MutableHistogram.getMax() reads FastLongHistogram bins that the metrics system resets to 0 on
+ // MutableHistogram.getMax() reads FastLongHistogram bins that the metrics system resets to 0
+ // on
// its periodic snapshot(), so a positive max is not stable to read after a multi-second real
// replay -- only the counter (a MutableFastCounter, never reset by snapshot) is reliable.
assertTrue("A successful multi-mutation file must record positive replay throughput",
@@ -1010,7 +1013,8 @@ public class ReplicationLogProcessorTestIT extends ParallelStatsDisabledIT {
*/
@Test
public void testProcessLogFileMidReplayFailureDoesNotCountMutations() throws Exception {
- final Path filePath = new Path(testFolder.newFile("testProcessLogFileMidReplayFailure").toURI());
+ final Path filePath =
+ new Path(testFolder.newFile("testProcessLogFileMidReplayFailure").toURI());
final String tableNameString = "T_" + generateUniqueName();
LogFileWriter writer = initLogFileWriter(filePath);
// Six single-mutation records with batch size 3 -> two full flushes; the second one throws.