This is an automated email from the ASF dual-hosted git repository. tkhurana pushed a commit to branch PHOENIX-7562-feature-new in repository https://gitbox.apache.org/repos/asf/phoenix.git
commit 8f9655a69ebededb9d66186cfa99736380bd13dd Merge: 4a08b1d10d 29fbbbcef0 Author: tkhurana <[email protected]> AuthorDate: Fri Jun 5 14:28:47 2026 -0700 Merge remote-tracking branch 'upstream/master' into PHOENIX-7562-feature-new .github/PULL_REQUEST_TEMPLATE | 54 + BUILDING.md | 33 + Jenkinsfile | 2 +- Jenkinsfile.github | 10 +- Jenkinsfile.yetus | 2 +- bin/phoenix_utils.py | 26 + bin/sqlline.py | 1 + dev/phoenix-vote.sh | 72 +- docs/phoenix.csv | 16 + phoenix-assembly/pom.xml | 78 +- .../src/build/components/all-common-jars.xml | 10 +- .../src/build/components/assembly-noop.xml | 31 + .../src/build/components/release-only-jars.xml | 40 + phoenix-assembly/src/build/package-to-tar-all.xml | 3 + phoenix-core-client/pom.xml | 4 + phoenix-core-client/src/main/antlr3/PhoenixSQL.g | 38 +- .../phoenix/compile/ExplainPlanAttributes.java | 13 + .../org/apache/phoenix/compile/JoinCompiler.java | 3 +- .../apache/phoenix/compile/RVCOffsetCompiler.java | 23 +- .../compile/RegionLocationsListSerializer.java | 61 + .../org/apache/phoenix/compile/ScanRanges.java | 35 +- .../compile/ServerMergeColumnsSerializer.java | 59 + .../apache/phoenix/compile/StatementContext.java | 83 + .../org/apache/phoenix/compile/UpsertCompiler.java | 334 +-- .../org/apache/phoenix/compile/WhereOptimizer.java | 26 +- .../BaseScannerRegionObserverConstants.java | 29 + .../coprocessorclient/MetaDataProtocol.java | 4 +- .../apache/phoenix/exception/SQLExceptionCode.java | 10 + .../org/apache/phoenix/execute/MutationState.java | 124 +- .../apache/phoenix/execute/SegmentInfoPlan.java | 7 + .../apache/phoenix/expression/ExpressionType.java | 5 +- .../function/ByteBasedRegexpLikeFunction.java | 59 + .../expression/function/RegexpLikeFunction.java | 244 ++ .../function/StringBasedRegexpLikeFunction.java | 47 + .../util/bson/UpdateExpressionUtils.java | 138 +- .../phoenix/filter/DistinctPrefixFilter.java | 42 + .../metrics/MetricsIndexCDCConsumerSource.java | 103 + .../metrics/MetricsIndexCDCConsumerSourceImpl.java | 120 + .../index/metrics/MetricsIndexerSourceFactory.java | 8 + .../org/apache/phoenix/index/IndexMaintainer.java | 17 + .../phoenix/iterate/BaseResultIterators.java | 9 +- .../apache/phoenix/iterate/ParallelIterators.java | 7 +- .../phoenix/iterate/ScanningResultIterator.java | 111 +- .../apache/phoenix/iterate/SerialIterators.java | 3 +- .../phoenix/iterate/UnionResultIterators.java | 4 +- .../phoenix/jdbc/MutationLimitBatchException.java | 50 + .../ParallelPhoenixNullComparingResultSet.java | 8 + .../phoenix/jdbc/ParallelPhoenixResultSet.java | 8 + .../org/apache/phoenix/jdbc/PhoenixConnection.java | 17 + .../phoenix/jdbc/PhoenixDatabaseMetaData.java | 12 + .../phoenix/jdbc/PhoenixMonitoredResultSet.java | 5 + .../org/apache/phoenix/jdbc/PhoenixResultSet.java | 7 + .../org/apache/phoenix/jdbc/PhoenixStatement.java | 85 +- .../mapreduce/index/IndexToolTableUtil.java | 165 ++ .../phoenix/monitoring/GlobalClientMetrics.java | 25 + .../org/apache/phoenix/monitoring/MetricType.java | 91 +- .../phoenix/monitoring/OverAllQueryMetrics.java | 14 +- .../apache/phoenix/monitoring/ReadMetricQueue.java | 15 +- .../phoenix/monitoring/ScanMetricsHolder.java | 181 +- .../monitoring/SlowestScanMetricsQueue.java | 81 + .../phoenix/monitoring/TopNTreeMultiMap.java | 196 ++ .../apache/phoenix/parse/AlterIndexStatement.java | 15 +- .../apache/phoenix/parse/CreateIndexStatement.java | 34 + .../org/apache/phoenix/parse/ParseNodeFactory.java | 24 +- .../apache/phoenix/parse/RegexpLikeParseNode.java | 53 + .../phoenix/parse/TruncateTableStatement.java | 59 + .../org/apache/phoenix/parse/UpsertStatement.java | 6 +- .../phoenix/query/ConnectionQueryServices.java | 9 +- .../phoenix/query/ConnectionQueryServicesImpl.java | 68 +- .../query/ConnectionlessQueryServicesImpl.java | 13 + .../query/DelegateConnectionQueryServices.java | 6 + .../org/apache/phoenix/query/QueryConstants.java | 27 +- .../org/apache/phoenix/query/QueryServices.java | 13 + .../apache/phoenix/query/QueryServicesOptions.java | 13 +- .../phoenix/schema/ConditionalTTLExpression.java | 12 +- .../org/apache/phoenix/schema/DelegateTable.java | 6 + .../phoenix/schema/LiteralTTLExpression.java | 2 +- .../org/apache/phoenix/schema/MetaDataClient.java | 273 +- .../MutationLimitReachedException.java} | 26 +- .../java/org/apache/phoenix/schema/PTable.java | 13 + .../java/org/apache/phoenix/schema/PTableImpl.java | 28 +- .../org/apache/phoenix/schema/TTLExpression.java | 8 +- .../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 | 86 +- .../main/java/org/apache/phoenix/util/CDCUtil.java | 29 + .../java/org/apache/phoenix/util/JDBCUtil.java | 48 +- .../java/org/apache/phoenix/util/MetaDataUtil.java | 3 +- .../apache/phoenix/util/PhoenixKeyValueUtil.java | 15 - .../org/apache/phoenix/util/PhoenixRuntime.java | 25 + .../org/apache/phoenix/util/SHA256DigestUtil.java | 77 + .../java/org/apache/phoenix/util/ScanUtil.java | 96 +- .../src/main/protobuf/IndexMutations.proto | 39 + phoenix-core-client/src/main/protobuf/PTable.proto | 1 + .../src/main/protobuf/ServerCachingService.proto | 1 + phoenix-core-server/pom.xml | 8 + .../coprocessor/CDCGlobalIndexRegionScanner.java | 294 ++- .../phoenix/coprocessor/DataTableScanMetrics.java | 5 +- .../phoenix/coprocessor/MetaDataEndpointImpl.java | 68 +- .../coprocessor/PhoenixSyncTableRegionScanner.java | 350 +++ .../UncoveredGlobalIndexRegionScanner.java | 4 +- .../coprocessor/UncoveredIndexRegionScanner.java | 38 +- .../UncoveredLocalIndexRegionScanner.java | 4 +- .../UngroupedAggregateRegionObserver.java | 9 + .../phoenix/hbase/index/IndexCDCConsumer.java | 1347 ++++++++++ .../phoenix/hbase/index/IndexRegionObserver.java | 427 ++- .../TrackingParallelWriterIndexCommitter.java | 7 +- .../apache/phoenix/index/GlobalIndexChecker.java | 23 +- .../phoenix/iterate/RegionScannerFactory.java | 22 +- .../phoenix/mapreduce/PhoenixInputSplit.java | 49 +- .../mapreduce/PhoenixNoOpSingleRecordReader.java | 114 + .../PhoenixServerBuildIndexInputFormat.java | 16 +- .../PhoenixSyncTableCheckpointOutputRow.java | 318 +++ .../mapreduce/PhoenixSyncTableInputFormat.java | 345 +++ .../phoenix/mapreduce/PhoenixSyncTableMapper.java | 782 ++++++ .../PhoenixSyncTableOutputRepository.java | 244 ++ .../phoenix/mapreduce/PhoenixSyncTableTool.java | 614 +++++ .../apache/phoenix/mapreduce/index/IndexTool.java | 34 +- .../index/IndexVerificationOutputRepository.java | 42 +- .../index/IndexVerificationResultRepository.java | 50 +- .../phoenix/mapreduce/transform/TransformTool.java | 8 +- .../mapreduce/util/PhoenixConfigurationUtil.java | 5 + .../mapreduce/util/PhoenixMapReduceUtil.java | 55 + phoenix-core/pom.xml | 5 + .../phoenix/end2end/AsyncIndexPermissionIT.java | 355 +++ .../phoenix/end2end/BackwardCompatibilityIT.java | 9 + .../end2end/BackwardCompatibilityTestUtil.java | 3 + .../apache/phoenix/end2end/BasePermissionsIT.java | 20 +- .../end2end/BaseTotalSegmentsFunctionIT.java | 17 +- .../java/org/apache/phoenix/end2end/Bson2IT.java | 101 + .../java/org/apache/phoenix/end2end/Bson4IT.java | 11 + .../java/org/apache/phoenix/end2end/Bson5IT.java | 108 +- .../apache/phoenix/end2end/CDCDefinitionIT.java | 115 + ...ncurrentMutationsCoveredEventualGenerateIT.java | 73 + .../ConcurrentMutationsCoveredEventualIT.java | 23 +- .../ConcurrentMutationsCoveredNonEventualIT.java | 23 +- ... => ConcurrentMutationsExtendedGenerateIT.java} | 36 +- .../end2end/ConcurrentMutationsExtendedIT.java | 114 +- .../ConcurrentMutationsExtendedIndexIT.java | 438 ++++ .../ConcurrentMutationsLazyPostBatchWriteIT.java | 18 +- ...urrentMutationsUncoveredEventualGenerateIT.java | 73 + .../ConcurrentMutationsUncoveredEventualIT.java | 23 +- .../ConcurrentMutationsUncoveredNonEventualIT.java | 23 +- .../org/apache/phoenix/end2end/EmptyColumnIT.java | 57 + .../end2end/IndexMaintainerConsistencyIT.java | 237 ++ .../end2end/IndexRepairRegionScannerIT.java | 21 +- ...xToolForNonTxGlobalIndexEventualGenerateIT.java | 93 + .../IndexToolForNonTxGlobalIndexEventualIT.java | 93 + .../end2end/IndexToolForNonTxGlobalIndexIT.java | 75 +- .../org/apache/phoenix/end2end/IndexToolIT.java | 166 +- .../end2end/LoadSystemTableSnapshotBase.java | 36 +- .../end2end/MetadataServerConnectionsIT.java | 2 + .../MigrateSystemTablesToSystemNamespaceIT.java | 28 +- ...ava => MultiTenantEventualIndexGenerateIT.java} | 33 +- .../end2end/MultiTenantEventualIndexIT.java | 626 +++++ .../apache/phoenix/end2end/MultipleUpsertIT.java | 123 + .../apache/phoenix/end2end/MutationStateIT.java | 526 ++++ .../ParallelStatsDisabledWithRegionMovesIT.java | 2 + .../phoenix/end2end/PhoenixSyncTableToolIT.java | 2761 ++++++++++++++++++++ .../phoenix/end2end/QueryDatabaseMetaDataIT.java | 5 + .../phoenix/end2end/RegexpLikeFunctionIT.java | 452 ++++ .../phoenix/end2end/RowValueConstructorIT.java | 863 ++++++ .../org/apache/phoenix/end2end/ServerPagingIT.java | 10 +- .../end2end/ServerPagingWithRegionMovesIT.java | 31 +- .../org/apache/phoenix/end2end/SetPropertyIT.java | 44 + .../end2end/SkipSystemTablesExistenceCheckIT.java | 3 +- .../SystemTablesCreationOnConnectionIT.java | 10 +- .../apache/phoenix/end2end/TTLAsPhoenixTTLIT.java | 19 +- .../it/java/org/apache/phoenix/end2end/TTLIT.java | 18 +- .../phoenix/end2end/TenantSpecificTablesDDLIT.java | 3 + .../apache/phoenix/end2end/TruncateTableIT.java | 534 ++++ .../phoenix/end2end/UCFWithServerMetadataIT.java | 2 + .../phoenix/end2end/UpgradeIndexToolTablesIT.java | 97 + ...eIndexToolTablesNameSpaceMappingDisabledIT.java | 19 +- .../org/apache/phoenix/end2end/UpsertSelectIT.java | 2 + .../end2end/UpsertSelectWithRegionMovesIT.java | 2 + .../phoenix/end2end/VarBinaryEncoded2IT.java | 97 +- .../end2end/WhereOptimizerForArrayAnyIT.java | 36 +- .../end2end/WhereOptimizerForArrayAnyITBase.java | 83 + .../WhereOptimizerForArrayAnyNullablePKIT.java | 1160 ++++++++ .../GlobalIndexCheckerEventualGenerateIT.java | 89 + .../index/GlobalIndexCheckerEventualIT.java | 89 + .../end2end/index/GlobalIndexCheckerIT.java | 252 +- .../index/GlobalIndexCheckerWithRegionMovesIT.java | 2 + .../end2end/index/IndexAsyncThresholdIT.java | 2 + .../index/IndexVerificationOutputRepositoryIT.java | 13 +- .../index/IndexVerificationResultRepositoryIT.java | 17 +- .../end2end/index/PartialSystemCatalogIndexIT.java | 4 +- .../index/ReplicationWithWALAnnotationIT.java | 3 + .../UncoveredGlobalIndexRegionScanner2IT.java | 2 + .../phoenix/jdbc/FailoverPhoenixConnection2IT.java | 4 + .../phoenix/jdbc/FailoverPhoenixConnectionIT.java | 5 + .../jdbc/HighAvailabilityTestingUtilityIT.java | 5 + .../jdbc/ParallelPhoenixConnectionFallbackIT.java | 5 + .../phoenix/jdbc/ParallelPhoenixConnectionIT.java | 78 + .../ParallelPhoenixNullComparingResultSetIT.java | 5 + .../phoenix/monitoring/BasePhoenixMetricsIT.java | 2 + .../monitoring/PhoenixTableLevelMetricsIT.java | 3 + .../phoenix/monitoring/SlowestScanMetricsIT.java | 955 +++++++ .../ConnectionQueryServicesMetricsIT.java | 4 + .../phoenix/query/MaxConcurrentConnectionsIT.java | 2 + .../apache/phoenix/query/MetaDataCachingIT.java | 2 + .../resources/gold_files/gold_query_distinct.txt | 26 +- .../indexToolsnapshot.tar.gz | Bin 0 -> 20480 bytes .../src/it/resources/sql_files/create_distinct.sql | 36 + .../src/it/resources/sql_files/query_distinct.sql | 16 +- .../apache/phoenix/compile/QueryCompilerTest.java | 189 ++ .../phoenix/compile/SlowestScanMetricsTest.java | 406 +++ .../apache/phoenix/compile/WhereOptimizerTest.java | 36 +- .../PhoenixSyncTableRegionScannerTest.java | 535 ++++ .../function/RegexpLikeFunctionTest.java | 241 ++ .../org/apache/phoenix/index/IndexToolTest.java | 2 +- .../phoenix/mapreduce/PhoenixInputSplitTest.java | 200 ++ .../PhoenixNoOpSingleRecordReaderTest.java | 72 + .../mapreduce/PhoenixSyncTableInputFormatTest.java | 587 +++++ .../mapreduce/PhoenixSyncTableMapperTest.java | 512 ++++ .../PhoenixSyncTableOutputRepositoryTest.java | 848 ++++++ .../mapreduce/PhoenixSyncTableToolTest.java | 593 +++++ .../mapreduce/util/SHA256DigestUtilTest.java | 277 ++ .../phoenix/parse/IndexConsistencyParseTest.java | 87 + .../org/apache/phoenix/parse/QueryParserTest.java | 36 + .../java/org/apache/phoenix/query/BaseTest.java | 2 +- .../phoenix/query/explain/ExplainChangeRule.java | 54 + .../query/explain/ExplainCompatibilityTest.java | 769 ++++++ .../query/explain/ExplainJsonNormalizer.java | 71 + .../phoenix/query/explain/ExplainOracle.java | 207 ++ .../query/explain/ExplainTextNormalizer.java | 68 + .../schema/ConditionalTTLExpressionTest.java | 97 +- .../org/apache/phoenix/util/IndexScrutiny.java | 18 +- .../java/org/apache/phoenix/util/TestUtil.java | 94 + .../util/bson/UpdateExpressionUtilsTest.java | 446 ++++ .../phoenix/compat/hbase/CompatScanMetrics.java | 67 +- .../phoenix/compat/hbase/CompatScanMetrics.java | 67 +- .../phoenix/compat/hbase/CompatScanMetrics.java | 67 +- phoenix-hbase-compat-2.6.4/pom.xml | 2 +- .../phoenix/compat/hbase/CompatScanMetrics.java | 109 +- pom.xml | 36 +- 238 files changed, 26675 insertions(+), 1339 deletions(-) diff --cc phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/BaseScannerRegionObserverConstants.java index eb0d012399,3badbbdebf..80901dc65b --- a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/BaseScannerRegionObserverConstants.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/BaseScannerRegionObserverConstants.java @@@ -200,8 -200,32 +200,37 @@@ public class BaseScannerRegionObserverC /** Exposed for testing */ public static final String SCANNER_OPENED_TRACE_INFO = "Scanner opened on server"; + /** + * Attribute name used to pass HAGroupName in Mutations and Scans. + */ + public static final String HA_GROUP_NAME_ATTRIB = "_HAGroupName"; ++ + /** + * The scan attribute to enable server-side chunk formation and checksum computation for + * PhoenixSyncTableTool. + */ + public static final String SYNC_TABLE_CHUNK_FORMATION = "_SyncTableChunkFormation"; + + /** + * The scan attribute to provide the target chunk size in bytes for PhoenixSyncTableTool. + */ + public static final String SYNC_TABLE_CHUNK_SIZE_BYTES = "_SyncTableChunkSizeBytes"; + + /** + * The scan attribute to provide the MessageDigest state for cross-region hash continuation in + * PhoenixSyncTableTool. + */ + public static final String SYNC_TABLE_CONTINUED_DIGEST_STATE = "_SyncTableContinuedDigestState"; + + /** + * PhoenixSyncTableTool chunk metadata cell qualifiers. These define the wire protocol between + * PhoenixSyncTableRegionScanner (server-side coprocessor) and PhoenixSyncTableMapper (client-side + * mapper). The coprocessor returns chunk metadata as HBase cells with these qualifiers, and the + * mapper parses them to extract chunk information. + */ + public static final byte[] SYNC_TABLE_START_KEY_QUALIFIER = Bytes.toBytes("START_KEY"); + public static final byte[] SYNC_TABLE_HASH_QUALIFIER = Bytes.toBytes("HASH"); + public static final byte[] SYNC_TABLE_ROW_COUNT_QUALIFIER = Bytes.toBytes("ROW_COUNT"); + public static final byte[] SYNC_TABLE_IS_PARTIAL_CHUNK_QUALIFIER = + Bytes.toBytes("IS_PARTIAL_CHUNK"); } diff --cc phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java index 6067269347,44ab9c99ae..07c4b2d419 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java @@@ -185,12 -185,10 +185,14 @@@ public class PhoenixConnectio private LogLevel auditLogLevel; private Double logSamplingRate; private String sourceOfOperation; + @Nullable // Only available for connection which is part of HA Connections + private String haGroupName; + @Nullable // Only available for connection which is part of HA Connections + private HighAvailabilityGroup haGroup; private volatile SQLException reasonForClose; private static final String[] CONNECTION_PROPERTIES; + private final int slowestScanMetricsCount; + private final boolean isScanMetricsByRegionEnabled; private final ConcurrentLinkedQueue<PhoenixConnection> childConnections = new ConcurrentLinkedQueue<>(); diff --cc phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java index d1bc5e6a85,5cd95b648e..13276b8163 --- 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 @@@ -4108,10 -4126,10 +4133,14 @@@ public class ConnectionQueryServicesImp return ddl + ",TTL='" + ttlExpression + "'"; } + protected String getHAGroupDDL() { + return setSystemDDLProperties(QueryConstants.CREATE_HA_GROUP_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, @@@ -4445,10 -4463,16 +4474,20 @@@ metaConnection.createStatement().executeUpdate(getCDCStreamDDL()); } catch (TableAlreadyExistsException ignore) { } + try (Statement stmt = metaConnection.createStatement()) { + stmt.executeUpdate(getHAGroupDDL()); + } 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 + IndexToolTableUtil.createNewIndexToolTables(metaConnection); + } catch (Exception ignore) { + } } /** @@@ -5030,7 -5053,7 +5077,8 @@@ metaConnection = upgradeSystemMutex(metaConnection); metaConnection = upgradeSystemCDCStreamStatus(metaConnection); metaConnection = upgradeSystemCDCStream(metaConnection); + metaConnection = upgradeSystemHAGroup(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. diff --cc phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java index 9c04f7d6c3,abbc08ac31..cf779f287e --- 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 @@@ -221,10 -219,10 +221,14 @@@ public class ConnectionlessQueryService return setSystemDDLProperties(QueryConstants.CREATE_CDC_STREAM_METADATA); } + protected String getHAGroupDDL() { + return setSystemDDLProperties(QueryConstants.CREATE_HA_GROUP_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, @@@ -508,10 -500,10 +512,14 @@@ metaConnection.createStatement().executeUpdate(getCDCStreamDDL()); } catch (TableAlreadyExistsException ignore) { } + try (Statement stmt = metaConnection.createStatement()) { + stmt.executeUpdate(getHAGroupDDL()); + } catch (TableAlreadyExistsException ignore) { + } + try { + metaConnection.createStatement().executeUpdate(getIdxCdcTrackerDDL()); + } catch (TableAlreadyExistsException ignore) { + } } catch (SQLException e) { sqlE = e; } finally { diff --cc phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java index 92f9de94af,aa73c83314..717eb2c603 --- 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 @@@ -58,9 -54,7 +58,10 @@@ import static org.apache.phoenix.jdbc.P 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.HA_GROUP_NAME; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.HDFS_URL_1; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.HDFS_URL_2; + 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; @@@ -141,7 -137,7 +145,8 @@@ import static org.apache.phoenix.jdbc.P 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_HA_GROUP_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; @@@ -599,27 -598,17 +610,41 @@@ public interface QueryConstants + ColumnFamilyDescriptorBuilder.KEEP_DELETED_CELLS + "=%s,\n" + TRANSACTIONAL + "=" + Boolean.FALSE + ",\n" + UPDATE_CACHE_FREQUENCY + "=" + "7200000"; + String CREATE_HA_GROUP_METADATA = + "CREATE TABLE " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_HA_GROUP_TABLE + "\"(\n" + + // PK column + HA_GROUP_NAME + " VARCHAR NOT NULL," + // This is the unique identifier for the HA group + // Non-PK columns + ZK_URL_1 + " VARCHAR," + // This will be used to associate the ClusterRole and URL with local + // or peer cluster + ZK_URL_2 + " VARCHAR," + // This will be used to associate the ClusterRole and URL with local + // or peer cluster + CLUSTER_URL_1 + " VARCHAR," + // This will be returned to the client + CLUSTER_URL_2 + " VARCHAR," + // This will be returned to the client + CLUSTER_ROLE_1 + " VARCHAR," + // Role for peer cluster might not be accurate, we will use + // only local role for recovery if needed + CLUSTER_ROLE_2 + " VARCHAR," + // Role for peer cluster might not be accurate, we will use + // only local role for recovery if needed + POLICY + " VARCHAR," + // There should be only one policy for an HA group + HDFS_URL_1 + " VARCHAR," // HDFS URL for local cluster + + HDFS_URL_2 + " VARCHAR," // HDFS URL for peer cluster + + VERSION + " BIGINT," // Version should be incremented for Admin Updates, only for + // CLUSTER_URLs and REGISTRY_TYPE + + "CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + HA_GROUP_NAME + "))\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 --cc phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java index 42514d7329,085ac34a64..0d2a4f33c2 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java @@@ -641,24 -645,12 +649,29 @@@ public interface QueryServices extends "phoenix.uncovered.index.threads.keepalive.sec"; String USE_BLOOMFILTER_FOR_MULTIKEY_POINTLOOKUP = "phoenix.bloomfilter.multikey.pointlookup"; + // Value of N for the top N slowest scan metrics to be collected. Accepts integer values. + String SLOWEST_SCAN_METRICS_COUNT = "phoenix.slowest.scan.metrics.count"; + // Whether to collect region name and server name in the slowest scan metrics. Accepts boolean + // values. + String SCAN_METRICS_BY_REGION_ENABLED = "phoenix.scan.metrics.by.region.enabled"; + String SYNCHRONOUS_REPLICATION_ENABLED = "phoenix.synchronous.replication.enabled"; + + // HA Group Store sync job interval in seconds + String HA_GROUP_STORE_SYNC_INTERVAL_SECONDS = "phoenix.ha.group.store.sync.interval.seconds"; + + // "CRR" = Cluster Role Record. Master switch for syncing the legacy /phoenix/ha cluster + // role record from /phoenix/consistentHA. When false, no legacy znode is read, written, or + // deleted by HAGroupStoreClient. + String PHOENIX_HA_LEGACY_CRR_SYNC_ENABLED = "phoenix.ha.legacy.crr.sync.enabled"; + + // Periodic reconciliation interval for the legacy /phoenix/ha cluster role record sync, in + // seconds. 0 disables the periodic loop only; event-driven sync still runs. + String PHOENIX_HA_LEGACY_CRR_RECONCILIATION_INTERVAL_SECONDS = + "phoenix.ha.legacy.crr.reconciliation.interval.seconds"; + + String REPLICATION_LOG_ROTATION_TIME_MS_KEY = "phoenix.replication.log.rotation.time.ms"; + /** * Get executor service used for parallel scans */ diff --cc phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java index 3e99b9fff6,4e3c29b6c3..b4c68f0741 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java @@@ -513,21 -513,9 +520,23 @@@ public class QueryServicesOptions public static final int DEFAULT_PHOENIX_UNCOVERED_INDEX_MAX_POOL_SIZE = 512; public static final int DEFAULT_PHOENIX_UNCOVERED_INDEX_KEEP_ALIVE_TIME_SEC = 60; // 1min + public static final int DEFAULT_SLOWEST_SCAN_METRICS_COUNT = 0; + public static final boolean DEFAULT_SCAN_METRICS_BY_REGION_ENABLED = false; + public static final Boolean DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED = false; + + // Default HA Group Store sync job interval in seconds (15 minutes = 900 seconds) + public static final int DEFAULT_HA_GROUP_STORE_SYNC_INTERVAL_SECONDS = 900; + + // Legacy /phoenix/ha CRR sync is opt-in (default off). + public static final boolean DEFAULT_PHOENIX_HA_LEGACY_CRR_SYNC_ENABLED = false; + + // Periodic reconciliation interval for legacy /phoenix/ha CRR sync, in seconds. + // 0 disables the periodic loop only; event-driven sync still runs. + public static final long DEFAULT_PHOENIX_HA_LEGACY_CRR_RECONCILIATION_INTERVAL_SECONDS = 60L; + + public static final long DEFAULT_REPLICATION_LOG_ROTATION_TIME_MS = 60 * 1000L; + private final Configuration config; private QueryServicesOptions(Configuration config) { diff --cc phoenix-core-server/pom.xml index 914a7078ee,8eb3c38413..a3b49eb537 --- a/phoenix-core-server/pom.xml +++ b/phoenix-core-server/pom.xml @@@ -193,10 -173,14 +193,18 @@@ <groupId>com.google.code.findbugs</groupId> <artifactId>jsr305</artifactId> </dependency> + <dependency> + <groupId>com.lmax</groupId> + <artifactId>disruptor</artifactId> + </dependency> + <dependency> + <groupId>org.bouncycastle</groupId> + <artifactId>bcprov-jdk18on</artifactId> + </dependency> + <dependency> + <groupId>org.xerial.snappy</groupId> + <artifactId>snappy-java</artifactId> + </dependency> </dependencies> <build> diff --cc phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java index 01f7ee898b,9763388eff..2547719695 --- 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 @@@ -25,11 -25,9 +25,12 @@@ import static org.apache.phoenix.coproc import static org.apache.phoenix.hbase.index.util.IndexManagementUtil.rethrowIndexingException; import static org.apache.phoenix.index.PhoenixIndexBuilderHelper.ATOMIC_OP_ATTRIB; import static org.apache.phoenix.index.PhoenixIndexBuilderHelper.RETURN_RESULT; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_NAME; +import static org.apache.phoenix.query.QueryServices.SYNCHRONOUS_REPLICATION_ENABLED; +import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED; import static org.apache.phoenix.util.ByteUtil.EMPTY_BYTE_ARRAY; + import com.google.protobuf.ByteString; import java.io.ByteArrayInputStream; import java.io.DataInputStream; import java.io.EOFException; @@@ -182,9 -172,64 +188,67 @@@ public class IndexRegionObserver implem 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; + // Mutation attribute to ignore the mutation for replication + public static final String IGNORE_REPLICATION_ATTRIB = "_IGNORE_REPLICATION"; + private static final byte[] IGNORE_REPLICATION_ATTRIB_VAL = new byte[] { 0 }; + 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; + public static final String PHOENIX_INDEX_CDC_MUTATIONS_COMPRESS_ENABLED = + "phoenix.index.cdc.mutations.compress.enabled"; + public static final boolean DEFAULT_PHOENIX_INDEX_CDC_MUTATIONS_COMPRESS_ENABLED = false; + /** + * Controls which approach is used for implementing eventually consistent global secondary indexes + * via the {@link IndexCDCConsumer}. + * <p> + * <b>Approach 1: Serialized mutations (value = true)</b> + * </p> + * <p> + * During {@code preBatchMutate}, {@link IndexRegionObserver} generates index mutations for each + * data table mutation and serializes them into a Protobuf {@code IndexMutations} message. This + * serialized payload is written as a column value in the CDC index table row alongside the CDC + * event. The {@link IndexCDCConsumer} later reads these pre-computed mutations from the CDC + * index, deserializes them, and applies them directly to the index table(s). In this approach, + * the consumer does not need to understand index structure or re-derive mutations — it simply + * replays what was already computed on the write path. The trade-off is increased CDC index row + * size due to the serialized mutation payload, and additional write IO on the CDC index table. + * </p> + * <p> + * <b>Approach 2: Generated mutations from data row states (default, value = false)</b> + * </p> + * <p> + * During {@code preBatchMutate}, {@link IndexRegionObserver} writes only a lightweight CDC index + * entry without serialized index mutations. Instead, the CDC event is created with the + * {@code DATA_ROW_STATE} scope. When the {@link IndexCDCConsumer} processes these events, it + * reads the CDC index rows which trigger a server-side scan of the data table (via + * {@code CDCGlobalIndexRegionScanner}) to reconstruct the before-image + * ({@code currentDataRowState}) and after-image ({@code nextDataRowState}) of the data row at the + * change timestamp. These raw row states are returned as a Protobuf {@code DataRowStates} + * message. The consumer then feeds these states into {@code generateIndexMutationsForRow()} — the + * same core utility used by {@link IndexRegionObserver#prepareIndexMutations} on the write path — + * to derive index mutations at consume time. This approach keeps CDC index rows small, avoids + * additional write IO, and generates mutations based on the current index definition, but + * requires an additional data table read per CDC event and is sensitive to data visibility + * timing. Make sure max lookback age is long enough to retain before and after images of the row. + * </p> + * <p> + * <b>When to use which approach:</b> + * </p> + * <ul> + * <li>Use <b>Approach 2</b> (serialize = false, default) to minimize write IO: no serialized + * mutations are written to the CDC index, keeping CDC index rows small and write latency uniform. + * The trade-off is higher read IO at consume time — the consumer performs an additional data + * table point-lookup with a raw scan per CDC event to reconstruct row states.</li> + * <li>Use <b>Approach 1</b> (serialize = true) to minimize read IO: the consumer reads + * pre-computed mutations from the CDC index and applies them directly, with no data table scan + * required at consume time. The trade-off is higher write IO — serialized index mutations are + * written alongside each CDC index entry, increasing CDC index row size and write-path latency. + * Although CDC index is expected to have TTL same as the data table max lookback age.</li> + * </ul> + */ + public static final String PHOENIX_INDEX_CDC_MUTATION_SERIALIZE = + "phoenix.index.cdc.mutation.serialize"; + public static final boolean DEFAULT_PHOENIX_INDEX_CDC_MUTATION_SERIALIZE = false; /** * Class to represent pending data table rows @@@ -537,20 -553,14 +619,28 @@@ BloomType bloomFilterType = tableDescriptor.getColumnFamilies()[0].getBloomFilterType(); // when the table descriptor changes, the coproc is reloaded this.useBloomFilter = bloomFilterType == BloomType.ROW; + byte[] tableName = env.getRegionInfo().getTable().getName(); + this.shouldReplicate = env.getConfiguration().getBoolean(SYNCHRONOUS_REPLICATION_ENABLED, + DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED); + if (this.shouldReplicate) { + // replication feature is enabled, check if it is enabled for the table + this.shouldReplicate = SchemaUtil.shouldReplicateTable(tableName); + } + if (this.shouldReplicate) { + this.ignoreReplicationFilter = getSynchronousReplicationFilter(tableName); + } + // @CoreCoprocessor guarantees HasRegionServerServices, but guard for testability + if (e instanceof HasRegionServerServices) { + this.abortable = ((HasRegionServerServices) e).getRegionServerServices(); + } + if ( + this.indexCDCConsumerEnabled && !this.dataTableName.startsWith("SYSTEM.") + && !this.dataTableName.startsWith("SYSTEM:") + ) { + this.indexCDCConsumer = + new IndexCDCConsumer(env, this.dataTableName, serverName, this.serializeCDCMutations); + this.indexCDCConsumer.start(); + } } catch (NoSuchMethodError ex) { disabled = true; LOG.error("Must be too early a version of HBase. Disabled coprocessor ", ex); diff --cc phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java index 624f404760,ac0002a479..74df66eed7 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java @@@ -63,14 -63,19 +63,21 @@@ import org.junit.experimental.categorie @Category(NeedsOwnMiniClusterTest.class) public class MigrateSystemTablesToSystemNamespaceIT extends BaseTest { + private static final Set<String> HBASE_SYSTEM_TABLES = + new HashSet<>(Arrays.asList("SYSTEM.PHOENIX_INDEX_TOOL_RESULT", "SYSTEM.PHOENIX_INDEX_TOOL")); + 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", "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 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.HA_GROUP")); ++ "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.HA_GROUP", ++ "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:HA_GROUP")); ++ "SYSTEM:TRANSFORM", "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM", "SYSTEM:HA_GROUP", ++ "SYSTEM:IDX_CDC_TRACKER")); private static final String SCHEMA_NAME = "MIGRATETEST"; private static final String TABLE_NAME = SCHEMA_NAME + "." + MigrateSystemTablesToSystemNamespaceIT.class.getSimpleName().toUpperCase(); diff --cc phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java index 791f9a476f,53adfd8554..60e0aba9f7 --- 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,10 -189,11 +189,15 @@@ public class QueryDatabaseMetaDataIT ex assertEquals(PTableType.SYSTEM.toString(), rs.getString("TABLE_TYPE")); assertTrue(rs.next()); assertEquals(SYSTEM_CATALOG_SCHEMA, rs.getString("TABLE_SCHEM")); + assertEquals(PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_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_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()); diff --cc phoenix-core/src/it/java/org/apache/phoenix/end2end/SkipSystemTablesExistenceCheckIT.java index b704fcb547,3568628ec7..909a900b90 --- 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,8 +67,8 @@@ public class SkipSystemTablesExistenceC 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.HA_GROUP")); - "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.IDX_CDC_TRACKER", - "SYSTEM.PHOENIX_INDEX_TOOL_RESULT", "SYSTEM.PHOENIX_INDEX_TOOL")); ++ "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.HA_GROUP", ++ "SYSTEM.IDX_CDC_TRACKER", "SYSTEM.PHOENIX_INDEX_TOOL_RESULT", "SYSTEM.PHOENIX_INDEX_TOOL")); private static class PhoenixSystemTablesCreationTestDriver extends PhoenixTestDriver { private static ConnectionQueryServices cqs; diff --cc phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemTablesCreationOnConnectionIT.java index dfa7fe9bfa,5d1cb86a2b..b90a98fe25 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemTablesCreationOnConnectionIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemTablesCreationOnConnectionIT.java @@@ -103,12 -106,15 +106,14 @@@ public class SystemTablesCreationOnConn 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.HA_GROUP")); - "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.IDX_CDC_TRACKER", - "SYSTEM.PHOENIX_INDEX_TOOL_RESULT", "SYSTEM.PHOENIX_INDEX_TOOL")); ++ "SYSTEM.TRANSFORM", "SYSTEM.CDC_STREAM_STATUS", "SYSTEM.CDC_STREAM", "SYSTEM.HA_GROUP", ++ "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:HA_GROUP")); - "SYSTEM:TRANSFORM", "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM", "SYSTEM:IDX_CDC_TRACKER", - "SYSTEM:PHOENIX_INDEX_TOOL_RESULT", "SYSTEM:PHOENIX_INDEX_TOOL")); ++ "SYSTEM:TRANSFORM", "SYSTEM:CDC_STREAM_STATUS", "SYSTEM:CDC_STREAM", "SYSTEM:HA_GROUP", ++ "SYSTEM:IDX_CDC_TRACKER", "SYSTEM:PHOENIX_INDEX_TOOL_RESULT", "SYSTEM:PHOENIX_INDEX_TOOL")); private static class PhoenixSysCatCreationServices extends ConnectionQueryServicesImpl { diff --cc phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java index 6ee7fa4358,07ca749575..8ad9e2fa3e --- 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,9 -578,9 +578,12 @@@ public class TenantSpecificTablesDDLIT assertTrue(rs.next()); assertTableMetaData(rs, SYSTEM_CATALOG_SCHEMA, SYSTEM_FUNCTION_TABLE, SYSTEM); assertTrue(rs.next()); + assertTableMetaData(rs, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA, + PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_TABLE, PTableType.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 --cc phoenix-core/src/it/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection2IT.java index ee320cd74b,68221e6340..4b330712c5 --- 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,18 -18,11 +18,19 @@@ 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.exception.SQLExceptionCode.FAILOVER_IN_PROGRESS; + import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_CONSUMER_ENABLED; +import static org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole.ACTIVE; +import static org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole.ACTIVE_TO_STANDBY; +import static org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole.STANDBY; +import static org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole.STANDBY_TO_ACTIVE; +import static org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_CRR_CACHE_FREQUENCY_MS_KEY; 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; import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.getHighAvailibilityGroup; +import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.sleepThreadFor; +import static org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@@ -87,9 -77,11 +88,12 @@@ public class FailoverPhoenixConnection2 @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); + // Disable replication replay service + conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); + conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); ++ conf1.setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, false); ++ conf2.setBoolean(PHOENIX_INDEX_CDC_CONSUMER_ENABLED, false); + CLUSTERS.start(); DriverManager.registerDriver(PhoenixDriver.INSTANCE); } diff --cc phoenix-core/src/it/java/org/apache/phoenix/jdbc/HighAvailabilityTestingUtilityIT.java index 86793f11ce,7453c59903..b4709f3669 --- 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,8 -17,11 +17,9 @@@ */ 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; -import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.HBaseTestingUtilityPair.doTestWhenOneZKDown; import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.doTestBasicOperationsWithConnection; import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.getHighAvailibilityGroup; import static org.junit.Assert.assertFalse; @@@ -68,8 -76,16 +69,12 @@@ public class HighAvailabilityTestingUti /** Table name per test case. */ private String tableName; - public HighAvailabilityTestingUtilityIT(ClusterRoleRecord.RegistryType registryType) { - this.registryType = registryType; - } - @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 --cc phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionFallbackIT.java index 61f694604a,3a1ae7d311..93ff8e846b --- 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,8 -18,10 +18,9 @@@ 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; import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.doTestBasicOperationsWithConnection; import static org.junit.Assert.assertTrue; diff --cc phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java index d600a5064a,4a0c518efe..38821276a4 --- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java @@@ -65,7 -69,10 +66,9 @@@ import java.util.concurrent.CountDownLa import java.util.concurrent.ThreadPoolExecutor; import org.apache.commons.lang3.RandomStringUtils; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.HConstants; + import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException; + import org.apache.hadoop.hbase.util.VersionInfo; import org.apache.hadoop.test.GenericTestUtils; import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest; import org.apache.phoenix.exception.MutationBlockedIOException; @@@ -78,7 -86,9 +82,8 @@@ import org.apache.phoenix.monitoring.HT import org.apache.phoenix.monitoring.HTableThreadPoolMetricsManager; import org.apache.phoenix.monitoring.HistogramDistribution; import org.apache.phoenix.monitoring.MetricType; + import org.apache.phoenix.monitoring.SlowestScanMetricsIT; import org.apache.phoenix.query.ConnectionQueryServices; -import org.apache.phoenix.query.ConnectionQueryServicesImpl; import org.apache.phoenix.query.QueryServices; import org.apache.phoenix.util.PhoenixRuntime; import org.apache.phoenix.util.QueryUtil;
