virajjasani commented on code in PR #2401: URL: https://github.com/apache/phoenix/pull/2401#discussion_r3082826678
########## phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java: ########## @@ -0,0 +1,1349 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.hbase.index; + +import com.google.protobuf.ByteString; +import java.io.IOException; +import java.sql.Date; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.client.ConnectionUtils; +import org.apache.hadoop.hbase.client.Mutation; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.util.Pair; +import org.apache.phoenix.coprocessor.DelegateRegionCoprocessorEnvironment; +import org.apache.phoenix.coprocessor.generated.IndexMutationsProtos; +import org.apache.phoenix.coprocessorclient.MetaDataProtocol; +import org.apache.phoenix.hbase.index.metrics.MetricsIndexCDCConsumerSource; +import org.apache.phoenix.hbase.index.metrics.MetricsIndexerSourceFactory; +import org.apache.phoenix.hbase.index.table.HTableInterfaceReference; +import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; +import org.apache.phoenix.hbase.index.write.IndexWriter; +import org.apache.phoenix.index.IndexMaintainer; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; +import org.apache.phoenix.query.QueryConstants; +import org.apache.phoenix.schema.PColumn; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.RowKeySchema; +import org.apache.phoenix.schema.TableNotFoundException; +import org.apache.phoenix.schema.types.IndexConsistency; +import org.apache.phoenix.schema.types.PDataType; +import org.apache.phoenix.util.ByteUtil; +import org.apache.phoenix.util.CDCUtil; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.QueryUtil; +import org.apache.phoenix.util.ServerUtil.ConnectionType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.phoenix.thirdparty.com.google.common.collect.ArrayListMultimap; +import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap; + +import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; + +/** + * A single-threaded background consumer that processes CDC mutations for eventually consistent + * indexes. This consumer reads mutations from the CDC index table and applies them to the + * appropriate secondary indexes. + * <p> + * The consumer tracks its progress in the SYSTEM.IDX_CDC_TRACKER table, allowing for proper + * handling of region splits and merges. + * </p> + * <p> + * <b>Lifecycle:</b> + * <ol> + * <li>On startup, the consumer first replays any remaining mutations from parent regions (if this + * region was created from a split or merge). A region will have multiple parent regions in case of + * region merge.</li> + * <li>After replaying all parent mutations, the consumer marks each parent region as COMPLETE in + * SYSTEM.IDX_CDC_TRACKER.</li> + * <li>The consumer then begins processing mutations for the current region. With every batch of + * rows consumed (e.g., 1000 rows), it updates LAST_TIMESTAMP in SYSTEM.IDX_CDC_TRACKER. The tracker + * entry with status IN_PROGRESS is created on the first batch update.</li> + * <li>When the region is closed (due to split, merge, or shutdown), the consumer stops but does NOT + * mark itself as COMPLETE. The COMPLETE status is only set by child regions after they have + * replayed all parent mutations.</li> + * </ol> + * </p> + */ +public class IndexCDCConsumer implements Runnable { + + private static final Logger LOG = LoggerFactory.getLogger(IndexCDCConsumer.class); + + public static final String INDEX_CDC_CONSUMER_BATCH_SIZE = + "phoenix.index.cdc.consumer.batch.size"; + private static final int DEFAULT_CDC_BATCH_SIZE = 100; + + public static final String INDEX_CDC_CONSUMER_STARTUP_DELAY_MS = + "phoenix.index.cdc.consumer.startup.delay.ms"; + private static final long DEFAULT_STARTUP_DELAY_MS = 10000; + + /** + * The interval in milliseconds between processing batches when mutations are found. + */ + public static final String INDEX_CDC_CONSUMER_POLL_INTERVAL_MS = + "phoenix.index.cdc.consumer.poll.interval.ms"; + private static final long DEFAULT_POLL_INTERVAL_MS = 1000; + + /** + * The time buffer in milliseconds subtracted from current time when querying CDC mutations to + * help avoid reading mutations that are too recent. + */ + public static final String INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS = + "phoenix.index.cdc.consumer.timestamp.buffer.ms"; + private static final long DEFAULT_TIMESTAMP_BUFFER_MS = 10000; + + /** + * Maximum number of retries when CDC events exist but the corresponding data table mutations are + * not yet visible (or permanently failed). After exceeding this limit, the consumer advances past + * the unprocessable events to avoid blocking indefinitely. This is only used for index mutation + * generation approach (serializeCDCMutations = false). + */ + public static final String INDEX_CDC_CONSUMER_MAX_DATA_VISIBILITY_RETRIES = + "phoenix.index.cdc.consumer.max.data.visibility.retries"; + private static final int DEFAULT_MAX_DATA_VISIBILITY_RETRIES = 15; + + public static final String INDEX_CDC_CONSUMER_RETRY_PAUSE_MS = + "phoenix.index.cdc.consumer.retry.pause.ms"; + private static final long DEFAULT_RETRY_PAUSE_MS = 2000; + + public static final String INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS = + "phoenix.index.cdc.consumer.parent.progress.pause.ms"; + private static final long DEFAULT_PARENT_PROGRESS_PAUSE_MS = 15000; + + private final RegionCoprocessorEnvironment env; + private final String dataTableName; + private final String encodedRegionName; + private final IndexWriter indexWriter; + private final long pause; + private final long startupDelayMs; + private final int batchSize; + private final long pollIntervalMs; + private final long timestampBufferMs; + private final int maxDataVisibilityRetries; + private final long parentProgressPauseMs; + private final Configuration config; + private final boolean serializeCDCMutations; + private final MetricsIndexCDCConsumerSource metricSource; + private volatile boolean stopped = false; + private Thread consumerThread; + private boolean hasParentPartitions = false; + private PTable cachedDataTable; + + private boolean tenantInit = false; + private boolean isMultiTenant = false; + private String tenantIdColName; + private PDataType<?> tenantIdDataType; + private TenantScanInfo ownRegionScanInfo; + + private final Map<String, TenantScanInfo> ancestorScanInfoCache = new HashMap<>(); + + private static class TenantScanInfo { + + private static final TenantScanInfo EMPTY = new TenantScanInfo("", "", null, null, null); + + private final String filter; + private final String orderBy; + private final Object startValue; + private final Object endValue; + private final PDataType<?> dataType; + + TenantScanInfo(String filter, String orderBy, Object startValue, Object endValue, + PDataType<?> dataType) { + this.filter = filter; + this.orderBy = orderBy; + this.startValue = startValue; + this.endValue = endValue; + this.dataType = dataType; + } + + int bindParams(PreparedStatement ps, int startIndex) throws SQLException { + int idx = startIndex; + if (startValue != null) { + ps.setObject(idx++, startValue, dataType.getSqlType()); + } + if (endValue != null) { + ps.setObject(idx++, endValue, dataType.getSqlType()); + } + return idx; + } + } + + /** + * Creates a new IndexCDCConsumer for the given region with configurable serialization mode. + * @param env region coprocessor environment. + * @param dataTableName name of the data table. + * @param serverName server name. + * @param serializeCDCMutations when true, consumes pre-serialized index mutations; when false, + * generates index mutations from data row states. + * @throws IOException if the IndexWriter cannot be created. + */ + public IndexCDCConsumer(RegionCoprocessorEnvironment env, String dataTableName, String serverName, + boolean serializeCDCMutations) throws IOException { + this.env = env; + this.dataTableName = dataTableName; + this.encodedRegionName = env.getRegion().getRegionInfo().getEncodedName(); + this.config = env.getConfiguration(); + this.serializeCDCMutations = serializeCDCMutations; + this.pause = config.getLong(INDEX_CDC_CONSUMER_RETRY_PAUSE_MS, DEFAULT_RETRY_PAUSE_MS); + this.startupDelayMs = + config.getLong(INDEX_CDC_CONSUMER_STARTUP_DELAY_MS, DEFAULT_STARTUP_DELAY_MS); + int baseBatchSize = config.getInt(INDEX_CDC_CONSUMER_BATCH_SIZE, DEFAULT_CDC_BATCH_SIZE); + int jitter = ThreadLocalRandom.current().nextInt(baseBatchSize / 5 + 1); + this.batchSize = baseBatchSize + jitter; + this.pollIntervalMs = + config.getLong(INDEX_CDC_CONSUMER_POLL_INTERVAL_MS, DEFAULT_POLL_INTERVAL_MS); + this.timestampBufferMs = + config.getLong(INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS, DEFAULT_TIMESTAMP_BUFFER_MS); + this.maxDataVisibilityRetries = config.getInt(INDEX_CDC_CONSUMER_MAX_DATA_VISIBILITY_RETRIES, + DEFAULT_MAX_DATA_VISIBILITY_RETRIES); + this.parentProgressPauseMs = + config.getLong(INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS, DEFAULT_PARENT_PROGRESS_PAUSE_MS); + this.metricSource = MetricsIndexerSourceFactory.getInstance().getIndexCDCConsumerSource(); + DelegateRegionCoprocessorEnvironment indexWriterEnv = + new DelegateRegionCoprocessorEnvironment(env, ConnectionType.INDEX_WRITER_CONNECTION); + this.indexWriter = + new IndexWriter(indexWriterEnv, serverName + "-index-eventual-writer", false); + } + + /** + * Starts the consumer thread in the background. + */ + public void start() { + consumerThread = + new Thread(this, "IndexCDCConsumer-" + dataTableName + "-" + encodedRegionName); + consumerThread.setDaemon(true); + consumerThread.start(); + } + + /** + * Stops the consumer thread gracefully. + */ + public void stop() { + stopped = true; + if (consumerThread != null) { + consumerThread.interrupt(); + } + if (indexWriter != null) { + indexWriter.stop("IndexCDCConsumer stopped for " + dataTableName); + } + } + + /** + * Sleeps for the specified duration if the consumer has not been stopped. + * @param millis the duration to sleep in milliseconds. + * @throws InterruptedException if the thread is interrupted while sleeping. + */ + private void sleepIfNotStopped(long millis) throws InterruptedException { + if (!stopped) { + Thread.sleep(millis); + } + } + + private PTable getDataTable(PhoenixConnection conn) throws SQLException { + PTable dataTable = cachedDataTable; + if (dataTable == null) { + dataTable = conn.getTable(dataTableName); + cachedDataTable = dataTable; + } + return dataTable; + } + + private void refreshDataTableCache(PhoenixConnection conn) throws SQLException { + cachedDataTable = conn.getTable(dataTableName); + } + + private void initTenantInfo(PhoenixConnection conn) throws SQLException { + if (tenantInit) { + return; + } + PTable dataTable = getDataTable(conn); + isMultiTenant = dataTable.isMultiTenant(); + if (!isMultiTenant) { + ownRegionScanInfo = TenantScanInfo.EMPTY; + tenantInit = true; + return; + } + int tenantColIndex = dataTable.getBucketNum() != null ? 1 : 0; + PColumn tenantCol = dataTable.getPKColumns().get(tenantColIndex); + tenantIdColName = tenantCol.getName().getString(); + tenantIdDataType = tenantCol.getDataType(); + + byte[] regionStartKey = env.getRegion().getRegionInfo().getStartKey(); + byte[] regionEndKey = env.getRegion().getRegionInfo().getEndKey(); + ownRegionScanInfo = buildTenantScanInfo(regionStartKey, regionEndKey, dataTable); + LOG.debug( + "Initialized multi-tenant scan for table {} region {}:" + + " tenantCol {}, startTenant {}, endTenant {}", + dataTableName, encodedRegionName, tenantIdColName, ownRegionScanInfo.startValue, + ownRegionScanInfo.endValue); + tenantInit = true; + } + + private TenantScanInfo buildTenantScanInfo(byte[] startKey, byte[] endKey, PTable dataTable) { + Object startVal = extractTenantIdFromRegionKey(startKey, dataTable); + Object endVal = extractTenantIdFromRegionKey(endKey, dataTable); + StringBuilder sb = new StringBuilder(); + if (startVal != null) { + sb.append("\"").append(tenantIdColName).append("\" >= ? AND "); + } + if (endVal != null) { + sb.append("\"").append(tenantIdColName).append("\" <= ? AND "); + } + String filter = sb.toString(); + String orderBy = filter.isEmpty() ? "" : "\"" + tenantIdColName + "\" ASC,"; + return new TenantScanInfo(filter, orderBy, startVal, endVal, tenantIdDataType); + } + + private Object extractTenantIdFromRegionKey(byte[] regionKey, PTable dataTable) { + if (regionKey == null || regionKey.length == 0) { + return null; + } + final RowKeySchema schema = dataTable.getRowKeySchema(); + int pkPos = dataTable.getBucketNum() != null ? 1 : 0; + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + int maxOffset = schema.iterator(regionKey, 0, regionKey.length, ptr); + for (int i = 0; i <= pkPos; i++) { + Boolean hasValue = schema.next(ptr, i, maxOffset); + if (!Boolean.TRUE.equals(hasValue)) { + return null; + } + } + byte[] tenantBytes = ByteUtil.copyKeyBytesIfNecessary(ptr); + PColumn tenantCol = dataTable.getPKColumns().get(pkPos); + return tenantCol.getDataType().toObject(tenantBytes, 0, tenantBytes.length, + tenantCol.getDataType(), tenantCol.getSortOrder(), tenantCol.getMaxLength(), + tenantCol.getScale()); + } + + private byte[][] lookupPartitionKeys(String partitionId) throws InterruptedException { + int retryCount = 0; + final String query = "SELECT PARTITION_START_KEY, PARTITION_END_KEY FROM " + + PhoenixDatabaseMetaData.SYSTEM_CDC_STREAM_NAME + + " WHERE TABLE_NAME = ? AND PARTITION_ID = ? LIMIT 1"; + while (!stopped) { + try ( + PhoenixConnection conn = + QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class); + PreparedStatement ps = conn.prepareStatement(query)) { + ps.setString(1, dataTableName); + ps.setString(2, partitionId); + try (ResultSet rs = ps.executeQuery()) { + if (rs.next()) { + byte[] startKey = rs.getBytes(1); + byte[] endKey = rs.getBytes(2); + return new byte[][] { startKey == null ? new byte[0] : startKey, + endKey == null ? new byte[0] : endKey }; + } + } + LOG.error("No CDC_STREAM entry found for partition {} table {}. This should not happen.", + partitionId, dataTableName); + return new byte[][] { new byte[0], new byte[0] }; + } catch (SQLException e) { + long sleepTime = ConnectionUtils.getPauseTime(pause, ++retryCount); + LOG.warn( + "Error while retrieving partition keys from CDC_STREAM for partition {} table {}. " + + "Retry #{}, sleeping {} ms before retrying...", + partitionId, dataTableName, retryCount, sleepTime, e); + sleepIfNotStopped(sleepTime); + } + } + return null; + } + + private TenantScanInfo getPartitionTenantScanInfo(String partitionId) + throws InterruptedException { + if (!isMultiTenant) { + return TenantScanInfo.EMPTY; + } + if (partitionId.equals(encodedRegionName)) { + return ownRegionScanInfo; + } + TenantScanInfo cached = ancestorScanInfoCache.get(partitionId); + if (cached != null) { + return cached; + } + byte[][] keys = lookupPartitionKeys(partitionId); + if (keys == null) { + return TenantScanInfo.EMPTY; + } + TenantScanInfo info = buildTenantScanInfo(keys[0], keys[1], cachedDataTable); + ancestorScanInfoCache.put(partitionId, info); + return info; + } + + @Override + public void run() { + try { + if (startupDelayMs > 0 && getCDCStreamNumPartitions() <= 1) { + sleepIfNotStopped(startupDelayMs); + } + if (stopped) { + return; + } + if (!hasEventuallyConsistentIndexes()) { + LOG.trace("No eventually consistent indexes found for table {}. Exiting consumer.", Review Comment: Yes, exactly. In prod, regions split, merge and move all the 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
