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 2023/11/21 06:45:08 UTC
(phoenix) branch PHOENIX-7001-feature updated (e3ca004315 -> f7a4feeccc)
This is an automated email from the ASF dual-hosted git repository.
vjasani pushed a change to branch PHOENIX-7001-feature
in repository https://gitbox.apache.org/repos/asf/phoenix.git
from e3ca004315 Merge branch 'master' into PHOENIX-7001-feature
add 5d5aaca213 PHOENIX-7038 : Implement Connection Query Service Metrics (#1682)
add 2feb70fc2e PHOENIX-7032 Partial Global Secondary Indexes (#1701)
add 20529a62b8 PHOENIX-7102 phoenix-connectors doesn't compile with core HEAD
add 62ccb3b815 PHOENIX-7067 View indexes should be created only on non overlapping updatable views (#1709)
add b64a9736b7 PHOENIX-7095 Implement Statement.closeOnCompletion() and fix related close() bugs
add 5128ad0e7d PHOENIX-7076 : MetaDataRegionObserver#postOpen hook improvements (#1735)
new f7a4feeccc Merge branch 'master' into PHOENIX-7001-feature
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/phoenix/end2end/AlterTableIT.java | 75 +-
.../phoenix/end2end/ClientHashAggregateIT.java | 38 +-
.../java/org/apache/phoenix/end2end/InListIT.java | 7 +-
.../phoenix/end2end/MetaDataEndpointImplIT.java | 26 +-
.../phoenix/end2end/QueryDatabaseMetaDataIT.java | 50 +
.../end2end/RebuildIndexConnectionPropsIT.java | 4 +-
.../end2end/ViewExtendsPkRestrictionsIT.java | 1169 ++++++++++++++++++++
.../java/org/apache/phoenix/end2end/ViewTTLIT.java | 65 +-
.../phoenix/end2end/index/PartialIndexIT.java | 767 +++++++++++++
.../index/UncoveredGlobalIndexRegionScannerIT.java | 9 +-
.../ConnectionQueryServicesMetricsIT.java | 366 ++++++
phoenix-core/src/main/antlr3/PhoenixSQL.g | 9 +-
.../phoenix/compile/CreateIndexCompiler.java | 186 +++-
.../phoenix/compile/CreateTableCompiler.java | 61 +
.../phoenix/compile/ServerBuildIndexCompiler.java | 20 +-
.../apache/phoenix/compile/StatementContext.java | 25 +-
.../org/apache/phoenix/compile/WhereCompiler.java | 613 +++++++++-
.../coprocessor/GlobalIndexRegionScanner.java | 69 +-
.../phoenix/coprocessor/MetaDataEndpointImpl.java | 14 +
.../coprocessor/MetaDataRegionObserver.java | 21 +-
.../coprocessor/UncoveredIndexRegionScanner.java | 15 +-
.../UngroupedAggregateRegionObserver.java | 2 +-
.../apache/phoenix/exception/SQLExceptionCode.java | 13 +-
.../execute/PhoenixTxIndexMutationGenerator.java | 5 +-
.../phoenix/expression/ComparisonExpression.java | 138 ++-
.../org/apache/phoenix/expression/Expression.java | 10 +
.../phoenix/expression/IsNullExpression.java | 16 +-
.../phoenix/hbase/index/IndexRegionObserver.java | 60 +-
.../org/apache/phoenix/index/IndexMaintainer.java | 123 +-
.../phoenix/iterate/RegionScannerFactory.java | 15 +-
.../phoenix/iterate/TableResultIterator.java | 23 +
.../org/apache/phoenix/jdbc/PhoenixConnection.java | 98 +-
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 507 +++++----
.../org/apache/phoenix/jdbc/PhoenixResultSet.java | 7 +-
.../org/apache/phoenix/jdbc/PhoenixStatement.java | 75 +-
.../apache/phoenix/log/BaseConnectionLimiter.java | 15 +-
.../org/apache/phoenix/log/ConnectionLimiter.java | 4 +
.../apache/phoenix/monitoring/AtomicMetric.java | 10 +
.../phoenix/monitoring/CombinableMetric.java | 8 +
.../phoenix/monitoring/CombinableMetricImpl.java | 10 +
...ric.java => ConnectionQueryServicesMetric.java} | 16 +-
...java => ConnectionQueryServicesMetricImpl.java} | 25 +-
.../phoenix/monitoring/GlobalMetricImpl.java | 10 +
.../java/org/apache/phoenix/monitoring/Metric.java | 4 +
.../phoenix/monitoring/NoOpGlobalMetricImpl.java | 10 +
.../apache/phoenix/monitoring/NonAtomicMetric.java | 10 +
.../phoenix/monitoring/PhoenixTableMetricImpl.java | 10 +
.../ConnectionQueryServicesHistogram.java} | 34 +-
.../ConnectionQueryServicesMetrics.java | 120 ++
.../ConnectionQueryServicesMetricsHistograms.java | 71 ++
.../ConnectionQueryServicesMetricsManager.java | 343 ++++++
.../NoOpConnectionQueryServicesMetricsManager.java | 62 ++
.../apache/phoenix/optimize/QueryOptimizer.java | 72 +-
.../apache/phoenix/parse/CreateIndexStatement.java | 12 +-
.../org/apache/phoenix/parse/ParseNodeFactory.java | 9 +-
.../phoenix/query/ConnectionQueryServices.java | 1 +
.../phoenix/query/ConnectionQueryServicesImpl.java | 26 +-
.../query/ConnectionlessQueryServicesImpl.java | 5 +
.../query/DelegateConnectionQueryServices.java | 7 +-
.../phoenix/query/DelegateQueryServices.java | 4 +-
.../org/apache/phoenix/query/QueryConstants.java | 4 +-
.../org/apache/phoenix/query/QueryServices.java | 24 +
.../apache/phoenix/query/QueryServicesOptions.java | 44 +
.../org/apache/phoenix/schema/DelegateTable.java | 25 +-
.../org/apache/phoenix/schema/MetaDataClient.java | 112 +-
.../java/org/apache/phoenix/schema/PTable.java | 32 +-
.../java/org/apache/phoenix/schema/PTableImpl.java | 96 +-
.../java/org/apache/phoenix/util/IndexUtil.java | 31 +
.../org/apache/phoenix/util/PhoenixRuntime.java | 18 +
phoenix-core/src/main/protobuf/PTable.proto | 3 +-
.../src/main/protobuf/ServerCachingService.proto | 2 +
.../apache/phoenix/compile/WhereCompilerTest.java | 88 ++
.../phoenix/index/VerifySingleIndexRowTest.java | 2 +-
.../apache/phoenix/jdbc/PhoenixStatementTest.java | 38 +
.../ConnectionQueryServicesHistogramTest.java | 73 ++
...nnectionQueryServicesMetricsHistogramsTest.java | 37 +
.../ConnectionQueryServicesMetricsManagerTest.java | 112 ++
.../ConnectionQueryServicesMetricsTest.java | 106 ++
.../ConnectionQueryServicesNameMetricsTest.java | 87 ++
79 files changed, 5910 insertions(+), 623 deletions(-)
create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewExtendsPkRestrictionsIT.java
create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexIT.java
create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsIT.java
copy phoenix-core/src/main/java/org/apache/phoenix/monitoring/{PhoenixTableMetric.java => ConnectionQueryServicesMetric.java} (71%)
copy phoenix-core/src/main/java/org/apache/phoenix/monitoring/{PhoenixTableMetricImpl.java => ConnectionQueryServicesMetricImpl.java} (78%)
copy phoenix-core/src/main/java/org/apache/phoenix/monitoring/{SizeHistogram.java => connectionqueryservice/ConnectionQueryServicesHistogram.java} (59%)
create mode 100644 phoenix-core/src/main/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetrics.java
create mode 100644 phoenix-core/src/main/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsHistograms.java
create mode 100644 phoenix-core/src/main/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsManager.java
create mode 100644 phoenix-core/src/main/java/org/apache/phoenix/monitoring/connectionqueryservice/NoOpConnectionQueryServicesMetricsManager.java
create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesHistogramTest.java
create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsHistogramsTest.java
create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsManagerTest.java
create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesMetricsTest.java
create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/monitoring/connectionqueryservice/ConnectionQueryServicesNameMetricsTest.java
(phoenix) 01/01: Merge branch 'master' into PHOENIX-7001-feature
Posted by vj...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vjasani pushed a commit to branch PHOENIX-7001-feature
in repository https://gitbox.apache.org/repos/asf/phoenix.git
commit f7a4feeccc2f0f7304bf589567e8b8feddda2b52
Merge: e3ca004315 5128ad0e7d
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Mon Nov 20 22:44:55 2023 -0800
Merge branch 'master' into PHOENIX-7001-feature
.../org/apache/phoenix/end2end/AlterTableIT.java | 75 +-
.../phoenix/end2end/ClientHashAggregateIT.java | 38 +-
.../java/org/apache/phoenix/end2end/InListIT.java | 7 +-
.../phoenix/end2end/MetaDataEndpointImplIT.java | 26 +-
.../phoenix/end2end/QueryDatabaseMetaDataIT.java | 50 +
.../end2end/RebuildIndexConnectionPropsIT.java | 4 +-
.../end2end/ViewExtendsPkRestrictionsIT.java | 1169 ++++++++++++++++++++
.../java/org/apache/phoenix/end2end/ViewTTLIT.java | 65 +-
.../phoenix/end2end/index/PartialIndexIT.java | 767 +++++++++++++
.../index/UncoveredGlobalIndexRegionScannerIT.java | 9 +-
.../ConnectionQueryServicesMetricsIT.java | 366 ++++++
phoenix-core/src/main/antlr3/PhoenixSQL.g | 9 +-
.../phoenix/compile/CreateIndexCompiler.java | 186 +++-
.../phoenix/compile/CreateTableCompiler.java | 61 +
.../phoenix/compile/ServerBuildIndexCompiler.java | 20 +-
.../apache/phoenix/compile/StatementContext.java | 25 +-
.../org/apache/phoenix/compile/WhereCompiler.java | 613 +++++++++-
.../coprocessor/GlobalIndexRegionScanner.java | 69 +-
.../phoenix/coprocessor/MetaDataEndpointImpl.java | 14 +
.../coprocessor/MetaDataRegionObserver.java | 21 +-
.../coprocessor/UncoveredIndexRegionScanner.java | 15 +-
.../UngroupedAggregateRegionObserver.java | 2 +-
.../apache/phoenix/exception/SQLExceptionCode.java | 13 +-
.../execute/PhoenixTxIndexMutationGenerator.java | 5 +-
.../phoenix/expression/ComparisonExpression.java | 138 ++-
.../org/apache/phoenix/expression/Expression.java | 10 +
.../phoenix/expression/IsNullExpression.java | 16 +-
.../phoenix/hbase/index/IndexRegionObserver.java | 60 +-
.../org/apache/phoenix/index/IndexMaintainer.java | 123 +-
.../phoenix/iterate/RegionScannerFactory.java | 15 +-
.../phoenix/iterate/TableResultIterator.java | 23 +
.../org/apache/phoenix/jdbc/PhoenixConnection.java | 98 +-
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 507 +++++----
.../org/apache/phoenix/jdbc/PhoenixResultSet.java | 7 +-
.../org/apache/phoenix/jdbc/PhoenixStatement.java | 75 +-
.../apache/phoenix/log/BaseConnectionLimiter.java | 15 +-
.../org/apache/phoenix/log/ConnectionLimiter.java | 4 +
.../apache/phoenix/monitoring/AtomicMetric.java | 10 +
.../phoenix/monitoring/CombinableMetric.java | 8 +
.../phoenix/monitoring/CombinableMetricImpl.java | 10 +
.../ConnectionQueryServicesMetric.java} | 29 +-
...java => ConnectionQueryServicesMetricImpl.java} | 25 +-
.../phoenix/monitoring/GlobalMetricImpl.java | 10 +
.../java/org/apache/phoenix/monitoring/Metric.java | 4 +
.../phoenix/monitoring/NoOpGlobalMetricImpl.java | 10 +
.../apache/phoenix/monitoring/NonAtomicMetric.java | 10 +
.../phoenix/monitoring/PhoenixTableMetricImpl.java | 10 +
.../ConnectionQueryServicesHistogram.java | 43 +
.../ConnectionQueryServicesMetrics.java | 120 ++
.../ConnectionQueryServicesMetricsHistograms.java | 71 ++
.../ConnectionQueryServicesMetricsManager.java | 343 ++++++
.../NoOpConnectionQueryServicesMetricsManager.java | 62 ++
.../apache/phoenix/optimize/QueryOptimizer.java | 72 +-
.../apache/phoenix/parse/CreateIndexStatement.java | 12 +-
.../org/apache/phoenix/parse/ParseNodeFactory.java | 9 +-
.../phoenix/query/ConnectionQueryServices.java | 1 +
.../phoenix/query/ConnectionQueryServicesImpl.java | 26 +-
.../query/ConnectionlessQueryServicesImpl.java | 5 +
.../query/DelegateConnectionQueryServices.java | 7 +-
.../phoenix/query/DelegateQueryServices.java | 4 +-
.../org/apache/phoenix/query/QueryConstants.java | 4 +-
.../org/apache/phoenix/query/QueryServices.java | 24 +
.../apache/phoenix/query/QueryServicesOptions.java | 44 +
.../org/apache/phoenix/schema/DelegateTable.java | 25 +-
.../org/apache/phoenix/schema/MetaDataClient.java | 112 +-
.../java/org/apache/phoenix/schema/PTable.java | 32 +-
.../java/org/apache/phoenix/schema/PTableImpl.java | 96 +-
.../java/org/apache/phoenix/util/IndexUtil.java | 31 +
.../org/apache/phoenix/util/PhoenixRuntime.java | 18 +
phoenix-core/src/main/protobuf/PTable.proto | 3 +-
.../src/main/protobuf/ServerCachingService.proto | 2 +
.../apache/phoenix/compile/WhereCompilerTest.java | 88 ++
.../phoenix/index/VerifySingleIndexRowTest.java | 2 +-
.../apache/phoenix/jdbc/PhoenixStatementTest.java | 38 +
.../ConnectionQueryServicesHistogramTest.java | 73 ++
...nnectionQueryServicesMetricsHistogramsTest.java | 37 +
.../ConnectionQueryServicesMetricsManagerTest.java | 112 ++
.../ConnectionQueryServicesMetricsTest.java | 106 ++
.../ConnectionQueryServicesNameMetricsTest.java | 87 ++
79 files changed, 5944 insertions(+), 611 deletions(-)
diff --cc phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index a0c8f6e594,4605554409..7c80c85c60
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@@ -367,8 -366,8 +368,10 @@@ TABLE_FAMILY_BYTES, TABLE_SEQ_NUM_BYTES
TABLE_FAMILY_BYTES, EXTERNAL_SCHEMA_ID_BYTES);
private static final Cell STREAMING_TOPIC_NAME_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY,
TABLE_FAMILY_BYTES, STREAMING_TOPIC_NAME_BYTES);
+ private static final Cell CDC_INCLUDE_KV = 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 List<Cell> TABLE_KV_COLUMNS = Lists.newArrayList(
EMPTY_KEYVALUE_KV,
@@@ -409,7 -408,7 +412,8 @@@
SCHEMA_VERSION_KV,
EXTERNAL_SCHEMA_ID_KV,
STREAMING_TOPIC_NAME_KV,
- INDEX_WHERE_KV
++ INDEX_WHERE_KV,
+ CDC_INCLUDE_KV
);
static {
@@@ -457,7 -456,8 +461,9 @@@
TABLE_KV_COLUMNS.indexOf(EXTERNAL_SCHEMA_ID_KV);
private static final int STREAMING_TOPIC_NAME_INDEX =
TABLE_KV_COLUMNS.indexOf(STREAMING_TOPIC_NAME_KV);
+ private static final int CDC_INCLUDE_INDEX = TABLE_KV_COLUMNS.indexOf(CDC_INCLUDE_KV);
+ private static final int INDEX_WHERE_INDEX =
+ TABLE_KV_COLUMNS.indexOf(INDEX_WHERE_KV);
// KeyValues for Column
private static final KeyValue DECIMAL_DIGITS_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, DECIMAL_DIGITS_BYTES);
private static final KeyValue COLUMN_SIZE_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, TABLE_FAMILY_BYTES, COLUMN_SIZE_BYTES);
@@@ -1429,15 -1429,13 +1435,23 @@@
builder.setStreamingTopicName(streamingTopicName != null ? streamingTopicName :
oldTable != null ? oldTable.getStreamingTopicName() : null);
+ Cell includeSpecKv = tableKeyValues[CDC_INCLUDE_INDEX];
+ String includeSpec = includeSpecKv != null ?
+ (String) PVarchar.INSTANCE.toObject(includeSpecKv.getValueArray(),
+ includeSpecKv.getValueOffset(), includeSpecKv.getValueLength())
+ : null;
+ builder.setCDCIncludeScopes(includeSpec != null ?
+ CDCUtil.makeChangeScopeEnumsFromString(includeSpec) :
+ oldTable != null ? oldTable.getCDCIncludeScopes() : null);
+
+ Cell indexWhereKv = tableKeyValues[INDEX_WHERE_INDEX];
+ String indexWhere = indexWhereKv != null
+ ? (String) PVarchar.INSTANCE.toObject(indexWhereKv.getValueArray(),
+ indexWhereKv.getValueOffset(), indexWhereKv.getValueLength())
+ : null;
+ builder.setIndexWhere(indexWhere != null ? indexWhere
+ : oldTable != null ? oldTable.getIndexWhere() : null);
++
// Check the cell tag to see whether the view has modified this property
final byte[] tagUseStatsForParallelization = (useStatsForParallelizationKv == null) ?
HConstants.EMPTY_BYTE_ARRAY :
diff --cc phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index 56c9a6dfb3,5796562566..5139eb5764
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@@ -352,8 -354,10 +355,12 @@@ public enum SQLExceptionCode
+ PhoenixDatabaseMetaData.PHOENIX_TTL + " property on an view when parent/child view has PHOENIX_TTL set,"),
CHANGE_DETECTION_SUPPORTED_FOR_TABLES_AND_VIEWS_ONLY(10954, "44A36",
CHANGE_DETECTION_ENABLED + " is only supported on tables and views"),
- UNKNOWN_INDEX_TYPE(1098,"44A37", "Unknown INDEX type: "),
- UNKNOWN_INCLUDE_CHANGE_SCOPE(1099,"44A38", "Unknown change scope for INCLUDE: "),
+ CANNOT_CREATE_INDEX_CHILD_VIEWS_EXTEND_PK(10955, "44A37", "Index can be created "
+ + "only if none of the child views extends primary key"),
+ VIEW_CANNOT_EXTEND_PK_WITH_PARENT_INDEXES(10956, "44A38", "View can extend parent primary key"
+ + " only if none of the parents have indexes in the parent hierarchy"),
++ UNKNOWN_INDEX_TYPE(1098, "44A39", "Unknown INDEX type: "),
++ UNKNOWN_INCLUDE_CHANGE_SCOPE(1099, "44A40", "Unknown change scope for INCLUDE: "),
/** Sequence related */
SEQUENCE_ALREADY_EXIST(1200, "42Z00", "Sequence already exists.", new Factory() {
diff --cc phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
index 2e45b2eb74,c8f392bdbc..390ae10ade
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
@@@ -367,19 -365,15 +367,24 @@@ public class ParseNodeFactory
return new CreateSchemaStatement(schemaName, ifNotExists);
}
- public CreateIndexStatement createIndex(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) {
- return new CreateIndexStatement(indexName, dataTable, ikConstraint, includeColumns, splits, props, ifNotExists, indexType, async, bindCount, udfParseNodes);
+ public CreateIndexStatement createIndex(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) {
+ return new CreateIndexStatement(indexName, dataTable, ikConstraint, includeColumns, splits,
+ props, ifNotExists, indexType, async, bindCount, udfParseNodes, where);
}
+ public CreateCDCStatement createCDC(NamedNode cdcObj, TableName dataTable,
+ ColumnName timeIdxColumn, FunctionParseNode timeIdxFunc,
+ Set<PTable.CDCChangeScope> includeScopes,
+ ListMultimap<String, Pair<String, Object>> props,
+ boolean ifNotExists, int bindCount) {
+ return new CreateCDCStatement(cdcObj, dataTable, timeIdxColumn, timeIdxFunc, includeScopes,
+ props, ifNotExists, bindCount);
+ }
+
public CreateSequenceStatement createSequence(TableName tableName, ParseNode startsWith,
ParseNode incrementBy, ParseNode cacheSize, ParseNode minValue, ParseNode maxValue,
boolean cycle, boolean ifNotExits, int bindCount) {
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index adcb93ea47,d73c1123ca..bf9352caad
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@@ -4138,22 -4149,22 +4151,25 @@@ public class ConnectionQueryServicesImp
}
if (currentServerSideTableTimeStamp < MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0) {
metaConnection = addColumnsIfNotExists(metaConnection,
-- PhoenixDatabaseMetaData.SYSTEM_CATALOG, MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 4,
++ PhoenixDatabaseMetaData.SYSTEM_CATALOG, MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 5,
PhoenixDatabaseMetaData.PHYSICAL_TABLE_NAME
+ " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
-- MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 3,
++ MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 4,
PhoenixDatabaseMetaData.SCHEMA_VERSION + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
-- MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 2,
++ MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 3,
PhoenixDatabaseMetaData.EXTERNAL_SCHEMA_ID + " " + PVarchar.INSTANCE.getSqlTypeName());
metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
-- MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 1,
++ MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 2,
PhoenixDatabaseMetaData.STREAMING_TOPIC_NAME + " " + PVarchar.INSTANCE.getSqlTypeName());
+ metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0,
++ MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 - 1,
+ PhoenixDatabaseMetaData.INDEX_WHERE + " " + PVarchar.INSTANCE.getSqlTypeName());
+ metaConnection = addColumnsIfNotExists(metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG, MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0,
+ PhoenixDatabaseMetaData.CDC_INCLUDE_TABLE + " " + PVarchar.INSTANCE.getSqlTypeName());
UpgradeUtil.bootstrapLastDDLTimestampForIndexes(metaConnection);
}
return metaConnection;
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index b3bab6b637,ac1ae6a255..ba9f3904a2
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@@ -346,7 -343,7 +347,8 @@@ public interface QueryConstants
SCHEMA_VERSION + " VARCHAR, \n" +
EXTERNAL_SCHEMA_ID + " VARCHAR, \n" +
STREAMING_TOPIC_NAME + " VARCHAR, \n" +
+ INDEX_WHERE + " VARCHAR, \n" +
+ CDC_INCLUDE_TABLE + " VARCHAR, \n" +
// Column metadata (will be null for table row)
DATA_TYPE + " INTEGER," +
COLUMN_SIZE + " INTEGER," +
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateTable.java
index ca62149975,1a5d83e728..d1ad19fce2
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateTable.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateTable.java
@@@ -404,10 -409,21 +409,26 @@@ public class DelegateTable implements P
public String getStreamingTopicName() { return delegate.getStreamingTopicName(); }
@Override
+ public Set<CDCChangeScope> getCDCIncludeScopes() {
+ return delegate.getCDCIncludeScopes();
+ }
+
+ public String getIndexWhere() {
+ return delegate.getIndexWhere();
+ }
+
+ @Override
+ public Expression getIndexWhereExpression(PhoenixConnection connection)
+ throws SQLException {
+ return delegate.getIndexWhereExpression(connection);
+ }
+
+ @Override
+ public Set<ColumnReference> getIndexWhereColumns(PhoenixConnection connection)
+ throws SQLException {
+ return delegate.getIndexWhereColumns(connection);
+ }
++
@Override public Map<String, String> getPropertyValues() { return delegate.getPropertyValues(); }
@Override public Map<String, String> getDefaultPropertyValues() { return delegate.getDefaultPropertyValues(); }
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index f190da84d5,fb33efbada..ebbe4bb7d4
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@@ -93,7 -80,11 +94,9 @@@ import static org.apache.phoenix.jdbc.P
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYNC_INDEX_CREATED_DATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
+ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_NAMESPACE_BYTES;
+ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_NAME_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_TABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_TASK_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SCHEM;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SEQ_NUM;
@@@ -108,6 -99,17 +111,8 @@@ import static org.apache.phoenix.jdbc.P
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_INDEX_ID_DATA_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_STATEMENT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TYPE;
-import static org.apache.phoenix.query.QueryConstants.SYSTEM_SCHEMA_NAME;
+ import static org.apache.phoenix.query.QueryServices.DEFAULT_DISABLE_VIEW_SUBTREE_VALIDATION;
+ import static org.apache.phoenix.query.QueryServices.DISABLE_VIEW_SUBTREE_VALIDATION;
-import static org.apache.phoenix.query.QueryServices.INDEX_CREATE_DEFAULT_STATE;
-import static org.apache.phoenix.thirdparty.com.google.common.collect.Sets.newLinkedHashSet;
-import static org.apache.phoenix.thirdparty.com.google.common.collect.Sets.newLinkedHashSetWithExpectedSize;
-import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.RUN_UPDATE_STATS_ASYNC_ATTRIB;
-import static org.apache.phoenix.coprocessor.tasks.IndexRebuildTask.INDEX_NAME;
-import static org.apache.phoenix.coprocessor.tasks.IndexRebuildTask.REBUILD_ALL;
-import static org.apache.phoenix.exception.SQLExceptionCode.INSUFFICIENT_MULTI_TENANT_COLUMNS;
-import static org.apache.phoenix.exception.SQLExceptionCode.PARENT_TABLE_NOT_FOUND;
import static org.apache.phoenix.monitoring.MetricType.NUM_METADATA_LOOKUP_FAILURES;
import static org.apache.phoenix.query.QueryConstants.BASE_TABLE_BASE_COLUMN_COUNT;
import static org.apache.phoenix.query.QueryConstants.DEFAULT_COLUMN_FAMILY;
@@@ -156,7 -157,10 +161,11 @@@ import java.util.Set
import java.util.HashSet;
import org.apache.hadoop.conf.Configuration;
+import org.apache.phoenix.parse.CreateCDCStatement;
+ import org.apache.hadoop.hbase.client.Table;
+ import org.apache.phoenix.coprocessor.TableInfo;
+ import org.apache.phoenix.query.ConnectionlessQueryServicesImpl;
+ import org.apache.phoenix.query.DelegateQueryServices;
import org.apache.phoenix.schema.task.SystemTaskParams;
import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.hbase.HConstants;
@@@ -351,9 -353,9 +360,10 @@@ public class MetaDataClient
PHYSICAL_TABLE_NAME + "," +
SCHEMA_VERSION + "," +
STREAMING_TOPIC_NAME + "," +
+ INDEX_WHERE +
+ CDC_INCLUDE_TABLE +
") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, " +
-- "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
++ "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String CREATE_SCHEMA = "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE
+ "\"( " + TABLE_SCHEM + "," + TABLE_NAME + ") VALUES (?,?)";
@@@ -1676,10 -1676,13 +1691,13 @@@
PrimaryKeyConstraint pk = FACTORY.primaryKey(null, allPkColumns);
tableProps.put(MetaDataUtil.DATA_TABLE_NAME_PROP_NAME, dataTable.getPhysicalName().getString());
- CreateTableStatement tableStatement = FACTORY.createTable(indexTableName, statement.getProps(), columnDefs, pk, statement.getSplitNodes(), PTableType.INDEX, statement.ifNotExists(), null, null, statement.getBindCount(), null);
+ CreateTableStatement tableStatement = FACTORY.createTable(indexTableName,
+ statement.getProps(), columnDefs, pk, statement.getSplitNodes(),
+ PTableType.INDEX, statement.ifNotExists(), null,
+ statement.getWhere(), statement.getBindCount(), null);
table = createTableInternal(tableStatement, splits, dataTable, null, null,
getViewIndexDataType() ,null, null, allocateIndexId,
- statement.getIndexType(), asyncCreatedDate, tableProps, commonFamilyProps);
+ statement.getIndexType(), asyncCreatedDate, null, tableProps, commonFamilyProps);
}
finally {
deleteMutexCells(physicalSchemaName, physicalTableName, acquiredColumnMutexSet);
@@@ -1711,94 -1714,65 +1729,152 @@@
return buildIndex(table, tableRef);
}
+ public MutationState createCDC(CreateCDCStatement statement) throws SQLException {
+ // TODO: Do we need to borrow the schema name of the table?
+ ColumnResolver resolver = FromCompiler.getResolver(NamedTableNode.create(statement.getDataTable()), connection);
+ TableRef tableRef = resolver.getTables().get(0);
+ PTable dataTable = tableRef.getTable();
+ // Check if data table is a view and give a not supported error.
+ if (dataTable.getType() != TABLE) {
+ throw new SQLExceptionInfo.Builder(INVALID_TABLE_TYPE_FOR_CDC).setTableType(
+ dataTable.getType()).build().buildException();
+ }
+
- Map<String,Object> tableProps = Maps.newHashMapWithExpectedSize(
++ Map<String, Object> tableProps = Maps.newHashMapWithExpectedSize(
+ statement.getProps().size());
- Map<String,Object> commonFamilyProps = Maps.newHashMapWithExpectedSize(
++ Map<String, Object> commonFamilyProps = Maps.newHashMapWithExpectedSize(
+ statement.getProps().size() + 1);
+ populatePropertyMaps(statement.getProps(), tableProps, commonFamilyProps, PTableType.CDC);
+
+ NamedNode indexName = FACTORY.indexName(CDCUtil.getCDCIndexName(
+ statement.getCdcObjName().getName()));
+ String timeIdxColName = statement.getTimeIdxColumn() != null ?
+ statement.getTimeIdxColumn().getColumnName() : null;
+ IndexKeyConstraint indexKeyConstraint =
- FACTORY.indexKey(Arrays.asList(new Pair[] { Pair.newPair(
++ FACTORY.indexKey(Arrays.asList(new Pair[]{Pair.newPair(
+ timeIdxColName != null ? FACTORY.column(statement.getDataTable(),
+ timeIdxColName, timeIdxColName) : statement.getTimeIdxFunc(),
- SortOrder.getDefault()) }));
++ SortOrder.getDefault())}));
+ IndexType indexType = (IndexType) TableProperty.INDEX_TYPE.getValue(tableProps);
+ ListMultimap<String, Pair<String, Object>> indexProps = ArrayListMultimap.create();
+ if (TableProperty.SALT_BUCKETS.getValue(tableProps) != null) {
+ indexProps.put(QueryConstants.ALL_FAMILY_PROPERTIES_KEY, new Pair<>(
+ TableProperty.SALT_BUCKETS.getPropertyName(),
+ TableProperty.SALT_BUCKETS.getValue(tableProps)));
+ }
+ // TODO: Transfer TTL and MaxLookback from statement.getProps() to indexProps.
+ CreateIndexStatement indexStatement = FACTORY.createIndex(indexName, FACTORY.namedTable(null,
+ statement.getDataTable(), (Double) null), indexKeyConstraint, null, null,
+ indexProps, statement.isIfNotExists(), indexType, false, 0,
- new HashMap<>());
++ new HashMap<>(), null);
+ // TODO: Currently index can be dropped, leaving the CDC dangling, DROP INDEX needs to
+ // protect based on CDCUtil.isACDCIndex().
+ // TODO: Should we also allow PTimestamp here?
+ MutationState indexMutationState;
+ try {
+ indexMutationState = createIndex(indexStatement, null, PDate.INSTANCE);
- }
- catch(SQLException e) {
++ } catch (SQLException e) {
+ if (e.getErrorCode() == TABLE_ALREADY_EXIST.getErrorCode()) {
+ throw new SQLExceptionInfo.Builder(TABLE_ALREADY_EXIST).setTableName(
+ statement.getCdcObjName().getName()).build().buildException();
+ }
+ // TODO: What about translating other index creation failures? E.g., bad TS column.
+ throw e;
+ }
+
+ List<PColumn> pkColumns = dataTable.getPKColumns();
+ List<ColumnDef> columnDefs = new ArrayList<>();
+ List<ColumnDefInPkConstraint> pkColumnDefs = new ArrayList<>();
+ // TODO: toString() on function will have an extra space at the beginning, but this may
+ // be OK as I see exactly the same with an index.
+ ColumnName timeIdxCol = statement.getTimeIdxColumn() != null ?
+ statement.getTimeIdxColumn() :
+ FACTORY.columnName(statement.getTimeIdxFunc().toString());
+ columnDefs.add(FACTORY.columnDef(timeIdxCol, PTimestamp.INSTANCE.getSqlTypeName(), false, null, false,
+ PTimestamp.INSTANCE.getMaxLength(null), PTimestamp.INSTANCE.getScale(null), false,
+ SortOrder.getDefault(), "", null, false));
+ pkColumnDefs.add(FACTORY.columnDefInPkConstraint(timeIdxCol, SortOrder.getDefault(), false));
- for (PColumn pcol: pkColumns) {
++ for (PColumn pcol : pkColumns) {
+ // TODO: Cross check with the ColumnName creation logic in createIndex (line ~1578).
+ columnDefs.add(FACTORY.columnDef(FACTORY.columnName(pcol.getName().getString()),
+ pcol.getDataType().getSqlTypeName(), false, null, false, pcol.getMaxLength(),
+ pcol.getScale(), false, pcol.getSortOrder(), "", null, false));
+ pkColumnDefs.add(FACTORY.columnDefInPkConstraint(FACTORY.columnName(
+ pcol.getName().getString()), pcol.getSortOrder(), pcol.isRowTimestamp()));
+ }
+ columnDefs.add(FACTORY.columnDef(FACTORY.columnName(QueryConstants.CDC_JSON_COL_NAME),
+ PVarchar.INSTANCE.getSqlTypeName(), false, null, true, null,
+ null, false, SortOrder.getDefault(), "", null, false));
+ CreateTableStatement tableStatement = FACTORY.createTable(
+ FACTORY.table(dataTable.getSchemaName().getString(), statement.getCdcObjName().getName()),
+ statement.getProps(), 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, false, null,
+ null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ return indexMutationState;
+ }
+
+ /**
+ * Go through all the descendent views from the child view hierarchy and find if any of the
+ * descendent views extends the primary key, throw error.
+ *
+ * @param tableOrView view or table on which the index is being created.
+ * @param config the configuration.
+ * @throws SQLException if any of the descendent views extends pk or if something goes wrong
+ * while querying descendent view hierarchy.
+ */
+ private void verifyIfDescendentViewsExtendPk(PTable tableOrView, Configuration config)
+ throws SQLException {
+ if (connection.getQueryServices() instanceof ConnectionlessQueryServicesImpl) {
+ return;
+ }
+ if (connection.getQueryServices() instanceof DelegateQueryServices) {
+ DelegateQueryServices services = (DelegateQueryServices) connection.getQueryServices();
+ if (services.getDelegate() instanceof ConnectionlessQueryServicesImpl) {
+ return;
+ }
+ }
+ byte[] systemChildLinkTable = SchemaUtil.isNamespaceMappingEnabled(null, config) ?
+ SYSTEM_CHILD_LINK_NAMESPACE_BYTES :
+ SYSTEM_CHILD_LINK_NAME_BYTES;
+ try (Table childLinkTable =
+ connection.getQueryServices().getTable(systemChildLinkTable)) {
+ byte[] tenantId = connection.getTenantId() == null ? null
+ : connection.getTenantId().getBytes();
+ byte[] schemaNameBytes = tableOrView.getSchemaName().getBytes();
+ byte[] viewOrTableName = tableOrView.getTableName().getBytes();
+ Pair<List<PTable>, List<TableInfo>> descViews =
+ ViewUtil.findAllDescendantViews(
+ childLinkTable,
+ config,
+ tenantId,
+ schemaNameBytes,
+ viewOrTableName,
+ HConstants.LATEST_TIMESTAMP,
+ false);
+ List<PTable> legitimateChildViews = descViews.getFirst();
+ int dataTableOrViewPkCols = tableOrView.getPKColumns().size();
+ if (legitimateChildViews != null && legitimateChildViews.size() > 0) {
+ for (PTable childView : legitimateChildViews) {
+ if (childView.getPKColumns().size() > dataTableOrViewPkCols) {
+ LOGGER.error("Creation of view index not allowed as child view {}"
+ + " extends pk", childView.getName());
+ throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode
+ .CANNOT_CREATE_INDEX_CHILD_VIEWS_EXTEND_PK)
+ .build()
+ .buildException();
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Error while retrieving descendent views", e);
+ throw new SQLException(e);
+ }
+ }
+
public MutationState dropSequence(DropSequenceStatement statement) throws SQLException {
Long scn = connection.getSCN();
long timestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
@@@ -3184,13 -3148,11 +3262,19 @@@
} else {
tableUpsert.setString(35, streamingTopicName);
}
+
- if (cdcIncludeScopesStr == null) {
+ if (tableType == INDEX && statement.getWhereClause() != null) {
+ tableUpsert.setString(36, statement.getWhereClause().toString());
+ } else {
tableUpsert.setNull(36, Types.VARCHAR);
+ }
++
++ if (cdcIncludeScopesStr == null) {
++ tableUpsert.setNull(37, Types.VARCHAR);
+ } else {
- tableUpsert.setString(36, cdcIncludeScopesStr);
++ tableUpsert.setString(37, cdcIncludeScopesStr);
+ }
+
tableUpsert.execute();
if (asyncCreatedDate != null) {
@@@ -3340,7 -3300,8 +3424,9 @@@
.setExternalSchemaId(result.getTable() != null ?
result.getTable().getExternalSchemaId() : null)
.setStreamingTopicName(streamingTopicName)
+ .setIndexWhere(statement.getWhereClause() == null ? null
+ : statement.getWhereClause().toString())
+ .setCDCIncludeScopes(cdcIncludeScopes)
.build();
result = new MetaDataMutationResult(code, result.getMutationTime(), table, true);
addTableToCache(result);
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
index bf0700c6fa,6fd89ad4bb..64ab453492
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
@@@ -970,11 -975,28 +975,34 @@@ public interface PTable extends PMetaDa
*/
String getStreamingTopicName();
+ /**
+ * @return Optional string that represents the default include scopes to be used for CDC queries.
+ */
+ Set<CDCChangeScope> getCDCIncludeScopes();
+
+ /**
+ *
+ * @return the optional where clause in string used for partial indexes
+ */
+ String getIndexWhere();
+
+
+ /**
+ *
+ * @param connection PhoenixConnection
+ * @return the optional where clause in DNF expression used for partial indexes
+ * @throws SQLException
+ */
+ Expression getIndexWhereExpression(PhoenixConnection connection) throws SQLException;
+
+ /**
+ *
+ * @param connection
+ * @return the set of column references for the columns included in the index where clause
+ * @throws SQLException
+ */
+ Set<ColumnReference> getIndexWhereColumns(PhoenixConnection connection) throws SQLException;
++
/**
* Class to help track encoded column qualifier counters per column family.
*/
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
index 2dde20a948,8b30976cf0..533f6dc2d9
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
@@@ -106,16 -110,7 +110,8 @@@ import org.apache.phoenix.schema.types.
import org.apache.phoenix.schema.types.PDouble;
import org.apache.phoenix.schema.types.PFloat;
import org.apache.phoenix.schema.types.PVarchar;
- import org.apache.phoenix.transaction.TransactionFactory;
- import org.apache.phoenix.util.ByteUtil;
- import org.apache.phoenix.util.CDCUtil;
- import org.apache.phoenix.util.EncodedColumnsUtil;
- import org.apache.phoenix.util.MetaDataUtil;
- import org.apache.phoenix.util.PhoenixRuntime;
- import org.apache.phoenix.util.SchemaUtil;
- import org.apache.phoenix.util.SizedUtil;
- import org.apache.phoenix.util.TrustedByteArrayOutputStream;
+
+ import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.phoenix.thirdparty.com.google.common.base.Objects;
import org.apache.phoenix.thirdparty.com.google.common.base.Preconditions;
import org.apache.phoenix.thirdparty.com.google.common.base.Strings;
@@@ -126,6 -121,15 +122,16 @@@ import org.apache.phoenix.thirdparty.co
import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
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.phoenix.transaction.TransactionFactory;
+ import org.apache.phoenix.util.ByteUtil;
++import org.apache.phoenix.util.CDCUtil;
+ import org.apache.phoenix.util.EncodedColumnsUtil;
+ import org.apache.phoenix.util.MetaDataUtil;
+ import org.apache.phoenix.util.PhoenixRuntime;
+ import org.apache.phoenix.util.SchemaUtil;
+ import org.apache.phoenix.util.SizedUtil;
+ import org.apache.phoenix.util.TrustedByteArrayOutputStream;
/**
*
@@@ -213,7 -217,9 +219,10 @@@ public class PTableImpl implements PTab
private String schemaVersion;
private String externalSchemaId;
private String streamingTopicName;
+ private String indexWhere;
+ private Expression indexWhereExpression;
+ private Set<ColumnReference> indexWhereColumns;
+ private Set<CDCChangeScope> cdcIncludeScopes;
public static class Builder {
private PTableKey key;
@@@ -279,7 -285,7 +288,8 @@@
private String schemaVersion;
private String externalSchemaId;
private String streamingTopicName;
+ private Set<CDCChangeScope> cdcIncludeScopes;
+ private String indexWhere;
// Used to denote which properties a view has explicitly modified
private BitSet viewModifiedPropSet = new BitSet(3);
@@@ -700,13 -706,13 +710,20 @@@
return this;
}
+ public Builder setIndexWhere(String indexWhere) {
+ if (indexWhere != null) {
+ this.indexWhere = indexWhere;
+ }
+ return this;
+ }
+
+ public Builder setCDCIncludeScopes(Set<CDCChangeScope> cdcIncludeScopes) {
+ if (cdcIncludeScopes != null) {
+ this.cdcIncludeScopes = cdcIncludeScopes;
+ }
+ return this;
+ }
+
/**
* Populate derivable attributes of the PTable
* @return PTableImpl.Builder object
@@@ -997,7 -1003,7 +1014,8 @@@
this.schemaVersion = builder.schemaVersion;
this.externalSchemaId = builder.externalSchemaId;
this.streamingTopicName = builder.streamingTopicName;
+ this.cdcIncludeScopes = builder.cdcIncludeScopes;
+ this.indexWhere = builder.indexWhere;
}
// When cloning table, ignore the salt column as it will be added back in the constructor
@@@ -2016,10 -2025,11 +2037,15 @@@
streamingTopicName =
(String) PVarchar.INSTANCE.toObject(table.getStreamingTopicName().toByteArray());
}
+ String indexWhere = null;
+ if (table.hasIndexWhere()) {
+ indexWhere =
+ (String) PVarchar.INSTANCE.toObject(table.getIndexWhere().toByteArray());
+ }
+ String cdcIncludeScopesStr = null;
+ if (table.hasCDCIncludeScopes()) {
+ cdcIncludeScopesStr = table.getCDCIncludeScopes();
+ }
try {
return new PTableImpl.Builder()
.setType(tableType)
@@@ -2077,8 -2087,7 +2103,9 @@@
.setSchemaVersion(schemaVersion)
.setExternalSchemaId(externalSchemaId)
.setStreamingTopicName(streamingTopicName)
+ .setCDCIncludeScopes(
+ CDCUtil.makeChangeScopeEnumsFromString(cdcIncludeScopesStr))
+ .setIndexWhere(indexWhere)
.build();
} catch (SQLException e) {
throw new RuntimeException(e); // Impossible
@@@ -2360,11 -2373,42 +2391,48 @@@
return streamingTopicName;
}
+ @Override
+ public Set<CDCChangeScope> getCDCIncludeScopes() {
+ return cdcIncludeScopes;
+ }
+
+ @Override
+ public String getIndexWhere() {
+ return indexWhere;
+ }
+
+ private void buildIndexWhereExpression(PhoenixConnection connection) throws SQLException {
+ PhoenixPreparedStatement
+ pstmt =
+ new PhoenixPreparedStatement(connection,
+ "select * from " + SchemaUtil.getTableName(parentSchemaName, parentTableName).getString() + " where " + indexWhere);
+ QueryPlan plan = pstmt.compileQuery();
+ ParseNode where = plan.getStatement().getWhere();
+ plan.getContext().setResolver(FromCompiler.getResolver(plan.getTableRef()));
+ indexWhereExpression = transformDNF(where, plan.getContext());
+ indexWhereColumns =
+ Sets.newHashSetWithExpectedSize(plan.getContext().getWhereConditionColumns().size());
+ for (Pair<byte[], byte[]> column : plan.getContext().getWhereConditionColumns()) {
+ indexWhereColumns.add(new ColumnReference(column.getFirst(), column.getSecond()));
+ }
+ }
+ @Override
+ public Expression getIndexWhereExpression(PhoenixConnection connection) throws SQLException {
+ if (indexWhereExpression == null && indexWhere != null) {
+ buildIndexWhereExpression(connection);
+ }
+ return indexWhereExpression;
+ }
+
+ @Override
+ public Set<ColumnReference> getIndexWhereColumns(PhoenixConnection connection)
+ throws SQLException {
+ if (indexWhereColumns == null && indexWhere != null) {
+ buildIndexWhereExpression(connection);
+ }
+ return indexWhereColumns;
+ }
++
private static final class KVColumnFamilyQualifier {
@Nonnull
private final String colFamilyName;
diff --cc phoenix-core/src/main/protobuf/PTable.proto
index cd8b69ecc8,88617e3636..283c391b3a
--- a/phoenix-core/src/main/protobuf/PTable.proto
+++ b/phoenix-core/src/main/protobuf/PTable.proto
@@@ -118,7 -117,7 +118,8 @@@ message PTable
optional bytes externalSchemaId=50;
optional PTable transformingNewTable=51;
optional bytes streamingTopicName=52;
- optional string CDCIncludeScopes=55;
+ optional bytes indexWhere=53;
++ optional string CDCIncludeScopes=54;
}
message EncodedCQCounter {