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/06 06:35:04 UTC
(phoenix) branch tmp-ec updated: changes
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
The following commit(s) were added to refs/heads/tmp-ec by this push:
new 5c281b4aa2 changes
5c281b4aa2 is described below
commit 5c281b4aa21de252b81fcd193dfa500e3624f667
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Thu Feb 5 22:30:14 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/FailoverPhoenixConnection2IT.java | 6 +
.../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 +
.../phoenix/query/MaxConcurrentConnectionsIT.java | 2 +
.../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 +++
73 files changed, 2860 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 0da85f851d..b1d2567d07 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;
@@ -279,6 +280,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;
@@ -347,9 +349,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 (?,?)";
@@ -401,6 +403,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
@@ -1065,7 +1071,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() /*
@@ -1898,7 +1904,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);
}
@@ -1908,6 +1914,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);
@@ -2021,7 +2034,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);
@@ -2444,12 +2457,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;
@@ -3399,9 +3437,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)
@@ -3706,6 +3745,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) {
@@ -3809,6 +3858,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)
@@ -5928,6 +5978,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)
@@ -5941,13 +6022,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);
@@ -5958,10 +6034,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 =
@@ -6050,8 +6122,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);
@@ -6060,10 +6132,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 9d3d0b3ca4..053976b586 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;
@@ -263,8 +264,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;
@@ -418,6 +421,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);
@@ -436,7 +441,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);
@@ -498,6 +503,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);
@@ -1586,6 +1592,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..e2a93572dc
--- /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(10000);
+ 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 = 5000000)
+ 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 db0ada9832..23f083f096 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
@@ -1155,15 +1155,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.getOutputTableName()));
+ if (dumpTables) {
+ TestUtil.dumpTable(conn,
+ TableName.valueOf(IndexVerificationOutputRepository.getOutputTableName()));
+ }
Counters counters = indexTool.getJob().getCounters();
LOGGER.info(counters.toString());
assertEquals(0, counters.findCounter(REBUILT_INDEX_ROW_COUNT).getValue());
@@ -1193,7 +1208,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());
}
@@ -1232,14 +1248,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/FailoverPhoenixConnection2IT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection2IT.java
index 7e348a0902..68221e6340 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection2IT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection2IT.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.doTestWhenOneZKDown;
import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.doTestBasicOperationsWithConnection;
@@ -76,6 +77,11 @@ public class FailoverPhoenixConnection2IT {
@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/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/MaxConcurrentConnectionsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
index e9df1d0018..2a28c61e78 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.query;
import static org.apache.phoenix.exception.SQLExceptionCode.NEW_INTERNAL_CONNECTION_THROTTLED;
+import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_PHOENIX_CONNECTIONS;
import static org.apache.phoenix.query.QueryServices.CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS;
@@ -60,6 +61,7 @@ public class MaxConcurrentConnectionsIT extends BaseTest {
@BeforeClass
public static void setUp() throws Exception {
final Configuration conf = HBaseConfiguration.create();
+ conf.set(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, Boolean.toString(false));
setUpConfigForMiniCluster(conf);
hbaseTestUtil = new HBaseTestingUtility(conf);
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);
+ }
+ }
+
}