From b7302df41a93e7366bf120c236e5a4a72b3f0fae Mon Sep 17 00:00:00 2001 From: lokesh-khurana Date: Mon, 24 Aug 2026 14:44:44 -0700 Subject: [PATCH] PHOENIX-7994 Add cross-cluster IT for server-side immutable-index regeneration on the standby Consistent Failover replicates only data-table mutations; secondary-index tables carry no capture coprocessor, so an index is never shipped as index-table entries. For an IMMUTABLE table the standby regenerates the index from the replayed data mutations via IndexRegionObserver, and that only happens when index maintenance is deferred to the server via phoenix.server.side.immutable.indexes.enabled. Adds ReplicationLogGroupImmutableIndexIT (extends ReplicationLogGroupBaseIT): creates an IMMUTABLE table + global index with the config enabled on the client connection, asserts the replication log carries data-table mutations only, replays the log on cluster 2, and asserts the standby index is fully regenerated (count served by the index, index in the explain plan, cell-equal across clusters). The cross-cluster index equality is the config gate: with the flag disabled the standby index is empty and the assertion fails. Fills the immutable gap next to the mutable sibling ReplicationLogGroupIT#testIndexRegenerationOnStandby. Generated-by: Claude Code (Opus 4.8) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../ReplicationLogGroupImmutableIndexIT.java | 179 ++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupImmutableIndexIT.java diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupImmutableIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupImmutableIndexIT.java new file mode 100644 index 00000000000..e98db898b08 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupImmutableIndexIT.java @@ -0,0 +1,179 @@ +/* + * 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.replication; + +import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED; +import static org.apache.phoenix.query.BaseTest.generateUniqueName; +import static org.apache.phoenix.query.QueryServices.SERVER_SIDE_IMMUTABLE_INDEXES_ENABLED_ATTRIB; +import static org.apache.phoenix.replication.CrossClusterReplicationTestUtil.findLogFiles; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.client.Mutation; +import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest; +import org.apache.phoenix.jdbc.FailoverPhoenixConnection; +import org.apache.phoenix.jdbc.PhoenixResultSet; +import org.apache.phoenix.replication.reader.ReplicationLogProcessor; +import org.apache.phoenix.util.QueryUtil; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Replication coverage for server-side immutable-index maintenance + * ({@code phoenix.server.side.immutable.indexes.enabled}). + *

+ * CCF replicates only data-table mutations; secondary-index tables have no capture coprocessor, so + * an index is never shipped as index-table entries. The standby instead regenerates the index from + * the replayed data mutations via {@code IndexRegionObserver} — but only for batches the active + * marked as server-maintained. For an IMMUTABLE table that requires this config: when it is enabled + * the client defers index maintenance to the server, the active stamps the data batch for standby + * regeneration, and the standby rebuilds the index. When it is disabled the client maintains the + * immutable index itself; those writes go to the active's index table only (never replicated) and + * the data batch is not marked for regeneration, so the standby index is left empty after failover + * — verified by flipping this flag on the client, which leaves the {@code indexName} scan on + * cluster 2 empty and fails the cross-cluster index equality below. + *

+ * The flag is read from the client connection config in {@code MutationState}, so it is set on the + * client properties here (as {@code ServerSideImmutableIndexIT}/{@code ClientSideImmutableIndexIT} + * do), not on the server {@code conf1}/{@code conf2}. The sibling mutable-table case is + * {@code ReplicationLogGroupIT#testIndexRegenerationOnStandby}; existing immutable-index coverage + * ({@code ServerSideImmutableIndexIT}) is single-cluster only. + */ +@Category(NeedsOwnMiniClusterTest.class) +public class ReplicationLogGroupImmutableIndexIT extends ReplicationLogGroupBaseIT { + + @BeforeClass + public static void doSetup() throws Exception { + // Match the sibling IT: the downstream CDC consumer is out of scope here. + conf1.setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, false); + conf2.setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, false); + setupClusters(); + } + + /** + * With server-side immutable-index maintenance enabled on the writing client, an IMMUTABLE + * table's global index is (a) never present as index-table mutations in the replication log + * (structural: index tables carry no capture coprocessor) and (b) regenerated on the standby by + * {@code IndexRegionObserver} from the replayed data mutations. The standby index being fully + * populated (assertions at the end) is the config-gated behavior: with the flag disabled on the + * client the standby index would be empty and the cross-cluster index equality would fail. + */ + @Test + public void testImmutableIndexRegeneratedOnStandby() throws Exception { + final String tableName = "T_" + generateUniqueName(); + final String indexName = "I_" + generateUniqueName(); + final int rowCount = 10; + + // The flag is a client-side decision (read in MutationState from the connection config), so it + // must be set on the client props, not on the server conf1/conf2. + clientProps.setProperty(SERVER_SIDE_IMMUTABLE_INDEXES_ENABLED_ATTRIB, Boolean.toString(true)); + + // Create an immutable table + global index on cluster 1 and insert data. The default storage + // scheme for IMMUTABLE_ROWS=true is inherited by the index, so schemes match and the flag + // genuinely governs client- vs server-side maintenance. + try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection) DriverManager + .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) { + conn.createStatement() + .execute(String.format("CREATE TABLE %s (ID1 INTEGER NOT NULL, ID2 INTEGER NOT NULL, " + + "VAL1 VARCHAR CONSTRAINT PK PRIMARY KEY (ID1, ID2)) IMMUTABLE_ROWS=true", tableName)); + conn.createStatement() + .execute(String.format("CREATE INDEX %s ON %s (VAL1)", indexName, tableName)); + conn.commit(); + PreparedStatement stmt = + conn.prepareStatement("UPSERT INTO " + tableName + " VALUES(?, ?, ?)"); + for (int i = 0; i < rowCount; i++) { + stmt.setInt(1, i); + stmt.setInt(2, i); + stmt.setString(3, "val_" + i); + stmt.executeUpdate(); + } + conn.commit(); + } + + // Capture the standby log dir before closing the group, then confirm the log carries data-table + // mutations only. The absence of index-table mutations is structural (index tables have no + // capture coprocessor), not the config gate. + Path standByLogDir = logGroup.getOrCreatePeerShardManager().getRootDirectoryPath(); + logGroup.close(); + Map> logsByTable = groupLogsByTable(); + dumpTableLogCount(logsByTable); + assertTrue("Replication log should contain data table mutations", + logsByTable.containsKey(tableName)); + assertFalse("Replication log is data-table-only; index tables are never captured", + logsByTable.containsKey(indexName)); + + // Recreate the schema on cluster 2 and replay the log there. + try (Connection conn2 = CLUSTERS.getCluster2Connection(haGroup)) { + conn2.createStatement().execute( + String.format("CREATE TABLE IF NOT EXISTS %s (ID1 INTEGER NOT NULL, ID2 INTEGER NOT NULL, " + + "VAL1 VARCHAR CONSTRAINT PK PRIMARY KEY (ID1, ID2)) IMMUTABLE_ROWS=true", tableName)); + conn2.createStatement() + .execute(String.format("CREATE INDEX IF NOT EXISTS %s ON %s (VAL1)", indexName, tableName)); + conn2.commit(); + } + + FileSystem fs = standByLogDir.getFileSystem(conf2); + List logFiles = findLogFiles(standByLogDir, fs); + LOG.info("Found {} log files to replay", logFiles.size()); + assertTrue("Should have at least one log file", !logFiles.isEmpty()); + ReplicationLogProcessor processor = ReplicationLogProcessor.get(conf2, haGroupName); + try { + for (Path logFile : logFiles) { + LOG.info("Replaying log file: {}", logFile); + processor.processLogFile(fs, logFile); + } + } finally { + processor.close(); + } + + // Config gate: the standby index must have been regenerated from the replayed data mutations by + // IndexRegionObserver. With server-side maintenance disabled on the client this would be empty. + try (Connection conn2 = CLUSTERS.getCluster2Connection(haGroup); + Statement stmt = conn2.createStatement()) { + try (ResultSet rs = stmt.executeQuery("SELECT /*+ NO_INDEX */ COUNT(*) FROM " + tableName)) { + assertTrue(rs.next()); + assertEquals("Data table on cluster 2 should have all rows", rowCount, rs.getInt(1)); + } + try (ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM " + tableName)) { + PhoenixResultSet prs = rs.unwrap(PhoenixResultSet.class); + String explainPlan = QueryUtil.getExplainPlan(prs.getUnderlyingIterator()); + assertTrue("Query should be served by the index on cluster 2", + explainPlan.contains(indexName)); + assertTrue(rs.next()); + assertEquals("Standby index should be regenerated with all rows (server-side maintenance)", + rowCount, rs.getInt(1)); + } + } + + // Deep cell-level equality of both the data table and the regenerated index across clusters. + // The index equality is the strong gate: an empty standby index (flag disabled) fails here. + assertTablesEqualAcrossClusters(tableName); + assertTablesEqualAcrossClusters(indexName); + } +}