You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by vj...@apache.org on 2026/02/03 21:16:46 UTC
(phoenix) branch tmp-ec updated (d0bab6c1d4 -> 0ef716267e)
This is an automated email from the ASF dual-hosted git repository.
vjasani pushed a change to branch tmp-ec
in repository https://gitbox.apache.org/repos/asf/phoenix.git
discard d0bab6c1d4 changes
new 0ef716267e changes
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (d0bab6c1d4)
\
N -- N -- N refs/heads/tmp-ec (0ef716267e)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../ConcurrentMutationsExtendedIndexIT.java | 62 +++++++++++-----------
.../org/apache/phoenix/end2end/UpsertSelectIT.java | 2 +
.../phoenix/monitoring/BasePhoenixMetricsIT.java | 2 +
.../monitoring/PhoenixTableLevelMetricsIT.java | 3 ++
.../ConnectionQueryServicesMetricsIT.java | 6 ++-
5 files changed, 43 insertions(+), 32 deletions(-)
(phoenix) 01/01: changes
Posted by vj...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vjasani pushed a commit to branch tmp-ec
in repository https://gitbox.apache.org/repos/asf/phoenix.git
commit 0ef716267ec39c585b8f94e1e203a379f4627022
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Tue Feb 3 13:14:44 2026 -0800
changes
---
Jenkinsfile | 2 +-
Jenkinsfile.yetus | 2 +-
phoenix-core-client/src/main/antlr3/PhoenixSQL.g | 14 +-
.../org/apache/phoenix/compile/JoinCompiler.java | 3 +-
.../coprocessorclient/MetaDataProtocol.java | 4 +-
.../org/apache/phoenix/index/IndexMaintainer.java | 17 +
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 12 +
.../org/apache/phoenix/jdbc/PhoenixStatement.java | 24 +-
.../apache/phoenix/parse/AlterIndexStatement.java | 15 +-
.../apache/phoenix/parse/CreateIndexStatement.java | 34 +
.../org/apache/phoenix/parse/ParseNodeFactory.java | 8 +
.../phoenix/query/ConnectionQueryServicesImpl.java | 30 +-
.../query/ConnectionlessQueryServicesImpl.java | 8 +
.../org/apache/phoenix/query/QueryConstants.java | 27 +-
.../org/apache/phoenix/schema/DelegateTable.java | 6 +
.../org/apache/phoenix/schema/MetaDataClient.java | 122 ++-
.../java/org/apache/phoenix/schema/PTable.java | 8 +
.../java/org/apache/phoenix/schema/PTableImpl.java | 28 +-
.../apache/phoenix/schema/tool/SchemaSQLUtil.java | 7 +
.../phoenix/schema/transform/TransformClient.java | 7 +-
.../phoenix/schema/types/IndexConsistency.java | 109 +++
.../org/apache/phoenix/util/CDCChangeBuilder.java | 15 +
.../main/java/org/apache/phoenix/util/CDCUtil.java | 29 +
.../java/org/apache/phoenix/util/MetaDataUtil.java | 3 +-
phoenix-core-client/src/main/protobuf/PTable.proto | 1 +
.../src/main/protobuf/ServerCachingService.proto | 1 +
phoenix-core-server/pom.xml | 4 +
.../coprocessor/CDCGlobalIndexRegionScanner.java | 105 ++-
.../phoenix/coprocessor/MetaDataEndpointImpl.java | 19 +-
.../coprocessor/UncoveredIndexRegionScanner.java | 17 +-
.../phoenix/hbase/index/IndexCDCConsumer.java | 857 +++++++++++++++++++++
.../phoenix/hbase/index/IndexRegionObserver.java | 199 +++++
.../TrackingParallelWriterIndexCommitter.java | 7 +-
.../apache/phoenix/end2end/BasePermissionsIT.java | 4 +-
.../java/org/apache/phoenix/end2end/Bson4IT.java | 11 +
.../java/org/apache/phoenix/end2end/Bson5IT.java | 86 ++-
.../ConcurrentMutationsCoveredEventualIT.java | 36 +
.../ConcurrentMutationsCoveredNonEventualIT.java | 36 +
.../end2end/ConcurrentMutationsExtendedIT.java | 108 ++-
.../ConcurrentMutationsExtendedIndexIT.java | 278 +++++++
.../ConcurrentMutationsLazyPostBatchWriteIT.java | 4 +-
.../ConcurrentMutationsUncoveredEventualIT.java | 36 +
.../ConcurrentMutationsUncoveredNonEventualIT.java | 36 +
.../end2end/IndexMaintainerConsistencyIT.java | 216 ++++++
.../org/apache/phoenix/end2end/IndexToolIT.java | 33 +-
.../end2end/MetadataServerConnectionsIT.java | 2 +
.../MigrateSystemTablesToSystemNamespaceIT.java | 16 +-
.../phoenix/end2end/QueryDatabaseMetaDataIT.java | 7 +-
.../end2end/SkipSystemTablesExistenceCheckIT.java | 2 +-
.../SystemTablesCreationOnConnectionIT.java | 4 +-
.../phoenix/end2end/TenantSpecificTablesDDLIT.java | 3 +
.../phoenix/end2end/UCFWithServerMetadataIT.java | 2 +
.../org/apache/phoenix/end2end/UpsertSelectIT.java | 2 +
.../phoenix/end2end/VarBinaryEncoded2IT.java | 89 ++-
.../index/GlobalIndexCheckerWithRegionMovesIT.java | 2 +
.../end2end/index/IndexAsyncThresholdIT.java | 2 +
.../end2end/index/PartialSystemCatalogIndexIT.java | 4 +-
.../index/ReplicationWithWALAnnotationIT.java | 3 +
.../UncoveredGlobalIndexRegionScanner2IT.java | 2 +
.../phoenix/jdbc/FailoverPhoenixConnectionIT.java | 5 +
.../HAConnectionWithMasterAndRPCRegistryIT.java | 5 +
.../jdbc/HighAvailabilityTestingUtilityIT.java | 5 +
.../jdbc/ParallelPhoenixConnectionFallbackIT.java | 5 +
.../ParallelPhoenixNullComparingResultSetIT.java | 5 +
.../phoenix/monitoring/BasePhoenixMetricsIT.java | 2 +
.../monitoring/PhoenixTableLevelMetricsIT.java | 3 +
.../ConnectionQueryServicesMetricsIT.java | 4 +
.../apache/phoenix/query/MetaDataCachingIT.java | 2 +
.../phoenix/parse/IndexConsistencyParseTest.java | 87 +++
.../org/apache/phoenix/util/IndexScrutiny.java | 18 +-
.../java/org/apache/phoenix/util/TestUtil.java | 94 +++
71 files changed, 2852 insertions(+), 151 deletions(-)
diff --git a/Jenkinsfile b/Jenkinsfile
index 46a0628388..0308d61c2f 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -66,7 +66,7 @@ pipeline {
stage('BuildAndTest') {
options {
- timeout(time: 5, unit: 'HOURS')
+ timeout(time: 7, unit: 'HOURS')
}
steps {
dir("HBASE_${HBASE_PROFILE}") {
diff --git a/Jenkinsfile.yetus b/Jenkinsfile.yetus
index a0e9c43f65..cdd97f283e 100644
--- a/Jenkinsfile.yetus
+++ b/Jenkinsfile.yetus
@@ -37,7 +37,7 @@ pipeline {
}
options {
- timeout(time: 5, unit: 'HOURS')
+ timeout(time: 7, unit: 'HOURS')
}
steps {
diff --git a/phoenix-core-client/src/main/antlr3/PhoenixSQL.g b/phoenix-core-client/src/main/antlr3/PhoenixSQL.g
index b905c43f3e..dac4c11be0 100644
--- a/phoenix-core-client/src/main/antlr3/PhoenixSQL.g
+++ b/phoenix-core-client/src/main/antlr3/PhoenixSQL.g
@@ -85,6 +85,7 @@ tokens
PRE='pre';
POST='post';
CHANGE='change';
+ IDX_MUTATIONS='idx_mutations';
LATEST='latest';
ALL='all';
INDEX='index';
@@ -164,6 +165,9 @@ tokens
TRUNCATE = 'truncate';
PRESERVE='preserve';
SPLITS='splits';
+ CONSISTENCY = 'consistency';
+ EVENTUAL = 'eventual';
+ STRONG = 'strong';
}
@@ -212,6 +216,7 @@ import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.IllegalDataException;
import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTable.CDCChangeScope;
@@ -603,7 +608,7 @@ cdc_change_scopes returns [Set<CDCChangeScope> ret]
;
cdc_change_scope returns [CDCChangeScope ret]
- : v=(PRE | POST | CHANGE)
+ : v=(PRE | POST | CHANGE | IDX_MUTATIONS)
{
ret = CDCChangeScope.valueOf(v.getText().toUpperCase());
}
@@ -717,8 +722,8 @@ drop_cdc_node returns [DropCDCStatement ret]
// Parse a alter index statement
alter_index_node returns [AlterIndexStatement ret]
: ALTER INDEX (IF ex=EXISTS)? i=index_name ON t=from_table_name
- ((s=(USABLE | UNUSABLE | REBUILD (isRebuildAll=ALL)? | DISABLE | ACTIVE)) (async=ASYNC)? ((SET?)p=fam_properties)?)
- {ret = factory.alterIndex(factory.namedTable(null, TableName.create(t.getSchemaName(), i.getName())), t.getTableName(), ex!=null, PIndexState.valueOf(SchemaUtil.normalizeIdentifier(s.getText())), isRebuildAll!=null, async!=null, p); }
+ ((s=(USABLE | UNUSABLE | REBUILD (isRebuildAll=ALL)? | DISABLE | ACTIVE)) (async=ASYNC)? ((SET?)p=fam_properties)? | (CONSISTENCY EQ c=(STRONG | EVENTUAL)))
+ {ret = factory.alterIndex(factory.namedTable(null, TableName.create(t.getSchemaName(), i.getName())), t.getTableName(), ex!=null, s!=null ? PIndexState.valueOf(SchemaUtil.normalizeIdentifier(s.getText())) : null, isRebuildAll!=null, async!=null, p, c!=null ? IndexConsistency.valueOf(SchemaUtil.normalizeIdentifier(c.getText())) : null); }
;
// Parse a trace statement.
@@ -1335,6 +1340,9 @@ identifier returns [String ret]
parseNoReserved returns [String ret]
: n=NAME { $ret = n.getText(); }
+ | CONSISTENCY { $ret = "CONSISTENCY"; }
+ | EVENTUAL { $ret = "EVENTUAL"; }
+ | STRONG { $ret = "STRONG"; }
;
case_statement returns [ParseNode ret]
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/JoinCompiler.java b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
index ad31baeaf8..8a091fbdc0 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
@@ -1544,7 +1544,8 @@ public class JoinCompiler {
.setDisableWAL(PTable.DEFAULT_DISABLE_WAL).setMultiTenant(left.isMultiTenant())
.setStoreNulls(left.getStoreNulls()).setViewType(left.getViewType())
.setViewIndexIdType(left.getviewIndexIdType()).setViewIndexId(left.getViewIndexId())
- .setIndexType(left.getIndexType()).setTransactionProvider(left.getTransactionProvider())
+ .setIndexType(left.getIndexType()).setIndexConsistency(left.getIndexConsistency())
+ .setTransactionProvider(left.getTransactionProvider())
.setUpdateCacheFrequency(left.getUpdateCacheFrequency())
.setNamespaceMapped(left.isNamespaceMapped())
.setAutoPartitionSeqName(left.getAutoPartitionSeqName())
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java
index b523a44de0..08cac7cbe9 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java
@@ -91,8 +91,8 @@ public abstract class MetaDataProtocol extends MetaDataService {
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_4_16_0 = MIN_TABLE_TIMESTAMP + 33;
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_1_0 = MIN_SYSTEM_TABLE_TIMESTAMP_4_16_0;
public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 = MIN_TABLE_TIMESTAMP + 38;
- public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_3_0 = MIN_TABLE_TIMESTAMP + 42;
- public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 = MIN_TABLE_TIMESTAMP + 43;
+ public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_3_0 = MIN_TABLE_TIMESTAMP + 44;
+ public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 = MIN_TABLE_TIMESTAMP + 45;
// MIN_SYSTEM_TABLE_TIMESTAMP needs to be set to the max of all the MIN_SYSTEM_TABLE_TIMESTAMP_*
// constants
public static final long MIN_SYSTEM_TABLE_TIMESTAMP = MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0;
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java b/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
index e2b33a17d7..70b83d1795 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
@@ -108,6 +108,7 @@ import org.apache.phoenix.schema.transform.TransformMaintainer;
import org.apache.phoenix.schema.tuple.BaseTuple;
import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
import org.apache.phoenix.schema.tuple.ValueGetterTuple;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PBoolean;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PVarbinaryEncoded;
@@ -477,10 +478,12 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
private Expression indexWhere;
private Set<ColumnReference> indexWhereColumns;
private boolean isCDCIndex;
+ private IndexConsistency indexConsistency;
protected IndexMaintainer(RowKeySchema dataRowKeySchema, boolean isDataTableSalted) {
this.dataRowKeySchema = dataRowKeySchema;
this.isDataTableSalted = isDataTableSalted;
+ this.indexConsistency = null;
}
private IndexMaintainer(final PTable dataTable, final PTable index, PhoenixConnection connection)
@@ -501,6 +504,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this.isUncovered = index.getIndexType() == IndexType.UNCOVERED_GLOBAL;
this.encodingScheme = index.getEncodingScheme();
this.isCDCIndex = CDCUtil.isCDCIndex(index);
+ this.indexConsistency = index.getIndexConsistency();
// null check for b/w compatibility
this.encodingScheme = index.getEncodingScheme() == null
@@ -1933,6 +1937,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this.encodingScheme = QualifierEncodingScheme.NON_ENCODED_QUALIFIERS;
this.dataImmutableStorageScheme = ImmutableStorageScheme.ONE_CELL_PER_COLUMN;
this.dataEncodingScheme = QualifierEncodingScheme.NON_ENCODED_QUALIFIERS;
+ this.indexConsistency = null;
initCachedState();
}
@@ -2068,6 +2073,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
} else {
maintainer.isCDCIndex = false;
}
+ if (proto.hasIndexConsistency()) {
+ maintainer.indexConsistency = IndexConsistency.valueOf(proto.getIndexConsistency());
+ } else {
+ maintainer.indexConsistency = null;
+ }
maintainer.nDataTableSaltBuckets =
proto.hasDataTableSaltBuckets() ? proto.getDataTableSaltBuckets() : -1;
maintainer.initCachedState();
@@ -2223,6 +2233,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
}
builder.setIsCDCIndex(maintainer.isCDCIndex);
+ if (maintainer.indexConsistency != null) {
+ builder.setIndexConsistency(maintainer.indexConsistency.name());
+ }
if (maintainer.isDataTableSalted) {
builder.setDataTableSaltBuckets(maintainer.nDataTableSaltBuckets);
}
@@ -2609,6 +2622,10 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
return indexedColumns;
}
+ public IndexConsistency getIndexConsistency() {
+ return indexConsistency;
+ }
+
public static class UDFParseNodeVisitor extends StatelessTraverseAllParseNodeVisitor {
private Map<String, UDFParseNode> udfParseNodes;
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 3bc68c5ddb..ed63485440 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -213,6 +213,8 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final byte[] VIEW_TYPE_BYTES = Bytes.toBytes(VIEW_TYPE);
public static final String INDEX_TYPE = "INDEX_TYPE";
public static final byte[] INDEX_TYPE_BYTES = Bytes.toBytes(INDEX_TYPE);
+ public static final String INDEX_CONSISTENCY = "INDEX_CONSISTENCY";
+ public static final byte[] INDEX_CONSISTENCY_BYTES = Bytes.toBytes(INDEX_CONSISTENCY);
public static final String LINK_TYPE = "LINK_TYPE";
public static final byte[] LINK_TYPE_BYTES = Bytes.toBytes(LINK_TYPE);
public static final String TASK_TYPE = "TASK_TYPE";
@@ -498,6 +500,16 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
"PARTITION_END_TIME IS NOT NULL "
+ "AND TO_NUMBER(CURRENT_TIME()) - TO_NUMBER(PHOENIX_ROW_TIMESTAMP()) >= %d";
+ public static final String SYSTEM_IDX_CDC_TRACKER_TABLE = "IDX_CDC_TRACKER";
+ public static final String SYSTEM_IDX_CDC_TRACKER_NAME =
+ SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA, SYSTEM_IDX_CDC_TRACKER_TABLE);
+ public static final String LAST_TIMESTAMP = "LAST_TIMESTAMP";
+ public static final String OWNER_PARTITION_ID = "OWNER_PARTITION_ID";
+ public static final String TRACKER_STATUS = "STATUS";
+ public static final String TRACKER_STATUS_IN_PROGRESS = "i";
+ public static final String TRACKER_STATUS_COMPLETE = "c";
+ public static final int IDX_CDC_TRACKER_TTL = 10 * 24 * 60 * 60;
+
public static final String QUERY_ID = "QUERY_ID";
public static final String USER = "USER";
public static final String CLIENT_IP = "CLIENT_IP";
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index b527ecd042..a62be697a9 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -220,6 +220,7 @@ import org.apache.phoenix.schema.stats.StatisticsCollectionScope;
import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
import org.apache.phoenix.schema.tuple.ResultTuple;
import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PLong;
import org.apache.phoenix.schema.types.PVarchar;
@@ -1650,9 +1651,10 @@ public class PhoenixStatement implements PhoenixMonitoredStatement, SQLCloseable
public ExecutableCreateIndexStatement(NamedNode indexName, NamedTableNode dataTable,
IndexKeyConstraint ikConstraint, List<ColumnName> includeColumns, List<ParseNode> splits,
ListMultimap<String, Pair<String, Object>> props, boolean ifNotExists, IndexType indexType,
- boolean async, int bindCount, Map<String, UDFParseNode> udfParseNodes, ParseNode where) {
+ boolean async, int bindCount, Map<String, UDFParseNode> udfParseNodes, ParseNode where,
+ IndexConsistency indexConsistency) {
super(indexName, dataTable, ikConstraint, includeColumns, splits, props, ifNotExists,
- indexType, async, bindCount, udfParseNodes, where);
+ indexType, async, bindCount, udfParseNodes, where, indexConsistency);
}
@SuppressWarnings("unchecked")
@@ -1890,6 +1892,13 @@ public class PhoenixStatement implements PhoenixMonitoredStatement, SQLCloseable
super(indexTableNode, dataTableName, ifExists, state, isRebuildAll, async, props);
}
+ public ExecutableAlterIndexStatement(NamedTableNode indexTableNode, String dataTableName,
+ boolean ifExists, PIndexState state, boolean isRebuildAll, boolean async,
+ ListMultimap<String, Pair<String, Object>> props, IndexConsistency indexConsistency) {
+ super(indexTableNode, dataTableName, ifExists, state, isRebuildAll, async, props,
+ indexConsistency);
+ }
+
@SuppressWarnings("unchecked")
@Override
public MutationPlan compilePlan(final PhoenixStatement stmt, Sequence.ValueOp seqAction)
@@ -2281,7 +2290,8 @@ public class PhoenixStatement implements PhoenixMonitoredStatement, SQLCloseable
ListMultimap<String, Pair<String, Object>> props, boolean ifNotExists, IndexType indexType,
boolean async, int bindCount, Map<String, UDFParseNode> udfParseNodes, ParseNode where) {
return new ExecutableCreateIndexStatement(indexName, dataTable, ikConstraint, includeColumns,
- splits, props, ifNotExists, indexType, async, bindCount, udfParseNodes, where);
+ splits, props, ifNotExists, indexType, async, bindCount, udfParseNodes, where,
+ CreateIndexStatement.getIndexConsistency(props));
}
@Override
@@ -2338,6 +2348,14 @@ public class PhoenixStatement implements PhoenixMonitoredStatement, SQLCloseable
isRebuildAll, async, props);
}
+ @Override
+ public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName,
+ boolean ifExists, PIndexState state, boolean isRebuildAll, boolean async,
+ ListMultimap<String, Pair<String, Object>> props, IndexConsistency indexConsistency) {
+ return new ExecutableAlterIndexStatement(indexTableNode, dataTableName, ifExists, state,
+ isRebuildAll, async, props, indexConsistency);
+ }
+
@Override
public TraceStatement trace(boolean isTraceOn, double samplingRate) {
return new ExecutableTraceStatement(isTraceOn, samplingRate);
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java
index 7ef5a61828..2b12e32ae6 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java
@@ -20,6 +20,7 @@ package org.apache.phoenix.parse;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.thirdparty.com.google.common.collect.ImmutableListMultimap;
import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
@@ -31,16 +32,23 @@ public class AlterIndexStatement extends SingleTableStatement {
private boolean async;
private boolean isRebuildAll;
private ListMultimap<String, Pair<String, Object>> props;
+ private final IndexConsistency indexConsistency;
private static final PTableType tableType = PTableType.INDEX;
public AlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists,
PIndexState indexState, boolean isRebuildAll, boolean async) {
- this(indexTableNode, dataTableName, ifExists, indexState, isRebuildAll, async, null);
+ this(indexTableNode, dataTableName, ifExists, indexState, isRebuildAll, async, null, null);
}
public AlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists,
PIndexState indexState, boolean isRebuildAll, boolean async,
ListMultimap<String, Pair<String, Object>> props) {
+ this(indexTableNode, dataTableName, ifExists, indexState, isRebuildAll, async, props, null);
+ }
+
+ public AlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists,
+ PIndexState indexState, boolean isRebuildAll, boolean async,
+ ListMultimap<String, Pair<String, Object>> props, IndexConsistency indexConsistency) {
super(indexTableNode, 0);
this.dataTableName = dataTableName;
this.ifExists = ifExists;
@@ -48,6 +56,7 @@ public class AlterIndexStatement extends SingleTableStatement {
this.async = async;
this.isRebuildAll = isRebuildAll;
this.props = props == null ? ImmutableListMultimap.<String, Pair<String, Object>> of() : props;
+ this.indexConsistency = indexConsistency;
}
public String getTableName() {
@@ -82,4 +91,8 @@ public class AlterIndexStatement extends SingleTableStatement {
public PTableType getTableType() {
return tableType;
}
+
+ public IndexConsistency getIndexConsistency() {
+ return indexConsistency;
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java
index ec71fb1024..8cc7685a31 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java
@@ -21,7 +21,9 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PTable.IndexType;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.thirdparty.com.google.common.collect.ArrayListMultimap;
import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
@@ -37,11 +39,21 @@ public class CreateIndexStatement extends SingleTableStatement {
private final boolean async;
private final Map<String, UDFParseNode> udfParseNodes;
private final ParseNode where;
+ private final IndexConsistency indexConsistency;
public CreateIndexStatement(NamedNode indexTableName, NamedTableNode dataTable,
IndexKeyConstraint indexKeyConstraint, List<ColumnName> includeColumns, List<ParseNode> splits,
ListMultimap<String, Pair<String, Object>> props, boolean ifNotExists, IndexType indexType,
boolean async, int bindCount, Map<String, UDFParseNode> udfParseNodes, ParseNode where) {
+ this(indexTableName, dataTable, indexKeyConstraint, includeColumns, splits, props, ifNotExists,
+ indexType, async, bindCount, udfParseNodes, where, getIndexConsistency(props));
+ }
+
+ public CreateIndexStatement(NamedNode indexTableName, NamedTableNode dataTable,
+ IndexKeyConstraint indexKeyConstraint, List<ColumnName> includeColumns, List<ParseNode> splits,
+ ListMultimap<String, Pair<String, Object>> props, boolean ifNotExists, IndexType indexType,
+ boolean async, int bindCount, Map<String, UDFParseNode> udfParseNodes, ParseNode where,
+ IndexConsistency indexConsistency) {
super(dataTable, bindCount);
this.indexTableName =
TableName.create(dataTable.getName().getSchemaName(), indexTableName.getName());
@@ -56,6 +68,7 @@ public class CreateIndexStatement extends SingleTableStatement {
this.async = async;
this.udfParseNodes = udfParseNodes;
this.where = where;
+ this.indexConsistency = indexConsistency;
}
public CreateIndexStatement(CreateIndexStatement createStmt,
@@ -71,6 +84,23 @@ public class CreateIndexStatement extends SingleTableStatement {
this.async = createStmt.isAsync();
this.udfParseNodes = createStmt.getUdfParseNodes();
this.where = createStmt.where;
+ this.indexConsistency = createStmt.getIndexConsistency();
+ }
+
+ public static IndexConsistency
+ getIndexConsistency(ListMultimap<String, Pair<String, Object>> props) {
+ IndexConsistency indexConsistency = null;
+ if (props != null) {
+ for (Pair<String, Object> prop : props.get(QueryConstants.ALL_FAMILY_PROPERTIES_KEY)) {
+ if (prop != null && "CONSISTENCY".equalsIgnoreCase(prop.getFirst())) {
+ Object value = prop.getSecond();
+ indexConsistency =
+ value == null ? null : IndexConsistency.valueOf(value.toString().toUpperCase());
+ break;
+ }
+ }
+ }
+ return indexConsistency;
}
public IndexKeyConstraint getIndexConstraint() {
@@ -112,4 +142,8 @@ public class CreateIndexStatement extends SingleTableStatement {
public ParseNode getWhere() {
return where;
}
+
+ public IndexConsistency getIndexConsistency() {
+ return indexConsistency;
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
index 716dbd566a..4fe036e492 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
@@ -49,6 +49,7 @@ import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TypeMismatchException;
import org.apache.phoenix.schema.stats.StatisticsCollectionScope;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PBinary;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PLong;
@@ -484,6 +485,13 @@ public class ParseNodeFactory {
async, props);
}
+ public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName,
+ boolean ifExists, PIndexState state, boolean isRebuildAll, boolean async,
+ ListMultimap<String, Pair<String, Object>> props, IndexConsistency indexConsistency) {
+ return new AlterIndexStatement(indexTableNode, dataTableName, ifExists, state, isRebuildAll,
+ async, props, indexConsistency);
+ }
+
public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName,
boolean ifExists, PIndexState state) {
return new AlterIndexStatement(indexTableNode, dataTableName, ifExists, state, false, false);
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 926089201e..5cd95b648e 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
@@ -60,6 +60,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_HBASE_TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_NAME_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_MUTEX_COLUMN_NAME_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_MUTEX_FAMILY_NAME_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_MUTEX_NAME;
@@ -4125,6 +4126,10 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
return ddl + ",TTL='" + ttlExpression + "'";
}
+ protected String getIdxCdcTrackerDDL() {
+ return setSystemDDLProperties(QueryConstants.CREATE_IDX_CDC_TRACKER_METADATA);
+ }
+
private String setSystemDDLProperties(String ddl) {
return String.format(ddl,
props.getInt(DEFAULT_SYSTEM_MAX_VERSIONS_ATTRIB,
@@ -4458,6 +4463,10 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
metaConnection.createStatement().executeUpdate(getCDCStreamDDL());
} catch (TableAlreadyExistsException ignore) {
}
+ try {
+ metaConnection.createStatement().executeUpdate(getIdxCdcTrackerDDL());
+ } catch (TableAlreadyExistsException ignore) {
+ }
try {
// check if we have old PHOENIX_INDEX_TOOL tables
// move data to the new tables under System, or simply create the new tables
@@ -4816,8 +4825,8 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 1,
PhoenixDatabaseMetaData.IS_STRICT_TTL + " " + PBoolean.INSTANCE.getSqlTypeName());
- // Add any 5.4.0 specific upgrade logic here if needed in the future
- // Currently no new schema changes required for 5.4.0
+ metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0, PhoenixDatabaseMetaData.INDEX_CONSISTENCY + " CHAR(1)");
// move TTL values stored in descriptor to SYSCAT TTL column.
moveTTLFromHBaseLevelTTLToPhoenixLevelTTL(metaConnection);
@@ -5044,6 +5053,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
metaConnection = upgradeSystemMutex(metaConnection);
metaConnection = upgradeSystemCDCStreamStatus(metaConnection);
metaConnection = upgradeSystemCDCStream(metaConnection);
+ metaConnection = upgradeSystemIdxCdcTracker(metaConnection);
// As this is where the most time will be spent during an upgrade,
// especially when there are large number of views.
@@ -5366,6 +5376,15 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
return metaConnection;
}
+ private PhoenixConnection upgradeSystemIdxCdcTracker(PhoenixConnection metaConnection)
+ throws SQLException {
+ try {
+ metaConnection.createStatement().executeUpdate(getIdxCdcTrackerDDL());
+ } catch (TableAlreadyExistsException ignored) {
+ }
+ return metaConnection;
+ }
+
// Special method for adding the column qualifier column for 4.10.
private PhoenixConnection addColumnQualifierColumn(PhoenixConnection oldMetaConnection,
Long timestamp) throws SQLException {
@@ -6966,6 +6985,8 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
"DELETE FROM " + SYSTEM_CDC_STREAM_STATUS_NAME + " WHERE TABLE_NAME = ?";
String deleteStreamPartitionsQuery =
"DELETE FROM " + SYSTEM_CDC_STREAM_NAME + " WHERE TABLE_NAME = ?";
+ String deleteIdxCdcTrackerQuery =
+ "DELETE FROM " + SYSTEM_IDX_CDC_TRACKER_NAME + " WHERE TABLE_NAME = ?";
LOGGER.info("Deleting Stream Metadata for table {}", tableName);
try (PreparedStatement ps = conn.prepareStatement(deleteStreamStatusQuery)) {
ps.setString(1, tableName);
@@ -6977,6 +6998,11 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices
ps.executeUpdate();
conn.commit();
}
+ try (PreparedStatement ps = conn.prepareStatement(deleteIdxCdcTrackerQuery)) {
+ ps.setString(1, tableName);
+ ps.executeUpdate();
+ conn.commit();
+ }
}
/**
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index d0fe82ee19..abbc08ac31 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@ -219,6 +219,10 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices
return setSystemDDLProperties(QueryConstants.CREATE_CDC_STREAM_METADATA);
}
+ protected String getIdxCdcTrackerDDL() {
+ return setSystemDDLProperties(QueryConstants.CREATE_IDX_CDC_TRACKER_METADATA);
+ }
+
private String setSystemDDLProperties(String ddl) {
return String.format(ddl,
props.getInt(DEFAULT_SYSTEM_MAX_VERSIONS_ATTRIB,
@@ -496,6 +500,10 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices
metaConnection.createStatement().executeUpdate(getCDCStreamDDL());
} catch (TableAlreadyExistsException ignore) {
}
+ try {
+ metaConnection.createStatement().executeUpdate(getIdxCdcTrackerDDL());
+ } catch (TableAlreadyExistsException ignore) {
+ }
} catch (SQLException e) {
sqlE = e;
} finally {
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java
index fe9ea6ef8b..aa73c83314 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -54,9 +54,11 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GLOBAL_SCAN_DETAIL
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GUIDE_POSTS_ROW_COUNT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GUIDE_POST_KEY;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IDX_CDC_TRACKER_TTL;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_ROWS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_STORAGE_SCHEME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INCREMENT_BY;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_CONSISTENCY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_STATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_TYPE;
@@ -73,6 +75,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.JAR_PATH;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.KEY_SEQ;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_DDL_TIMESTAMP;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_STATS_UPDATE_TIME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_TIMESTAMP;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LINK_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LOGICAL_PARENT_NAME;
@@ -88,6 +91,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NUM_ARGS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NUM_PREC_RADIX;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.OLD_METADATA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ORDINAL_POSITION;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.OWNER_PARTITION_ID;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PARENT_PARTITION_ID;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PARENT_PARTITION_START_TIME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PARTITION_END_KEY;
@@ -133,6 +137,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_TABLE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_MUTEX_TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_STATS_TABLE;
@@ -150,6 +155,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TASK_TABLE_TTL;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TASK_TS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TASK_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRACKER_STATUS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSACTIONAL;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSACTION_PROVIDER;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSFORM_FUNCTION;
@@ -376,12 +382,17 @@ public interface QueryConstants {
String CDC_PRE_IMAGE = "pre_image";
String CDC_POST_IMAGE = "post_image";
String CDC_CHANGE_IMAGE = "change_image";
+ String CDC_IDX_MUTATIONS = "idx_mutations";
String CDC_UPSERT_EVENT_TYPE = "upsert";
String CDC_DELETE_EVENT_TYPE = "delete";
String SPLITS_FILE = "SPLITS_FILE";
String CDC_TTL_DELETE_EVENT_TYPE = "ttl_delete";
String CDC_IMAGE_CQ = "_CDC_IMG_";
byte[] CDC_IMAGE_CQ_BYTES = Bytes.toBytes(CDC_IMAGE_CQ);
+ String CDC_INDEX_PRE_MUTATIONS_CQ = "_IDX_PRE_";
+ byte[] CDC_INDEX_PRE_MUTATIONS_CQ_BYTES = Bytes.toBytes(CDC_INDEX_PRE_MUTATIONS_CQ);
+ String CDC_INDEX_POST_MUTATIONS_CQ = "_IDX_POST_";
+ byte[] CDC_INDEX_POST_MUTATIONS_CQ_BYTES = Bytes.toBytes(CDC_INDEX_POST_MUTATIONS_CQ);
/**
* We mark counter values 0 to 10 as reserved. Value 0 is used by
@@ -408,7 +419,7 @@ public interface QueryConstants {
+ " BOOLEAN, \n" + SCHEMA_VERSION + " VARCHAR, \n" + EXTERNAL_SCHEMA_ID + " VARCHAR, \n"
+ STREAMING_TOPIC_NAME + " VARCHAR, \n" + INDEX_WHERE + " VARCHAR, \n" + CDC_INCLUDE_TABLE
+ " VARCHAR, \n" + TTL + " VARCHAR, \n" + ROW_KEY_MATCHER + " VARBINARY_ENCODED, \n"
- + IS_STRICT_TTL + " BOOLEAN, \n" +
+ + IS_STRICT_TTL + " BOOLEAN, \n" + INDEX_CONSISTENCY + " CHAR(1), \n" +
// Column metadata (will be null for table row)
DATA_TYPE + " INTEGER," + COLUMN_SIZE + " INTEGER," + DECIMAL_DIGITS + " INTEGER," + NULLABLE
+ " INTEGER," + ORDINAL_POSITION + " INTEGER," + SORT_ORDER + " INTEGER," + ARRAY_SIZE
@@ -586,4 +597,18 @@ public interface QueryConstants {
+ PARENT_PARTITION_ID + "))\n" + HConstants.VERSIONS + "=%s,\n"
+ ColumnFamilyDescriptorBuilder.KEEP_DELETED_CELLS + "=%s,\n" + TRANSACTIONAL + "="
+ Boolean.FALSE + ",\n" + UPDATE_CACHE_FREQUENCY + "=" + "7200000";
+
+ String CREATE_IDX_CDC_TRACKER_METADATA =
+ "CREATE TABLE " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_IDX_CDC_TRACKER_TABLE + "\"(\n" +
+ // PK columns
+ TABLE_NAME + " VARCHAR NOT NULL," + PARTITION_ID + " VARCHAR NOT NULL," + OWNER_PARTITION_ID
+ + " VARCHAR NOT NULL," +
+ // Non-PK columns
+ LAST_TIMESTAMP + " BIGINT," + TRACKER_STATUS + " CHAR(1),\n" + "CONSTRAINT "
+ + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TABLE_NAME + "," + PARTITION_ID + ","
+ + OWNER_PARTITION_ID + "))\n" + HConstants.VERSIONS + "=%s,\n"
+ + ColumnFamilyDescriptorBuilder.KEEP_DELETED_CELLS + "=%s,\n" + TRANSACTIONAL + "="
+ + Boolean.FALSE + ",\n" + UPDATE_CACHE_FREQUENCY + "=" + "7200000" + ",\n"
+ + ColumnFamilyDescriptorBuilder.TTL + "=" + IDX_CDC_TRACKER_TTL + ",\n"
+ + "\"phoenix.max.lookback.age.seconds\"=0";
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java
index 44df43c929..b41da68782 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/DelegateTable.java
@@ -28,6 +28,7 @@ import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.transform.TransformMaintainer;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.transaction.TransactionFactory;
@@ -283,6 +284,11 @@ public class DelegateTable implements PTable {
return delegate.getIndexType();
}
+ @Override
+ public IndexConsistency getIndexConsistency() {
+ return delegate.getIndexConsistency();
+ }
+
private final PTable delegate;
public DelegateTable(PTable delegate) {
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 1b8e898bd3..aede85a2c5 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
@@ -59,6 +59,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.FUNCTION_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_ROWS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_STORAGE_SCHEME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_CONSISTENCY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_STATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_TYPE;
@@ -278,6 +279,7 @@ import org.apache.phoenix.schema.stats.StatisticsUtil;
import org.apache.phoenix.schema.task.SystemTaskParams;
import org.apache.phoenix.schema.task.Task;
import org.apache.phoenix.schema.transform.TransformClient;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PBson;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PDate;
@@ -346,9 +348,9 @@ public class MetaDataClient {
+ IMMUTABLE_STORAGE_SCHEME + "," + ENCODING_SCHEME + "," + USE_STATS_FOR_PARALLELIZATION + ","
+ VIEW_INDEX_ID_DATA_TYPE + "," + CHANGE_DETECTION_ENABLED + "," + PHYSICAL_TABLE_NAME + ","
+ SCHEMA_VERSION + "," + STREAMING_TOPIC_NAME + "," + INDEX_WHERE + "," + CDC_INCLUDE_TABLE
- + "," + TTL + "," + ROW_KEY_MATCHER + "," + IS_STRICT_TTL
+ + "," + TTL + "," + ROW_KEY_MATCHER + "," + IS_STRICT_TTL + "," + INDEX_CONSISTENCY
+ ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, "
- + "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+ + "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String CREATE_SCHEMA = "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\""
+ SYSTEM_CATALOG_TABLE + "\"( " + TABLE_SCHEM + "," + TABLE_NAME + ") VALUES (?,?)";
@@ -400,6 +402,10 @@ public class MetaDataClient {
+ INDEX_STATE + "," + INDEX_DISABLE_TIMESTAMP + "," + ASYNC_REBUILD_TIMESTAMP + " "
+ PLong.INSTANCE.getSqlTypeName() + ") VALUES (?, ?, ?, ?, ?, ?)";
+ public static final String UPDATE_INDEX_CONSISTENCY =
+ "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " + TENANT_ID + ","
+ + TABLE_SCHEM + "," + TABLE_NAME + "," + INDEX_CONSISTENCY + ") VALUES (?, ?, ?, ?)";
+
/*
* Custom sql to add a column to SYSTEM.CATALOG table during upgrade. We can't use the regular
* ColumnMetaDataOps.UPSERT_COLUMN sql because the COLUMN_QUALIFIER column was added in 4.10. And
@@ -1064,7 +1070,7 @@ public class MetaDataClient {
}
}
table = createTableInternal(statement, splits, parent, viewStatement, viewType, viewIndexIdType,
- rowKeyMatcher, viewColumnConstants, isViewColumnReferenced, false, null, null, null,
+ rowKeyMatcher, viewColumnConstants, isViewColumnReferenced, false, null, null, null, null,
tableProps, commonFamilyProps);
if (table == null || table.getType() == PTableType.VIEW || statement.isNoVerify() /*
@@ -1897,7 +1903,7 @@ public class MetaDataClient {
statement.ifNotExists(), null, statement.getWhere(), statement.getBindCount(), null);
table = createTableInternal(tableStatement, splits, dataTable, null, null,
getViewIndexDataType(), null, null, null, allocateIndexId, statement.getIndexType(),
- asyncCreatedDate, null, tableProps, commonFamilyProps);
+ statement.getIndexConsistency(), asyncCreatedDate, null, tableProps, commonFamilyProps);
} finally {
deleteMutexCells(physicalSchemaName, physicalTableName, acquiredColumnMutexSet);
}
@@ -1907,6 +1913,13 @@ public class MetaDataClient {
if (LOGGER.isInfoEnabled())
LOGGER.info("Created index " + table.getName().getString() + " at " + table.getTimeStamp());
+
+ if (
+ statement.getIndexConsistency() != null && statement.getIndexConsistency().isAsynchronous()
+ ) {
+ createCDCForEventuallyConsistentIndex(dataTable);
+ }
+
boolean asyncIndexBuildEnabled =
connection.getQueryServices().getProps().getBoolean(QueryServices.INDEX_ASYNC_BUILD_ENABLED,
QueryServicesOptions.DEFAULT_INDEX_ASYNC_BUILD_ENABLED);
@@ -2020,7 +2033,7 @@ public class MetaDataClient {
columnDefs, FACTORY.primaryKey(null, pkColumnDefs), Collections.emptyList(), PTableType.CDC,
statement.isIfNotExists(), null, null, statement.getBindCount(), null);
createTableInternal(tableStatement, null, dataTable, null, null, null, null, null, null, false,
- null, null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ null, null, null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
// for now, only track stream partition metadata for tables, TODO: updatable views
if (PTableType.TABLE == dataTable.getType()) {
updateStreamPartitionMetadata(dataTableFullName, cdcObjName);
@@ -2440,12 +2453,37 @@ public class MetaDataClient {
return parentName.getString().equals(view.getPhysicalName().getString());
}
+ /**
+ * Creates a CDC stream on the data table for eventually consistent indexes if not already
+ * present.
+ * @param dataTable The data table PTable object for which to create the CDC stream.
+ * @throws SQLException If there is an error executing the CREATE CDC statement or accessing the
+ * database connection.
+ */
+ private void createCDCForEventuallyConsistentIndex(PTable dataTable) throws SQLException {
+ String dataTableName = dataTable.getTableName().getString();
+ String schemaName = dataTable.getSchemaName().getString();
+ String cdcName = "CDC_" + dataTableName;
+ String fullTableName = SchemaUtil.getCaseSensitiveColumnDisplayName(schemaName, dataTableName);
+ String cdcSql = "CREATE CDC IF NOT EXISTS \"" + cdcName + "\" ON " + fullTableName;
+ try {
+ connection.createStatement().execute(cdcSql);
+ } catch (SQLException e) {
+ if (SQLExceptionCode.CDC_ALREADY_ENABLED.getErrorCode() != e.getErrorCode()) {
+ LOGGER.error("Error while creating CDC for {}", fullTableName, e);
+ throw e;
+ }
+ LOGGER.debug("CDC already exists for {}", fullTableName, e);
+ }
+ }
+
private PTable createTableInternal(CreateTableStatement statement, byte[][] splits,
final PTable parent, String viewStatement, ViewType viewType, PDataType viewIndexIdType,
final byte[] rowKeyMatcher, final byte[][] viewColumnConstants,
final BitSet isViewColumnReferenced, boolean allocateIndexId, IndexType indexType,
- Date asyncCreatedDate, Set<PTable.CDCChangeScope> cdcIncludeScopes,
- Map<String, Object> tableProps, Map<String, Object> commonFamilyProps) throws SQLException {
+ IndexConsistency indexConsistency, Date asyncCreatedDate,
+ Set<PTable.CDCChangeScope> cdcIncludeScopes, Map<String, Object> tableProps,
+ Map<String, Object> commonFamilyProps) throws SQLException {
final PTableType tableType = statement.getTableType();
boolean wasAutoCommit = connection.getAutoCommit();
TableName tableNameNode = null;
@@ -3395,9 +3433,10 @@ public class MetaDataClient {
.setTimeStamp(MetaDataProtocol.MIN_TABLE_TIMESTAMP).setIndexDisableTimestamp(0L)
.setSequenceNumber(PTable.INITIAL_SEQ_NUM).setImmutableRows(isImmutableRows)
.setDisableWAL(Boolean.TRUE.equals(disableWAL)).setMultiTenant(false).setStoreNulls(false)
- .setViewIndexIdType(viewIndexIdType).setIndexType(indexType).setUpdateCacheFrequency(0)
- .setNamespaceMapped(isNamespaceMapped).setAutoPartitionSeqName(autoPartitionSeq)
- .setAppendOnlySchema(isAppendOnlySchema).setImmutableStorageScheme(ONE_CELL_PER_COLUMN)
+ .setViewIndexIdType(viewIndexIdType).setIndexType(indexType).setIndexConsistency(null)
+ .setUpdateCacheFrequency(0).setNamespaceMapped(isNamespaceMapped)
+ .setAutoPartitionSeqName(autoPartitionSeq).setAppendOnlySchema(isAppendOnlySchema)
+ .setImmutableStorageScheme(ONE_CELL_PER_COLUMN)
.setQualifierEncodingScheme(NON_ENCODED_QUALIFIERS)
.setBaseColumnCount(QueryConstants.BASE_TABLE_BASE_COLUMN_COUNT)
.setEncodedCQCounter(PTable.EncodedCQCounter.NULL_COUNTER)
@@ -3702,6 +3741,16 @@ public class MetaDataClient {
tableUpsert.setBoolean(38, isStrictTTL);
+ if (tableType == PTableType.INDEX) {
+ if (indexConsistency == null) {
+ tableUpsert.setNull(39, Types.CHAR);
+ } else {
+ tableUpsert.setString(39, indexConsistency.getCodeAsString());
+ }
+ } else {
+ tableUpsert.setNull(39, Types.CHAR);
+ }
+
tableUpsert.execute();
if (asyncCreatedDate != null) {
@@ -3805,6 +3854,7 @@ public class MetaDataClient {
.setDisableWAL(Boolean.TRUE.equals(disableWAL)).setMultiTenant(multiTenant)
.setStoreNulls(storeNulls).setViewType(viewType).setViewIndexIdType(viewIndexIdType)
.setViewIndexId(result.getViewIndexId()).setIndexType(indexType)
+ .setIndexConsistency(tableType == PTableType.INDEX ? indexConsistency : null)
.setTransactionProvider(transactionProvider).setUpdateCacheFrequency(updateCacheFrequency)
.setNamespaceMapped(isNamespaceMapped).setAutoPartitionSeqName(autoPartitionSeq)
.setAppendOnlySchema(isAppendOnlySchema).setImmutableStorageScheme(immutableStorageScheme)
@@ -5924,6 +5974,37 @@ public class MetaDataClient {
metaPropertiesEvaluated, table, schemaName, tableName, new MutableBoolean(false));
PIndexState newIndexState = statement.getIndexState();
+ IndexConsistency newIndexConsistency = statement.getIndexConsistency();
+
+ if (newIndexConsistency != null) {
+ try (PreparedStatement consistencyUpsert =
+ connection.prepareStatement(UPDATE_INDEX_CONSISTENCY)) {
+ consistencyUpsert.setString(1, tenantId);
+ consistencyUpsert.setString(2, schemaName);
+ consistencyUpsert.setString(3, indexName);
+ consistencyUpsert.setString(4, newIndexConsistency.getCodeAsString());
+ consistencyUpsert.execute();
+ connection.commit();
+
+ if (newIndexConsistency == IndexConsistency.EVENTUAL) {
+ PTable dataTable =
+ connection.getTable(tenantId, SchemaUtil.getTableName(schemaName, dataTableName));
+ createCDCForEventuallyConsistentIndex(dataTable);
+ }
+
+ connection.getQueryServices().clearTableFromCache(
+ connection.getTenantId() == null
+ ? ByteUtil.EMPTY_BYTE_ARRAY
+ : connection.getTenantId().getBytes(),
+ schemaName == null ? ByteUtil.EMPTY_BYTE_ARRAY : Bytes.toBytes(schemaName),
+ Bytes.toBytes(indexName), HConstants.LATEST_TIMESTAMP);
+ connection.removeTable(connection.getTenantId(),
+ SchemaUtil.getTableName(schemaName, indexName),
+ SchemaUtil.getTableName(schemaName, dataTableName), HConstants.LATEST_TIMESTAMP);
+
+ return new MutationState(1, 1000, connection);
+ }
+ }
if (isAsync && newIndexState != PIndexState.REBUILD) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.ASYNC_NOT_ALLOWED)
@@ -5937,13 +6018,8 @@ public class MetaDataClient {
connection.setAutoCommit(false);
// Confirm index table is valid and up-to-date
TableRef indexRef = FromCompiler.getResolver(statement, connection).getTables().get(0);
- PreparedStatement tableUpsert = null;
- try {
- if (newIndexState == PIndexState.ACTIVE) {
- tableUpsert = connection.prepareStatement(UPDATE_INDEX_STATE_TO_ACTIVE);
- } else {
- tableUpsert = connection.prepareStatement(UPDATE_INDEX_STATE);
- }
+ try (PreparedStatement tableUpsert = connection.prepareStatement(
+ newIndexState == PIndexState.ACTIVE ? UPDATE_INDEX_STATE_TO_ACTIVE : UPDATE_INDEX_STATE)) {
tableUpsert.setString(1,
connection.getTenantId() == null ? null : connection.getTenantId().getString());
tableUpsert.setString(2, schemaName);
@@ -5954,10 +6030,6 @@ public class MetaDataClient {
tableUpsert.setLong(6, 0);
}
tableUpsert.execute();
- } finally {
- if (tableUpsert != null) {
- tableUpsert.close();
- }
}
Long timeStamp = indexRef.getTable().isTransactional() ? indexRef.getTimeStamp() : null;
List<Mutation> tableMetadata =
@@ -6046,8 +6118,8 @@ public class MetaDataClient {
}
}
} else {
- try {
- tableUpsert = connection.prepareStatement(UPDATE_INDEX_REBUILD_ASYNC_STATE);
+ try (PreparedStatement tableUpsert =
+ connection.prepareStatement(UPDATE_INDEX_REBUILD_ASYNC_STATE)) {
tableUpsert.setString(1,
connection.getTenantId() == null ? null : connection.getTenantId().getString());
tableUpsert.setString(2, schemaName);
@@ -6056,10 +6128,6 @@ public class MetaDataClient {
tableUpsert.setLong(4, beginTimestamp);
tableUpsert.execute();
connection.commit();
- } finally {
- if (tableUpsert != null) {
- tableUpsert.close();
- }
}
}
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java
index 36a9ab739f..bd314e00c9 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java
@@ -39,6 +39,7 @@ import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.transform.TransformMaintainer;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PArrayDataType;
import org.apache.phoenix.schema.types.PArrayDataTypeDecoder;
import org.apache.phoenix.schema.types.PArrayDataTypeEncoder;
@@ -931,6 +932,8 @@ public interface PTable extends PMetaDataEntity {
IndexType getIndexType();
+ IndexConsistency getIndexConsistency();
+
int getBaseColumnCount();
/**
@@ -1154,5 +1157,10 @@ public interface PTable extends PMetaDataEntity {
* Include only the post image (state past the change) of the row.
*/
POST,
+
+ /**
+ * Include index mutations for eventually consistent indexes.
+ */
+ IDX_MUTATIONS,
}
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java
index 027d031705..dee96ceafc 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTableImpl.java
@@ -103,6 +103,7 @@ import org.apache.phoenix.protobuf.ProtobufUtil;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.RowKeySchema.RowKeySchemaBuilder;
import org.apache.phoenix.schema.transform.TransformMaintainer;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PBinary;
import org.apache.phoenix.schema.types.PChar;
import org.apache.phoenix.schema.types.PDataType;
@@ -193,6 +194,7 @@ public class PTableImpl implements PTable {
private final Long viewIndexId;
private final int estimatedSize;
private final IndexType indexType;
+ private final IndexConsistency indexConsistency;
private final int baseColumnCount;
private final boolean rowKeyOrderOptimizable; // TODO: remove when required that tables have been
// upgrade for PHOENIX-2067
@@ -267,6 +269,7 @@ public class PTableImpl implements PTable {
private Long viewIndexId;
private int estimatedSize;
private IndexType indexType;
+ private IndexConsistency indexConsistency;
private int baseColumnCount;
private boolean rowKeyOrderOptimizable;
private boolean hasColumnsRequiringUpgrade;
@@ -534,6 +537,11 @@ public class PTableImpl implements PTable {
return this;
}
+ public Builder setIndexConsistency(IndexConsistency indexConsistency) {
+ this.indexConsistency = indexConsistency;
+ return this;
+ }
+
public Builder setBaseColumnCount(int baseColumnCount) {
this.baseColumnCount = baseColumnCount;
return this;
@@ -994,6 +1002,7 @@ public class PTableImpl implements PTable {
this.viewIndexId = builder.viewIndexId;
this.estimatedSize = builder.estimatedSize;
this.indexType = builder.indexType;
+ this.indexConsistency = builder.indexConsistency;
this.baseColumnCount = builder.baseColumnCount;
this.rowKeyOrderOptimizable = builder.rowKeyOrderOptimizable;
this.hasColumnsRequiringUpgrade = builder.hasColumnsRequiringUpgrade;
@@ -1052,6 +1061,7 @@ public class PTableImpl implements PTable {
.setMultiTenant(table.isMultiTenant()).setStoreNulls(table.getStoreNulls())
.setViewType(table.getViewType()).setViewIndexIdType(table.getviewIndexIdType())
.setViewIndexId(table.getViewIndexId()).setIndexType(table.getIndexType())
+ .setIndexConsistency(table.getIndexConsistency())
.setTransactionProvider(table.getTransactionProvider())
.setUpdateCacheFrequency(table.getUpdateCacheFrequency())
.setNamespaceMapped(table.isNamespaceMapped())
@@ -1890,6 +1900,11 @@ public class PTableImpl implements PTable {
return indexType;
}
+ @Override
+ public IndexConsistency getIndexConsistency() {
+ return type == PTableType.INDEX ? indexConsistency : null;
+ }
+
/**
* Construct a PTable instance from ProtoBuffered PTable instance
*/
@@ -2096,6 +2111,14 @@ public class PTableImpl implements PTable {
if (table.hasRowKeyMatcher()) {
rowKeyMatcher = table.getRowKeyMatcher().toByteArray();
}
+ IndexConsistency indexConsistency = null;
+ if (tableType == PTableType.INDEX) {
+ if (table.hasIndexConsistency()) {
+ indexConsistency = IndexConsistency.valueOf(table.getIndexConsistency());
+ } else {
+ indexConsistency = null;
+ }
+ }
try {
return new PTableImpl.Builder().setType(tableType).setState(indexState)
@@ -2103,7 +2126,7 @@ public class PTableImpl implements PTable {
.setSequenceNumber(sequenceNumber).setImmutableRows(isImmutableRows)
.setViewStatement(viewStatement).setDisableWAL(disableWAL).setMultiTenant(multiTenant)
.setStoreNulls(storeNulls).setViewType(viewType).setViewIndexIdType(viewIndexIdType)
- .setViewIndexId(viewIndexId).setIndexType(indexType)
+ .setViewIndexId(viewIndexId).setIndexType(indexType).setIndexConsistency(indexConsistency)
.setTransactionProvider(transactionProvider).setUpdateCacheFrequency(updateCacheFrequency)
.setNamespaceMapped(isNamespaceMapped).setAutoPartitionSeqName(autoPartitionSeqName)
.setAppendOnlySchema(isAppendOnlySchema)
@@ -2166,6 +2189,9 @@ public class PTableImpl implements PTable {
builder.setIndexType(
ByteStringer.wrap(new byte[] { table.getIndexType().getSerializedValue() }));
}
+ if (table.getIndexConsistency() != null) {
+ builder.setIndexConsistency(table.getIndexConsistency().name());
+ }
}
builder.setSequenceNumber(table.getSequenceNumber());
builder.setTimeStamp(table.getTimeStamp());
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/tool/SchemaSQLUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/tool/SchemaSQLUtil.java
index 6fc4c1c98a..d6b513ad26 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/tool/SchemaSQLUtil.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/tool/SchemaSQLUtil.java
@@ -26,6 +26,7 @@ import org.apache.phoenix.parse.CreateIndexStatement;
import org.apache.phoenix.parse.CreateTableStatement;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
@@ -74,6 +75,12 @@ public class SchemaSQLUtil {
if (createStmt.isAsync()) {
sb.append(" ASYNC");
}
+ if (
+ createStmt.getIndexConsistency() != null
+ && createStmt.getIndexConsistency() != IndexConsistency.getDefault()
+ ) {
+ sb.append(" CONSISTENCY=").append(createStmt.getIndexConsistency().name());
+ }
appendProperties(sb, createStmt.getProps());
return sb.toString();
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java
index ebcd9f034c..9d1145afd5 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java
@@ -258,9 +258,10 @@ public class TransformClient {
.setEstimatedSize(table.getEstimatedSize()).setFamilies(table.getColumnFamilies())
.setImmutableRows(table.isImmutableRows())
.setIsChangeDetectionEnabled(table.isChangeDetectionEnabled())
- .setIndexType(table.getIndexType()).setIndexes(Collections.<PTable> emptyList())
- .setName(newTableName).setMultiTenant(table.isMultiTenant())
- .setParentName(table.getParentName()).setParentSchemaName(table.getParentSchemaName())
+ .setIndexType(table.getIndexType()).setIndexConsistency(table.getIndexConsistency())
+ .setIndexes(Collections.<PTable> emptyList()).setName(newTableName)
+ .setMultiTenant(table.isMultiTenant()).setParentName(table.getParentName())
+ .setParentSchemaName(table.getParentSchemaName())
.setNamespaceMapped(table.isNamespaceMapped()).setSchemaName(table.getSchemaName())
.setPkColumns(table.getPKColumns()).setPkName(table.getPKName())
.setRowKeySchema(table.getRowKeySchema()).setStoreNulls(table.getStoreNulls())
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/IndexConsistency.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/IndexConsistency.java
new file mode 100644
index 0000000000..13b61fab4e
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/types/IndexConsistency.java
@@ -0,0 +1,109 @@
+/*
+ * 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.schema.types;
+
+/**
+ * Enum representing the consistency level for Phoenix secondary indexes.
+ * <p>
+ * <b>STRONG</b>: Synchronous index maintenance where index mutations are applied synchronously with
+ * data table mutations. This provides immediate consistency but may impact write performance and
+ * availability.
+ * </p>
+ * <p>
+ * <b>EVENTUAL</b>: Asynchronous index maintenance where index mutations are applied in the
+ * background. This provides better write performance and availability but with eventual consistency
+ * guarantees.
+ * </p>
+ */
+public enum IndexConsistency {
+ /**
+ * Strong consistency - synchronous index maintenance (default behavior). Index updates are
+ * applied synchronously with data table updates.
+ */
+ STRONG('s'),
+
+ /**
+ * Eventual consistency - asynchronous index maintenance using CDC. Index updates are applied in
+ * the background, providing eventual consistency.
+ */
+ EVENTUAL('e');
+
+ private final char code;
+
+ IndexConsistency(char code) {
+ this.code = code;
+ }
+
+ /**
+ * Returns the single character code for the consistency.
+ * @return 's' for STRONG, 'e' for EVENTUAL
+ */
+ public char getCode() {
+ return code;
+ }
+
+ /**
+ * Returns the string representation of the code for storage.
+ * @return single character string for storage
+ */
+ public String getCodeAsString() {
+ return String.valueOf(code);
+ }
+
+ /**
+ * Parses an IndexConsistency from its stored code.
+ * @param code the stored code.
+ * @return the corresponding IndexConsistency, or null if code is null/empty/invalid.
+ */
+ public static IndexConsistency fromCode(String code) {
+ if (code == null || code.isEmpty()) {
+ return null;
+ }
+ char c = code.charAt(0);
+ for (IndexConsistency ic : values()) {
+ if (ic.code == c) {
+ return ic;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Returns the default consistency level for indexes.
+ * @return STRONG consistency as the default
+ */
+ public static IndexConsistency getDefault() {
+ return STRONG;
+ }
+
+ /**
+ * Checks if this consistency level requires synchronous index maintenance.
+ * @return true if synchronous maintenance is required, false otherwise
+ */
+ public boolean isSynchronous() {
+ return this == STRONG;
+ }
+
+ /**
+ * Checks if this consistency level uses asynchronous index maintenance.
+ * @return true if asynchronous maintenance is used, false otherwise
+ */
+ public boolean isAsynchronous() {
+ return this == EVENTUAL;
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCChangeBuilder.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCChangeBuilder.java
index 45a7e957e2..34408cfbf1 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCChangeBuilder.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCChangeBuilder.java
@@ -36,6 +36,7 @@ public class CDCChangeBuilder {
private final boolean isChangeImageInScope;
private final boolean isPreImageInScope;
private final boolean isPostImageInScope;
+ private final boolean isIdxMutationsInScope;
private final CDCTableInfo cdcDataTableInfo;
private String changeType;
private long lastDeletedTimestamp;
@@ -49,6 +50,7 @@ public class CDCChangeBuilder {
isChangeImageInScope = changeScopes.contains(PTable.CDCChangeScope.CHANGE);
isPreImageInScope = changeScopes.contains(PTable.CDCChangeScope.PRE);
isPostImageInScope = changeScopes.contains(PTable.CDCChangeScope.POST);
+ isIdxMutationsInScope = changeScopes.contains(PTable.CDCChangeScope.IDX_MUTATIONS);
}
public void initChange(long ts) {
@@ -149,4 +151,17 @@ public class CDCChangeBuilder {
public boolean isPreImageInScope() {
return isPreImageInScope;
}
+
+ public boolean isPostImageInScope() {
+ return isPostImageInScope;
+ }
+
+ public boolean isChangeImageInScope() {
+ return isChangeImageInScope;
+ }
+
+ public boolean isIdxMutationsInScope() {
+ return isIdxMutationsInScope;
+ }
+
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java
index 68c9b48aa7..d36c957f7a 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.util.StringUtils;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.execute.DescVarLengthFastByteComparisons;
+import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.types.PDataType;
@@ -148,6 +149,34 @@ public class CDCUtil {
return table.getIndexes().stream().anyMatch(CDCUtil::isCDCIndex);
}
+ /**
+ * Get the CDC object name for the given data table that can be used in SQL queries.
+ * @param dataTable The data table PTable object.
+ * @param withSchema If true, returns the CDC object name with schema prefix. If false, returns
+ * only the CDC object name.
+ * @return The CDC object name that can be used in SQL queries, optionally prefixed with schema
+ * name, or null if no active CDC index exists.
+ */
+ public static String getCDCObjectName(PTable dataTable, boolean withSchema) {
+ if (dataTable == null) {
+ return null;
+ }
+ PTable cdcIndex = getActiveCDCIndex(dataTable);
+ if (cdcIndex == null) {
+ return null;
+ }
+ String cdcIndexName = cdcIndex.getTableName().getString();
+ String schemaName = dataTable.getSchemaName().getString();
+ if (isCDCIndex(cdcIndexName)) {
+ String cdcObj = cdcIndexName.substring(CDC_INDEX_PREFIX.length());
+ if (!withSchema || schemaName == null || schemaName.isEmpty()) {
+ return cdcObj;
+ }
+ return schemaName + QueryConstants.NAME_SEPARATOR + cdcObj;
+ }
+ return null;
+ }
+
public static Scan setupScanForCDC(Scan scan) {
scan.setRaw(true);
scan.readAllVersions();
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
index 5396fc70f8..9249a8ad68 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
@@ -126,7 +126,8 @@ public class MetaDataUtil {
public static final List<String> SYSTEM_TABLES_WITH_TTL_SUPPORTED =
Arrays.asList(SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA, SYSTEM_CDC_STREAM_TABLE),
- SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA, SYSTEM_CDC_STREAM_STATUS_TABLE));
+ SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA, SYSTEM_CDC_STREAM_STATUS_TABLE),
+ SchemaUtil.getTableName(SYSTEM_CATALOG_SCHEMA, SYSTEM_IDX_CDC_TRACKER_TABLE));
public static Put getLastDDLTimestampUpdate(byte[] tableHeaderRowKey, long clientTimestamp,
long lastDDLTimestamp) {
diff --git a/phoenix-core-client/src/main/protobuf/PTable.proto b/phoenix-core-client/src/main/protobuf/PTable.proto
index 43f5315114..a98239f9ed 100644
--- a/phoenix-core-client/src/main/protobuf/PTable.proto
+++ b/phoenix-core-client/src/main/protobuf/PTable.proto
@@ -124,6 +124,7 @@ message PTable {
optional bytes ttl = 55;
optional bytes rowKeyMatcher = 56;
optional bool isStrictTTL = 57;
+ optional string indexConsistency = 58;
}
message EncodedCQCounter {
diff --git a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
index d8e1945527..2c4f358ccc 100644
--- a/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
+++ b/phoenix-core-client/src/main/protobuf/ServerCachingService.proto
@@ -73,6 +73,7 @@ message IndexMaintainer {
repeated ColumnReference indexWhereColumns = 30;
optional bool isCDCIndex = 31;
optional int32 dataTableSaltBuckets = 32;
+ optional string indexConsistency = 33;
}
message TransformMaintainer {
diff --git a/phoenix-core-server/pom.xml b/phoenix-core-server/pom.xml
index e3abcb3558..043cb5c31e 100644
--- a/phoenix-core-server/pom.xml
+++ b/phoenix-core-server/pom.xml
@@ -173,6 +173,10 @@
<groupId>com.google.code.findbugs</groupId>
<artifactId>jsr305</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.mongodb</groupId>
+ <artifactId>bson</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CDCGlobalIndexRegionScanner.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CDCGlobalIndexRegionScanner.java
index 0b49441824..3e82409498 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CDCGlobalIndexRegionScanner.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/CDCGlobalIndexRegionScanner.java
@@ -48,13 +48,18 @@ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.index.CDCTableInfo;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.tuple.ResultTuple;
+import org.apache.phoenix.schema.types.PBson;
import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PVarchar;
import org.apache.phoenix.util.CDCChangeBuilder;
import org.apache.phoenix.util.CDCUtil;
import org.apache.phoenix.util.EncodedColumnsUtil;
import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.JacksonUtil;
+import org.bson.BsonArray;
+import org.bson.BsonDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,6 +92,9 @@ public class CDCGlobalIndexRegionScanner extends UncoveredGlobalIndexRegionScann
private CDCTableInfo cdcDataTableInfo;
private CDCChangeBuilder changeBuilder;
+ private static final byte[] EMPTY_IDX_MUTATIONS =
+ PVarchar.INSTANCE.toBytes(new BsonDocument().toJson());
+
public CDCGlobalIndexRegionScanner(final RegionScanner innerScanner, final Region region,
final Scan scan, final RegionCoprocessorEnvironment env, final Scan dataTableScan,
final TupleProjector tupleProjector, final IndexMaintainer indexMaintainer,
@@ -102,6 +110,12 @@ public class CDCGlobalIndexRegionScanner extends UncoveredGlobalIndexRegionScann
@Override
protected Scan prepareDataTableScan(Collection<byte[]> dataRowKeys) throws IOException {
+ if (
+ changeBuilder.isIdxMutationsInScope() && !changeBuilder.isChangeImageInScope()
+ && !changeBuilder.isPreImageInScope() && !changeBuilder.isPostImageInScope()
+ ) {
+ return null;
+ }
// TODO: Get Timerange from the start row and end row of the index scan object
// and set it in the datatable scan object.
// if (scan.getStartRow().length == 8) {
@@ -120,9 +134,11 @@ public class CDCGlobalIndexRegionScanner extends UncoveredGlobalIndexRegionScann
List<Cell> indexRow = indexRowIterator.next();
Cell indexCell = indexRow.get(0);
byte[] indexRowKey = ImmutableBytesPtr.cloneCellRowIfNecessary(indexCell);
+ if (handleIdxMutationsCDCEvent(indexRow, indexRowKey, indexCell, result)) {
+ return true;
+ }
if (indexRow.size() > 1) {
- boolean success = handlePreImageCDCEvent(indexRow, indexRowKey, indexCell, result);
- if (success) {
+ if (handlePreImageCDCEvent(indexRow, indexRowKey, indexCell, result)) {
return true;
}
}
@@ -317,18 +333,93 @@ public class CDCGlobalIndexRegionScanner extends UncoveredGlobalIndexRegionScann
cdcJson.remove(QueryConstants.CDC_PRE_IMAGE);
cdcEventBytes = JacksonUtil.getObjectWriter(HashMap.class).writeValueAsBytes(cdcJson);
}
+ addResult(indexRowKey, indexCell, result, cdcDataCell, cdcEventBytes);
+ return true;
+ }
+
+ /**
+ * Handles CDC event when IDX_MUTATIONS scope is enabled. Returns the raw index mutations bytes
+ * directly, or empty BsonDocument bytes if no mutations are present. Skip the data table scan.
+ * @param indexRow The CDC index row.
+ * @param indexRowKey The CDC index row key.
+ * @param indexCell The primary index cell.
+ * @param result The result list to populate.
+ * @return true if IDX_MUTATIONS scope is enabled, false otherwise.
+ */
+ private boolean handleIdxMutationsCDCEvent(List<Cell> indexRow, byte[] indexRowKey,
+ Cell indexCell, List<Cell> result) {
+ if (!changeBuilder.isIdxMutationsInScope()) {
+ return false;
+ }
+ byte[] idxMutationsBytes = EMPTY_IDX_MUTATIONS;
+ Cell idxMutationsCell = indexCell;
+ BsonDocument preDoc = null;
+ BsonDocument postDoc = null;
+ Cell preCell = null;
+ for (Cell cell : indexRow) {
+ if (
+ Bytes.equals(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
+ QueryConstants.CDC_INDEX_PRE_MUTATIONS_CQ_BYTES, 0,
+ QueryConstants.CDC_INDEX_PRE_MUTATIONS_CQ_BYTES.length)
+ ) {
+ byte[] rawBytes = CellUtil.cloneValue(cell);
+ preDoc = (BsonDocument) PBson.INSTANCE.toObject(rawBytes, 0, rawBytes.length,
+ PBson.INSTANCE, SortOrder.ASC, null, null);
+ preCell = cell;
+ } else if (
+ Bytes.equals(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
+ QueryConstants.CDC_INDEX_POST_MUTATIONS_CQ_BYTES, 0,
+ QueryConstants.CDC_INDEX_POST_MUTATIONS_CQ_BYTES.length)
+ ) {
+ byte[] rawBytes = CellUtil.cloneValue(cell);
+ postDoc = (BsonDocument) PBson.INSTANCE.toObject(rawBytes, 0, rawBytes.length,
+ PBson.INSTANCE, SortOrder.ASC, null, null);
+ }
+ }
+ if (preDoc != null) {
+ BsonDocument mergedDoc = mergeBsonDocuments(preDoc, postDoc);
+ String jsonString = mergedDoc.toJson();
+ idxMutationsBytes = PVarchar.INSTANCE.toBytes(jsonString);
+ idxMutationsCell = preCell;
+ }
+ addResult(indexRowKey, indexCell, result, idxMutationsCell, idxMutationsBytes);
+ return true;
+ }
+
+ private BsonDocument mergeBsonDocuments(BsonDocument preDoc, BsonDocument postDoc) {
+ if (postDoc == null) {
+ return preDoc;
+ }
+ BsonArray mergedTables = new BsonArray();
+ BsonArray mergedMutations = new BsonArray();
+ if (preDoc.containsKey("tables")) {
+ mergedTables.addAll(preDoc.getArray("tables"));
+ }
+ if (preDoc.containsKey("mutations")) {
+ mergedMutations.addAll(preDoc.getArray("mutations"));
+ }
+ if (postDoc.containsKey("tables")) {
+ mergedTables.addAll(postDoc.getArray("tables"));
+ }
+ if (postDoc.containsKey("mutations")) {
+ mergedMutations.addAll(postDoc.getArray("mutations"));
+ }
+ BsonDocument mergedDoc = new BsonDocument();
+ mergedDoc.append("tables", mergedTables);
+ mergedDoc.append("mutations", mergedMutations);
+ return mergedDoc;
+ }
+
+ private void addResult(byte[] indexRowKey, Cell indexCell, List<Cell> result, Cell cdcDataCell,
+ byte[] cdcEventBytes) {
Result cdcRow =
createCDCResult(indexRowKey, indexCell, cdcDataCell.getTimestamp(), cdcEventBytes);
-
if (tupleProjector != null) {
result.add(indexCell);
IndexUtil.addTupleAsOneCell(result, new ResultTuple(cdcRow), tupleProjector, ptr);
} else {
result.clear();
}
- LOGGER.debug(
- "Processed CDC event with embedded data, skipped data table scan for" + " row key: {}",
- Bytes.toStringBinary(indexRowKey));
- return true;
}
+
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index c8ee55138c..62547021fb 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -39,6 +39,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DISABLE_WAL_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ENCODING_SCHEME_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.EXTERNAL_SCHEMA_ID_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_ROWS_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_CONSISTENCY_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_STATE_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_TYPE_BYTES;
@@ -261,8 +262,10 @@ import org.apache.phoenix.schema.metrics.MetricsMetadataSource;
import org.apache.phoenix.schema.metrics.MetricsMetadataSourceFactory;
import org.apache.phoenix.schema.task.ServerTask;
import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.types.IndexConsistency;
import org.apache.phoenix.schema.types.PBinary;
import org.apache.phoenix.schema.types.PBoolean;
+import org.apache.phoenix.schema.types.PChar;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PInteger;
import org.apache.phoenix.schema.types.PLong;
@@ -416,6 +419,8 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, CDC_INCLUDE_BYTES);
private static final Cell INDEX_WHERE_KV =
createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, INDEX_WHERE_BYTES);
+ private static final Cell INDEX_CONSISTENCY_KV =
+ createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, INDEX_CONSISTENCY_BYTES);
private static final Cell TTL_KV =
createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, TTL_BYTES);
@@ -434,7 +439,7 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
AUTO_PARTITION_SEQ_KV, APPEND_ONLY_SCHEMA_KV, STORAGE_SCHEME_KV, ENCODING_SCHEME_KV,
USE_STATS_FOR_PARALLELIZATION_KV, LAST_DDL_TIMESTAMP_KV, CHANGE_DETECTION_ENABLED_KV,
SCHEMA_VERSION_KV, EXTERNAL_SCHEMA_ID_KV, STREAMING_TOPIC_NAME_KV, INDEX_WHERE_KV,
- CDC_INCLUDE_KV, TTL_KV, ROW_KEY_MATCHER_KV, IS_STRICT_TTL_KV);
+ CDC_INCLUDE_KV, TTL_KV, ROW_KEY_MATCHER_KV, IS_STRICT_TTL_KV, INDEX_CONSISTENCY_KV);
static {
Collections.sort(TABLE_KV_COLUMNS, CellComparatorImpl.COMPARATOR);
@@ -496,6 +501,7 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
private static final int TTL_INDEX = TABLE_KV_COLUMNS.indexOf(TTL_KV);
private static final int ROW_KEY_MATCHER_INDEX = TABLE_KV_COLUMNS.indexOf(ROW_KEY_MATCHER_KV);
private static final int IS_STRICT_TTL_INDEX = TABLE_KV_COLUMNS.indexOf(IS_STRICT_TTL_KV);
+ private static final int INDEX_CONSISTENCY_INDEX = TABLE_KV_COLUMNS.indexOf(INDEX_CONSISTENCY_KV);
// KeyValues for Column
private static final KeyValue DECIMAL_DIGITS_KV =
createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, DECIMAL_DIGITS_BYTES);
@@ -1584,6 +1590,17 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
builder.setIsStrictTTL(
isStrictTTLKv != null ? isStrictTTL : oldTable == null || oldTable.isStrictTTL());
+ Cell indexConsistencyKv = tableKeyValues[INDEX_CONSISTENCY_INDEX];
+ IndexConsistency indexConsistency = null;
+ if (indexConsistencyKv != null) {
+ String consistencyStr = (String) PChar.INSTANCE.toObject(indexConsistencyKv.getValueArray(),
+ indexConsistencyKv.getValueOffset(), indexConsistencyKv.getValueLength());
+ indexConsistency = IndexConsistency.fromCode(consistencyStr);
+ }
+ builder.setIndexConsistency(indexConsistency != null ? indexConsistency
+ : oldTable != null ? oldTable.getIndexConsistency()
+ : null);
+
// Check the cell tag to see whether the view has modified this property
final byte[] tagUseStatsForParallelization = (useStatsForParallelizationKv == null)
? HConstants.EMPTY_BYTE_ARRAY
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredIndexRegionScanner.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredIndexRegionScanner.java
index 1b126a6591..56ec3cd6da 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredIndexRegionScanner.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredIndexRegionScanner.java
@@ -63,6 +63,7 @@ import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.PhoenixKeyValueUtil;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.ServerUtil;
import org.slf4j.Logger;
@@ -271,7 +272,21 @@ public abstract class UncoveredIndexRegionScanner extends BaseRegionScanner {
indexToDataRowKeyMap.put(offset == 0 ? lastIndexRowKey : CellUtil.cloneRow(firstCell),
indexMaintainer.buildDataRowKey(new ImmutableBytesWritable(lastIndexRowKey),
viewConstants));
- indexRows.add(row);
+ // When a row has at most one cell, we can safely reuse it as we don't look at cell value,
+ // and we proceed with data table scan.
+ // For rows with multiple cells, we must copy each cell to avoid issues where the
+ // underlying byte buffers for cell values may be reused or invalidated by the scanner
+ // on subsequent nextRaw() calls and hence the cell value could be garbage under
+ // high GC pressure.
+ if (row.size() <= 1) {
+ indexRows.add(row);
+ } else {
+ List<Cell> dupRow = new ArrayList<>(row.size());
+ for (Cell cell : row) {
+ dupRow.add(PhoenixKeyValueUtil.maybeCopyCell(cell));
+ }
+ indexRows.add(dupRow);
+ }
indexRowCount++;
if (
hasMore && (PhoenixScannerContext.isTimedOut(scannerContext, pageSizeMs)
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java
new file mode 100644
index 0000000000..6a8aa55108
--- /dev/null
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java
@@ -0,0 +1,857 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.sql.Date;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.ConnectionUtils;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.coprocessor.DelegateRegionCoprocessorEnvironment;
+import org.apache.phoenix.coprocessorclient.MetaDataProtocol;
+import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.hbase.index.write.IndexWriter;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.TableNotFoundException;
+import org.apache.phoenix.schema.types.IndexConsistency;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
+import org.bson.BsonArray;
+import org.bson.BsonDocument;
+import org.bson.BsonValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.ArrayListMultimap;
+import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
+
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
+
+/**
+ * A single-threaded background consumer that processes CDC mutations for eventually consistent
+ * indexes. This consumer reads mutations from the CDC index table and applies them to the
+ * appropriate secondary indexes.
+ * <p>
+ * The consumer tracks its progress in the SYSTEM.IDX_CDC_TRACKER table, allowing for proper
+ * handling of region splits and merges.
+ * </p>
+ * <p>
+ * <b>Lifecycle:</b>
+ * <ol>
+ * <li>On startup, the consumer first replays any remaining mutations from parent regions (if this
+ * region was created from a split or merge). A region will have multiple parent regions in case of
+ * region merge.</li>
+ * <li>After replaying all parent mutations, the consumer marks each parent region as COMPLETE in
+ * SYSTEM.IDX_CDC_TRACKER.</li>
+ * <li>The consumer then begins processing mutations for the current region. With every batch of
+ * rows consumed (e.g., 1000 rows), it updates LAST_TIMESTAMP in SYSTEM.IDX_CDC_TRACKER. The tracker
+ * entry with status IN_PROGRESS is created on the first batch update.</li>
+ * <li>When the region is closed (due to split, merge, or shutdown), the consumer stops but does NOT
+ * mark itself as COMPLETE. The COMPLETE status is only set by child regions after they have
+ * replayed all parent mutations.</li>
+ * </ol>
+ * </p>
+ */
+public class IndexCDCConsumer implements Runnable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IndexCDCConsumer.class);
+
+ public static final String INDEX_CDC_CONSUMER_BATCH_SIZE =
+ "phoenix.index.cdc.consumer.batch.size";
+ private static final int DEFAULT_CDC_BATCH_SIZE = 1000;
+
+ public static final String INDEX_CDC_CONSUMER_STARTUP_DELAY_MS =
+ "phoenix.index.cdc.consumer.startup.delay.ms";
+ private static final long DEFAULT_STARTUP_DELAY_MS = 10000;
+
+ /**
+ * The interval in milliseconds between processing batches when mutations are found.
+ */
+ public static final String INDEX_CDC_CONSUMER_POLL_INTERVAL_MS =
+ "phoenix.index.cdc.consumer.poll.interval.ms";
+ private static final long DEFAULT_POLL_INTERVAL_MS = 1000;
+
+ /**
+ * The time buffer in milliseconds subtracted from current time when querying CDC mutations. This
+ * buffer helps avoid reading mutations that are too recent.
+ */
+ public static final String INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS =
+ "phoenix.index.cdc.consumer.timestamp.buffer.ms";
+ private static final long DEFAULT_TIMESTAMP_BUFFER_MS = 1000;
+
+ private final RegionCoprocessorEnvironment env;
+ private final String dataTableName;
+ private final String encodedRegionName;
+ private final IndexWriter indexWriter;
+ private final long pause;
+ private final long startupDelayMs;
+ private final int batchSize;
+ private final long pollIntervalMs;
+ private final long timestampBufferMs;
+ private final Configuration config;
+ private volatile boolean stopped = false;
+ private Thread consumerThread;
+ private boolean hasParentPartitions = false;
+ private PTable cachedDataTable;
+
+ /**
+ * Creates a new IndexCDCConsumer for the given region.
+ * @param env region coprocessor environment.
+ * @param dataTableName name of the data table.
+ * @param serverName server name.
+ * @throws IOException if the IndexWriter cannot be created.
+ */
+ public IndexCDCConsumer(RegionCoprocessorEnvironment env, String dataTableName, String serverName)
+ throws IOException {
+ this.env = env;
+ this.dataTableName = dataTableName;
+ this.encodedRegionName = env.getRegion().getRegionInfo().getEncodedName();
+ this.config = env.getConfiguration();
+ this.pause = config.getLong(HConstants.HBASE_CLIENT_PAUSE, 300);
+ this.startupDelayMs =
+ config.getLong(INDEX_CDC_CONSUMER_STARTUP_DELAY_MS, DEFAULT_STARTUP_DELAY_MS);
+ this.batchSize = config.getInt(INDEX_CDC_CONSUMER_BATCH_SIZE, DEFAULT_CDC_BATCH_SIZE);
+ this.pollIntervalMs =
+ config.getLong(INDEX_CDC_CONSUMER_POLL_INTERVAL_MS, DEFAULT_POLL_INTERVAL_MS);
+ this.timestampBufferMs =
+ config.getLong(INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS, DEFAULT_TIMESTAMP_BUFFER_MS);
+ DelegateRegionCoprocessorEnvironment indexWriterEnv =
+ new DelegateRegionCoprocessorEnvironment(env, ConnectionType.INDEX_WRITER_CONNECTION);
+ this.indexWriter =
+ new IndexWriter(indexWriterEnv, serverName + "-index-eventual-writer", false);
+ }
+
+ /**
+ * Starts the consumer thread in the background.
+ */
+ public void start() {
+ consumerThread =
+ new Thread(this, "IndexCDCConsumer-" + dataTableName + "-" + encodedRegionName);
+ consumerThread.setDaemon(true);
+ consumerThread.start();
+ }
+
+ /**
+ * Stops the consumer thread gracefully.
+ */
+ public void stop() {
+ stopped = true;
+ if (consumerThread != null) {
+ consumerThread.interrupt();
+ }
+ if (indexWriter != null) {
+ indexWriter.stop("IndexCDCConsumer stopped for " + dataTableName);
+ }
+ LOG.info("Stopped IndexCDCConsumer for table {} region {}", dataTableName, encodedRegionName);
+ }
+
+ /**
+ * Sleeps for the specified duration if the consumer has not been stopped.
+ * @param millis the duration to sleep in milliseconds.
+ * @throws InterruptedException if the thread is interrupted while sleeping.
+ */
+ private void sleepIfNotStopped(long millis) throws InterruptedException {
+ if (!stopped) {
+ Thread.sleep(millis);
+ }
+ }
+
+ private PTable getDataTable(PhoenixConnection conn) throws SQLException {
+ PTable dataTable = cachedDataTable;
+ if (dataTable == null) {
+ dataTable = conn.getTable(dataTableName);
+ cachedDataTable = dataTable;
+ }
+ return dataTable;
+ }
+
+ private void refreshDataTableCache(PhoenixConnection conn) throws SQLException {
+ cachedDataTable = conn.getTable(dataTableName);
+ }
+
+ @Override
+ public void run() {
+ try {
+ if (startupDelayMs > 0 && getCDCStreamNumPartitions() <= 1) {
+ sleepIfNotStopped(startupDelayMs);
+ }
+ if (stopped) {
+ return;
+ }
+ if (!hasEventuallyConsistentIndexes()) {
+ LOG.trace("No eventually consistent indexes found for table {}. Exiting consumer.",
+ dataTableName);
+ return;
+ }
+ LOG.info("IndexCDCConsumer started for table {} region {}", dataTableName, encodedRegionName);
+ if (!waitForCDCStreamEntry()) {
+ LOG.error(
+ "IndexCDCConsumer stopped while waiting for CDC_STREAM entry for table {} region {}",
+ dataTableName, encodedRegionName);
+ return;
+ }
+ long lastProcessedTimestamp = checkTrackerStatus(encodedRegionName, encodedRegionName);
+ if (lastProcessedTimestamp == -1) {
+ // should never happen as COMPLETE is only set by child regions for their parent regions
+ LOG.error(
+ "Unexpected COMPLETE status in IDX_CDC_TRACKER for current region. "
+ + "Table: {}, Partition: {}. Exiting consumer thread.",
+ dataTableName, encodedRegionName);
+ return;
+ } else if (lastProcessedTimestamp > 0) {
+ LOG.info(
+ "Found existing tracker entry for table {} region {} with lastTimestamp {}. "
+ + "Resuming from last position (region movement scenario).",
+ dataTableName, encodedRegionName, lastProcessedTimestamp);
+ } else {
+ if (hasParentPartitions) {
+ replayAndCompleteParentRegions(encodedRegionName);
+ } else {
+ LOG.info("No parent partitions for table {} region {}, skipping parent replay",
+ dataTableName, encodedRegionName);
+ }
+ }
+ int retryCount = 0;
+ while (!stopped) {
+ try {
+ long previousTimestamp = lastProcessedTimestamp;
+ lastProcessedTimestamp =
+ processCDCBatch(encodedRegionName, encodedRegionName, lastProcessedTimestamp);
+ if (lastProcessedTimestamp == previousTimestamp) {
+ sleepIfNotStopped(ConnectionUtils.getPauseTime(pause, ++retryCount));
+ } else {
+ retryCount = 0;
+ sleepIfNotStopped(pollIntervalMs);
+ }
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ throw (InterruptedException) e;
+ }
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.error(
+ "Error processing CDC mutations for table {} region {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ dataTableName, encodedRegionName, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ } catch (InterruptedException e) {
+ if (!stopped) {
+ LOG.warn("IndexCDCConsumer interrupted unexpectedly for table {} region {}", dataTableName,
+ encodedRegionName, e);
+ }
+ Thread.currentThread().interrupt();
+ }
+ LOG.info("IndexCDCConsumer exiting for table {} region {}", dataTableName, encodedRegionName);
+ }
+
+ private boolean hasEventuallyConsistentIndexes() throws InterruptedException {
+ int retryCount = 0;
+ while (!stopped) {
+ try (PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(config).unwrap(PhoenixConnection.class)) {
+ refreshDataTableCache(conn);
+ PTable dataTable = getDataTable(conn);
+ String cdcObjectName = CDCUtil.getCDCObjectName(dataTable, false);
+ if (cdcObjectName == null) {
+ LOG.debug("No CDC index found for table {}. Exiting consumer.", dataTableName);
+ return false;
+ }
+ for (PTable index : dataTable.getIndexes()) {
+ IndexConsistency consistency = index.getIndexConsistency();
+ if (consistency != null && consistency.isAsynchronous()) {
+ LOG.debug("Found eventually consistent index {} for table {}",
+ index.getName().getString(), dataTableName);
+ return true;
+ }
+ }
+ return false;
+ } catch (SQLException e) {
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error checking for eventually consistent indexes for table {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ dataTableName, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Retrieves the count of partitions for the given table.
+ * @return the count of CDC_STREAM rows for this table.
+ * @throws InterruptedException if the thread is interrupted while waiting.
+ */
+ private long getCDCStreamNumPartitions() throws InterruptedException {
+ int retryCount = 0;
+ String query = "SELECT COUNT(*) FROM " + PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME
+ + " WHERE TABLE_NAME = ?";
+ while (!stopped) {
+ try (
+ PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(config).unwrap(PhoenixConnection.class);
+ PreparedStatement ps = conn.prepareStatement(query)) {
+ ps.setString(1, dataTableName);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (rs.next()) {
+ return rs.getLong(1);
+ }
+ }
+ return 0;
+ } catch (SQLException e) {
+ if (e instanceof TableNotFoundException) {
+ TableNotFoundException tnfe = (TableNotFoundException) e;
+ // 5.3.0+ server with old metadata tables (i.e. EXECUTE_UPGRADE is not yet run)
+ if (PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_TABLE.equals(tnfe.getTableName())) {
+ stopped = true;
+ return -1;
+ }
+ }
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error getting CDC_STREAM row count for table {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ dataTableName, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ return -1;
+ }
+
+ /**
+ * Waits for the CDC_STREAM entry for the given table and partition to be available.
+ * @return true if the entry was found, false if the consumer was stopped before finding the
+ * entry.
+ * @throws InterruptedException if the thread is interrupted while waiting.
+ */
+ private boolean waitForCDCStreamEntry() throws InterruptedException {
+ int retryCount = 0;
+ String query =
+ "SELECT PARENT_PARTITION_ID FROM " + PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME
+ + " WHERE TABLE_NAME = ? AND PARTITION_ID = ? LIMIT 1";
+ while (!stopped) {
+ try (
+ PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ PreparedStatement ps = conn.prepareStatement(query)) {
+ ps.setString(1, dataTableName);
+ ps.setString(2, encodedRegionName);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (rs.next()) {
+ String parentPartitionId = rs.getString(1);
+ hasParentPartitions = parentPartitionId != null;
+ LOG.debug("Found CDC_STREAM entry for table {} partition {}, hasParentPartitions={}",
+ dataTableName, encodedRegionName, hasParentPartitions);
+ return true;
+ }
+ }
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.info(
+ "CDC_STREAM entry not found for table {} partition {}. "
+ + "Attempt #{}, sleeping {} ms before retrying...",
+ dataTableName, encodedRegionName, retryCount, sleepTime);
+ sleepIfNotStopped(sleepTime);
+ } catch (SQLException e) {
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error checking CDC_STREAM for table {} partition {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ dataTableName, encodedRegionName, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Checks for an existing entry for the given partition and owner.
+ * @param partitionId the partition ID to check.
+ * @param ownerPartitionId the owner partition ID.
+ * @return the last processed timestamp if an entry exists with status IN_PROGRESS, -1 if the
+ * entry exists with status COMPLETE, or 0 if no entry exists.
+ * @throws InterruptedException if the thread is interrupted while waiting
+ */
+ private long checkTrackerStatus(String partitionId, String ownerPartitionId)
+ throws InterruptedException {
+ int retryCount = 0;
+ String query =
+ "SELECT LAST_TIMESTAMP, STATUS FROM " + PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_NAME
+ + " WHERE TABLE_NAME = ? AND PARTITION_ID = ? AND OWNER_PARTITION_ID = ?";
+ while (!stopped) {
+ try (
+ PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ PreparedStatement ps = conn.prepareStatement(query)) {
+ ps.setString(1, dataTableName);
+ ps.setString(2, partitionId);
+ ps.setString(3, ownerPartitionId);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (rs.next()) {
+ long lastTimestamp = rs.getLong(1);
+ String status = rs.getString(2);
+ LOG.debug(
+ "Found IDX_CDC_TRACKER entry for table {} partition {} owner {} "
+ + "with status={}, lastTimestamp={}",
+ dataTableName, partitionId, ownerPartitionId, status, lastTimestamp);
+ if (PhoenixDatabaseMetaData.TRACKER_STATUS_IN_PROGRESS.equals(status)) {
+ return lastTimestamp;
+ } else if (PhoenixDatabaseMetaData.TRACKER_STATUS_COMPLETE.equals(status)) {
+ return -1;
+ } else {
+ // Unknown status - should not happen
+ throw new IllegalStateException(
+ String.format("Unknown tracker status '%s' for table %s partition %s owner %s.",
+ status, dataTableName, partitionId, ownerPartitionId));
+ }
+ } else {
+ LOG.debug("No IDX_CDC_TRACKER entry found for table {} partition {} owner {}",
+ dataTableName, partitionId, ownerPartitionId);
+ return 0;
+ }
+ }
+ } catch (SQLException e) {
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error checking IDX_CDC_TRACKER for table {} partition {} owner {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ dataTableName, partitionId, ownerPartitionId, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ return 0;
+ }
+
+ /**
+ * Checks if any consumer has already completed processing the given partition.
+ * @param partitionId the partition ID to check
+ * @return true if the partition is done processing.
+ * @throws InterruptedException if the thread is interrupted while waiting.
+ */
+ private boolean isPartitionCompleted(String partitionId) throws InterruptedException {
+ int retryCount = 0;
+ String query = "SELECT * FROM " + PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_NAME
+ + " WHERE TABLE_NAME = ? AND PARTITION_ID = ? AND STATUS = ?";
+ while (!stopped) {
+ try (
+ PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ PreparedStatement ps = conn.prepareStatement(query)) {
+ ps.setString(1, dataTableName);
+ ps.setString(2, partitionId);
+ ps.setString(3, PhoenixDatabaseMetaData.TRACKER_STATUS_COMPLETE);
+ try (ResultSet rs = ps.executeQuery()) {
+ return rs.next();
+ }
+ } catch (SQLException e) {
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error checking if partition {} is completed for table {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ partitionId, dataTableName, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Gets the maximum last processed timestamp from any consumer's tracker record for the given
+ * partition. This is used when a child region needs to continue processing from where any
+ * previous consumer left off.
+ * @param partitionId the partition ID to get progress for.
+ * @return the max last processed timestamp across all owners, or 0 if not found.
+ * @throws InterruptedException if the thread is interrupted while waiting.
+ */
+ private long getParentProgress(String partitionId) throws InterruptedException {
+ int retryCount = 0;
+ String query =
+ "SELECT MAX(LAST_TIMESTAMP) FROM " + PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_NAME
+ + " WHERE TABLE_NAME = ? AND PARTITION_ID = ? AND STATUS = ?";
+ while (!stopped) {
+ try (
+ PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ PreparedStatement ps = conn.prepareStatement(query)) {
+ ps.setString(1, dataTableName);
+ ps.setString(2, partitionId);
+ ps.setString(3, PhoenixDatabaseMetaData.TRACKER_STATUS_IN_PROGRESS);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (rs.next()) {
+ long maxTimestamp = rs.getLong(1);
+ if (maxTimestamp > 0) {
+ LOG.debug("Found max progress {} for partition {} from previous consumers",
+ maxTimestamp, partitionId);
+ return maxTimestamp;
+ }
+ }
+ return 0;
+ }
+ } catch (SQLException e) {
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error getting parent progress for partition {} table {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ partitionId, dataTableName, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ throw new InterruptedException("IndexCDCConsumer stopped while getting parent progress.");
+ }
+
+ /**
+ * Retrieves all parent partition IDs for the given partition.
+ * @param partitionId the partition ID to find parents for.
+ * @return list of parent partition IDs.
+ * @throws InterruptedException if the thread is interrupted.
+ */
+ private List<String> getParentPartitionIds(String partitionId) throws InterruptedException {
+ int retryCount = 0;
+ String query =
+ "SELECT PARENT_PARTITION_ID FROM " + PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME
+ + " WHERE TABLE_NAME = ? AND PARTITION_ID = ? AND PARENT_PARTITION_ID IS NOT NULL";
+ while (!stopped) {
+ try (
+ PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ PreparedStatement ps = conn.prepareStatement(query)) {
+ ps.setString(1, dataTableName);
+ ps.setString(2, partitionId);
+ List<String> parentIds = new ArrayList<>();
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ parentIds.add(rs.getString(1));
+ }
+ }
+ LOG.debug("Found {} parent partition(s) for table {} partition {}: {}", parentIds.size(),
+ dataTableName, partitionId, parentIds);
+ return parentIds;
+ } catch (SQLException e) {
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error querying parent partitions from CDC_STREAM for table {} partition {}. "
+ + "Retry #{}, sleeping {} ms before retrying...",
+ dataTableName, partitionId, retryCount, sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ return Collections.emptyList();
+ }
+
+ /**
+ * Processes all remaining CDC mutations for the given partition until completion, then marks the
+ * partition as COMPLETE. Used to replay ancestor partitions (regions) after they are split or
+ * merged into new regions.
+ * @param partitionId the partition ID to process.
+ * @param ownerPartitionId the owner partition ID.
+ * @param lastProcessedTimestamp the timestamp to start processing from.
+ * @throws InterruptedException if the thread is interrupted.
+ */
+ private void processPartitionToCompletion(String partitionId, String ownerPartitionId,
+ long lastProcessedTimestamp) throws InterruptedException {
+ LOG.debug(
+ "Processing partition {} owner {} to completion for table {}, starting from timestamp {}",
+ partitionId, ownerPartitionId, dataTableName, lastProcessedTimestamp);
+ long currentLastProcessedTimestamp = lastProcessedTimestamp;
+ int retryCount = 0;
+ int batchCount = 0;
+ while (!stopped) {
+ try {
+ if (batchCount > 0 && batchCount % 5 == 0) {
+ if (isPartitionCompleted(partitionId)) {
+ return;
+ }
+ long otherProgress = getParentProgress(partitionId);
+ if (otherProgress > currentLastProcessedTimestamp) {
+ // other owner has already made some progress, pause for a while before resuming
+ sleepIfNotStopped(10000);
+ if (isPartitionCompleted(partitionId)) {
+ return;
+ }
+ currentLastProcessedTimestamp = getParentProgress(partitionId);
+ }
+ }
+ long newTimestamp =
+ processCDCBatch(partitionId, ownerPartitionId, currentLastProcessedTimestamp);
+ batchCount++;
+ retryCount = 0;
+ if (newTimestamp == currentLastProcessedTimestamp) {
+ if (isPartitionCompleted(partitionId)) {
+ LOG.info(
+ "Partition {} for table {} was completed by another consumer before {} could mark it",
+ partitionId, dataTableName, ownerPartitionId);
+ return;
+ }
+ LOG.info("Partition {} owner {} for table {} fully processed, marking as COMPLETE",
+ partitionId, ownerPartitionId, dataTableName);
+ try (PhoenixConnection conn = QueryUtil.getConnectionOnServer(env.getConfiguration())
+ .unwrap(PhoenixConnection.class)) {
+ updateTrackerProgress(conn, partitionId, ownerPartitionId,
+ currentLastProcessedTimestamp, PhoenixDatabaseMetaData.TRACKER_STATUS_COMPLETE);
+ }
+ return;
+ }
+ currentLastProcessedTimestamp = newTimestamp;
+ } catch (SQLException | IOException e) {
+ long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount);
+ LOG.warn(
+ "Error processing CDC batch for partition {} owner {} table {} "
+ + "lastProcessedTimestamp {}. Retry #{}, sleeping {} ms",
+ partitionId, ownerPartitionId, dataTableName, currentLastProcessedTimestamp, retryCount,
+ sleepTime, e);
+ sleepIfNotStopped(sleepTime);
+ }
+ }
+ LOG.info("Processing partition {} (owner {}) stopped before completion for table {}",
+ partitionId, ownerPartitionId, dataTableName);
+ }
+
+ /**
+ * Recursive helper method to replay and complete parent regions. Replays all remaining mutations
+ * from parent regions and marks them as COMPLETE. This is called during initialization when a
+ * region has parent regions (split/merge scenarios).
+ * @param partitionId the partition (region) ID to find and process parents for.
+ */
+ private void replayAndCompleteParentRegions(String partitionId) throws InterruptedException {
+ List<String> parentIds = getParentPartitionIds(partitionId);
+ if (parentIds.isEmpty()) {
+ LOG.debug("No parent partitions found for table {} partition {}", dataTableName, partitionId);
+ return;
+ }
+ for (String parentId : parentIds) {
+ if (stopped) {
+ return;
+ }
+ if (isPartitionCompleted(parentId)) {
+ LOG.debug("Parent partition {} for table {} already has a COMPLETE record, skipping",
+ parentId, dataTableName);
+ continue;
+ }
+ replayAndCompleteParentRegions(parentId);
+ if (isPartitionCompleted(parentId)) {
+ LOG.debug(
+ "Parent partition {} for table {} was completed by sibling during ancestor processing, "
+ + "skipping",
+ parentId, dataTableName);
+ continue;
+ }
+ long parentProgress = getParentProgress(parentId);
+ LOG.debug("Processing/Resuming parent partition {} for table {} from timestamp {} owner: {}",
+ parentId, dataTableName, parentProgress, encodedRegionName);
+ processPartitionToCompletion(parentId, encodedRegionName, parentProgress);
+ }
+ }
+
+ /**
+ * Processes a batch of CDC mutations for the given partition starting from the specified
+ * timestamp. This method reads mutations from the CDC index, applies them to the respective
+ * eventually consistent index tables, and updates the progress in SYSTEM.IDX_CDC_TRACKER.
+ * @param partitionId the partition (region) ID to process CDC mutations for.
+ * @param ownerPartitionId the owner partition ID.
+ * @param lastProcessedTimestamp the timestamp to start reading CDC mutations from.
+ * @return the new last processed timestamp after this batch, or the same timestamp if no new
+ * records were found.
+ * @throws SQLException if SQL error occurs.
+ * @throws IOException if an I/O error occurs.
+ * @throws InterruptedException if the thread is interrupted while waiting.
+ */
+ private long processCDCBatch(String partitionId, String ownerPartitionId,
+ long lastProcessedTimestamp) throws SQLException, IOException, InterruptedException {
+ LOG.debug("Processing CDC batch for table {} partition {} owner {} from timestamp {}",
+ dataTableName, partitionId, ownerPartitionId, lastProcessedTimestamp);
+ try (PhoenixConnection conn =
+ QueryUtil.getConnectionOnServer(config).unwrap(PhoenixConnection.class)) {
+ PTable dataTable = getDataTable(conn);
+ String cdcObjectName = CDCUtil.getCDCObjectName(dataTable, false);
+ if (cdcObjectName == null) {
+ throw new SQLException("No CDC object found for table " + dataTableName);
+ }
+ String schemaName = dataTable.getSchemaName().getString();
+ if (schemaName == null || schemaName.isEmpty()) {
+ cdcObjectName = "\"" + cdcObjectName + "\"";
+ } else {
+ cdcObjectName =
+ "\"" + schemaName + "\"" + QueryConstants.NAME_SEPARATOR + "\"" + cdcObjectName + "\"";
+ }
+ String cdcQuery = String
+ .format("SELECT /*+ CDC_INCLUDE(IDX_MUTATIONS) */ PHOENIX_ROW_TIMESTAMP(), \"CDC JSON\" "
+ + "FROM %s WHERE PARTITION_ID() = ? AND PHOENIX_ROW_TIMESTAMP() > ? "
+ + "AND PHOENIX_ROW_TIMESTAMP() < ? "
+ + "ORDER BY PARTITION_ID() ASC, PHOENIX_ROW_TIMESTAMP() ASC LIMIT ?", cdcObjectName);
+ List<Pair<Long, BsonDocument>> batchMutations = new ArrayList<>();
+ long newLastTimestamp = lastProcessedTimestamp;
+ boolean hasMoreRows = true;
+ int retryCount = 0;
+ while (hasMoreRows && batchMutations.isEmpty()) {
+ long currentTime = EnvironmentEdgeManager.currentTimeMillis() - timestampBufferMs;
+ try (PreparedStatement ps = conn.prepareStatement(cdcQuery)) {
+ ps.setString(1, partitionId);
+ ps.setDate(2, new Date(newLastTimestamp));
+ ps.setDate(3, new Date(currentTime));
+ ps.setInt(4, batchSize);
+ Pair<Long, Boolean> result =
+ getMutationsAndTimestamp(ps, newLastTimestamp, batchMutations);
+ hasMoreRows = result.getSecond();
+ if (hasMoreRows) {
+ newLastTimestamp = result.getFirst();
+ if (batchMutations.isEmpty()) {
+ sleepIfNotStopped(ConnectionUtils.getPauseTime(pause, ++retryCount));
+ }
+ }
+ }
+ }
+ // With predefined LIMIT, there might be more rows with the same timestamp that were not
+ // included in this batch.
+ if (newLastTimestamp > lastProcessedTimestamp) {
+ String sameTimestampQuery = String
+ .format("SELECT /*+ CDC_INCLUDE(IDX_MUTATIONS) */ PHOENIX_ROW_TIMESTAMP(), \"CDC JSON\" "
+ + "FROM %s WHERE PARTITION_ID() = ? AND PHOENIX_ROW_TIMESTAMP() = ? "
+ + "ORDER BY PARTITION_ID() ASC, PHOENIX_ROW_TIMESTAMP() ASC", cdcObjectName);
+ final long timestampToRefetch = newLastTimestamp;
+ batchMutations.removeIf(pair -> pair.getFirst() == timestampToRefetch);
+ try (PreparedStatement ps = conn.prepareStatement(sameTimestampQuery)) {
+ ps.setString(1, partitionId);
+ ps.setDate(2, new Date(newLastTimestamp));
+ Pair<Long, Boolean> result =
+ getMutationsAndTimestamp(ps, newLastTimestamp, batchMutations);
+ newLastTimestamp = result.getFirst();
+ if (newLastTimestamp != timestampToRefetch) {
+ throw new IOException("Unexpected timestamp mismatch: expected " + timestampToRefetch
+ + " but got " + newLastTimestamp);
+ }
+ }
+ }
+ executeIndexMutations(partitionId, batchMutations, ownerPartitionId, newLastTimestamp);
+ if (!batchMutations.isEmpty()) {
+ updateTrackerProgress(conn, partitionId, ownerPartitionId, newLastTimestamp,
+ PhoenixDatabaseMetaData.TRACKER_STATUS_IN_PROGRESS);
+ }
+ return newLastTimestamp;
+ }
+ }
+
+ private void executeIndexMutations(String partitionId,
+ List<Pair<Long, BsonDocument>> batchMutations, String ownerPartitionId,
+ long lastProcessedTimestamp) throws SQLException, IOException {
+ if (!batchMutations.isEmpty()) {
+ ListMultimap<HTableInterfaceReference, Mutation> indexUpdates = ArrayListMultimap.create();
+ Map<ImmutableBytesPtr, HTableInterfaceReference> tableRefCache = new HashMap<>();
+ int totalMutations = 0;
+ for (Pair<Long, BsonDocument> batchMutation : batchMutations) {
+ BsonDocument mutationsDoc = batchMutation.getSecond();
+ BsonArray tables = mutationsDoc.getArray("tables");
+ BsonArray mutations = mutationsDoc.getArray("mutations");
+ if (tables.size() != mutations.size()) {
+ throw new SQLException("Tables and mutations sizes do not match. Tables size: "
+ + tables.size() + ", mutations size: " + mutations.size());
+ }
+ for (int i = 0; i < tables.size(); i++) {
+ BsonValue tableValue = tables.get(i);
+ BsonValue mutationValue = mutations.get(i);
+ byte[] indexTableName = tableValue.asBinary().getData();
+ byte[] mutationBytes = mutationValue.asBinary().getData();
+ ImmutableBytesPtr tableNamePtr = new ImmutableBytesPtr(indexTableName);
+ HTableInterfaceReference tableRef =
+ tableRefCache.computeIfAbsent(tableNamePtr, HTableInterfaceReference::new);
+ ClientProtos.MutationProto mProto = ClientProtos.MutationProto.parseFrom(mutationBytes);
+ Mutation mutation = ProtobufUtil.toMutation(mProto);
+ indexUpdates.put(tableRef, mutation);
+ }
+ if (indexUpdates.size() >= batchSize) {
+ indexWriter.write(indexUpdates, false, MetaDataProtocol.PHOENIX_VERSION);
+ totalMutations += indexUpdates.size();
+ indexUpdates.clear();
+ }
+ }
+ if (!indexUpdates.isEmpty()) {
+ indexWriter.write(indexUpdates, false, MetaDataProtocol.PHOENIX_VERSION);
+ totalMutations += indexUpdates.size();
+ }
+ if (totalMutations > 0) {
+ LOG.debug(
+ "Applied total {} index mutations for table {} partition {} owner {} "
+ + ", last processed timestamp {}",
+ totalMutations, dataTableName, partitionId, ownerPartitionId, lastProcessedTimestamp);
+ }
+ }
+ }
+
+ private void updateTrackerProgress(PhoenixConnection conn, String partitionId,
+ String ownerPartitionId, long lastTimestamp, String status) throws SQLException {
+ String upsertSql = "UPSERT INTO " + PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_NAME
+ + " (TABLE_NAME, PARTITION_ID, OWNER_PARTITION_ID, LAST_TIMESTAMP, STATUS) "
+ + "VALUES (?, ?, ?, ?, ?)";
+ try (PreparedStatement ps = conn.prepareStatement(upsertSql)) {
+ ps.setString(1, dataTableName);
+ ps.setString(2, partitionId);
+ ps.setString(3, ownerPartitionId);
+ ps.setLong(4, lastTimestamp);
+ ps.setString(5, status);
+ ps.executeUpdate();
+ conn.commit();
+ LOG.debug("Updated tracker for table {} partition {} owner {} to timestamp {} with status {}",
+ dataTableName, partitionId, ownerPartitionId, lastTimestamp, status);
+ }
+ }
+
+ /**
+ * Executes the prepared statement and extracts mutations and timestamps from the result set.
+ * @param ps the prepared statement to execute.
+ * @param initialLastTimestamp the initial last timestamp to use if no rows are found.
+ * @param batchMutations list to add mutations to.
+ * @return Pair of last processed timestamp and boolean to indicate if any rows were returned.
+ */
+ private static Pair<Long, Boolean> getMutationsAndTimestamp(PreparedStatement ps,
+ long initialLastTimestamp, List<Pair<Long, BsonDocument>> batchMutations) throws SQLException {
+ boolean hasRows = false;
+ long lastTimestamp = initialLastTimestamp;
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ hasRows = true;
+ lastTimestamp = rs.getDate(1).getTime();
+ String cdcJson = rs.getString(2);
+ if (cdcJson != null && !cdcJson.isEmpty()) {
+ BsonDocument mutationsDoc = BsonDocument.parse(cdcJson);
+ if (mutationsDoc.containsKey("tables") && mutationsDoc.containsKey("mutations")) {
+ batchMutations.add(Pair.newPair(lastTimestamp, mutationsDoc));
+ }
+ }
+ }
+ }
+ return Pair.newPair(lastTimestamp, hasRows);
+ }
+
+}
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 fc55b2435d..163b3ad881 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
@@ -127,6 +127,7 @@ import org.apache.phoenix.schema.TTLExpressionFactory;
import org.apache.phoenix.schema.transform.TransformMaintainer;
import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
import org.apache.phoenix.schema.types.PBoolean;
+import org.apache.phoenix.schema.types.PBson;
import org.apache.phoenix.schema.types.PInteger;
import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.trace.TracingUtils;
@@ -141,6 +142,9 @@ import org.apache.phoenix.util.PhoenixKeyValueUtil;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.ServerIndexUtil;
import org.apache.phoenix.util.ServerUtil.ConnectionType;
+import org.bson.BsonArray;
+import org.bson.BsonBinary;
+import org.bson.BsonDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -151,6 +155,9 @@ import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
import org.apache.phoenix.thirdparty.com.google.common.collect.Sets;
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
+
/**
* Do all the work of managing index updates from a single coprocessor. All Puts/Delets are passed
* to an {@link IndexBuilder} to determine the actual updates to make. We don't need to implement
@@ -166,6 +173,9 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
private static final OperationStatus NOWRITE = new OperationStatus(SUCCESS);
public static final String PHOENIX_APPEND_METADATA_TO_WAL = "phoenix.append.metadata.to.wal";
public static final boolean DEFAULT_PHOENIX_APPEND_METADATA_TO_WAL = false;
+ public static final String PHOENIX_INDEX_CDC_CONSUMER_ENABLED =
+ "phoenix.index.cdc.consumer.enabled";
+ public static final boolean DEFAULT_PHOENIX_INDEX_CDC_CONSUMER_ENABLED = true;
/**
* Class to represent pending data table rows
@@ -274,6 +284,13 @@ 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;
+ // Map of data table row key to BsonDocument containing pre-phase mutations for eventually
+ // consistent indexes (index mutations with UNVERIFIED Puts only, no Deletes)
+ private Map<ImmutableBytesPtr, BsonDocument> cdcPreMutationsDoc;
+ // Map of data table row key to BsonDocument containing post-phase mutations for eventually
+ // consistent indexes (index mutations with VERIFIED Puts for covered,
+ // no Put mutations for uncovered, and Deletes if needed)
+ private Map<ImmutableBytesPtr, BsonDocument> cdcPostMutationsDoc;
private List<RowLock> rowLocks =
Lists.newArrayListWithExpectedSize(QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE);
// TreeSet to improve locking efficiency and avoid deadlock (PHOENIX-6871 and HBASE-17924)
@@ -410,10 +427,12 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
private int concurrentMutationWaitDuration;
private String dataTableName;
private boolean shouldWALAppend = DEFAULT_PHOENIX_APPEND_METADATA_TO_WAL;
+ private boolean indexCDCConsumerEnabled = DEFAULT_PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
private boolean isNamespaceEnabled = false;
private boolean useBloomFilter = false;
private long lastTimestamp = 0;
private List<Set<ImmutableBytesPtr>> batchesWithLastTimestamp = new ArrayList<>();
+ private IndexCDCConsumer indexCDCConsumer;
private static final int DEFAULT_ROWLOCK_WAIT_DURATION = 30000;
private static final int DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS = 100;
private byte[] encodedRegionName;
@@ -465,12 +484,21 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
this.dataTableName = env.getRegionInfo().getTable().getNameAsString();
this.shouldWALAppend = env.getConfiguration().getBoolean(PHOENIX_APPEND_METADATA_TO_WAL,
DEFAULT_PHOENIX_APPEND_METADATA_TO_WAL);
+ this.indexCDCConsumerEnabled = env.getConfiguration()
+ .getBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, DEFAULT_PHOENIX_INDEX_CDC_CONSUMER_ENABLED);
this.isNamespaceEnabled =
SchemaUtil.isNamespaceMappingEnabled(PTableType.INDEX, env.getConfiguration());
TableDescriptor tableDescriptor = env.getRegion().getTableDescriptor();
BloomType bloomFilterType = tableDescriptor.getColumnFamilies()[0].getBloomFilterType();
// when the table descriptor changes, the coproc is reloaded
this.useBloomFilter = bloomFilterType == BloomType.ROW;
+ if (
+ this.indexCDCConsumerEnabled && !this.dataTableName.startsWith("SYSTEM.")
+ && !this.dataTableName.startsWith("SYSTEM:")
+ ) {
+ this.indexCDCConsumer = new IndexCDCConsumer(env, this.dataTableName, serverName);
+ this.indexCDCConsumer.start();
+ }
} catch (NoSuchMethodError ex) {
disabled = true;
LOG.error("Must be too early a version of HBase. Disabled coprocessor ", ex);
@@ -507,6 +535,9 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
this.builder.stop(msg);
this.preWriter.stop(msg);
this.postWriter.stop(msg);
+ if (this.indexCDCConsumer != null) {
+ this.indexCDCConsumer.stop();
+ }
}
/**
@@ -1287,9 +1318,17 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
ArrayListMultimap.<HTableInterfaceReference, Pair<Mutation, byte[]>> create();
prepareIndexMutations(context, maintainers, batchTimestamp);
+ prepareEventuallyConsistentIndexMutations(context, batchTimestamp, maintainers);
+
context.preIndexUpdates = ArrayListMultimap.<HTableInterfaceReference, Mutation> create();
int updateCount = 0;
for (IndexMaintainer indexMaintainer : maintainers) {
+ if (
+ indexMaintainer.getIndexConsistency() != null
+ && indexMaintainer.getIndexConsistency().isAsynchronous()
+ ) {
+ continue;
+ }
updateCount++;
byte[] emptyCF = indexMaintainer.getEmptyKeyValueFamily().copyBytesIfNecessary();
byte[] emptyCQ = indexMaintainer.getEmptyKeyValueQualifier();
@@ -1299,6 +1338,16 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
for (Pair<Mutation, byte[]> update : updates) {
Mutation m = update.getFirst();
if (m instanceof Put) {
+ if (indexMaintainer.isCDCIndex() && context.cdcPreMutationsDoc != null) {
+ ImmutableBytesPtr dataRowKeyPtr = new ImmutableBytesPtr(update.getSecond());
+ BsonDocument cdcDoc = context.cdcPreMutationsDoc.get(dataRowKeyPtr);
+ if (cdcDoc != null) {
+ byte[] cdcMutationsBytes = PBson.INSTANCE.toBytes(cdcDoc);
+ ((Put) m).addColumn(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
+ QueryConstants.CDC_INDEX_PRE_MUTATIONS_CQ_BYTES, batchTimestamp,
+ cdcMutationsBytes);
+ }
+ }
// This will be done before the data table row is updated (i.e., in the first write
// phase)
context.preIndexUpdates.put(hTableInterfaceReference, m);
@@ -1309,6 +1358,16 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
Put unverifiedPut = new Put(m.getRow());
unverifiedPut.addColumn(emptyCF, emptyCQ, batchTimestamp,
QueryConstants.UNVERIFIED_BYTES);
+ if (indexMaintainer.isCDCIndex() && context.cdcPreMutationsDoc != null) {
+ ImmutableBytesPtr dataRowKeyPtr = new ImmutableBytesPtr(update.getSecond());
+ BsonDocument cdcDoc = context.cdcPreMutationsDoc.get(dataRowKeyPtr);
+ if (cdcDoc != null) {
+ byte[] cdcMutationsBytes = PBson.INSTANCE.toBytes(cdcDoc);
+ unverifiedPut.addColumn(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
+ QueryConstants.CDC_INDEX_PRE_MUTATIONS_CQ_BYTES, batchTimestamp,
+ cdcMutationsBytes);
+ }
+ }
// This will be done before the data table row is updated (i.e., in the first write
// phase)
context.preIndexUpdates.put(hTableInterfaceReference, unverifiedPut);
@@ -1319,6 +1378,114 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
}
}
+ /**
+ * Prepares pre-phase and post-phase cdc mutations for eventually consistent indexes.
+ * @param context batch mutate context.
+ * @param batchTimestamp the timestamp to use for mutations.
+ * @param maintainers the list of index maintainers.
+ * @throws IOException if there is an error.
+ */
+ private static void prepareEventuallyConsistentIndexMutations(BatchMutateContext context,
+ long batchTimestamp, List<IndexMaintainer> maintainers) throws IOException {
+ // Store pre-index updates
+ Map<ImmutableBytesPtr, BsonArray> preTablesMap = new HashMap<>();
+ Map<ImmutableBytesPtr, BsonArray> preMutationsMap = new HashMap<>();
+ // Store post-index updates
+ Map<ImmutableBytesPtr, BsonArray> postTablesMap = new HashMap<>();
+ Map<ImmutableBytesPtr, BsonArray> postMutationsMap = new HashMap<>();
+
+ for (IndexMaintainer indexMaintainer : maintainers) {
+ if (
+ indexMaintainer.getIndexConsistency() == null
+ || !indexMaintainer.getIndexConsistency().isAsynchronous()
+ ) {
+ continue;
+ }
+ byte[] emptyCF = indexMaintainer.getEmptyKeyValueFamily().copyBytesIfNecessary();
+ byte[] emptyCQ = indexMaintainer.getEmptyKeyValueQualifier();
+ HTableInterfaceReference hTableInterfaceReference =
+ new HTableInterfaceReference(new ImmutableBytesPtr(indexMaintainer.getIndexTableName()));
+ List<Pair<Mutation, byte[]>> updates = context.indexUpdates.get(hTableInterfaceReference);
+ for (Pair<Mutation, byte[]> update : updates) {
+ Mutation m = update.getFirst();
+ byte[] dataRowKey = update.getSecond();
+ ImmutableBytesPtr rowKeyPtr = new ImmutableBytesPtr(dataRowKey);
+ BsonArray preTables = preTablesMap.computeIfAbsent(rowKeyPtr, k -> new BsonArray());
+ BsonArray preMutations = preMutationsMap.computeIfAbsent(rowKeyPtr, k -> new BsonArray());
+ BsonArray postTables = postTablesMap.computeIfAbsent(rowKeyPtr, k -> new BsonArray());
+ BsonArray postMutations = postMutationsMap.computeIfAbsent(rowKeyPtr, k -> new BsonArray());
+ if (m instanceof Put) {
+ preTables.add(new BsonBinary(indexMaintainer.getIndexTableName()));
+ byte[] preMutation =
+ ProtobufUtil.toMutation(ClientProtos.MutationProto.MutationType.PUT, m).toByteArray();
+ preMutations.add(new BsonBinary(preMutation));
+ if (!indexMaintainer.isUncovered()) {
+ Put verifiedPut = new Put(m.getRow());
+ verifiedPut.addColumn(emptyCF, emptyCQ, batchTimestamp, QueryConstants.VERIFIED_BYTES);
+ postTables.add(new BsonBinary(indexMaintainer.getIndexTableName()));
+ byte[] postMutation = ProtobufUtil
+ .toMutation(ClientProtos.MutationProto.MutationType.PUT, verifiedPut).toByteArray();
+ postMutations.add(new BsonBinary(postMutation));
+ }
+ } else {
+ if (IndexUtil.isDeleteFamily(m)) {
+ Put unverifiedPut = new Put(m.getRow());
+ unverifiedPut.addColumn(emptyCF, emptyCQ, batchTimestamp,
+ QueryConstants.UNVERIFIED_BYTES);
+ preTables.add(new BsonBinary(indexMaintainer.getIndexTableName()));
+ byte[] preMutation = ProtobufUtil
+ .toMutation(ClientProtos.MutationProto.MutationType.PUT, unverifiedPut).toByteArray();
+ preMutations.add(new BsonBinary(preMutation));
+ }
+ postTables.add(new BsonBinary(indexMaintainer.getIndexTableName()));
+ byte[] deleteMutation = ProtobufUtil
+ .toMutation(ClientProtos.MutationProto.MutationType.DELETE, m).toByteArray();
+ postMutations.add(new BsonBinary(deleteMutation));
+ }
+ }
+ }
+
+ if (!preTablesMap.isEmpty()) {
+ context.cdcPreMutationsDoc = new HashMap<>();
+ for (Map.Entry<ImmutableBytesPtr, BsonArray> entry : preTablesMap.entrySet()) {
+ ImmutableBytesPtr rowKey = entry.getKey();
+ BsonArray tables = entry.getValue();
+ BsonArray mutations = preMutationsMap.get(rowKey);
+ if (tables.size() != mutations.size()) {
+ throw new DoNotRetryIOException(
+ "Pre-phase tables and mutations sizes do not match for row key. Tables size: "
+ + tables.size() + " , mutations size: " + mutations.size());
+ }
+ if (!tables.isEmpty()) {
+ BsonDocument doc = new BsonDocument();
+ doc.append("tables", tables);
+ doc.append("mutations", mutations);
+ context.cdcPreMutationsDoc.put(rowKey, doc);
+ }
+ }
+ }
+
+ if (!postTablesMap.isEmpty()) {
+ context.cdcPostMutationsDoc = new HashMap<>();
+ for (Map.Entry<ImmutableBytesPtr, BsonArray> entry : postTablesMap.entrySet()) {
+ ImmutableBytesPtr rowKey = entry.getKey();
+ BsonArray tables = entry.getValue();
+ BsonArray mutations = postMutationsMap.get(rowKey);
+ if (tables.size() != mutations.size()) {
+ throw new DoNotRetryIOException(
+ "Post-phase tables and mutations sizes do not match for row key. Tables size: "
+ + tables.size() + " , mutations size: " + mutations.size());
+ }
+ if (!tables.isEmpty()) {
+ BsonDocument doc = new BsonDocument();
+ doc.append("tables", tables);
+ doc.append("mutations", mutations);
+ context.cdcPostMutationsDoc.put(rowKey, doc);
+ }
+ }
+ }
+ }
+
protected PhoenixIndexMetaData getPhoenixIndexMetaData(
ObserverContext<RegionCoprocessorEnvironment> observerContext,
MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
@@ -1337,6 +1504,12 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
context.postIndexUpdates = ArrayListMultimap.<HTableInterfaceReference, Mutation> create();
List<IndexMaintainer> maintainers = indexMetaData.getIndexMaintainers();
for (IndexMaintainer indexMaintainer : maintainers) {
+ if (
+ indexMaintainer.getIndexConsistency() != null
+ && indexMaintainer.getIndexConsistency().isAsynchronous()
+ ) {
+ continue;
+ }
byte[] emptyCF = indexMaintainer.getEmptyKeyValueFamily().copyBytesIfNecessary();
byte[] emptyCQ = indexMaintainer.getEmptyKeyValueQualifier();
HTableInterfaceReference hTableInterfaceReference =
@@ -1356,6 +1529,32 @@ public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
}
}
}
+
+ if (context.cdcPostMutationsDoc != null && !context.cdcPostMutationsDoc.isEmpty()) {
+ for (IndexMaintainer indexMaintainer : maintainers) {
+ if (!indexMaintainer.isCDCIndex()) {
+ continue;
+ }
+ HTableInterfaceReference hTableInterfaceReference =
+ new HTableInterfaceReference(new ImmutableBytesPtr(indexMaintainer.getIndexTableName()));
+ List<Pair<Mutation, byte[]>> updates = context.indexUpdates.get(hTableInterfaceReference);
+ for (Pair<Mutation, byte[]> update : updates) {
+ Mutation m = update.getFirst();
+ if (m instanceof Put) {
+ ImmutableBytesPtr dataRowKeyPtr = new ImmutableBytesPtr(update.getSecond());
+ BsonDocument cdcDoc = context.cdcPostMutationsDoc.get(dataRowKeyPtr);
+ if (cdcDoc != null) {
+ Put postPut = new Put(m.getRow());
+ byte[] cdcMutationsBytes = PBson.INSTANCE.toBytes(cdcDoc);
+ postPut.addColumn(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
+ QueryConstants.CDC_INDEX_POST_MUTATIONS_CQ_BYTES, batchTimestamp,
+ cdcMutationsBytes);
+ context.postIndexUpdates.put(hTableInterfaceReference, postPut);
+ }
+ }
+ }
+ }
+ }
// all cleanup will be done in postBatchMutateIndispensably()
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
index 69d782e078..fcf839b7d3 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
@@ -151,11 +151,13 @@ public class TrackingParallelWriterIndexCommitter implements IndexCommitter {
Set<Entry<HTableInterfaceReference, Collection<Mutation>>> entries = toWrite.asMap().entrySet();
TaskBatch<Boolean> tasks = new TaskBatch<Boolean>(entries.size());
List<HTableInterfaceReference> tables = new ArrayList<HTableInterfaceReference>(entries.size());
+ int totalMutations = 0;
for (Entry<HTableInterfaceReference, Collection<Mutation>> entry : entries) {
// get the mutations for each table. We leak the implementation here a little bit to save
// doing a complete copy over of all the index update for each table.
final List<Mutation> mutations =
kvBuilder.cloneIfNecessary((List<Mutation>) entry.getValue());
+ totalMutations += mutations.size();
// track each reference so we can get at it easily later, when determing failures
final HTableInterfaceReference tableReference = entry.getKey();
final RegionCoprocessorEnvironment env = this.env;
@@ -249,7 +251,10 @@ public class TrackingParallelWriterIndexCommitter implements IndexCommitter {
Pair<List<Boolean>, List<Future<Boolean>>> resultsAndFutures = null;
try {
- LOGGER.debug("Waiting on index update tasks to complete...");
+ LOGGER.debug(
+ "Waiting on index update tasks to complete... "
+ + "tasks size {}, total mutations {}, num tables {}",
+ tasks.size(), totalMutations, tables.size());
resultsAndFutures = this.pool.submitUninterruptible(tasks);
} catch (ExecutionException e) {
throw new RuntimeException(
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
index e111304e68..7264e3a8b7 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
@@ -105,7 +105,7 @@ public abstract class BasePermissionsIT extends BaseTest {
private static final Set<String> PHOENIX_SYSTEM_TABLES =
new HashSet<>(Arrays.asList("SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS",
"SYSTEM.FUNCTION", "SYSTEM.MUTEX", "SYSTEM.CHILD_LINK", "SYSTEM.TRANSFORM",
- "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM"));
+ "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.IDX_CDC_TRACKER"));
private static final Set<String> PHOENIX_SYSTEM_TABLES_IDENTIFIERS =
new HashSet<>(Arrays.asList("SYSTEM.\"CATALOG\"", "SYSTEM.\"SEQUENCE\"", "SYSTEM.\"STATS\"",
@@ -123,7 +123,7 @@ public abstract class BasePermissionsIT extends BaseTest {
static final Set<String> PHOENIX_NAMESPACE_MAPPED_SYSTEM_TABLES =
new HashSet<>(Arrays.asList("SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS",
"SYSTEM:FUNCTION", "SYSTEM:MUTEX", "SYSTEM:CHILD_LINK", "SYSTEM:TRANSFORM",
- "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM"));
+ "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM", "SYSTEM:IDX_CDC_TRACKER"));
// Create Multiple users so that we can use Hadoop UGI to run tasks as various users
// Permissions can be granted or revoke by superusers and admins only
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson4IT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson4IT.java
index 103f0e6df5..f35cc50f5c 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson4IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson4IT.java
@@ -37,6 +37,7 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.Collection;
+import java.util.List;
import java.util.Properties;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hbase.util.Bytes;
@@ -47,9 +48,13 @@ import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.hbase.index.metrics.GlobalIndexCheckerSource;
import org.apache.phoenix.hbase.index.metrics.GlobalIndexCheckerSourceImpl;
import org.apache.phoenix.hbase.index.metrics.MetricsIndexerSourceFactory;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
import org.apache.phoenix.jdbc.PhoenixStatement;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.schema.types.PDouble;
+import org.apache.phoenix.util.CDCUtil;
import org.apache.phoenix.util.PropertiesUtil;
import org.bson.BsonArray;
import org.bson.BsonBinary;
@@ -145,6 +150,12 @@ public class Bson4IT extends ParallelStatsDisabledIT {
conn.commit();
+ PhoenixConnection pcon = conn.unwrap(PhoenixConnection.class);
+ PTable dataTable = pcon.getTable(new PTableKey(pcon.getTenantId(), tableName));
+ List<PTable> indexes = dataTable.getIndexes();
+ assertEquals("Data table should have exactly 2 indexes", 2, indexes.size());
+ assertFalse("Data table should not have CDC index", CDCUtil.hasCDCIndex(dataTable));
+
ResultSet rs = conn.createStatement().executeQuery("SELECT count(*) FROM " + tableName);
assertTrue(rs.next());
assertEquals(3, rs.getInt(1));
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson5IT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson5IT.java
index 99f4642baf..5ca60cbd9c 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson5IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson5IT.java
@@ -40,9 +40,11 @@ import java.util.Map;
import java.util.Properties;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.compile.ExplainPlan;
+import org.apache.phoenix.compile.ExplainPlanAttributes;
+import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.types.PDouble;
-import org.apache.phoenix.util.CDCUtil;
import org.apache.phoenix.util.PropertiesUtil;
import org.bson.BsonArray;
import org.bson.BsonBinary;
@@ -79,14 +81,24 @@ public class Bson5IT extends ParallelStatsDisabledIT {
Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
String tableName = generateUniqueName();
String cdcName = generateUniqueName();
+ String indexName1 = "IDX1_" + tableName;
+ String indexName2 = "IDX2_" + tableName;
try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
String ddl = "CREATE TABLE " + tableName + " (PK1 VARCHAR NOT NULL, C1 VARCHAR, COL BSON"
+ " CONSTRAINT pk PRIMARY KEY(PK1))";
String cdcDdl = "CREATE CDC " + cdcName + " ON " + tableName;
+ String indexDdl1 = "CREATE UNCOVERED INDEX " + indexName1 + " ON " + tableName
+ + "(BSON_VALUE(COL, 'rather[3].outline.clock', 'VARCHAR')) "
+ + "WHERE BSON_VALUE(COL, 'rather[3].outline.clock', 'VARCHAR') IS NOT NULL "
+ + "CONSISTENCY = EVENTUAL";
+ String indexDdl2 = "CREATE UNCOVERED INDEX " + indexName2 + " ON " + tableName
+ + "(BSON_VALUE(COL, 'result[1].location.coordinates.longitude', 'DOUBLE')) "
+ + "WHERE BSON_VALUE(COL, 'result[1].location.coordinates.longitude', 'DOUBLE') IS NOT NULL "
+ + "CONSISTENCY = EVENTUAL";
conn.createStatement().execute(ddl);
conn.createStatement().execute(cdcDdl);
- IndexToolIT.runIndexTool(false, "", tableName,
- "\"" + CDCUtil.getCDCIndexName(cdcName) + "\"");
+ conn.createStatement().execute(indexDdl1);
+ conn.createStatement().execute(indexDdl2);
Timestamp ts1 = new Timestamp(System.currentTimeMillis());
Thread.sleep(100);
@@ -118,6 +130,30 @@ public class Bson5IT extends ParallelStatsDisabledIT {
Thread.sleep(100);
Timestamp ts2 = new Timestamp(System.currentTimeMillis());
+ PreparedStatement indexPs = conn.prepareStatement("SELECT PK1, COL FROM " + tableName
+ + " WHERE BSON_VALUE(COL, 'rather[3].outline.clock', 'VARCHAR') = ?");
+ indexPs.setString(1, "personal");
+ ResultSet indexRs = indexPs.executeQuery();
+ assertFalse(indexRs.next());
+
+ Thread.sleep(11000);
+ indexRs = indexPs.executeQuery();
+ assertTrue(indexRs.next());
+ assertEquals("pk1010", indexRs.getString(1));
+ BsonDocument actualDoc = (BsonDocument) indexRs.getObject(2);
+ assertEquals(bsonDocument2, actualDoc);
+ assertFalse(indexRs.next());
+ validateExplainPlan(indexPs, indexName1, "RANGE SCAN ");
+
+ indexPs = conn.prepareStatement("SELECT PK1 FROM " + tableName
+ + " WHERE BSON_VALUE(COL, 'result[1].location.coordinates.longitude', 'DOUBLE') = ?");
+ indexPs.setDouble(1, 52.3736);
+ indexRs = indexPs.executeQuery();
+ assertTrue(indexRs.next());
+ assertEquals("pk1011", indexRs.getString(1));
+ assertFalse(indexRs.next());
+ validateExplainPlan(indexPs, indexName2, "RANGE SCAN ");
+
testCDCAfterFirstUpsert(conn, cdcName, ts1, ts2, bsonDocument1, bsonDocument2, bsonDocument3);
ts1 = new Timestamp(System.currentTimeMillis());
@@ -312,14 +348,24 @@ public class Bson5IT extends ParallelStatsDisabledIT {
Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
String tableName = generateUniqueName();
String cdcName = generateUniqueName();
+ String indexName1 = "IDX1_" + tableName;
+ String indexName2 = "IDX2_" + tableName;
try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
String ddl = "CREATE TABLE " + tableName + " (PK1 VARCHAR NOT NULL, C1 VARCHAR, COL BSON"
+ " CONSTRAINT pk PRIMARY KEY(PK1))";
String cdcDdl = "CREATE CDC " + cdcName + " ON " + tableName;
+ String indexDdl1 = "CREATE INDEX " + indexName1 + " ON " + tableName
+ + "(BSON_VALUE(COL, 'rather[3].outline.clock', 'VARCHAR')) INCLUDE(COL) "
+ + "WHERE BSON_VALUE(COL, 'rather[3].outline.clock', 'VARCHAR') IS NOT NULL "
+ + "CONSISTENCY = EVENTUAL";
+ String indexDdl2 = "CREATE UNCOVERED INDEX " + indexName2 + " ON " + tableName
+ + "(BSON_VALUE(COL, 'result[1].location.coordinates.longitude', 'DOUBLE')) "
+ + "WHERE BSON_VALUE(COL, 'result[1].location.coordinates.longitude', 'DOUBLE') IS NOT NULL "
+ + "CONSISTENCY = EVENTUAL";
conn.createStatement().execute(ddl);
conn.createStatement().execute(cdcDdl);
- IndexToolIT.runIndexTool(false, "", tableName,
- "\"" + CDCUtil.getCDCIndexName(cdcName) + "\"");
+ conn.createStatement().execute(indexDdl1);
+ conn.createStatement().execute(indexDdl2);
Timestamp ts1 = new Timestamp(System.currentTimeMillis());
Thread.sleep(100);
@@ -352,6 +398,27 @@ public class Bson5IT extends ParallelStatsDisabledIT {
Thread.sleep(100);
Timestamp ts2 = new Timestamp(System.currentTimeMillis());
+ Thread.sleep(11000);
+ PreparedStatement indexPs = conn.prepareStatement("SELECT PK1, COL FROM " + tableName
+ + " WHERE BSON_VALUE(COL, 'rather[3].outline.clock', 'VARCHAR') = ?");
+ indexPs.setString(1, "personal");
+ ResultSet indexRs = indexPs.executeQuery();
+ assertTrue(indexRs.next());
+ assertEquals("pk1010", indexRs.getString(1));
+ BsonDocument actualDoc = (BsonDocument) indexRs.getObject(2);
+ assertEquals(bsonDocument2, actualDoc);
+ assertFalse(indexRs.next());
+ validateExplainPlan(indexPs, indexName1, "RANGE SCAN ");
+
+ indexPs = conn.prepareStatement("SELECT PK1 FROM " + tableName
+ + " WHERE BSON_VALUE(COL, 'result[1].location.coordinates.longitude', 'DOUBLE') = ?");
+ indexPs.setDouble(1, 52.3736);
+ indexRs = indexPs.executeQuery();
+ assertTrue(indexRs.next());
+ assertEquals("pk1011", indexRs.getString(1));
+ assertFalse(indexRs.next());
+ validateExplainPlan(indexPs, indexName2, "RANGE SCAN ");
+
testCDCAfterFirstUpsert(conn, cdcName, ts1, ts2, bsonDocument1, bsonDocument2, bsonDocument3);
Thread.sleep(100);
@@ -654,4 +721,13 @@ public class Bson5IT extends ParallelStatsDisabledIT {
}
}
+ private static void validateExplainPlan(PreparedStatement ps, String tableName, String scanType)
+ throws SQLException {
+ ExplainPlan plan = ps.unwrap(PhoenixPreparedStatement.class).optimizeQuery().getExplainPlan();
+ ExplainPlanAttributes explainPlanAttributes = plan.getPlanStepsAsAttributes();
+ assertEquals(tableName, explainPlanAttributes.getTableName());
+ assertEquals("PARALLEL 1-WAY", explainPlanAttributes.getIteratorTypeAndScanSize());
+ assertEquals(scanType, explainPlanAttributes.getExplainScanType());
+ }
+
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredEventualIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredEventualIT.java
new file mode 100644
index 0000000000..f83da1bab1
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredEventualIT.java
@@ -0,0 +1,36 @@
+/*
+ * 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.end2end;
+
+import java.util.Arrays;
+import java.util.Collection;
+import org.junit.experimental.categories.Category;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class ConcurrentMutationsCoveredEventualIT extends ConcurrentMutationsExtendedIndexIT {
+
+ public ConcurrentMutationsCoveredEventualIT(boolean uncovered, boolean eventual) {
+ super(uncovered, eventual);
+ }
+
+ @Parameterized.Parameters(name = "uncovered={0}, eventual={1}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { false, true } });
+ }
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredNonEventualIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredNonEventualIT.java
new file mode 100644
index 0000000000..7d7c2c1e5b
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredNonEventualIT.java
@@ -0,0 +1,36 @@
+/*
+ * 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.end2end;
+
+import java.util.Arrays;
+import java.util.Collection;
+import org.junit.experimental.categories.Category;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class ConcurrentMutationsCoveredNonEventualIT extends ConcurrentMutationsExtendedIndexIT {
+
+ public ConcurrentMutationsCoveredNonEventualIT(boolean uncovered, boolean eventual) {
+ super(uncovered, eventual);
+ }
+
+ @Parameterized.Parameters(name = "uncovered={0}, eventual={1}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { false, false } });
+ }
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
index 03ca9e1a68..39d6486336 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.end2end;
import static org.apache.phoenix.end2end.IndexToolIT.verifyIndexTable;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_BATCH_SIZE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -41,6 +42,7 @@ import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.SimpleRegionObserver;
import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.coprocessor.PhoenixMasterObserver;
import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryServices;
@@ -64,6 +66,7 @@ import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
private static final Logger LOGGER = LoggerFactory.getLogger(ConcurrentMutationsExtendedIT.class);
private final boolean uncovered;
+ private final boolean eventual;
private static final Random RAND = new Random(5);
private static final String MVCC_LOCK_TEST_TABLE_PREFIX = "MVCCLOCKTEST_";
private static final String LOCK_TEST_TABLE_PREFIX = "LOCKTEST_";
@@ -71,8 +74,9 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
protected static final int MAX_LOOKBACK_AGE = 1000000;
private final Object lock = new Object();
- public ConcurrentMutationsExtendedIT(boolean uncovered) {
+ public ConcurrentMutationsExtendedIT(boolean uncovered, boolean eventual) {
this.uncovered = uncovered;
+ this.eventual = eventual;
}
@BeforeClass
@@ -88,12 +92,17 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
// The following sets the wait duration for the previous concurrent batch to 10 ms to test
// the code path handling timeouts
props.put("phoenix.index.concurrent.wait.duration.ms", "10");
+ props.put(QueryServices.TASK_HANDLING_INTERVAL_MS_ATTRIB, Long.toString(2));
+ props.put(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB, Long.toString(1));
+ props.put(INDEX_CDC_CONSUMER_BATCH_SIZE, Integer.toString(4500));
+ props.put("hbase.coprocessor.master.classes", PhoenixMasterObserver.class.getName());
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
- @Parameterized.Parameters(name = "uncovered={0}")
- public static synchronized Collection<Boolean> data() {
- return Arrays.asList(true, false);
+ @Parameterized.Parameters(name = "uncovered={0}, eventual={1}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(
+ new Object[][] { { false, false }, { false, true }, { true, false }, { true, true } });
}
@Test
@@ -105,7 +114,7 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
+ "(k1 INTEGER NOT NULL, k2 INTEGER NOT NULL, v1 INTEGER, CONSTRAINT pk PRIMARY KEY (k1,k2)) COLUMN_ENCODED_BYTES = 0");
TestUtil.addCoprocessor(conn, tableName, DelayingRegionObserver.class);
conn.createStatement().execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX "
- + indexName + " ON " + tableName + "(v1)");
+ + indexName + " ON " + tableName + "(v1)" + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
final CountDownLatch doneSignal = new CountDownLatch(2);
Runnable r1 = new Runnable() {
@@ -167,6 +176,9 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
t2.start();
doneSignal.await(60, TimeUnit.SECONDS);
+ if (eventual) {
+ Thread.sleep(10000);
+ }
verifyIndexTable(tableName, indexName, conn);
}
@@ -180,11 +192,11 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
+ "(k1 INTEGER NOT NULL, k2 INTEGER NOT NULL, v1 INTEGER, CONSTRAINT pk PRIMARY KEY (k1,k2))");
TestUtil.addCoprocessor(conn, tableName, DelayingRegionObserver.class);
conn.createStatement().execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX "
- + indexName + " ON " + tableName + "(v1)");
+ + indexName + " ON " + tableName + "(v1)" + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
conn.createStatement()
.execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX " + singleCellindexName
- + " ON " + tableName
- + "(v1) IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2");
+ + " ON " + tableName + "(v1)" + (eventual ? " CONSISTENCY = EVENTUAL," : "")
+ + " IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS," + " COLUMN_ENCODED_BYTES=2");
final CountDownLatch doneSignal = new CountDownLatch(2);
Runnable r1 = new Runnable() {
@@ -236,6 +248,9 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
t2.start();
doneSignal.await(60, TimeUnit.SECONDS);
+ if (eventual) {
+ Thread.sleep(10000);
+ }
verifyIndexTable(tableName, indexName, conn);
verifyIndexTable(tableName, singleCellindexName, conn);
}
@@ -252,51 +267,56 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
conn.createStatement().execute("CREATE TABLE " + tableName
+ "(k1 INTEGER NOT NULL, k2 INTEGER NOT NULL, a.v1 INTEGER, b.v2 INTEGER, c.v3 INTEGER, d.v4 INTEGER,"
+ "CONSTRAINT pk PRIMARY KEY (k1,k2)) COLUMN_ENCODED_BYTES = 0, VERSIONS=1");
- conn.createStatement().execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX "
- + indexName + " ON " + tableName + "(v1)" + (uncovered ? "" : "INCLUDE(v2, v3)"));
+ conn.createStatement()
+ .execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX " + indexName + " ON "
+ + tableName + "(v1)" + (uncovered ? "" : " INCLUDE(v2, v3)")
+ + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
final CountDownLatch doneSignal = new CountDownLatch(nThreads);
Runnable[] runnables = new Runnable[nThreads];
long startTime = EnvironmentEdgeManager.currentTimeMillis();
- for (int i = 0; i < nThreads; i++) {
- runnables[i] = new Runnable() {
-
- @Override
- public void run() {
- try {
- Connection conn = DriverManager.getConnection(getUrl());
- for (int i = 0; i < 10000; i++) {
- conn.createStatement()
- .execute("UPSERT INTO " + tableName + " VALUES (" + (i % nRows) + ", 0, "
- + (RAND.nextBoolean() ? null : (RAND.nextInt() % nIndexValues)) + ", "
- + (RAND.nextBoolean() ? null : RAND.nextInt()) + ", "
- + (RAND.nextBoolean() ? null : RAND.nextInt()) + ", "
- + (RAND.nextBoolean() ? null : RAND.nextInt()) + ")");
- if ((i % batchSize) == 0) {
- conn.commit();
- }
- }
- conn.commit();
- } catch (SQLException e) {
- LOGGER.warn("Exception during upsert : " + e);
- } finally {
- doneSignal.countDown();
- }
- }
-
- };
- }
+ buildRunnables(nThreads, runnables, tableName, nRows, nIndexValues, batchSize, doneSignal,
+ 10000);
for (int i = 0; i < nThreads; i++) {
Thread t = new Thread(runnables[i]);
t.start();
}
-
assertTrue("Ran out of time", doneSignal.await(120, TimeUnit.SECONDS));
LOGGER.info(
"Total upsert time in ms : " + (EnvironmentEdgeManager.currentTimeMillis() - startTime));
+ if (eventual) {
+ Thread.sleep(350000);
+ }
long actualRowCount = verifyIndexTable(tableName, indexName, conn);
assertEquals(nRows, actualRowCount);
}
+ static void buildRunnables(int nThreads, Runnable[] runnables, String tableName, int nRows,
+ int nIndexValues, int batchSize, CountDownLatch doneSignal, int totalUpserts) {
+ for (int i = 0; i < nThreads; i++) {
+ runnables[i] = () -> {
+ try {
+ Connection conn = DriverManager.getConnection(getUrl());
+ for (int j = 0; j < totalUpserts; j++) {
+ conn.createStatement()
+ .execute("UPSERT INTO " + tableName + " VALUES (" + (j % nRows) + ", 0, "
+ + (RAND.nextBoolean() ? null : (RAND.nextInt() % nIndexValues)) + ", "
+ + (RAND.nextBoolean() ? null : RAND.nextInt()) + ", "
+ + (RAND.nextBoolean() ? null : RAND.nextInt()) + ", "
+ + (RAND.nextBoolean() ? null : RAND.nextInt()) + ")");
+ if ((j % batchSize) == 0) {
+ conn.commit();
+ }
+ }
+ conn.commit();
+ } catch (SQLException e) {
+ LOGGER.warn("Exception during upsert : " + e);
+ } finally {
+ doneSignal.countDown();
+ }
+ };
+ }
+ }
+
@Test
public void testRowLockDuringPreBatchMutateWhenIndexed() throws Exception {
final String tableName = LOCK_TEST_TABLE_PREFIX + generateUniqueName();
@@ -307,7 +327,7 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
"CREATE TABLE " + tableName + "(k VARCHAR PRIMARY KEY, v INTEGER) COLUMN_ENCODED_BYTES = 0");
TestUtil.addCoprocessor(conn, tableName, DelayingRegionObserver.class);
conn.createStatement().execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX "
- + indexName + " ON " + tableName + "(v)");
+ + indexName + " ON " + tableName + "(v)" + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
final CountDownLatch doneSignal = new CountDownLatch(2);
final String[] failedMsg = new String[1];
Runnable r1 = new Runnable() {
@@ -353,6 +373,9 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
doneSignal.await(ROW_LOCK_WAIT_TIME + 5000, TimeUnit.SECONDS);
assertNull(failedMsg[0], failedMsg[0]);
+ if (eventual) {
+ Thread.sleep(10000);
+ }
long actualRowCount = IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);
assertEquals(1, actualRowCount);
}
@@ -365,7 +388,7 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
conn.createStatement().execute(
"CREATE TABLE " + tableName + "(k VARCHAR PRIMARY KEY, v INTEGER) COLUMN_ENCODED_BYTES = 0");
conn.createStatement().execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX "
- + indexName + " ON " + tableName + "(v,k)");
+ + indexName + " ON " + tableName + "(v,k)" + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('foo',0)");
conn.commit();
TestUtil.addCoprocessor(conn, tableName, DelayingRegionObserver.class);
@@ -411,6 +434,9 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
t2.start();
doneSignal.await(ROW_LOCK_WAIT_TIME + 5000, TimeUnit.SECONDS);
+ if (eventual) {
+ Thread.sleep(10000);
+ }
long actualRowCount = IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);
assertEquals(1, actualRowCount);
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIndexIT.java
new file mode 100644
index 0000000000..792d412a3a
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIndexIT.java
@@ -0,0 +1,278 @@
+/*
+ * 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.end2end;
+
+import static org.apache.phoenix.end2end.IndexToolIT.verifyIndexTable;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(Parameterized.class)
+public abstract class ConcurrentMutationsExtendedIndexIT extends ParallelStatsDisabledIT {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ConcurrentMutationsExtendedIndexIT.class);
+
+ private final boolean uncovered;
+ private final boolean eventual;
+
+ protected ConcurrentMutationsExtendedIndexIT(boolean uncovered, boolean eventual) {
+ this.uncovered = uncovered;
+ this.eventual = eventual;
+ }
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ ConcurrentMutationsExtendedIT.doSetup();
+ }
+
+ @Test(timeout = 2000000)
+ public void testConcurrentUpsertsWithTableSplits() throws Exception {
+ int nThreads = 12;
+ final int batchSize = 100;
+ final int nRows = 777;
+ final int nIndexValues = 23;
+ final int nSplits = 3;
+ final String schemaName = "SchemaN_01-" + generateUniqueName();
+ final String tableName = "Table.Name-00_001_" + generateUniqueName();
+ final String indexName1 = "IndexTable.Name-00_001_" + generateUniqueName();
+ final String indexName2 = "IndexTable.Name-00_002_" + generateUniqueName();
+ final String indexName3 = "IndexTable.Name-00_003_" + generateUniqueName();
+ final String indexName4 = "IndexTable.Name-00_004_" + generateUniqueName();
+ final String indexName5 = "IndexTable.Name-00_005_" + generateUniqueName();
+ Connection conn = DriverManager.getConnection(getUrl());
+ conn.createStatement().execute("CREATE TABLE " + "\"" + schemaName + "\".\"" + tableName + "\""
+ + "(k1 INTEGER NOT NULL, k2 INTEGER NOT NULL, v1 INTEGER, v2 INTEGER, v3 INTEGER, v4 INTEGER,"
+ + "CONSTRAINT pk PRIMARY KEY (k1,k2))");
+ conn.createStatement()
+ .execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX " + "\"" + indexName1 + "\""
+ + " ON " + "\"" + schemaName + "\".\"" + tableName + "\"" + "(v1)"
+ + (uncovered ? "" : " INCLUDE(v2, v3)")
+ + (eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG"));
+ conn.createStatement()
+ .execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX " + "\"" + indexName2 + "\""
+ + " ON " + "\"" + schemaName + "\".\"" + tableName + "\"" + "(v2)"
+ + (uncovered ? "" : " INCLUDE(v1, v4)") + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+ conn.createStatement()
+ .execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX " + "\"" + indexName3 + "\""
+ + " ON " + "\"" + schemaName + "\".\"" + tableName + "\"" + "(v3)"
+ + (uncovered ? "" : " INCLUDE(v1, v2)")
+ + (eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG"));
+ conn.createStatement()
+ .execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX " + "\"" + indexName4 + "\""
+ + " ON " + "\"" + schemaName + "\".\"" + tableName + "\"" + "(v4)"
+ + (uncovered ? "" : " INCLUDE(v2, v3)") + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+ conn.createStatement()
+ .execute("CREATE " + (uncovered ? "UNCOVERED " : " ") + "INDEX " + "\"" + indexName5 + "\""
+ + " ON " + "\"" + schemaName + "\".\"" + tableName + "\"" + "(v1, v2)"
+ + (uncovered ? "" : " INCLUDE(v3, v4)")
+ + (eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG"));
+
+ final CountDownLatch doneSignal = new CountDownLatch(nThreads);
+ Runnable[] runnables = new Runnable[nThreads];
+ Thread.sleep(3000);
+ long startTime = EnvironmentEdgeManager.currentTimeMillis();
+
+ ConcurrentMutationsExtendedIT.buildRunnables(nThreads, runnables,
+ "\"" + schemaName + "\".\"" + tableName + "\"", nRows, nIndexValues, batchSize, doneSignal,
+ 10000);
+ Thread splitThread = new Thread(() -> TestUtil.splitTable(getUrl(),
+ schemaName + "." + tableName, nSplits, eventual ? 8000 : 2000));
+ splitThread.start();
+ for (int i = 0; i < nThreads; i++) {
+ if (i >= (nThreads - 4)) {
+ Thread.sleep(12000);
+ }
+ Thread t = new Thread(runnables[i]);
+ t.start();
+ }
+ assertTrue("Ran out of time", doneSignal.await(350, TimeUnit.SECONDS));
+ splitThread.join(10000);
+ LOGGER.info(
+ "Total upsert time in ms : " + (EnvironmentEdgeManager.currentTimeMillis() - startTime));
+ List<String> allIndexes =
+ new ArrayList<>(Arrays.asList(indexName1, indexName2, indexName3, indexName4, indexName5));
+ verifyRandomIndexes(allIndexes, schemaName, tableName, conn, nRows);
+ }
+
+ private void verifyRandomIndexes(List<String> allIndexes, String schemaName, String tableName,
+ Connection conn, int nRows) throws Exception {
+ Collections.shuffle(allIndexes, ThreadLocalRandom.current());
+ if (!eventual) {
+ verifyIndexes(allIndexes, schemaName, tableName, conn, nRows, true);
+ return;
+ }
+ boolean isComplete;
+ int tries = 0;
+ do {
+ Thread.sleep(50000);
+ try {
+ verifyIndexes(allIndexes, schemaName, tableName, conn, nRows, false);
+ isComplete = true;
+ } catch (AssertionError e) {
+ isComplete = false;
+ LOGGER.error("Index verification failed. Tries: {}", tries, e);
+ } finally {
+ dumpTableRows(conn, "SYSTEM.CDC_STREAM");
+ dumpTableRows(conn, "SYSTEM.IDX_CDC_TRACKER");
+ }
+ tries++;
+ } while (!isComplete);
+ LOGGER.info("Index verification complete. Tries: {}", tries);
+ }
+
+ private static void verifyIndexes(List<String> allIndexes, String schemaName, String tableName,
+ Connection conn, int nRows, boolean dumpTables) throws Exception {
+ LOGGER.info("Randomly selected indexes to verify: {}, {}", allIndexes.get(0),
+ allIndexes.get(1));
+ long actualRowCount = verifyIndexTable("\"" + schemaName + "\".\"" + tableName + "\"",
+ "\"" + schemaName + "\".\"" + allIndexes.get(0) + "\"", conn, dumpTables);
+ assertEquals(nRows, actualRowCount);
+ actualRowCount = verifyIndexTable("\"" + schemaName + "\".\"" + tableName + "\"",
+ "\"" + schemaName + "\".\"" + allIndexes.get(1) + "\"", conn, dumpTables);
+ assertEquals(nRows, actualRowCount);
+ }
+
+ @Test(timeout = 3000000)
+ public void testConcurrentUpsertsWithTableSplitsMerges() throws Exception {
+ Assume.assumeFalse(uncovered);
+ int nThreads = 13;
+ final int batchSize = 100;
+ final int nRows = 1451;
+ final int nIndexValues = 28;
+ final int nSplits = 3;
+ final String schemaName = "SchemaN_01-" + generateUniqueName();
+ final String tableName = "Table.Name-00_001_" + generateUniqueName();
+ final String indexName1 = "IndexTable.Name-00_001_" + generateUniqueName();
+ final String indexName2 = "IndexTable.Name-00_002_" + generateUniqueName();
+ final String indexName3 = "IndexTable.Name-00_003_" + generateUniqueName();
+ final String indexName4 = "IndexTable.Name-00_004_" + generateUniqueName();
+ final String indexName5 = "IndexTable.Name-00_005_" + generateUniqueName();
+ final String indexName6 = "IndexTable.Name-00_006_" + generateUniqueName();
+ Connection conn = DriverManager.getConnection(getUrl());
+ conn.createStatement().execute("CREATE TABLE " + "\"" + schemaName + "\".\"" + tableName + "\""
+ + "(k1 INTEGER NOT NULL, k2 INTEGER NOT NULL, v1 INTEGER, v2 INTEGER, v3 INTEGER, v4 INTEGER,"
+ + "CONSTRAINT pk PRIMARY KEY (k1,k2))");
+ conn.createStatement()
+ .execute("CREATE INDEX " + "\"" + indexName1 + "\"" + " ON " + "\"" + schemaName + "\".\""
+ + tableName + "\"" + "(v1)" + " INCLUDE(v2, v3)"
+ + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+ conn.createStatement()
+ .execute("CREATE UNCOVERED INDEX " + "\"" + indexName2 + "\"" + " ON " + "\"" + schemaName
+ + "\".\"" + tableName + "\"" + "(v2)" + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+ conn.createStatement()
+ .execute("CREATE INDEX " + "\"" + indexName3 + "\"" + " ON " + "\"" + schemaName + "\".\""
+ + tableName + "\"" + "(v3)" + " INCLUDE(v1, v2)"
+ + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+ conn.createStatement()
+ .execute("CREATE UNCOVERED INDEX " + "\"" + indexName4 + "\"" + " ON " + "\"" + schemaName
+ + "\".\"" + tableName + "\"" + "(v4)" + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+ conn.createStatement()
+ .execute("CREATE INDEX " + "\"" + indexName5 + "\"" + " ON " + "\"" + schemaName + "\".\""
+ + tableName + "\"" + "(v1, v2)" + " INCLUDE(v3, v4)"
+ + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+ conn.createStatement()
+ .execute("CREATE UNCOVERED INDEX " + "\"" + indexName6 + "\"" + " ON " + "\"" + schemaName
+ + "\".\"" + tableName + "\"" + "(v3, v1)" + (eventual ? " CONSISTENCY = EVENTUAL" : ""));
+
+ final CountDownLatch doneSignal = new CountDownLatch(nThreads);
+ Runnable[] runnables = new Runnable[nThreads];
+ Thread.sleep(3000);
+ long startTime = EnvironmentEdgeManager.currentTimeMillis();
+
+ ConcurrentMutationsExtendedIT.buildRunnables(nThreads, runnables,
+ "\"" + schemaName + "\".\"" + tableName + "\"", nRows, nIndexValues, batchSize, doneSignal,
+ 10000);
+ Thread splitThread = new Thread(() -> TestUtil.splitTable(getUrl(),
+ schemaName + "." + tableName, nSplits, eventual ? 3500 : 2000));
+ splitThread.start();
+ for (int i = 0; i < (nThreads - 5); i++) {
+ if (i >= (nThreads - 8)) {
+ Thread.sleep(9000);
+ }
+ Thread t = new Thread(runnables[i]);
+ t.start();
+ }
+ splitThread.join(10000);
+ Thread mergeThread =
+ new Thread(() -> TestUtil.mergeTableRegions(getUrl(), schemaName + "." + tableName, 4, 1000));
+ for (int i = nThreads - 5; i < nThreads; i++) {
+ Thread t = new Thread(runnables[i]);
+ t.start();
+ if (i == nThreads - 3) {
+ mergeThread.start();
+ }
+ }
+ mergeThread.join(10000);
+ assertTrue("Ran out of time", doneSignal.await(1500, TimeUnit.SECONDS));
+ LOGGER.info(
+ "Total upsert time in ms : " + (EnvironmentEdgeManager.currentTimeMillis() - startTime));
+ List<String> allIndexes = new ArrayList<>(
+ Arrays.asList(indexName1, indexName2, indexName3, indexName4, indexName5, indexName6));
+ verifyRandomIndexes(allIndexes, schemaName, tableName, conn, nRows);
+ }
+
+ private void dumpTableRows(Connection conn, String tableName) throws SQLException {
+ LOGGER.info("Dumping {} table:", tableName);
+ try (ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM " + tableName)) {
+ ResultSetMetaData metaData = rs.getMetaData();
+ int columnCount = metaData.getColumnCount();
+ StringBuilder header = new StringBuilder();
+ for (int i = 1; i <= columnCount; i++) {
+ if (i > 1) {
+ header.append("\t|\t");
+ }
+ header.append(metaData.getColumnName(i));
+ }
+ LOGGER.info("{}", header);
+ LOGGER.info("_____________________________________________________________________________");
+ while (rs.next()) {
+ StringBuilder row = new StringBuilder();
+ for (int i = 1; i <= columnCount; i++) {
+ if (i > 1) {
+ row.append("\t|\t");
+ }
+ row.append(rs.getString(i));
+ }
+ LOGGER.info("{}", row);
+ }
+ }
+ }
+
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsLazyPostBatchWriteIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsLazyPostBatchWriteIT.java
index 761c8658eb..fd54b57ffb 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsLazyPostBatchWriteIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsLazyPostBatchWriteIT.java
@@ -34,8 +34,8 @@ import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
@Category(NeedsOwnMiniClusterTest.class)
public class ConcurrentMutationsLazyPostBatchWriteIT extends ConcurrentMutationsExtendedIT {
- public ConcurrentMutationsLazyPostBatchWriteIT(boolean uncovered) {
- super(uncovered);
+ public ConcurrentMutationsLazyPostBatchWriteIT(boolean uncovered, boolean eventual) {
+ super(uncovered, eventual);
Assume.assumeFalse("Only covered index supports lazy post batch write mode", uncovered);
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredEventualIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredEventualIT.java
new file mode 100644
index 0000000000..9887f8c546
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredEventualIT.java
@@ -0,0 +1,36 @@
+/*
+ * 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.end2end;
+
+import java.util.Arrays;
+import java.util.Collection;
+import org.junit.experimental.categories.Category;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class ConcurrentMutationsUncoveredEventualIT extends ConcurrentMutationsExtendedIndexIT {
+
+ public ConcurrentMutationsUncoveredEventualIT(boolean uncovered, boolean eventual) {
+ super(uncovered, eventual);
+ }
+
+ @Parameterized.Parameters(name = "uncovered={0}, eventual={1}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { true, true } });
+ }
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredNonEventualIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredNonEventualIT.java
new file mode 100644
index 0000000000..70bb2ac995
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredNonEventualIT.java
@@ -0,0 +1,36 @@
+/*
+ * 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.end2end;
+
+import java.util.Arrays;
+import java.util.Collection;
+import org.junit.experimental.categories.Category;
+import org.junit.runners.Parameterized;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class ConcurrentMutationsUncoveredNonEventualIT extends ConcurrentMutationsExtendedIndexIT {
+
+ public ConcurrentMutationsUncoveredNonEventualIT(boolean uncovered, boolean eventual) {
+ super(uncovered, eventual);
+ }
+
+ @Parameterized.Parameters(name = "uncovered={0}, eventual={1}")
+ public static synchronized Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] { { true, false } });
+ }
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexMaintainerConsistencyIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexMaintainerConsistencyIT.java
new file mode 100644
index 0000000000..8c529a1aad
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexMaintainerConsistencyIT.java
@@ -0,0 +1,216 @@
+/*
+ * 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.end2end;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.List;
+import org.apache.phoenix.coprocessor.generated.ServerCachingProtos;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableKey;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.RowKeySchema;
+import org.apache.phoenix.schema.types.IndexConsistency;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/**
+ * IndexMaintainer consistency tests.
+ */
+@Category(ParallelStatsDisabledTest.class)
+public class IndexMaintainerConsistencyIT extends ParallelStatsDisabledIT {
+
+ @After
+ public synchronized void afterTest() throws Exception {
+ boolean refCountLeaked = isAnyStoreRefCountLeaked();
+ Assert.assertFalse("refCount leaked", refCountLeaked);
+ }
+
+ @Test
+ public void testIndexMaintainerConsistencyFromPTable() throws Exception {
+ String dataTableName = "XYZ.\"tablENam1.1_0-001\"";
+ String dataTableNameWithoutQuotes = "XYZ.tablENam1.1_0-001";
+ String strongIndexName1 = "\"idXNam1.12._0-001\"";
+ String strongIndexName2 = "\"idXNam1.12._0-002\"";
+ String eventualIndexName1 = "\"idXNam2.12._0-002\"";
+ String eventualIndexName2 = "\"idXNam3.12._0-003\"";
+
+ try (Connection conn =
+ DriverManager.getConnection(getUrl(), PropertiesUtil.deepCopy(TEST_PROPERTIES))) {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ conn.createStatement().execute(
+ "CREATE TABLE " + dataTableName + " (id INTEGER PRIMARY KEY, name VARCHAR, city VARCHAR)"
+ + " IS_STRICT_TTL=false, UPDATE_CACHE_FREQUENCY=2, "
+ + "\"phoenix.max.lookback.age.seconds\"=97200");
+ conn.createStatement().execute(
+ "CREATE INDEX " + strongIndexName1 + " ON " + dataTableName + " (name) INCLUDE (city)");
+ conn.createStatement().execute("CREATE INDEX " + strongIndexName2 + " ON " + dataTableName
+ + " (name) INCLUDE (city) CONSISTENCY=STRONG");
+ conn.createStatement().execute("CREATE INDEX " + eventualIndexName1 + " ON " + dataTableName
+ + " (name) INCLUDE (city) CONSISTENCY=EVENTUAL");
+ conn.createStatement().execute("CREATE UNCOVERED INDEX " + eventualIndexName2 + " ON "
+ + dataTableName + " (city, name) CONSISTENCY=EVENTUAL");
+ PTable strongIndex1 =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), "XYZ.idXNam1.12._0-001"));
+ PTable strongIndex2 =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), "XYZ.idXNam1.12._0-002"));
+ PTable eventualIndex1 =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), "XYZ.idXNam2.12._0-002"));
+ PTable eventualIndex2 =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), "XYZ.idXNam3.12._0-003"));
+ PTable dataTable =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), dataTableNameWithoutQuotes));
+ assertNull(strongIndex1.getIndexConsistency());
+ assertEquals(IndexConsistency.STRONG, strongIndex2.getIndexConsistency());
+ assertEquals(IndexConsistency.EVENTUAL, eventualIndex1.getIndexConsistency());
+ assertEquals(IndexConsistency.EVENTUAL, eventualIndex2.getIndexConsistency());
+ assertEquals(2, strongIndex1.getUpdateCacheFrequency());
+ assertEquals(2, strongIndex2.getUpdateCacheFrequency());
+ assertEquals(2, eventualIndex1.getUpdateCacheFrequency());
+ assertEquals(2, eventualIndex2.getUpdateCacheFrequency());
+ assertEquals(2, dataTable.getUpdateCacheFrequency());
+ assertFalse(eventualIndex1.isStrictTTL());
+ assertFalse(eventualIndex2.isStrictTTL());
+ assertFalse(strongIndex1.isStrictTTL());
+ assertFalse(strongIndex2.isStrictTTL());
+ assertFalse(dataTable.isStrictTTL());
+ assertNull(dataTable.getIndexConsistency());
+
+ IndexMaintainer strongMaintainer1 = strongIndex1.getIndexMaintainer(dataTable, pconn);
+ IndexMaintainer strongMaintainer2 = strongIndex2.getIndexMaintainer(dataTable, pconn);
+ IndexMaintainer eventualMaintainer1 = eventualIndex1.getIndexMaintainer(dataTable, pconn);
+ IndexMaintainer eventualMaintainer2 = eventualIndex2.getIndexMaintainer(dataTable, pconn);
+
+ assertNotNull(strongMaintainer1);
+ assertNotNull(strongMaintainer2);
+ assertNotNull(eventualMaintainer1);
+ assertNotNull(eventualMaintainer2);
+
+ verifyIndexMaintainerConsistency(strongMaintainer1, null, dataTable);
+ verifyIndexMaintainerConsistency(strongMaintainer2, IndexConsistency.STRONG, dataTable);
+ verifyIndexMaintainerConsistency(eventualMaintainer1, IndexConsistency.EVENTUAL, dataTable);
+ verifyIndexMaintainerConsistency(eventualMaintainer2, IndexConsistency.EVENTUAL, dataTable);
+
+ RowKeySchema schema = RowKeySchema.EMPTY_SCHEMA;
+ IndexMaintainer maintainer = new IndexMaintainer(schema, false) {
+ };
+
+ assertNull(maintainer.getIndexConsistency());
+
+ assertTrue("Data table should have CDC index after creating eventually consistent index",
+ CDCUtil.hasCDCIndex(dataTable));
+ assertTrue("Data table should have active CDC index", CDCUtil.hasActiveCDCIndex(dataTable));
+
+ String cdcObjectName = CDCUtil.getCDCObjectName(dataTable, false);
+ assertNotNull("CDC object name should be retrievable from data table", cdcObjectName);
+
+ String expectedCDCName = "CDC_tablENam1.1_0-001";
+ assertEquals("CDC object name should follow expected convention", expectedCDCName,
+ cdcObjectName);
+
+ PTable cdcIndex = CDCUtil.getActiveCDCIndex(dataTable);
+ String expectedCDCIndexName = CDCUtil.getCDCIndexName(cdcObjectName);
+ assertEquals("CDC index name should follow expected convention", expectedCDCIndexName,
+ cdcIndex.getTableName().getString());
+
+ PTable cdcTable = pconn.getTable(new PTableKey(pconn.getTenantId(), "XYZ." + cdcObjectName));
+ assertEquals("CDC table should have CDC type", PTableType.CDC, cdcTable.getType());
+
+ List<PTable> indexes = dataTable.getIndexes();
+ assertEquals("Data table should have total 5 indexes", 5, indexes.size());
+ }
+ }
+
+ @Test
+ public void testAlterIndexConsistency() throws Exception {
+ String dataTableName = "\"ScheMa1234-_\".\"alterTable1.1_0-001\"";
+ String dataTableNameWithoutQuotes = "ScheMa1234-_.alterTable1.1_0-001";
+ String indexName = "\"alterIdx1.12._0-001\"";
+ String fullIdxNameWithoutQuotes = "ScheMa1234-_.alterIdx1.12._0-001";
+
+ try (Connection conn =
+ DriverManager.getConnection(getUrl(), PropertiesUtil.deepCopy(TEST_PROPERTIES))) {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+
+ conn.createStatement().execute(
+ "CREATE TABLE " + dataTableName + " (id INTEGER PRIMARY KEY, name VARCHAR, city VARCHAR)");
+ conn.createStatement()
+ .execute("CREATE INDEX " + indexName + " ON " + dataTableName + " (name) INCLUDE (city)");
+
+ PTable strongIndex =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), fullIdxNameWithoutQuotes));
+ PTable dataTable =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), dataTableNameWithoutQuotes));
+ assertNull(strongIndex.getIndexConsistency());
+ assertFalse("Data table should not have CDC index initially", CDCUtil.hasCDCIndex(dataTable));
+
+ IndexMaintainer strongMaintainer = strongIndex.getIndexMaintainer(dataTable, pconn);
+ verifyIndexMaintainerConsistency(strongMaintainer, null, dataTable);
+
+ conn.createStatement()
+ .execute("ALTER INDEX " + indexName + " ON " + dataTableName + " CONSISTENCY=EVENTUAL");
+ PTable eventualIndex =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), fullIdxNameWithoutQuotes));
+ dataTable = pconn.getTable(new PTableKey(pconn.getTenantId(), dataTableNameWithoutQuotes));
+
+ assertEquals(IndexConsistency.EVENTUAL, eventualIndex.getIndexConsistency());
+ assertTrue("Data table should have CDC index after altering to eventual consistency",
+ CDCUtil.hasCDCIndex(dataTable));
+
+ IndexMaintainer eventualMaintainer = eventualIndex.getIndexMaintainer(dataTable, pconn);
+ verifyIndexMaintainerConsistency(eventualMaintainer, IndexConsistency.EVENTUAL, dataTable);
+ conn.createStatement()
+ .execute("ALTER INDEX " + indexName + " ON " + dataTableName + " CONSISTENCY=STRONG");
+ PTable strongIdx =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), fullIdxNameWithoutQuotes));
+ assertEquals(IndexConsistency.STRONG, strongIdx.getIndexConsistency());
+ IndexMaintainer strongIdxMaintainer = strongIdx.getIndexMaintainer(dataTable, pconn);
+ verifyIndexMaintainerConsistency(strongIdxMaintainer, IndexConsistency.STRONG, dataTable);
+ }
+ }
+
+ private void verifyIndexMaintainerConsistency(IndexMaintainer maintainer,
+ IndexConsistency expectedConsistency, PTable dataTable) throws IOException {
+ assertEquals(expectedConsistency, maintainer.getIndexConsistency());
+ ServerCachingProtos.IndexMaintainer proto = IndexMaintainer.toProto(maintainer);
+ if (expectedConsistency != null) {
+ assertTrue(proto.hasIndexConsistency());
+ assertEquals(expectedConsistency.toString(), proto.getIndexConsistency());
+ } else {
+ assertFalse(proto.hasIndexConsistency());
+ }
+ IndexMaintainer deserializedMaintainer = IndexMaintainer.fromProto(proto,
+ dataTable.getRowKeySchema(), dataTable.getBucketNum() != null);
+ assertEquals(expectedConsistency, deserializedMaintainer.getIndexConsistency());
+ }
+
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
index 8533b59c7c..8c08bc0740 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
@@ -1073,15 +1073,30 @@ public class IndexToolIT extends BaseTest {
public static long verifyIndexTable(String fullTableName, String fullIndexName, Connection conn)
throws Exception {
+ return verifyIndexTable(fullTableName, fullIndexName, conn, true);
+ }
+
+ public static long verifyIndexTable(String fullTableName, String fullIndexName, Connection conn,
+ boolean dumpTables) throws Exception {
String schemaName = SchemaUtil.getSchemaNameFromFullName(fullTableName);
String tableName = SchemaUtil.getTableNameFromFullName(fullTableName);
String indexName = SchemaUtil.getTableNameFromFullName(fullIndexName);
+ String fullTableNameNormalized = null;
+ String fullIndexNameNormalized = null;
+ if (schemaName != null && !schemaName.isEmpty()) {
+ fullTableNameNormalized = SchemaUtil.normalizeIdentifier(schemaName) + "."
+ + SchemaUtil.normalizeIdentifier(tableName);
+ fullIndexNameNormalized = SchemaUtil.normalizeIdentifier(schemaName) + "."
+ + SchemaUtil.normalizeIdentifier(indexName);
+ }
try {
// This checks the state of every raw index row without rebuilding any row
IndexTool indexTool = IndexToolIT.runIndexTool(false, schemaName, tableName, indexName, null,
0, IndexTool.IndexVerifyType.ONLY);
- TestUtil.dumpTable(conn,
- TableName.valueOf(IndexVerificationOutputRepository.OUTPUT_TABLE_NAME));
+ if (dumpTables) {
+ TestUtil.dumpTable(conn,
+ TableName.valueOf(IndexVerificationOutputRepository.OUTPUT_TABLE_NAME));
+ }
Counters counters = indexTool.getJob().getCounters();
LOGGER.info(counters.toString());
assertEquals(0, counters.findCounter(REBUILT_INDEX_ROW_COUNT).getValue());
@@ -1111,7 +1126,8 @@ public class IndexToolIT extends BaseTest {
// The index scrutiny run will trigger index repair on all unverified rows, and they will be
// repaired or
// deleted (since the age threshold is set to zero ms for these tests
- PTable pIndexTable = conn.unwrap(PhoenixConnection.class).getTable(fullIndexName);
+ PTable pIndexTable = conn.unwrap(PhoenixConnection.class)
+ .getTable(fullIndexNameNormalized != null ? fullIndexNameNormalized : fullIndexName);
if (pIndexTable.getIndexType() != PTable.IndexType.UNCOVERED_GLOBAL) {
assertEquals(0, counters.findCounter(BEFORE_REBUILD_UNVERIFIED_INDEX_ROW_COUNT).getValue());
}
@@ -1150,14 +1166,19 @@ public class IndexToolIT extends BaseTest {
counters.findCounter(AFTER_REBUILD_BEYOND_MAXLOOKBACK_MISSING_INDEX_ROW_COUNT).getValue());
assertEquals(0,
counters.findCounter(AFTER_REBUILD_BEYOND_MAXLOOKBACK_INVALID_INDEX_ROW_COUNT).getValue());
- pConn.getQueryServices().clearTableRegionCache(TableName.valueOf(fullIndexName));
+ pConn.getQueryServices().clearTableRegionCache(TableName
+ .valueOf(fullIndexNameNormalized != null ? fullIndexNameNormalized : fullIndexName));
long actualRowCountAfterCompaction =
IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
assertEquals(actualRowCount, actualRowCountAfterCompaction);
return actualRowCount;
} catch (AssertionError e) {
- TestUtil.dumpTable(conn, TableName.valueOf(fullTableName));
- TestUtil.dumpTable(conn, TableName.valueOf(fullIndexName));
+ if (dumpTables) {
+ TestUtil.dumpTable(conn, TableName
+ .valueOf(fullTableNameNormalized != null ? fullTableNameNormalized : fullTableName));
+ TestUtil.dumpTable(conn, TableName
+ .valueOf(fullIndexNameNormalized != null ? fullIndexNameNormalized : fullIndexName));
+ }
throw e;
}
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataServerConnectionsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataServerConnectionsIT.java
index ecebf38016..bebabc9bbf 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataServerConnectionsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataServerConnectionsIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.query.QueryServices.DISABLE_VIEW_SUBTREE_VALIDATION;
import static org.apache.phoenix.query.QueryServicesTestImpl.DEFAULT_HCONNECTION_POOL_CORE_SIZE;
import static org.apache.phoenix.query.QueryServicesTestImpl.DEFAULT_HCONNECTION_POOL_MAX_SIZE;
@@ -81,6 +82,7 @@ public class MetadataServerConnectionsIT extends BaseTest {
Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
props.put(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
props.put(DISABLE_VIEW_SUBTREE_VALIDATION, "true");
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpTestDriver(new ReadOnlyProps(props));
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java
index 9299bc3816..ac0002a479 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java
@@ -68,14 +68,14 @@ public class MigrateSystemTablesToSystemNamespaceIT extends BaseTest {
private static final Set<String> HBASE_NAMESPACE_MAPPED_SYSTEM_TABLES =
new HashSet<>(Arrays.asList("SYSTEM:PHOENIX_INDEX_TOOL_RESULT", "SYSTEM:PHOENIX_INDEX_TOOL"));
- private static final Set<String> PHOENIX_SYSTEM_TABLES =
- new HashSet<>(Arrays.asList("SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS",
- "SYSTEM.FUNCTION", "SYSTEM.MUTEX", "SYSTEM.LOG", "SYSTEM.CHILD_LINK", "SYSTEM.TASK",
- "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM"));
- private static final Set<String> PHOENIX_NAMESPACE_MAPPED_SYSTEM_TABLES =
- new HashSet<>(Arrays.asList("SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS",
- "SYSTEM:FUNCTION", "SYSTEM:MUTEX", "SYSTEM:LOG", "SYSTEM:CHILD_LINK", "SYSTEM:TASK",
- "SYSTEM:TRANSFORM", "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM"));
+ private static final Set<String> PHOENIX_SYSTEM_TABLES = new HashSet<>(
+ Arrays.asList("SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS", "SYSTEM.FUNCTION",
+ "SYSTEM.MUTEX", "SYSTEM.LOG", "SYSTEM.CHILD_LINK", "SYSTEM.TASK", "SYSTEM.TRANSFORM",
+ "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.IDX_CDC_TRACKER"));
+ private static final Set<String> PHOENIX_NAMESPACE_MAPPED_SYSTEM_TABLES = new HashSet<>(
+ Arrays.asList("SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS", "SYSTEM:FUNCTION",
+ "SYSTEM:MUTEX", "SYSTEM:LOG", "SYSTEM:CHILD_LINK", "SYSTEM:TASK", "SYSTEM:TRANSFORM",
+ "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM", "SYSTEM:IDX_CDC_TRACKER"));
private static final String SCHEMA_NAME = "MIGRATETEST";
private static final String TABLE_NAME =
SCHEMA_NAME + "." + MigrateSystemTablesToSystemNamespaceIT.class.getSimpleName().toUpperCase();
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java
index 7d839f4c16..53adfd8554 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java
@@ -189,6 +189,11 @@ public class QueryDatabaseMetaDataIT extends ParallelStatsDisabledIT {
assertEquals(PTableType.SYSTEM.toString(), rs.getString("TABLE_TYPE"));
assertTrue(rs.next());
assertEquals(SYSTEM_CATALOG_SCHEMA, rs.getString("TABLE_SCHEM"));
+ assertEquals(PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_TABLE,
+ rs.getString("TABLE_NAME"));
+ assertEquals(PTableType.SYSTEM.toString(), rs.getString("TABLE_TYPE"));
+ assertTrue(rs.next());
+ assertEquals(SYSTEM_CATALOG_SCHEMA, rs.getString("TABLE_SCHEM"));
assertEquals(PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE, rs.getString("TABLE_NAME"));
assertEquals(PTableType.SYSTEM.toString(), rs.getString("TABLE_TYPE"));
assertTrue(rs.next());
@@ -425,7 +430,7 @@ public class QueryDatabaseMetaDataIT extends ParallelStatsDisabledIT {
tables.add(rs.getString("TABLE_NAME"));
assertEquals("SYSTEM", rs.getString("TABLE_SCHEM"));
}
- assertEquals(11, tables.size());
+ assertEquals(12, tables.size());
assertTrue(tables.contains("CATALOG"));
assertTrue(tables.contains("FUNCTION"));
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SkipSystemTablesExistenceCheckIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SkipSystemTablesExistenceCheckIT.java
index e24f0f69ff..3568628ec7 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SkipSystemTablesExistenceCheckIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SkipSystemTablesExistenceCheckIT.java
@@ -67,7 +67,7 @@ public class SkipSystemTablesExistenceCheckIT {
private static final Set<String> PHOENIX_SYSTEM_TABLES =
new HashSet<>(Arrays.asList("SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS",
"SYSTEM.FUNCTION", "SYSTEM.MUTEX", "SYSTEM.LOG", "SYSTEM.CHILD_LINK", "SYSTEM.TASK",
- "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM",
+ "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.IDX_CDC_TRACKER",
"SYSTEM.PHOENIX_INDEX_TOOL_RESULT", "SYSTEM.PHOENIX_INDEX_TOOL"));
private static class PhoenixSystemTablesCreationTestDriver extends PhoenixTestDriver {
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemTablesCreationOnConnectionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemTablesCreationOnConnectionIT.java
index f88385faa6..5d1cb86a2b 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemTablesCreationOnConnectionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemTablesCreationOnConnectionIT.java
@@ -107,13 +107,13 @@ public class SystemTablesCreationOnConnectionIT {
private static final Set<String> PHOENIX_SYSTEM_TABLES =
new HashSet<>(Arrays.asList("SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS",
"SYSTEM.FUNCTION", "SYSTEM.MUTEX", "SYSTEM.LOG", "SYSTEM.CHILD_LINK", "SYSTEM.TASK",
- "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM",
+ "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.IDX_CDC_TRACKER",
"SYSTEM.PHOENIX_INDEX_TOOL_RESULT", "SYSTEM.PHOENIX_INDEX_TOOL"));
private static final Set<String> PHOENIX_NAMESPACE_MAPPED_SYSTEM_TABLES =
new HashSet<>(Arrays.asList("SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS",
"SYSTEM:FUNCTION", "SYSTEM:MUTEX", "SYSTEM:LOG", "SYSTEM:CHILD_LINK", "SYSTEM:TASK",
- "SYSTEM:TRANSFORM", "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM",
+ "SYSTEM:TRANSFORM", "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM", "SYSTEM:IDX_CDC_TRACKER",
"SYSTEM:PHOENIX_INDEX_TOOL_RESULT", "SYSTEM:PHOENIX_INDEX_TOOL"));
private static class PhoenixSysCatCreationServices extends ConnectionQueryServicesImpl {
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java
index bf396d25d9..07ca749575 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java
@@ -578,6 +578,9 @@ public class TenantSpecificTablesDDLIT extends BaseTenantSpecificTablesIT {
assertTrue(rs.next());
assertTableMetaData(rs, SYSTEM_CATALOG_SCHEMA, SYSTEM_FUNCTION_TABLE, SYSTEM);
assertTrue(rs.next());
+ assertTableMetaData(rs, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
+ PhoenixDatabaseMetaData.SYSTEM_IDX_CDC_TRACKER_TABLE, PTableType.SYSTEM);
+ assertTrue(rs.next());
assertTableMetaData(rs, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE, PTableType.SYSTEM);
assertTrue(rs.next());
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UCFWithServerMetadataIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UCFWithServerMetadataIT.java
index 6a19bec2e6..fe29929456 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UCFWithServerMetadataIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UCFWithServerMetadataIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.junit.Assert.assertEquals;
import com.google.protobuf.RpcCallback;
@@ -100,6 +101,7 @@ public class UCFWithServerMetadataIT extends BaseTest {
public static synchronized void doSetup() throws Exception {
Map<String, String> props = new HashMap<>(1);
props.put(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpTestDriver(new ReadOnlyProps(props));
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
index 588e08b6b6..1bf090cdae 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_PHOENIX_CONNECTIONS;
import static org.apache.phoenix.util.PhoenixRuntime.TENANT_ID_ATTRIB;
import static org.apache.phoenix.util.PhoenixRuntime.UPSERT_BATCH_SIZE_ATTRIB;
@@ -88,6 +89,7 @@ public class UpsertSelectIT extends ParallelStatsDisabledIT {
// any addition to GLOBAL_OPEN_PHOENIX_CONNECTIONS
props.put(QueryServices.TASK_HANDLING_INTERVAL_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
props.put(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded2IT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded2IT.java
index c5213df866..bddd8d384a 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded2IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/VarBinaryEncoded2IT.java
@@ -24,40 +24,77 @@ import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
-import java.util.Arrays;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.compile.ExplainPlan;
import org.apache.phoenix.compile.ExplainPlanAttributes;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
+import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
import org.junit.Assert;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
@Category(ParallelStatsDisabledTest.class)
@RunWith(Parameterized.class)
public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
+ private static final long EVENTUAL_CONSISTENCY_WAIT_MS = 17000;
+
private final boolean columnEncoded;
private final boolean coveredIndex;
private final boolean isBindStatement;
+ private final boolean eventual;
- public VarBinaryEncoded2IT(boolean columnEncoded, boolean coveredIndex, boolean isBindStatement) {
+ public VarBinaryEncoded2IT(boolean columnEncoded, boolean coveredIndex, boolean isBindStatement,
+ boolean eventual) {
this.columnEncoded = columnEncoded;
this.coveredIndex = coveredIndex;
this.isBindStatement = isBindStatement;
+ this.eventual = eventual;
}
- @Parameterized.Parameters(name = "VarBinary2IT_columnEncoded={0}, coveredIndex={1}")
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
+ props.put(BaseScannerRegionObserverConstants.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY,
+ Integer.toString(60 * 60));
+ props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION, Boolean.toString(false));
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Parameterized.Parameters(
+ name = "VarBinary2IT_columnEncoded={0}, coveredIndex={1}, isBindStatement={2}, eventual={3}")
public static synchronized Collection<Object[]> data() {
- return Arrays.asList(new Object[][] { { false, false, false }, { false, true, false },
- { true, false, false }, { true, true, false }, { false, false, true }, { false, true, true },
- { true, false, true }, { true, true, true } });
+ List<Object[]> params = new ArrayList<>();
+ for (boolean columnEncoded : new boolean[] { false, true }) {
+ for (boolean coveredIndex : new boolean[] { false, true }) {
+ for (boolean isBindStatement : new boolean[] { false, true }) {
+ for (boolean eventual : new boolean[] { false, true }) {
+ params.add(new Object[] { columnEncoded, coveredIndex, isBindStatement, eventual });
+ }
+ }
+ }
+ }
+ return params;
+ }
+
+ private void waitForEventualConsistency() throws InterruptedException {
+ if (eventual) {
+ Thread.sleep(EVENTUAL_CONSISTENCY_WAIT_MS);
+ }
}
@Test
@@ -73,12 +110,13 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
+ " COL3 VARBINARY_ENCODED CONSTRAINT pk PRIMARY KEY(PK1, PK2, PK3)) "
+ (this.columnEncoded ? "" : "COLUMN_ENCODED_BYTES=0"));
+ String consistencyClause = eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG";
if (this.coveredIndex) {
- conn.createStatement().execute(
- "CREATE INDEX " + indexName + " ON " + tableName + " (COL1, COL2) INCLUDE (COL3)");
+ conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2) INCLUDE (COL3)" + consistencyClause);
} else {
- conn.createStatement()
- .execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName + " (COL1, COL2)");
+ conn.createStatement().execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2)" + consistencyClause);
}
byte[] b1 = new byte[] { 1, 1, 19, -28, 24, 1, 1, -11, -21, 1 };
@@ -152,6 +190,7 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
upsertRow(preparedStatement, b16, b26, b36, b46, b56, b66);
}
conn.commit();
+ waitForEventualConsistency();
PreparedStatement preparedStatement =
conn.prepareStatement("SELECT * FROM " + tableName + " WHERE COL1 = ? AND COL2 = ?");
@@ -367,12 +406,13 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
+ " COL3 VARBINARY_ENCODED CONSTRAINT pk PRIMARY KEY(PK1, PK2, PK3)) "
+ (this.columnEncoded ? "" : "COLUMN_ENCODED_BYTES=0"));
+ String consistencyClause = eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG";
if (this.coveredIndex) {
- conn.createStatement().execute(
- "CREATE INDEX " + indexName + " ON " + tableName + " (COL1, COL2) INCLUDE (COL3)");
+ conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2) INCLUDE (COL3)" + consistencyClause);
} else {
- conn.createStatement()
- .execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName + " (COL1, COL2)");
+ conn.createStatement().execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2)" + consistencyClause);
}
byte[] b1 = new byte[] { 1, 1, 19, -28, 24, 1, 1, -11, -21, 1 };
@@ -449,6 +489,7 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
upsertRow(preparedStatement, b16, b26, b36, b46, b56, b66);
}
conn.commit();
+ waitForEventualConsistency();
PreparedStatement preparedStatement =
conn.prepareStatement("SELECT * FROM " + tableName + " WHERE COL1 = ? AND COL2 = ?");
@@ -664,12 +705,13 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
+ " COL3 VARBINARY_ENCODED CONSTRAINT pk PRIMARY KEY(PK1, PK2, PK3)) "
+ (this.columnEncoded ? "" : "COLUMN_ENCODED_BYTES=0"));
+ String consistencyClause = eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG";
if (this.coveredIndex) {
- conn.createStatement().execute(
- "CREATE INDEX " + indexName + " ON " + tableName + " (COL1, COL2) INCLUDE (COL3)");
+ conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2) INCLUDE (COL3)" + consistencyClause);
} else {
- conn.createStatement()
- .execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName + " (COL1, COL2)");
+ conn.createStatement().execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2)" + consistencyClause);
}
byte[] b1 = new byte[] { 1, 1, 19, -28, 24, 1, 1, -11, -21, 1 };
@@ -743,6 +785,7 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
upsertRow(preparedStatement, b16, b26, b36, b46, b56, b66);
}
conn.commit();
+ waitForEventualConsistency();
PreparedStatement preparedStatement =
conn.prepareStatement("SELECT * FROM " + tableName + " WHERE COL1 = ? AND COL2 = ?");
@@ -951,12 +994,13 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
+ " COL3 VARBINARY_ENCODED CONSTRAINT pk PRIMARY KEY(PK1, PK2, PK3)) "
+ (this.columnEncoded ? "" : "COLUMN_ENCODED_BYTES=0"));
+ String consistencyClause = eventual ? " CONSISTENCY = EVENTUAL" : " CONSISTENCY = STRONG";
if (this.coveredIndex) {
- conn.createStatement().execute(
- "CREATE INDEX " + indexName + " ON " + tableName + " (COL1, COL2) INCLUDE (COL3)");
+ conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2) INCLUDE (COL3)" + consistencyClause);
} else {
- conn.createStatement()
- .execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName + " (COL1, COL2)");
+ conn.createStatement().execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName
+ + " (COL1, COL2)" + consistencyClause);
}
byte[] b1 = new byte[] { 1, 1, 19, -28, 24, 1, 1, -11, -21, 1 };
@@ -1027,6 +1071,7 @@ public class VarBinaryEncoded2IT extends ParallelStatsDisabledIT {
upsertRow(preparedStatement, b16, b26, b36, b46, b56, b66);
}
conn.commit();
+ waitForEventualConsistency();
PreparedStatement preparedStatement =
conn.prepareStatement("SELECT * FROM " + tableName + " WHERE COL1 = ? AND COL2 = ?");
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/GlobalIndexCheckerWithRegionMovesIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/GlobalIndexCheckerWithRegionMovesIT.java
index 8632eaf013..6583df2d84 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/GlobalIndexCheckerWithRegionMovesIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/GlobalIndexCheckerWithRegionMovesIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end.index;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.mapreduce.PhoenixJobCounters.INPUT_RECORDS;
import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_BEYOND_MAXLOOKBACK_INVALID_INDEX_ROW_COUNT;
import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_BEYOND_MAXLOOKBACK_MISSING_INDEX_ROW_COUNT;
@@ -142,6 +143,7 @@ public class GlobalIndexCheckerWithRegionMovesIT extends BaseTest {
props.put(QueryServices.PHOENIX_POST_DUMMY_PROCESS,
TestScanningResultPostDummyResultCaller.class.getName());
props.put(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB, Long.toString(0));
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexAsyncThresholdIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexAsyncThresholdIT.java
index 4db0b3c5bd..6b6f3059b3 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexAsyncThresholdIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexAsyncThresholdIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end.index;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_STARTUP_DELAY_MS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -100,6 +101,7 @@ public class IndexAsyncThresholdIT extends BaseTest {
Configuration conf = HBaseConfiguration.create();
Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
props.put(QueryServices.CLIENT_INDEX_ASYNC_THRESHOLD, Long.toString(threshold));
+ props.put(INDEX_CDC_CONSUMER_STARTUP_DELAY_MS, Integer.toString(Integer.MAX_VALUE));
url = setUpTestCluster(conf, new ReadOnlyProps(props.entrySet().iterator()));
driver = initAndRegisterTestDriver(url, new ReadOnlyProps(props.entrySet().iterator()));
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialSystemCatalogIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialSystemCatalogIndexIT.java
index 2a46162e8b..8184f92396 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialSystemCatalogIndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialSystemCatalogIndexIT.java
@@ -18,7 +18,7 @@
package org.apache.phoenix.end2end.index;
import static org.apache.phoenix.exception.SQLExceptionCode.CANNOT_INDEX_SYSTEM_TABLE;
-import static org.apache.phoenix.exception.SQLExceptionCode.MISMATCHED_TOKEN;
+import static org.apache.phoenix.exception.SQLExceptionCode.PARSER_ERROR;
import static org.apache.phoenix.query.PhoenixTestBuilder.DDLDefaults.COLUMN_TYPES;
import static org.apache.phoenix.query.PhoenixTestBuilder.DDLDefaults.TENANT_VIEW_COLUMNS;
import static org.apache.phoenix.query.QueryServices.SYSTEM_CATALOG_INDEXES_ENABLED;
@@ -907,7 +907,7 @@ public class PartialSystemCatalogIndexIT extends ParallelStatsDisabledIT {
"CREATE INDEX IF NOT EXISTS SYS_INDEX_FUNCTION_IDX ON SYSTEM.FUNCTION(CLASS_NAME,JAR_PATH) WHERE FUNCTION_NAME = '4'");
fail();
} catch (SQLException sqle) {
- Assert.assertEquals(MISMATCHED_TOKEN.getErrorCode(), sqle.getErrorCode());
+ Assert.assertEquals(PARSER_ERROR.getErrorCode(), sqle.getErrorCode());
}
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ReplicationWithWALAnnotationIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ReplicationWithWALAnnotationIT.java
index 3293a9525a..9848452bd8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ReplicationWithWALAnnotationIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/ReplicationWithWALAnnotationIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end.index;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_STARTUP_DELAY_MS;
import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -151,6 +152,7 @@ public class ReplicationWithWALAnnotationIT extends BaseTest {
ReplicationSinkEndpoint.class.getName());
conf1.setBoolean(IndexRegionObserver.PHOENIX_APPEND_METADATA_TO_WAL, true);
conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
+ conf1.set(INDEX_CDC_CONSUMER_STARTUP_DELAY_MS, Integer.toString(Integer.MAX_VALUE));
setUpConfigForMiniCluster(conf1);
utility1 = new HBaseTestingUtility(conf1);
@@ -165,6 +167,7 @@ public class ReplicationWithWALAnnotationIT extends BaseTest {
conf2.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 6);
conf2.setBoolean("dfs.support.append", true);
conf2.setBoolean("hbase.tests.use.shortcircuit.reads", false);
+ conf2.set(INDEX_CDC_CONSUMER_STARTUP_DELAY_MS, Integer.toString(Integer.MAX_VALUE));
utility2 = new HBaseTestingUtility(conf2);
utility2.startMiniZKCluster();
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScanner2IT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScanner2IT.java
index a008150d6b..555f87ff0c 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScanner2IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScanner2IT.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.end2end.index;
import static org.apache.phoenix.end2end.index.GlobalIndexCheckerIT.assertExplainPlan;
import static org.apache.phoenix.end2end.index.GlobalIndexCheckerIT.assertExplainPlanWithLimit;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -133,6 +134,7 @@ public class UncoveredGlobalIndexRegionScanner2IT extends BaseTest {
props.put(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB, Long.toString(0));
props.put(QueryServices.PHOENIX_POST_DUMMY_PROCESS,
TestScanningResultPostDummyResultCaller.class.getName());
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnectionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnectionIT.java
index f633294df2..7ee3f49b9b 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnectionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnectionIT.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.jdbc;
import static org.apache.hadoop.test.GenericTestUtils.waitFor;
import static org.apache.phoenix.exception.SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_STARTUP_DELAY_MS;
import static org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
import static org.apache.phoenix.jdbc.HighAvailabilityGroup.URLS;
import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.HBaseTestingUtilityPair;
@@ -99,6 +100,10 @@ public class FailoverPhoenixConnectionIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
+ CLUSTERS.getHBaseCluster1().getConfiguration().setInt(INDEX_CDC_CONSUMER_STARTUP_DELAY_MS,
+ Integer.MAX_VALUE);
+ CLUSTERS.getHBaseCluster2().getConfiguration().setInt(INDEX_CDC_CONSUMER_STARTUP_DELAY_MS,
+ Integer.MAX_VALUE);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAConnectionWithMasterAndRPCRegistryIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAConnectionWithMasterAndRPCRegistryIT.java
index 1464db36df..46674b61db 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAConnectionWithMasterAndRPCRegistryIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAConnectionWithMasterAndRPCRegistryIT.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.jdbc;
import static org.apache.hadoop.test.GenericTestUtils.waitFor;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.HBaseTestingUtilityPair.doTestWhenOneHBaseDown;
import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.doTestBasicOperationsWithConnection;
@@ -86,6 +87,10 @@ public class HAConnectionWithMasterAndRPCRegistryIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
+ CLUSTERS.getHBaseCluster1().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
+ CLUSTERS.getHBaseCluster2().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HighAvailabilityTestingUtilityIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HighAvailabilityTestingUtilityIT.java
index 1564beda21..7453c59903 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HighAvailabilityTestingUtilityIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HighAvailabilityTestingUtilityIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.jdbc;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.HBaseTestingUtilityPair;
import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.HBaseTestingUtilityPair.doTestWhenOneHBaseDown;
@@ -81,6 +82,10 @@ public class HighAvailabilityTestingUtilityIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
+ CLUSTERS.getHBaseCluster1().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
+ CLUSTERS.getHBaseCluster2().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionFallbackIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionFallbackIT.java
index 076e418636..3a1ae7d311 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionFallbackIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionFallbackIT.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.jdbc;
import static org.apache.hadoop.test.GenericTestUtils.waitFor;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
import static org.apache.phoenix.jdbc.HighAvailabilityPolicy.PARALLEL;
import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.HBaseTestingUtilityPair;
@@ -57,6 +58,10 @@ public class ParallelPhoenixConnectionFallbackIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
haGroupName = ParallelPhoenixConnectionFallbackIT.class.getSimpleName();
+ CLUSTERS.getHBaseCluster1().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
+ CLUSTERS.getHBaseCluster2().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
PROPERTIES.setProperty(PHOENIX_HA_GROUP_ATTR, haGroupName);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSetIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSetIT.java
index d14f00ae8a..04b385f762 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSetIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSetIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.jdbc;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.jdbc.ClusterRoleRecordGeneratorTool.PHOENIX_HA_GROUP_STORE_PEER_ID_DEFAULT;
import static org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
import static org.apache.phoenix.jdbc.HighAvailabilityPolicy.PARALLEL;
@@ -72,6 +73,10 @@ public class ParallelPhoenixNullComparingResultSetIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
String haGroupName = ParallelPhoenixNullComparingResultSetIT.class.getSimpleName();
+ CLUSTERS.getHBaseCluster1().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
+ CLUSTERS.getHBaseCluster2().getConfiguration().setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED,
+ false);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
PROPERTIES.setProperty(PHOENIX_HA_GROUP_ATTR, haGroupName);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java
index 05538f5da4..fe059f8451 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.monitoring;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
@@ -65,6 +66,7 @@ public abstract class BasePhoenixMetricsIT extends BaseTest {
props.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS, String.valueOf(true));
// disable renewing leases as this will force spooling to happen.
props.put(QueryServices.RENEW_LEASE_ENABLED, String.valueOf(false));
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
// need the non-test driver for some tests that check number of hconnections, etc.
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java
index fdaf8a8639..4ca910964a 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java
@@ -20,6 +20,7 @@ package org.apache.phoenix.monitoring;
import static org.apache.phoenix.exception.SQLExceptionCode.DATA_EXCEEDS_MAX_CAPACITY;
import static org.apache.phoenix.exception.SQLExceptionCode.GET_TABLE_REGIONS_FAIL;
import static org.apache.phoenix.exception.SQLExceptionCode.OPERATION_TIMED_OUT;
+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.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BYTES;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_QUERY_TIME;
@@ -158,6 +159,7 @@ public class PhoenixTableLevelMetricsIT extends BaseTest {
@BeforeClass
public static void doSetup() throws Exception {
final Configuration conf = HBaseConfiguration.create();
+ conf.set(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpConfigForMiniCluster(conf);
conf.set(QueryServices.TABLE_LEVEL_METRICS_ENABLED, String.valueOf(true));
conf.set(QueryServices.METRIC_PUBLISHER_ENABLED, String.valueOf(true));
@@ -189,6 +191,7 @@ public class PhoenixTableLevelMetricsIT extends BaseTest {
Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
props.put(BaseTest.DRIVER_CLASS_NAME_ATTRIB, PhoenixMetricsTestingDriver.class.getName());
props.put(ENABLE_SERVER_UPSERT_SELECT, "true");
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
initAndRegisterTestDriver(url, new ReadOnlyProps(props.entrySet().iterator()));
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsIT.java
index 54d53afdf6..57791072cd 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.monitoring.connectionqueryservice;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.monitoring.MetricType.OPEN_INTERNAL_PHOENIX_CONNECTIONS_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.OPEN_PHOENIX_CONNECTIONS_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.PHOENIX_CONNECTIONS_THROTTLED_COUNTER;
@@ -98,6 +99,7 @@ public class ConnectionQueryServicesMetricsIT extends BaseTest {
// counter won't be increased at all. So we need to set max allowed connection count
conf.set(CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS, "2");
conf.set(INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS, "1");
+ conf.set(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
return conf;
}
@@ -109,6 +111,7 @@ public class ConnectionQueryServicesMetricsIT extends BaseTest {
// counter won't be increased at all. So we need to set max allowed connection count
conf.set(CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS, "2");
conf.set(INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS, "1");
+ conf.set(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
Configuration copy = new Configuration(conf);
copy.addResource(confToClone);
return copy;
@@ -117,6 +120,7 @@ public class ConnectionQueryServicesMetricsIT extends BaseTest {
Configuration conf = HBaseFactoryProvider.getConfigurationFactory().getConfiguration();
conf.set(QueryServices.TASK_HANDLING_INTERVAL_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
conf.set(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
+ conf.set(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
hbaseTestUtil = new HBaseTestingUtility(conf);
setUpConfigForMiniCluster(conf);
conf.set(QueryServices.EXTRA_JDBC_ARGUMENTS_ATTRIB,
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/query/MetaDataCachingIT.java b/phoenix-core/src/it/java/org/apache/phoenix/query/MetaDataCachingIT.java
index 03e70ddf77..c84d62c76a 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/query/MetaDataCachingIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/query/MetaDataCachingIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.query;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.junit.Assert.*;
import java.sql.Connection;
@@ -67,6 +68,7 @@ public class MetaDataCachingIT extends BaseTest {
props.put(QueryServices.CLIENT_CACHE_ENCODING, "object");
props.put(QueryServices.TASK_HANDLING_INTERVAL_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
props.put(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB, Long.toString(Long.MAX_VALUE));
+ props.put(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/parse/IndexConsistencyParseTest.java b/phoenix-core/src/test/java/org/apache/phoenix/parse/IndexConsistencyParseTest.java
new file mode 100644
index 0000000000..559a57d3f8
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/parse/IndexConsistencyParseTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.parse;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import org.apache.phoenix.schema.types.IndexConsistency;
+import org.junit.Test;
+
+/**
+ * Test class for parsing CREATE INDEX statements with CONSISTENCY clause.
+ */
+public class IndexConsistencyParseTest {
+
+ @Test
+ public void testCreateIndexWithStrongConsistency() throws Exception {
+ String sql = "CREATE INDEX idx ON mytable (col1) CONSISTENCY=STRONG";
+ SQLParser parser = new SQLParser(sql);
+ CreateIndexStatement stmt = (CreateIndexStatement) parser.parseStatement();
+
+ assertNotNull("Statement should not be null", stmt);
+ assertEquals("Should have STRONG consistency", IndexConsistency.STRONG,
+ stmt.getIndexConsistency());
+ }
+
+ @Test
+ public void testCreateIndexWithEventualConsistency() throws Exception {
+ String sql = "CREATE INDEX idx ON mytable (col1) CONSISTENCY=EVENTUAL";
+ SQLParser parser = new SQLParser(sql);
+ CreateIndexStatement stmt = (CreateIndexStatement) parser.parseStatement();
+
+ assertNotNull("Statement should not be null", stmt);
+ assertEquals("Should have EVENTUAL consistency", IndexConsistency.EVENTUAL,
+ stmt.getIndexConsistency());
+ }
+
+ @Test
+ public void testCreateIndexWithoutConsistency() throws Exception {
+ String sql = "CREATE INDEX idx ON mytable (col1)";
+ SQLParser parser = new SQLParser(sql);
+ CreateIndexStatement stmt = (CreateIndexStatement) parser.parseStatement();
+
+ assertNotNull("Statement should not be null", stmt);
+ assertNull("Should default to STRONG consistency", stmt.getIndexConsistency());
+ }
+
+ @Test
+ public void testCreateIndexWithConsistencyAndOtherClauses() throws Exception {
+ String sql = "CREATE INDEX idx ON mytable (col1) INCLUDE (col2) ASYNC CONSISTENCY=EVENTUAL";
+ SQLParser parser = new SQLParser(sql);
+ CreateIndexStatement stmt = (CreateIndexStatement) parser.parseStatement();
+
+ assertNotNull("Statement should not be null", stmt);
+ assertEquals("Should have EVENTUAL consistency", IndexConsistency.EVENTUAL,
+ stmt.getIndexConsistency());
+ assertEquals("Should have include columns", 1, stmt.getIncludeColumns().size());
+ assertEquals("Should be async", true, stmt.isAsync());
+ }
+
+ @Test
+ public void testLocalIndexWithConsistency() throws Exception {
+ String sql = "CREATE LOCAL INDEX idx ON mytable (col1) CONSISTENCY=EVENTUAL";
+ SQLParser parser = new SQLParser(sql);
+ CreateIndexStatement stmt = (CreateIndexStatement) parser.parseStatement();
+
+ assertNotNull("Statement should not be null", stmt);
+ assertEquals("Should have EVENTUAL consistency", IndexConsistency.EVENTUAL,
+ stmt.getIndexConsistency());
+ }
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/IndexScrutiny.java b/phoenix-core/src/test/java/org/apache/phoenix/util/IndexScrutiny.java
index 75cc9e3843..565d7aec7d 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/IndexScrutiny.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/IndexScrutiny.java
@@ -39,7 +39,22 @@ public class IndexScrutiny {
public static long scrutinizeIndex(Connection conn, String fullTableName, String fullIndexName)
throws SQLException {
PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
- PTable ptable = pconn.getTable(new PTableKey(pconn.getTenantId(), fullTableName));
+ String schemaName = SchemaUtil.getSchemaNameFromFullName(fullTableName);
+ String tableName = SchemaUtil.getTableNameFromFullName(fullTableName);
+ String indexName = SchemaUtil.getTableNameFromFullName(fullIndexName);
+ PTable ptable;
+ PTable pindex;
+ if (schemaName != null && !schemaName.isEmpty()) {
+ ptable =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), SchemaUtil.normalizeIdentifier(schemaName)
+ + "." + SchemaUtil.normalizeIdentifier(tableName)));
+ pindex =
+ pconn.getTable(new PTableKey(pconn.getTenantId(), SchemaUtil.normalizeIdentifier(schemaName)
+ + "." + SchemaUtil.normalizeIdentifier(indexName)));
+ } else {
+ ptable = pconn.getTable(new PTableKey(pconn.getTenantId(), fullTableName));
+ pindex = pconn.getTable(new PTableKey(pconn.getTenantId(), fullIndexName));
+ }
int tableColumnOffset = 0;
List<PColumn> tableColumns = ptable.getColumns();
List<PColumn> tablePKColumns = ptable.getPKColumns();
@@ -48,7 +63,6 @@ public class IndexScrutiny {
tableColumns = tableColumns.subList(tableColumnOffset, tableColumns.size());
tablePKColumns = tablePKColumns.subList(tableColumnOffset, tablePKColumns.size());
}
- PTable pindex = pconn.getTable(new PTableKey(pconn.getTenantId(), fullIndexName));
List<PColumn> indexColumns = pindex.getColumns();
int indexColumnOffset = 0;
if (pindex.getBucketNum() != null) {
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java b/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java
index 6541dd7338..69a191b5c2 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/TestUtil.java
@@ -78,6 +78,7 @@ import org.apache.hadoop.hbase.client.CoprocessorDescriptorBuilder;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionLocator;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
@@ -1694,4 +1695,97 @@ public class TestUtil {
return sb.toString();
}
+ public static void splitTable(String url, String tableName, int nSplits,
+ long sleepBetweenSplitsMs) {
+ try (
+ PhoenixConnection phoenixConn =
+ DriverManager.getConnection(url).unwrap(PhoenixConnection.class);
+ Admin admin = phoenixConn.getQueryServices().getAdmin()) {
+ TableName hTableName = TableName.valueOf(tableName);
+ int splitCount = 0;
+ while (splitCount < nSplits) {
+ Thread.sleep(sleepBetweenSplitsMs);
+ try {
+ int regionsBefore = admin.getRegions(hTableName).size();
+ admin.split(hTableName);
+ int retries = 0;
+ while (retries < 50) {
+ int regionsAfter = admin.getRegions(hTableName).size();
+ if (regionsAfter >= regionsBefore * 2) {
+ LOGGER.info("Split {} completed, regions: {} -> {}", splitCount + 1, regionsBefore,
+ regionsAfter);
+ break;
+ }
+ retries++;
+ Thread.sleep(100);
+ }
+ splitCount++;
+ } catch (Exception e) {
+ LOGGER.warn("Exception during split: " + e.getMessage());
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error in split thread", e);
+ }
+ }
+
+ /**
+ * Merges table regions in groups of nMerges consecutive regions.
+ * @param url connection URL.
+ * @param tableName name of the table.
+ * @param nMerges number of consecutive regions to merge together in each group.
+ * @param sleepBetweenMergesMs time to sleep between merge operations.
+ */
+ public static void mergeTableRegions(String url, String tableName, int nMerges,
+ long sleepBetweenMergesMs) {
+ if (nMerges < 2) {
+ throw new IllegalArgumentException("nMerges must be at least 2");
+ }
+ try (
+ PhoenixConnection phoenixConn =
+ DriverManager.getConnection(url).unwrap(PhoenixConnection.class);
+ Admin admin = phoenixConn.getQueryServices().getAdmin()) {
+ TableName hTableName = TableName.valueOf(tableName);
+ List<RegionInfo> regions = admin.getRegions(hTableName);
+ if (regions.size() < 2) {
+ return;
+ }
+ int regionIndex = 0;
+ while (regionIndex < regions.size()) {
+ int remainingRegions = regions.size() - regionIndex;
+ int groupSize = Math.min(nMerges, remainingRegions);
+ if (groupSize < 2) {
+ break;
+ }
+ byte[][] regionsToMerge = new byte[groupSize][];
+ for (int i = 0; i < groupSize; i++) {
+ regionsToMerge[i] = regions.get(regionIndex + i).getRegionName();
+ }
+ try {
+ int regionsBefore = admin.getRegions(hTableName).size();
+ admin.mergeRegionsAsync(regionsToMerge, false).get();
+ int retries = 0;
+ while (retries < 50) {
+ List<RegionInfo> currentRegions = admin.getRegions(hTableName);
+ if (currentRegions.size() <= regionsBefore - groupSize + 1) {
+ LOGGER.info("Merge completed, regions: {} -> {}", regionsBefore,
+ currentRegions.size());
+ break;
+ }
+ retries++;
+ Thread.sleep(100);
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Exception during merge: " + e.getMessage(), e);
+ }
+ regionIndex += groupSize;
+ if (regionIndex < regions.size()) {
+ Thread.sleep(sleepBetweenMergesMs);
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error in mergeTableRegions", e);
+ }
+ }
+
}