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
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 26951424fe PHOENIX-8014 Add support for writes to SYSTEM.SEQUENCE
table (#2628)
26951424fe is described below
commit 26951424fe97e3781ef691627720734e7e36c03e
Author: tkhurana <[email protected]>
AuthorDate: Fri Oct 2 10:23:28 2026 -0700
PHOENIX-8014 Add support for writes to SYSTEM.SEQUENCE table (#2628)
---
.../apache/phoenix/compile/SequenceManager.java | 8 +-
.../phoenix/query/ConnectionQueryServices.java | 25 ++++
.../phoenix/query/ConnectionQueryServicesImpl.java | 38 +++++-
.../query/DelegateConnectionQueryServices.java | 15 +++
.../org/apache/phoenix/schema/MetaDataClient.java | 5 +-
.../java/org/apache/phoenix/util/SchemaUtil.java | 5 +-
.../coprocessor/SequenceRegionObserver.java | 20 +++
.../phoenix/replication/ReplicationLogGroupIT.java | 146 +++++++++++++++++++++
.../org/apache/phoenix/schema/SchemaUtilTest.java | 18 +++
9 files changed, 273 insertions(+), 7 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/SequenceManager.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/SequenceManager.java
index 4c9e205599..ada0a708eb 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/SequenceManager.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/SequenceManager.java
@@ -107,7 +107,13 @@ public class SequenceManager {
Long scn = statement.getConnection().getSCN();
long timestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
ConnectionQueryServices services =
statement.getConnection().getQueryServices();
- services.incrementSequences(nextSequences, timestamp, srcSequenceValues,
sqlExceptions);
+ // Resolve the HA group name (null for a non-HA connection) and pass it
down so the INCREMENT
+ // mutations can be tagged for replication when the connection is part
of an HA group.
+ String haGroupName = statement.getConnection().getHAGroup() != null
+ ? statement.getConnection().getHAGroup().getName()
+ : null;
+ services.incrementSequences(nextSequences, timestamp, srcSequenceValues,
sqlExceptions,
+ haGroupName);
setSequenceValues(srcSequenceValues, dstSequenceValues, sqlExceptions);
int offset = nextSequences.size();
for (int i = 0; i < currentSequences.size(); i++) {
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
index 37e0f1d229..8c8cdd5e86 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
@@ -212,6 +212,20 @@ public interface ConnectionQueryServices extends
QueryServices, MetaDataMutated
long incrementBy, long cacheSize, long minValue, long maxValue, boolean
cycle, long timestamp)
throws SQLException;
+ /**
+ * Overload of
+ * {@link #createSequence(String, String, String, long, long, long, long,
long, boolean, long)}
+ * that carries the HA group name so the CREATE mutation can be replicated
when the connection is
+ * part of an HA group. The default ignores the name and preserves the
non-replicating behavior;
+ * implementations that support HA replication override it.
+ */
+ default long createSequence(String tenantId, String schemaName, String
sequenceName,
+ long startWith, long incrementBy, long cacheSize, long minValue, long
maxValue, boolean cycle,
+ long timestamp, String haGroupName) throws SQLException {
+ return createSequence(tenantId, schemaName, sequenceName, startWith,
incrementBy, cacheSize,
+ minValue, maxValue, cycle, timestamp);
+ }
+
long dropSequence(String tenantId, String schemaName, String sequenceName,
long timestamp)
throws SQLException;
@@ -221,6 +235,17 @@ public interface ConnectionQueryServices extends
QueryServices, MetaDataMutated
void incrementSequences(List<SequenceAllocation> sequenceAllocation, long
timestamp,
long[] values, SQLException[] exceptions) throws SQLException;
+ /**
+ * Overload of {@link #incrementSequences(List, long, long[],
SQLException[])} that carries the HA
+ * group name so the INCREMENT mutations can be replicated when the
connection is part of an HA
+ * group. The default ignores the name and preserves the non-replicating
behavior; implementations
+ * that support HA replication override it.
+ */
+ default void incrementSequences(List<SequenceAllocation> sequenceAllocation,
long timestamp,
+ long[] values, SQLException[] exceptions, String haGroupName) throws
SQLException {
+ incrementSequences(sequenceAllocation, timestamp, values, exceptions);
+ }
+
long currentSequenceValue(SequenceKey sequenceKey, long timestamp) throws
SQLException;
void returnSequences(List<SequenceKey> sequenceKeys, long timestamp,
SQLException[] exceptions)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 13276b8163..ef7ab66434 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -227,6 +227,7 @@ import
org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataResponse;
import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataService;
import
org.apache.phoenix.coprocessor.generated.MetaDataProtos.UpdateIndexStateRequest;
import org.apache.phoenix.coprocessor.generated.RegionServerEndpointProtos;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import
org.apache.phoenix.coprocessorclient.InvalidateServerMetadataCacheRequest;
import org.apache.phoenix.coprocessorclient.MetaDataProtocol;
import
org.apache.phoenix.coprocessorclient.MetaDataProtocol.MetaDataMutationResult;
@@ -6026,6 +6027,14 @@ public class ConnectionQueryServicesImpl extends
DelegateQueryServices
public long createSequence(String tenantId, String schemaName, String
sequenceName,
long startWith, long incrementBy, long cacheSize, long minValue, long
maxValue, boolean cycle,
long timestamp) throws SQLException {
+ return createSequence(tenantId, schemaName, sequenceName, startWith,
incrementBy, cacheSize,
+ minValue, maxValue, cycle, timestamp, null);
+ }
+
+ @Override
+ public long createSequence(String tenantId, String schemaName, String
sequenceName,
+ long startWith, long incrementBy, long cacheSize, long minValue, long
maxValue, boolean cycle,
+ long timestamp, String haGroupName) throws SQLException {
SequenceKey sequenceKey =
new SequenceKey(tenantId, schemaName, sequenceName,
nSequenceSaltBuckets);
Sequence newSequences = new Sequence(sequenceKey);
@@ -6038,6 +6047,7 @@ public class ConnectionQueryServicesImpl extends
DelegateQueryServices
// Now that we have the lock we need, create the sequence
Append append = sequence.createSequence(startWith, incrementBy,
cacheSize, timestamp,
minValue, maxValue, cycle);
+ annotateSequenceMutationWithHAGroup(append, haGroupName);
Table htable = this.getTable(SchemaUtil
.getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES,
this.getProps())
.getName());
@@ -6114,7 +6124,8 @@ public class ConnectionQueryServicesImpl extends
DelegateQueryServices
@Override
public void validateSequences(List<SequenceAllocation> sequenceAllocations,
long timestamp,
long[] values, SQLException[] exceptions, Sequence.ValueOp action) throws
SQLException {
- incrementSequenceValues(sequenceAllocations, timestamp, values,
exceptions, action);
+ // VALIDATE_SEQUENCE only reads (never persists a value), so it has
nothing to replicate.
+ incrementSequenceValues(sequenceAllocations, timestamp, values,
exceptions, action, null);
}
/**
@@ -6127,12 +6138,32 @@ public class ConnectionQueryServicesImpl extends
DelegateQueryServices
@Override
public void incrementSequences(List<SequenceAllocation> sequenceAllocations,
long timestamp,
long[] values, SQLException[] exceptions) throws SQLException {
+ incrementSequences(sequenceAllocations, timestamp, values, exceptions,
null);
+ }
+
+ @Override
+ public void incrementSequences(List<SequenceAllocation> sequenceAllocations,
long timestamp,
+ long[] values, SQLException[] exceptions, String haGroupName) throws
SQLException {
incrementSequenceValues(sequenceAllocations, timestamp, values, exceptions,
- Sequence.ValueOp.INCREMENT_SEQUENCE);
+ Sequence.ValueOp.INCREMENT_SEQUENCE, haGroupName);
+ }
+
+ /**
+ * Tag a sequence CREATE/INCREMENT mutation with the HA group name so that,
server-side,
+ * {@code SequenceRegionObserver} can copy it onto the Put it applies and
{@code
+ * IndexRegionObserver} ships the write to the HA replication log. A null
name (non-HA connection)
+ * leaves the mutation untagged, so it is not replicated.
+ */
+ private void annotateSequenceMutationWithHAGroup(Mutation mutation, String
haGroupName) {
+ if (haGroupName != null) {
+
mutation.setAttribute(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB,
+ Bytes.toBytes(haGroupName));
+ }
}
private void incrementSequenceValues(List<SequenceAllocation>
sequenceAllocations, long timestamp,
- long[] values, SQLException[] exceptions, Sequence.ValueOp op) throws
SQLException {
+ long[] values, SQLException[] exceptions, Sequence.ValueOp op, String
haGroupName)
+ throws SQLException {
List<Sequence> sequences =
Lists.newArrayListWithExpectedSize(sequenceAllocations.size());
for (SequenceAllocation sequenceAllocation : sequenceAllocations) {
SequenceKey key = sequenceAllocation.getSequenceKey();
@@ -6161,6 +6192,7 @@ public class ConnectionQueryServicesImpl extends
DelegateQueryServices
toIncrementList.add(sequence);
Increment inc =
sequence.newIncrement(timestamp, op,
sequenceAllocations.get(i).getNumAllocations());
+ annotateSequenceMutationWithHAGroup(inc, haGroupName);
incrementBatch.add(inc);
} catch (SQLException e) {
exceptions[i] = e;
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
index 24506ee9a6..2a79cfeb42 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
@@ -264,6 +264,14 @@ public class DelegateConnectionQueryServices extends
DelegateQueryServices
cacheSize, minValue, maxValue, cycle, timestamp);
}
+ @Override
+ public long createSequence(String tenantId, String schemaName, String
sequenceName,
+ long startWith, long incrementBy, long cacheSize, long minValue, long
maxValue, boolean cycle,
+ long timestamp, String haGroupName) throws SQLException {
+ return getDelegate().createSequence(tenantId, schemaName, sequenceName,
startWith, incrementBy,
+ cacheSize, minValue, maxValue, cycle, timestamp, haGroupName);
+ }
+
@Override
public long dropSequence(String tenantId, String schemaName, String
sequenceName, long timestamp)
throws SQLException {
@@ -282,6 +290,13 @@ public class DelegateConnectionQueryServices extends
DelegateQueryServices
getDelegate().incrementSequences(sequenceAllocations, timestamp, values,
exceptions);
}
+ @Override
+ public void incrementSequences(List<SequenceAllocation> sequenceAllocations,
long timestamp,
+ long[] values, SQLException[] exceptions, String haGroupName) throws
SQLException {
+ getDelegate().incrementSequences(sequenceAllocations, timestamp, values,
exceptions,
+ haGroupName);
+ }
+
@Override
public long currentSequenceValue(SequenceKey sequenceKey, long timestamp)
throws SQLException {
return getDelegate().currentSequenceValue(sequenceKey, timestamp);
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 22833e2494..b9d917cd3b 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -2193,9 +2193,12 @@ public class MetaDataClient {
private MutationState createSequence(String tenantId, String schemaName,
String sequenceName,
boolean ifNotExists, long startWith, long incrementBy, long cacheSize,
boolean cycle,
long minValue, long maxValue, long timestamp) throws SQLException {
+ // Resolve the HA group name (null for a non-HA connection) and pass it
down so the CREATE
+ // SEQUENCE mutation can be tagged for replication when the connection is
part of an HA group.
+ String haGroupName = connection.getHAGroup() != null ?
connection.getHAGroup().getName() : null;
try {
connection.getQueryServices().createSequence(tenantId, schemaName,
sequenceName, startWith,
- incrementBy, cacheSize, minValue, maxValue, cycle, timestamp);
+ incrementBy, cacheSize, minValue, maxValue, cycle, timestamp,
haGroupName);
} catch (SequenceAlreadyExistsException e) {
if (ifNotExists) {
return new MutationState(0, 0, connection);
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/util/SchemaUtil.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index 33f8370b48..5190d5381b 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@ -1523,7 +1523,8 @@ public class SchemaUtil {
* because this data is very specific to the cluster except for
SYSTEM.CATALOG and
* SYSTEM.CHILD_LINK where we only replicate rows with tenant information.
Non tenant (Global)
* rows are assumed to be executed by an admin or an admin process in each
cluster separately and
- * thus not replicated.
+ * thus not replicated. SYSTEM.SEQUENCE replicates all rows (not just tenant
rows) so that
+ * sequence values consumed on the active are known on the standby after
failover.
* @param tableName full name of the table
* @return true if the table data should be replicated, else false
*/
@@ -1531,7 +1532,7 @@ public class SchemaUtil {
if (!isSystemTable(tableName)) {
return true;
}
- if (isMetaTable(tableName) || isChildLinkTable(tableName)) {
+ if (isMetaTable(tableName) || isChildLinkTable(tableName) ||
isSequenceTable(tableName)) {
return true;
}
return false;
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
index 0bbe171486..87d606580d 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
@@ -48,6 +48,7 @@ import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.Region.RowLock;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import org.apache.phoenix.coprocessorclient.MetaDataProtocol;
import org.apache.phoenix.coprocessorclient.SequenceRegionObserverConstants;
import org.apache.phoenix.exception.SQLExceptionCode;
@@ -292,6 +293,7 @@ public class SequenceRegionObserver implements
RegionObserver, RegionCoprocessor
PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, limitReached,
timestamp);
put.add(newLimitReachedKV);
}
+ copyHAGroupNameAttribute(increment, put);
// update the KeyValues on the server
Mutation[] mutations = new Mutation[] { put };
region.batchMutate(mutations);
@@ -308,6 +310,19 @@ public class SequenceRegionObserver implements
RegionObserver, RegionCoprocessor
}
}
+ /**
+ * Copies the {@code _HAGroupName} attribute (when present) from the client
sequence operation
+ * onto the Put applied server-side. The client only sets it for HA
connections, so a non-HA write
+ * leaves the Put untagged and IndexRegionObserver does not ship it.
+ */
+ private static void copyHAGroupNameAttribute(Mutation source, Mutation
target) {
+ byte[] haGroupName =
+
source.getAttribute(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB);
+ if (haGroupName != null) {
+
target.setAttribute(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB,
haGroupName);
+ }
+ }
+
/**
* Creates a new KeyValue for a long value
* @param key key used while creating KeyValue
@@ -437,6 +452,11 @@ public class SequenceRegionObserver implements
RegionObserver, RegionCoprocessor
case CREATE_SEQUENCE:
m = new Put(row, clientTimestamp);
m.getFamilyCellMap().putAll(append.getFamilyCellMap());
+ // Propagate the HA group name from the client Append onto the Put
so that
+ // IndexRegionObserver ships the sequence creation to the HA
replication log. Only
+ // CREATE is replicated; RETURN_SEQUENCE and DROP_SEQUENCE are
intentionally left
+ // untagged so they are not shipped.
+ copyHAGroupNameAttribute(append, m);
break;
}
if (!hadClientTimestamp) {
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 ebc5a1573f..b72634458d 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
@@ -20,6 +20,7 @@ package org.apache.phoenix.replication;
import static
org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME;
import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_NAME;
+import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME;
import static org.apache.phoenix.query.BaseTest.generateUniqueName;
import static
org.apache.phoenix.replication.CrossClusterReplicationTestUtil.findLogFiles;
import static org.junit.Assert.assertEquals;
@@ -1234,6 +1235,151 @@ public class ReplicationLogGroupIT extends
ReplicationLogGroupBaseIT {
}
}
+ /**
+ * End-to-end replication of SYSTEM.SEQUENCE. A sequence created and
advanced on the active over
+ * an HA connection replicates its CREATE and INCREMENT writes; the standby
never runs CREATE
+ * SEQUENCE, so its SYSTEM.SEQUENCE row must materialize purely from replay.
After replaying the
+ * log onto cluster 2, NEXT VALUE FOR on the standby must resume from the
replicated value rather
+ * than restart from START -- i.e. no duplicate sequence values after
failover.
+ * <p>
+ * This is the bare-minimum verbatim path (see {@code
SchemaUtil.shouldReplicateTable}), which
+ * assumes ordered replay: because every sequence cell shares one constant
timestamp, the
+ * last-applied CURRENT_VALUE Put wins, and sequential replay applies them
in log order.
+ */
+ @Test
+ public void testSystemSequence() throws Exception {
+ final String sequenceName = "SEQ_" + generateUniqueName();
+ final int nextCount = 5;
+ final String createSeqDdl =
+ "CREATE SEQUENCE " + sequenceName + " START WITH 1 INCREMENT BY 1 CACHE
1";
+
+ long lastActiveValue = -1;
+ // Create + advance the sequence on the active over an HA connection so
the CREATE Append and
+ // the INCREMENT mutations carry _HAGroupName and replicate. The sequence
is NOT created on
+ // cluster 2; its SYSTEM.SEQUENCE row must materialize purely from replay.
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(createSeqDdl);
+ conn.commit();
+ for (int i = 0; i < nextCount; i++) {
+ try (ResultSet rs =
+ conn.createStatement().executeQuery("SELECT NEXT VALUE FOR " +
sequenceName)) {
+ assertTrue(rs.next());
+ lastActiveValue = rs.getLong(1);
+ }
+ }
+ }
+
+ // The CREATE and INCREMENT writes to SYSTEM.SEQUENCE must have been
shipped.
+ logGroup.close();
+ Map<String, List<Mutation>> logsByTable = groupLogsByTable();
+ dumpTableLogCount(logsByTable);
+ assertTrue("SYSTEM.SEQUENCE writes should replicate",
+ getCountForTable(logsByTable, SYSTEM_SEQUENCE_NAME) > 0);
+
+ // Replay onto cluster 2 without creating the sequence there; the row is
built from replay.
+ replayAndVerifyAcrossClusters(Collections.emptyList());
+
+ // On the standby, NEXT VALUE FOR resumes from the replicated value: the
next value is one past
+ // the last value handed out on the active, not START (which would collide
with 1..lastActive).
+ try (Connection conn2 = CLUSTERS.getCluster2Connection(haGroup); ResultSet
rs =
+ conn2.createStatement().executeQuery("SELECT NEXT VALUE FOR " +
sequenceName)) {
+ assertTrue(rs.next());
+ assertEquals("Standby must resume from the replicated sequence value,
not restart from START",
+ lastActiveValue + 1, rs.getLong(1));
+ }
+ }
+
+ /**
+ * Same as {@link #testSystemSequence} but with CACHE > 1, the realistic
production path. The
+ * server reserves a whole block per round trip, so the replicated
CURRENT_VALUE jumps ahead of
+ * the last value handed out on the active. The standby therefore resumes
strictly past
+ * lastActiveValue (a benign gap), never duplicating a value -- so the
invariant is {@code
+ * standbyNext > lastActiveValue}, not the exact {@code lastActiveValue + 1}
that only holds for
+ * CACHE 1.
+ */
+ @Test
+ public void testSystemSequenceWithCache() throws Exception {
+ final String sequenceName = "SEQ_" + generateUniqueName();
+ final int nextCount = 5;
+ final String createSeqDdl =
+ "CREATE SEQUENCE " + sequenceName + " START WITH 1 INCREMENT BY 1 CACHE
100";
+
+ long lastActiveValue = -1;
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(createSeqDdl);
+ conn.commit();
+ for (int i = 0; i < nextCount; i++) {
+ try (ResultSet rs =
+ conn.createStatement().executeQuery("SELECT NEXT VALUE FOR " +
sequenceName)) {
+ assertTrue(rs.next());
+ lastActiveValue = rs.getLong(1);
+ }
+ }
+ }
+
+ logGroup.close();
+ Map<String, List<Mutation>> logsByTable = groupLogsByTable();
+ dumpTableLogCount(logsByTable);
+ assertTrue("SYSTEM.SEQUENCE writes should replicate",
+ getCountForTable(logsByTable, SYSTEM_SEQUENCE_NAME) > 0);
+
+ replayAndVerifyAcrossClusters(Collections.emptyList());
+
+ try (Connection conn2 = CLUSTERS.getCluster2Connection(haGroup); ResultSet
rs =
+ conn2.createStatement().executeQuery("SELECT NEXT VALUE FOR " +
sequenceName)) {
+ assertTrue(rs.next());
+ assertTrue("Standby must resume past the last active value (gap allowed,
never a duplicate)",
+ rs.getLong(1) > lastActiveValue);
+ }
+ }
+
+ /**
+ * DROP SEQUENCE must NOT replicate: the drop's Delete is left untagged (see
+ * SequenceRegionObserver.preAppend), so only the CREATE and INCREMENT
writes ship. After dropping
+ * on the active and replaying onto cluster 2, the standby's SYSTEM.SEQUENCE
row must survive --
+ * NEXT VALUE FOR there still resumes from the replicated value. A
regression that shipped the
+ * drop would wipe the standby row and this NEXT VALUE FOR would fail.
+ */
+ @Test
+ public void testSystemSequenceDropNotReplicated() throws Exception {
+ final String sequenceName = "SEQ_" + generateUniqueName();
+ final int nextCount = 5;
+ final String createSeqDdl =
+ "CREATE SEQUENCE " + sequenceName + " START WITH 1 INCREMENT BY 1 CACHE
1";
+
+ long lastActiveValue = -1;
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(createSeqDdl);
+ conn.commit();
+ for (int i = 0; i < nextCount; i++) {
+ try (ResultSet rs =
+ conn.createStatement().executeQuery("SELECT NEXT VALUE FOR " +
sequenceName)) {
+ assertTrue(rs.next());
+ lastActiveValue = rs.getLong(1);
+ }
+ }
+ // Drop on the active. The Delete is untagged, so it must not ship.
+ conn.createStatement().execute("DROP SEQUENCE " + sequenceName);
+ conn.commit();
+ }
+
+ logGroup.close();
+ replayAndVerifyAcrossClusters(Collections.emptyList());
+
+ // The standby row survived replay: NEXT VALUE FOR resumes from the
replicated value rather than
+ // failing (which it would if the drop's Delete had been shipped and wiped
the row).
+ try (Connection conn2 = CLUSTERS.getCluster2Connection(haGroup); ResultSet
rs =
+ conn2.createStatement().executeQuery("SELECT NEXT VALUE FOR " +
sequenceName)) {
+ assertTrue(rs.next());
+ assertEquals(
+ "Standby row must survive a non-replicated DROP and resume from
replicated value",
+ lastActiveValue + 1, rs.getLong(1));
+ }
+ }
+
/**
* Verifies that when data mutations are replayed on the standby via
ReplicationLogProcessor,
* IndexRegionObserver on the standby generates index mutations from the
data mutations. The
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
index d03a588b70..c8bc3e5093 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/schema/SchemaUtilTest.java
@@ -18,9 +18,12 @@
package org.apache.phoenix.schema;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.util.SchemaUtil;
import org.junit.Test;
@@ -78,4 +81,19 @@ public class SchemaUtilTest {
String tableDisplayName =
SchemaUtil.getTableNameFromFullName("schemaName.tableName");
assertEquals(tableDisplayName, "tableName");
}
+
+ @Test
+ public void testShouldReplicateTable() {
+ // Non-system tables always replicate.
+
assertTrue(SchemaUtil.shouldReplicateTable(Bytes.toBytes("schemaName.tableName")));
+ // SYSTEM.CATALOG, SYSTEM.CHILD_LINK and SYSTEM.SEQUENCE replicate.
+
assertTrue(SchemaUtil.shouldReplicateTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES));
+ assertTrue(
+
SchemaUtil.shouldReplicateTable(PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_NAME_BYTES));
+
assertTrue(SchemaUtil.shouldReplicateTable(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES));
+ // Other system tables do not replicate.
+
assertFalse(SchemaUtil.shouldReplicateTable(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES));
+ assertFalse(
+
SchemaUtil.shouldReplicateTable(PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME_BYTES));
+ }
}
