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