You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@phoenix.apache.org by "palashc (via GitHub)" <gi...@apache.org> on 2024/12/03 08:24:24 UTC
[PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
palashc opened a new pull request, #2033:
URL: https://github.com/apache/phoenix/pull/2033
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877437795
##########
phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java:
##########
@@ -655,11 +655,11 @@ enum JoinType {INNER, LEFT_OUTER}
SYSTEM_CDC_STREAM_STATUS_TABLE + "\"(\n" +
// PK columns
TABLE_NAME + " VARCHAR NOT NULL," +
- STREAM_STATUS + " VARCHAR NOT NULL," +
+ STREAM_NAME + " VARCHAR NOT NULL," +
// Non-PK columns
- STREAM_NAME + " VARCHAR,\n" +
+ STREAM_STATUS + " VARCHAR,\n" +
Review Comment:
For querying, since there should only be one stream per table, the query should still be efficient?
SELECT STREAM_NAME FROM SYSTEM.CDC_STREAM_STATUS WHERE TABLE_NAME = <table-name> AND STREAM_STATUS = 'ENABLED'
##########
phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java:
##########
@@ -655,11 +655,11 @@ enum JoinType {INNER, LEFT_OUTER}
SYSTEM_CDC_STREAM_STATUS_TABLE + "\"(\n" +
// PK columns
TABLE_NAME + " VARCHAR NOT NULL," +
- STREAM_STATUS + " VARCHAR NOT NULL," +
+ STREAM_NAME + " VARCHAR NOT NULL," +
// Non-PK columns
- STREAM_NAME + " VARCHAR,\n" +
+ STREAM_STATUS + " VARCHAR,\n" +
Review Comment:
@virajjasani I changed it because the update pattern is
1. UPSERT (table_name, stream_name, ENABLING)
2. UPSERT (table_name, stream_name, ENABLED)
For querying, since there should only be one stream per table, the query should still be efficient?
`SELECT STREAM_NAME FROM SYSTEM.CDC_STREAM_STATUS WHERE TABLE_NAME = <table-name> AND STREAM_STATUS = 'ENABLED'`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1879793010
##########
phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java:
##########
@@ -150,4 +154,30 @@ public static boolean isBinaryType(PDataType dataType) {
|| sqlType == Types.LONGVARBINARY
|| dataType.getSqlType() == PDataType.VARBINARY_ENCODED_TYPE);
}
+
+ public enum CdcStreamStatus {
+ ENABLED("ENABLED"),
+ ENABLING("ENABLING"),
+ DISABLED("DISABLED"),
+ DISABLING("DISABLING");
+
+ private final String serializedValue;
+
+ private CdcStreamStatus(String value) {
+ this.serializedValue = value;
+ }
+
+ public String getSerializedValue() {
+ return serializedValue;
+ }
+ }
+
+ public static long getCDCCreationTimestamp(PTable table) {
+ for (PTable index : table.getIndexes()) {
+ if (CDCUtil.isCDCIndex(index)) {
+ return index.getTimeStamp();
+ }
+ }
Review Comment:
Please correct me if I am wrong but we want to move towards single stream and single CDC index, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#issuecomment-2532072435
DROP CDC can be separate sub task IMO. Keeping individual sub tasks make it much simpler to follow the change history.
I think we should be good to go unless there is any significant concern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1873042035
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTable(tableName));
+ String streamStatusSQL = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+ PreparedStatement ps = connection.prepareStatement(streamStatusSQL);
+ String streamName = String.format(CDC_STREAM_NAME_FORMAT, tableName, cdcIndexTimestamp);
+ ps.setString(1, tableName);
+ ps.setString(2, streamName);
+ ps.setString(3, CDCUtil.CdcStreamStatus.ENABLING.getSerializedValue());
+ ps.executeUpdate();
+ connection.commit();
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1901259078
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.task.ServerTask;
+import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
Review Comment:
We have seen issues with range scan queries timing out once or twice due to large num of regions. To handle them efficiently, precise region boundaries are now used: https://github.com/apache/phoenix/commit/0b9d8201875c27a8b8a1084680104057bcd46e60
However, even with large tables, CQSI#getTableRegions never timed out infinitely. It only used to timeout 1 or 2 times. This is because it never has to re-do the whole work, as every RPC call will be done only once for each region location. Once region location is retrieved, HBase client caches it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1868807959
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTableNoCache(tableName));
Review Comment:
I will check if the index ptable is updated in the data table ptable in the client's cache.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1876017456
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
+ upsertPartitionMetadata(pconn, tableName, streamName, tableRegions);
+ updateStreamStatus(pconn, tableName, streamName);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SUCCESS, "");
+ } catch (Throwable t) {
+ LOGGER.error("Exception while bootstrapping CDC Stream Partition Metadata for "
+ + taskRecord.getTableName() + " and timestamp " + timestamp.toString(), t);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.FAIL, t.toString());
Review Comment:
TODO: Consider a RETRY strategy here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877803588
##########
phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java:
##########
@@ -655,11 +655,11 @@ enum JoinType {INNER, LEFT_OUTER}
SYSTEM_CDC_STREAM_STATUS_TABLE + "\"(\n" +
// PK columns
TABLE_NAME + " VARCHAR NOT NULL," +
- STREAM_STATUS + " VARCHAR NOT NULL," +
+ STREAM_NAME + " VARCHAR NOT NULL," +
// Non-PK columns
- STREAM_NAME + " VARCHAR,\n" +
+ STREAM_STATUS + " VARCHAR,\n" +
Review Comment:
Also for the fact that you can can have multiple streams in the same state, see my latest comment on the corresponding [thread](https://docs.google.com/document/d/157KXONp4ZoWmhUBsy-lK3LNAanCsuJ3cb7qYTmozgZo/edit?disco=AAABZrpxMfI) in the design doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1901255834
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.task.ServerTask;
+import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
Review Comment:
@haridsv I just had offline discussion with @palashc. Since he is already using the existing API which is already used by `BaseResultIterators` for each query. The timeout is not for HBase ConnectionRegistry API but its for Phoenix client to internally figure out if meta RPC calls are exceeding some timeout.
For very large tables with full table scan, while going through the loop (CQSI#getTableRegions), the timeout can happen only if the region locations are not cached. Let's assume we have 500k regions with 2 min timeout set by the CDC Stream Metadata Task. Let's say none of those regions are cached at the moment. So, CQSI#getTableRegions starts RPC calls for each region and somehow meta is slow to respond and hence after 200k region locations, CQSI exceeds 2 min timeout and throws error. This means we already have 200k region locations cached at the client side. Hence, next time when the task is retried, CQSI#getTableRegions will go through 200k region location loop in few milliseconds because no RPC call is required. For the remaining regions, it will start creating RPC calls, until every region location is cached.
Given that this API is already being used by Phoenix client with every query, we are confident that it is already working fine and even for large tables, it can potentially timeout only few times, not all time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877437795
##########
phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java:
##########
@@ -655,11 +655,11 @@ enum JoinType {INNER, LEFT_OUTER}
SYSTEM_CDC_STREAM_STATUS_TABLE + "\"(\n" +
// PK columns
TABLE_NAME + " VARCHAR NOT NULL," +
- STREAM_STATUS + " VARCHAR NOT NULL," +
+ STREAM_NAME + " VARCHAR NOT NULL," +
// Non-PK columns
- STREAM_NAME + " VARCHAR,\n" +
+ STREAM_STATUS + " VARCHAR,\n" +
Review Comment:
For querying, since there should only be one stream per table, the query should still be efficient?
SELECT STREAM_NAME FROM SYSTEM.CDC_STREAM_STATUS WHERE TABLE_NAME = <table-name> AND STREAM_STATUS = 'ENABLED'
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877074035
##########
phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java:
##########
@@ -655,11 +655,11 @@ enum JoinType {INNER, LEFT_OUTER}
SYSTEM_CDC_STREAM_STATUS_TABLE + "\"(\n" +
// PK columns
TABLE_NAME + " VARCHAR NOT NULL," +
- STREAM_STATUS + " VARCHAR NOT NULL," +
+ STREAM_NAME + " VARCHAR NOT NULL," +
// Non-PK columns
- STREAM_NAME + " VARCHAR,\n" +
+ STREAM_STATUS + " VARCHAR,\n" +
Review Comment:
@palashc btw the query pattern is going to be:
```
Step 1: Get the active table stream from the metadata.
SELECT STREAM_NAME FROM SYSTEM.CDC_STREAM_STATUS WHERE TABLE_NAME = <table-name> AND STREAM_STATUS = 'ACTIVE'
```
How would keeping STREAM_NAME help here? I missed this earlier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#issuecomment-2522187225
When will the DROP CDC be hooked up to change the stream status?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#issuecomment-2568642182
Latest Test Report: https://ci-hadoop.apache.org/job/Phoenix/job/Phoenix-PreCommit-GitHub-PR/job/PR-2033/13/testReport/
Failures look unrelated to code change and pass locally.
```
org.apache.phoenix.end2end.AggregateQueryWithRegionMoves2IT.testSplitWithCachedMeta[AggregateQueryWithRegionMoves2IT_3]
org.apache.phoenix.end2end.PermissionNSDisabledWithCustomAccessControllerIT.testSuperUserCanChangePerms
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc merged PR #2033:
URL: https://github.com/apache/phoenix/pull/2033
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1882541480
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.task.ServerTask;
+import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
Review Comment:
We can use existing method that is already used by getAllTableRegions() API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1868041757
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamIT.java:
##########
@@ -0,0 +1,104 @@
+package org.apache.phoenix.end2end;
Review Comment:
Same here
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,100 @@
+package org.apache.phoenix.coprocessor.tasks;
Review Comment:
ASF Lincense missing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1868098330
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,100 @@
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.util.CDCUtil.CDC_STREAM_NAME_FORMAT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ public static String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
Review Comment:
nit: public static final
##########
phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java:
##########
@@ -150,4 +152,31 @@ public static boolean isBinaryType(PDataType dataType) {
|| sqlType == Types.LONGVARBINARY
|| dataType.getSqlType() == PDataType.VARBINARY_ENCODED_TYPE);
}
+
+ public enum CdcStreamStatus {
+ ENABLED("ENABLED"),
+ ENABLING("ENABLING"),
+ DISABLED("DISABLED"),
+ DISABLING("DISABLING");
+
+ private final String serializedValue;
+
+ private CdcStreamStatus(String value) {
+ this.serializedValue = value;
+ }
+
+ public String getSerializedValue() {
+ return serializedValue;
+ }
+ }
+
+ public static long getCDCCreationTimestamp(PTable table) {
+ long ts = 0;
+ for (PTable index : table.getIndexes()) {
+ if (CDCUtil.isCDCIndex(index)) {
+ ts = index.getTimeStamp();
Review Comment:
nit: return value directly from here?
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTableNoCache(tableName));
Review Comment:
Can we live with getTable() instead of getTableNoCache()?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1872792630
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTable(tableName));
+ String streamStatusSQL = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+ PreparedStatement ps = connection.prepareStatement(streamStatusSQL);
+ String streamName = String.format(CDC_STREAM_NAME_FORMAT, tableName, cdcIndexTimestamp);
+ ps.setString(1, tableName);
+ ps.setString(2, streamName);
+ ps.setString(3, CDCUtil.CdcStreamStatus.ENABLING.getSerializedValue());
+ ps.executeUpdate();
+ connection.commit();
+
+ try {
+ List<Mutation> sysTaskUpsertMutations = Task.getMutationsForAddTask(
+ new SystemTaskParams.SystemTaskParamsBuilder()
+ .setConn(connection)
+ .setTaskType(PTable.TaskType.CDC_STREAM_PARTITION)
+ .setTableName(tableName) //give full table name
+ .setSchemaName(streamName) // use schemaName to pass streamName
+ .build());
+ byte[] rowKey = sysTaskUpsertMutations
+ .get(0).getRow();
+ MetaDataProtocol.MetaDataMutationResult metaDataMutationResult =
+ Task.taskMetaDataCoprocessorExec(connection, rowKey,
+ new TaskMetaDataServiceCallBack(sysTaskUpsertMutations));
+ if (MetaDataProtocol.MutationCode.UNABLE_TO_UPSERT_TASK.equals(
+ metaDataMutationResult.getMutationCode())) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.UNABLE_TO_UPSERT_TASK)
+ .setSchemaName(SYSTEM_SCHEMA_NAME)
+ .setTableName(SYSTEM_TASK_TABLE).build().buildException();
+ }
+ } catch (IOException ioe) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.UNABLE_TO_UPSERT_TASK)
+ .setRootCause(ioe)
+ .setMessage(ioe.getMessage())
+ .setSchemaName(SYSTEM_SCHEMA_NAME)
+ .setTableName(SYSTEM_TASK_TABLE).build().buildException();
+ }
Review Comment:
@haridsv [PHOENIX-7460](https://issues.apache.org/jira/browse/PHOENIX-7460) for keeping region metadata in sync.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#issuecomment-2532193942
LThough if the PR changes are manageable as single sub task, we can include them here as well. Not a problem. It's upto you @palashc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#issuecomment-2532083276
We even need more validation changes for CDC creation anyway e.g. allow only one CDC Index to be created per table, and therefore allow only one CDC object per table.
Multiple CDC objects are redundant entries in SYSTEM.CATALOG if they only point to single CDC Index.
I mean to say there will always be some improvements, so I believe we should follow up with separate sub tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877772933
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -1963,6 +1964,15 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
ColumnResolver resolver = FromCompiler.getResolver(NamedTableNode.create(statement.getDataTable()), connection);
TableRef tableRef = resolver.getTables().get(0);
PTable dataTable = tableRef.getTable();
+ String
+ dataTableFullName =
+ SchemaUtil.getTableName(statement.getDataTable().getSchemaName(),
+ statement.getDataTable().getTableName());
+
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
Review Comment:
nit: with enums you can do equality check
```suggestion
if (PTableType.TABLE == dataTable.getType()) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1879791747
##########
phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java:
##########
@@ -150,4 +154,30 @@ public static boolean isBinaryType(PDataType dataType) {
|| sqlType == Types.LONGVARBINARY
|| dataType.getSqlType() == PDataType.VARBINARY_ENCODED_TYPE);
}
+
+ public enum CdcStreamStatus {
+ ENABLED("ENABLED"),
+ ENABLING("ENABLING"),
+ DISABLED("DISABLED"),
+ DISABLING("DISABLING");
+
+ private final String serializedValue;
+
+ private CdcStreamStatus(String value) {
+ this.serializedValue = value;
+ }
+
+ public String getSerializedValue() {
+ return serializedValue;
+ }
+ }
+
+ public static long getCDCCreationTimestamp(PTable table) {
+ for (PTable index : table.getIndexes()) {
+ if (CDCUtil.isCDCIndex(index)) {
+ return index.getTimeStamp();
+ }
+ }
Review Comment:
@haridsv As we discussed I added the check to only allow a single (enable/enabling) stream in createCDC. This also has the effect of only allowing a single CDC index since I throw a SQLException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877436958
##########
phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java:
##########
@@ -655,11 +655,11 @@ enum JoinType {INNER, LEFT_OUTER}
SYSTEM_CDC_STREAM_STATUS_TABLE + "\"(\n" +
// PK columns
TABLE_NAME + " VARCHAR NOT NULL," +
- STREAM_STATUS + " VARCHAR NOT NULL," +
+ STREAM_NAME + " VARCHAR NOT NULL," +
// Non-PK columns
- STREAM_NAME + " VARCHAR,\n" +
+ STREAM_STATUS + " VARCHAR,\n" +
Review Comment:
@virajjasani I changed it because the update pattern is
1. UPSERT (table_name, stream_name, ENABLING)
2. UPSERT (table_name, stream_name, ENABLED)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1868812174
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTableNoCache(tableName));
Review Comment:
confirmed that getTable should work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1868807959
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTableNoCache(tableName));
Review Comment:
I will check if the index ptable is updated in the data table ptable in the client's cache after the cdc index is created.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877805984
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
+ upsertPartitionMetadata(pconn, tableName, streamName, tableRegions);
+ updateStreamStatus(pconn, tableName, streamName);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SUCCESS, "");
+ } catch (Throwable t) {
+ LOGGER.error("Exception while bootstrapping CDC Stream Partition Metadata for "
+ + taskRecord.getTableName() + " and timestamp " + timestamp.toString(), t);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.FAIL, t.toString());
Review Comment:
I think it is safe to retry on all `SQLException` scenarios, which can only happen when there is an issue with the system table availability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1879672456
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.task.ServerTask;
+import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
Review Comment:
If there is a timeout due to the table having a large number of regions, it may continue to timeout even in retries. How about we query for the PARTITION_END_KEY of the last partition and use `getTableRegions` to start from that key? We could load any existing partition records for this stream first (which will be empty the first time so we will default to `HConstants.EMPTY_START_ROW`), sort them and get the highest end key. The end key for `getTableRegions` would also be `HConstants.EMPTY_START_ROW`.
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2032,16 +2038,81 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
tableProps.put(TableProperty.MULTI_TENANT.getPropertyName(), Boolean.TRUE);
}
CreateTableStatement tableStatement = FACTORY.createTable(
- FACTORY.table(dataTable.getSchemaName().getString(), statement.getCdcObjName().getName()),
+ FACTORY.table(dataTable.getSchemaName().getString(), cdcObjName),
null, columnDefs, FACTORY.primaryKey(null, pkColumnDefs),
Collections.emptyList(), PTableType.CDC, statement.isIfNotExists(), null, null,
statement.getBindCount(), null);
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE == dataTable.getType()) {
+ updateStreamPartitionMetadata(dataTableFullName, cdcObjName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName, String cdcObjName) throws SQLException {
+ // create Stream with ENABLING status
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTable(tableName));
+ String streamStatusSQL = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+ String streamName = String.format(CDC_STREAM_NAME_FORMAT, tableName, cdcObjName, cdcIndexTimestamp);
+ try (PreparedStatement ps = connection.prepareStatement(streamStatusSQL)) {
+ ps.setString(1, tableName);
+ ps.setString(2, streamName);
+ ps.setString(3, CDCUtil.CdcStreamStatus.ENABLING.getSerializedValue());
+ ps.executeUpdate();
+ connection.commit();
+ LOGGER.info("Marked stream {} for table {} as ENABLING", streamName, tableName);
+ }
+
+ // insert task to update partition metadata for stream
+ try {
+ List<Mutation> sysTaskUpsertMutations = Task.getMutationsForAddTask(
+ new SystemTaskParams.SystemTaskParamsBuilder()
+ .setConn(connection)
+ .setTaskType(PTable.TaskType.CDC_STREAM_PARTITION)
+ .setTableName(tableName) //give full table name
+ .setSchemaName(streamName) // use schemaName to pass streamName
+ .build());
Review Comment:
Shouldn't we set the status to CREATED?
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -3942,11 +4013,22 @@ public MutationState dropCDC(DropCDCStatement statement) throws SQLException {
String schemaName = statement.getTableName().getSchemaName();
String cdcTableName = statement.getCdcObjName().getName();
String parentTableName = statement.getTableName().getTableName();
+ String indexName = CDCUtil.getCDCIndexName(statement.getCdcObjName().getName());
+ // Mark CDC Stream as Disabled
+ long cdcIndexTimestamp = connection.getTable(indexName).getTimeStamp();
+ String streamStatusSQL = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+ String streamName = String.format(CDC_STREAM_NAME_FORMAT, parentTableName, cdcTableName, cdcIndexTimestamp);
+ try (PreparedStatement ps = connection.prepareStatement(streamStatusSQL)) {
+ ps.setString(1, parentTableName);
+ ps.setString(2, streamName);
+ ps.setString(3, CDCUtil.CdcStreamStatus.DISABLED.getSerializedValue());
+ ps.executeUpdate();
+ connection.commit();
+ LOGGER.info("Marked stream {} for table {} as DISABLED", streamName, parentTableName);
+ }
Review Comment:
We should do this after the CDC drop is successful, may be even after the index drop.
##########
phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java:
##########
@@ -150,4 +154,30 @@ public static boolean isBinaryType(PDataType dataType) {
|| sqlType == Types.LONGVARBINARY
|| dataType.getSqlType() == PDataType.VARBINARY_ENCODED_TYPE);
}
+
+ public enum CdcStreamStatus {
+ ENABLED("ENABLED"),
+ ENABLING("ENABLING"),
+ DISABLED("DISABLED"),
+ DISABLING("DISABLING");
+
+ private final String serializedValue;
+
+ private CdcStreamStatus(String value) {
+ this.serializedValue = value;
+ }
+
+ public String getSerializedValue() {
+ return serializedValue;
+ }
+ }
+
+ public static long getCDCCreationTimestamp(PTable table) {
+ for (PTable index : table.getIndexes()) {
+ if (CDCUtil.isCDCIndex(index)) {
+ return index.getTimeStamp();
+ }
+ }
Review Comment:
This method is misleading, it is returning the timestamp of the first index it finds. Since currently we can have multiple CDC indexes, it may return different value at different times. In theory, this can make `updateStreamPartitionMetadata` reuse a previous stream name instead of assigning a new one and `dropCDC` try to update the status for a wrong stream.
I am thinking, we can do the following as a quick workaround:
1. We can change `CDC_STREAM_NAME_FORMAT` to just "<table name>-UUID"` and avoid depending on this method in `updateStreamPartitionMetadata`.
2. In `dropCDC`, query for the name of the ENABLED/ENABLING stream to change its status, rather than constructing the name from `CDC_STREAM_NAME_FORMAT`.
3. In the ITs, we would assert on the existence of a stream of specific status rather than on the specific name.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.task.ServerTask;
+import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
+ upsertPartitionMetadata(pconn, tableName, streamName, tableRegions);
+ updateStreamStatus(pconn, tableName, streamName);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SUCCESS, "");
+ } catch (SQLException e) {
+ try {
+ // Update task status to RETRY so that it is retried
+ ServerTask.addTask(new SystemTaskParams.SystemTaskParamsBuilder()
+ .setConn(pconn)
+ .setTaskType(taskRecord.getTaskType())
+ .setSchemaName(taskRecord.getSchemaName())
+ .setTableName(taskRecord.getTableName())
+ .setTaskStatus(PTable.TaskStatus.RETRY.toString())
+ .setStartTs(taskRecord.getTimeStamp())
+ .setEndTs(null)
+ .build());
+ LOGGER.error("Marking task as RETRY. " +
Review Comment:
Shouldn't this be a warning, since we are not giving up yet?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1879672456
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.task.ServerTask;
+import org.apache.phoenix.schema.task.SystemTaskParams;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
Review Comment:
If there is a timeout due to the table having a large number of regions, it may continue to timeout even in retries. How about we query for the PARTITION_END_KEY of the last partition and use `getTableRegions` to start from that key? We could load any existing partition records for this stream first (which will be empty the first time so we will default to `HConstants.EMPTY_START_ROW`), sort them and get the highest end key. The end key for `getTableRegions` would always be `HConstants.EMPTY_START_ROW`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1880382665
##########
phoenix-core-client/src/main/java/org/apache/phoenix/util/CDCUtil.java:
##########
@@ -150,4 +154,30 @@ public static boolean isBinaryType(PDataType dataType) {
|| sqlType == Types.LONGVARBINARY
|| dataType.getSqlType() == PDataType.VARBINARY_ENCODED_TYPE);
}
+
+ public enum CdcStreamStatus {
+ ENABLED("ENABLED"),
+ ENABLING("ENABLING"),
+ DISABLED("DISABLED"),
+ DISABLING("DISABLING");
+
+ private final String serializedValue;
+
+ private CdcStreamStatus(String value) {
+ this.serializedValue = value;
+ }
+
+ public String getSerializedValue() {
+ return serializedValue;
+ }
+ }
+
+ public static long getCDCCreationTimestamp(PTable table) {
+ for (PTable index : table.getIndexes()) {
+ if (CDCUtil.isCDCIndex(index)) {
+ return index.getTimeStamp();
+ }
+ }
Review Comment:
Yes you are right, since we are not allowing more than 1 CDC, the logic is harmless, sorry about the spam.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "virajjasani (via GitHub)" <gi...@apache.org>.
virajjasani commented on PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#issuecomment-2568495423
+1, pending jenkins build results.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "haridsv (via GitHub)" <gi...@apache.org>.
haridsv commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1871021076
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTable(tableName));
+ String streamStatusSQL = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+ PreparedStatement ps = connection.prepareStatement(streamStatusSQL);
+ String streamName = String.format(CDC_STREAM_NAME_FORMAT, tableName, cdcIndexTimestamp);
+ ps.setString(1, tableName);
+ ps.setString(2, streamName);
+ ps.setString(3, CDCUtil.CdcStreamStatus.ENABLING.getSerializedValue());
+ ps.executeUpdate();
+ connection.commit();
Review Comment:
We can use try-with-resource block here to ensure PreparedStatement will always be closed and also that it will be released quickly.
##########
phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java:
##########
@@ -2039,9 +2041,57 @@ public MutationState createCDC(CreateCDCStatement statement) throws SQLException
createTableInternal(tableStatement, null, dataTable, null, null, null, null,
null, null, false, null,
null, statement.getIncludeScopes(), tableProps, commonFamilyProps);
+ // for now, only track stream partition metadata for tables, TODO: updatable views
+ if (PTableType.TABLE.equals(dataTable.getType())) {
+ updateStreamPartitionMetadata(dataTableFullName);
+ }
return new MutationState(0, 0, connection);
}
+ /**
+ * Trigger CDC Stream Partition metadata bootstrap for the given table in the background.
+ * Mark status as ENABLING in SYSTEM.CDC_STREAM_STATUS and add {@link CdcStreamPartitionMetadataTask}
+ * to SYSTEM.TASK which updates partition metadata based on table regions.
+ */
+ private void updateStreamPartitionMetadata(String tableName) throws SQLException {
+ long cdcIndexTimestamp = CDCUtil.getCDCCreationTimestamp(connection.getTable(tableName));
+ String streamStatusSQL = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+ PreparedStatement ps = connection.prepareStatement(streamStatusSQL);
+ String streamName = String.format(CDC_STREAM_NAME_FORMAT, tableName, cdcIndexTimestamp);
+ ps.setString(1, tableName);
+ ps.setString(2, streamName);
+ ps.setString(3, CDCUtil.CdcStreamStatus.ENABLING.getSerializedValue());
+ ps.executeUpdate();
+ connection.commit();
+
+ try {
+ List<Mutation> sysTaskUpsertMutations = Task.getMutationsForAddTask(
+ new SystemTaskParams.SystemTaskParamsBuilder()
+ .setConn(connection)
+ .setTaskType(PTable.TaskType.CDC_STREAM_PARTITION)
+ .setTableName(tableName) //give full table name
+ .setSchemaName(streamName) // use schemaName to pass streamName
+ .build());
+ byte[] rowKey = sysTaskUpsertMutations
+ .get(0).getRow();
+ MetaDataProtocol.MetaDataMutationResult metaDataMutationResult =
+ Task.taskMetaDataCoprocessorExec(connection, rowKey,
+ new TaskMetaDataServiceCallBack(sysTaskUpsertMutations));
+ if (MetaDataProtocol.MutationCode.UNABLE_TO_UPSERT_TASK.equals(
+ metaDataMutationResult.getMutationCode())) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.UNABLE_TO_UPSERT_TASK)
+ .setSchemaName(SYSTEM_SCHEMA_NAME)
+ .setTableName(SYSTEM_TASK_TABLE).build().buildException();
+ }
+ } catch (IOException ioe) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.UNABLE_TO_UPSERT_TASK)
+ .setRootCause(ioe)
+ .setMessage(ioe.getMessage())
+ .setSchemaName(SYSTEM_SCHEMA_NAME)
+ .setTableName(SYSTEM_TASK_TABLE).build().buildException();
+ }
Review Comment:
Shouldn't we do this before creating the CDC object itself to be technically correct and avoid the window when the CDC is active but the stream is not? Also, how do we plan to synchronize this with the ongoing splits and merges?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#issuecomment-2522478906
@haridsv Since DROP CDC needs more changes like not dropping the index but only stopping the writes to the index, I was thinking we can do that part as a separate JIRA. I can create another subtask under PHOENIX-7456, @virajjasani wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1878687040
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
+ upsertPartitionMetadata(pconn, tableName, streamName, tableRegions);
+ updateStreamStatus(pconn, tableName, streamName);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SUCCESS, "");
+ } catch (Throwable t) {
+ LOGGER.error("Exception while bootstrapping CDC Stream Partition Metadata for "
+ + taskRecord.getTableName() + " and timestamp " + timestamp.toString(), t);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.FAIL, t.toString());
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] PHOENIX-7459 : Bootstrap stream metadata when CDC is enabled on a table [phoenix]
Posted by "palashc (via GitHub)" <gi...@apache.org>.
palashc commented on code in PR #2033:
URL: https://github.com/apache/phoenix/pull/2033#discussion_r1877465052
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/CdcStreamPartitionMetadataTask.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.coprocessor.tasks;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.task.Task;
+import org.apache.phoenix.util.CDCUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_STATUS_NAME;
+import static org.apache.phoenix.query.QueryServices.PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT;
+
+/**
+ * Task to bootstrap partition metadata when CDC is enabled on a table.
+ * Upserts one row for each region of the table into SYSTEM.CDC_STREAM and marks the status as
+ * ENABLED in SYSTEM.CDC_STREAM_STATUS.
+ */
+public class CdcStreamPartitionMetadataTask extends BaseTask {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(CdcStreamPartitionMetadataTask.class);
+ private static final String CDC_STREAM_STATUS_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_STATUS_NAME + " VALUES (?, ?, ?)";
+
+ // parent_partition_id will be null, set partition_end_time to -1
+ private static final String CDC_STREAM_PARTITION_UPSERT_SQL
+ = "UPSERT INTO " + SYSTEM_CDC_STREAM_NAME + " VALUES (?,?,?,null,?,-1,?,?)";
+
+ @Override
+ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
+ Configuration conf = HBaseConfiguration.create(env.getConfiguration());
+ Configuration configuration = HBaseConfiguration.addHbaseResources(conf);
+ int getTableRegionsTimeout = configuration.getInt(PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT,
+ DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT);
+ PhoenixConnection pconn = null;
+ String tableName = taskRecord.getTableName();
+ String streamName = taskRecord.getSchemaName();
+ Timestamp timestamp = taskRecord.getTimeStamp();
+ try {
+ pconn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ List<HRegionLocation> tableRegions = pconn.getQueryServices().getAllTableRegions(
+ tableName.getBytes(), getTableRegionsTimeout);
+ upsertPartitionMetadata(pconn, tableName, streamName, tableRegions);
+ updateStreamStatus(pconn, tableName, streamName);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SUCCESS, "");
+ } catch (Throwable t) {
+ LOGGER.error("Exception while bootstrapping CDC Stream Partition Metadata for "
+ + taskRecord.getTableName() + " and timestamp " + timestamp.toString(), t);
+ return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.FAIL, t.toString());
Review Comment:
@haridsv @virajjasani Not sure under what kind of failures, we should be retrying the task here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@phoenix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org