leonardBang commented on code in PR #3952: URL: https://github.com/apache/flink-cdc/pull/3952#discussion_r4228558523
########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/TiDBSourceFetchTaskContext.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.flink.cdc.connectors.tidb.source.fetch; + +import org.apache.flink.cdc.connectors.base.WatermarkDispatcher; +import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig; +import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect; +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset; +import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase; +import org.apache.flink.cdc.connectors.base.source.reader.external.JdbcSourceFetchTaskContext; +import org.apache.flink.cdc.connectors.base.utils.SplitKeyUtils; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.connection.TiDBConnection; +import org.apache.flink.cdc.connectors.tidb.source.handler.TiDBErrorHandler; +import org.apache.flink.cdc.connectors.tidb.source.handler.TiDBSchemaChangeEventHandler; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetFactory; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetUtils; +import org.apache.flink.cdc.connectors.tidb.source.schema.TiDBDatabaseSchema; +import org.apache.flink.cdc.connectors.tidb.utils.TiDBUtils; +import org.apache.flink.table.types.logical.RowType; + +import io.debezium.connector.base.ChangeEventQueue; +import io.debezium.connector.tidb.TiDBEventMetadataProvider; +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.connector.tidb.TiDBTaskContext; +import io.debezium.pipeline.DataChangeEvent; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.metrics.DefaultChangeEventSourceMetricsFactory; +import io.debezium.pipeline.metrics.SnapshotChangeEventSourceMetrics; +import io.debezium.pipeline.metrics.spi.ChangeEventSourceMetricsFactory; +import io.debezium.pipeline.source.spi.EventMetadataProvider; +import io.debezium.relational.Table; +import io.debezium.relational.TableId; +import io.debezium.relational.Tables; +import io.debezium.schema.TopicSelector; +import org.apache.kafka.connect.source.SourceRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** TiDB source fetch task context. */ +public class TiDBSourceFetchTaskContext extends JdbcSourceFetchTaskContext { + + private static final Logger LOG = LoggerFactory.getLogger(TiDBSourceFetchTaskContext.class); + + private TiDBTaskContext tidbTaskContext; + + private final TiDBConnection connection; + private TiDBDatabaseSchema tiDBDatabaseSchema; + private EventOffsetContext offsetContext; + private SnapshotChangeEventSourceMetrics<TiDBPartition> snapshotChangeEventSourceMetrics; + private TopicSelector<TableId> topicSelector; + private JdbcSourceEventDispatcher<TiDBPartition> dispatcher; + private TiDBPartition tiDBPartition; + private ChangeEventQueue<DataChangeEvent> queue; + private ErrorHandler errorHandler; + private EventMetadataProvider metadataProvider; + + public TiDBSourceFetchTaskContext( + JdbcSourceConfig sourceConfig, + JdbcDataSourceDialect dataSourceDialect, + TiDBConnection connection) { + super(sourceConfig, dataSourceDialect); + this.connection = connection; + this.metadataProvider = new TiDBEventMetadataProvider(); + } + + @Override + public void configure(SourceSplitBase sourceSplitBase) { + final TiDBConnectorConfig connectorConfig = getDbzConnectorConfig(); + final boolean tableIdCaseInsensitive = + dataSourceDialect.isDataCollectionIdCaseSensitive(sourceConfig); Review Comment: Thanks. The concern here is specifically the boolean polarity rather than MySQL syntax compatibility. `isDataCollectionIdCaseSensitive()` returns `true` when table identifiers are case-sensitive, while the local variable and the `TiDBUtils.newSchema(...)` parameter represent `tableIdCaseInsensitive`. Passing the value through unchanged therefore reverses schema-key normalization in both modes. Would it make sense to invert the value at this boundary, or change the downstream API to consistently accept case-sensitive semantics? It would also be helpful to cover both `lower_case_table_names = 0` and a non-zero value with mixed-case table identifiers. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/table/TiDBTableSource.java: ########## @@ -70,102 +95,195 @@ public class TiDBTableSource implements ScanTableSource, SupportsReadingMetadata public TiDBTableSource( ResolvedSchema physicalSchema, + int port, + String hostName, String database, String tableName, + String username, + String password, + String serverTimeZone, + Properties jdbcProperties, + boolean enableParallelRead, + Duration heartbeatInterval, String pdAddresses, String hostMapping, - StartupOptions startupOptions, - Map<String, String> options) { + Duration connectTimeout, + Map<String, String> options, + int splitSize, + int splitMetaGroupSize, + int fetchSize, + int connectMaxRetries, + int connectionPoolSize, + double distributionFactorUpper, + double distributionFactorLower, + @Nullable String chunkKeyColumn, + @Nullable Map<ObjectPath, String> chunkKeyColumns, + String jdbcDriver, + StartupOptions startupOptions) { this.physicalSchema = physicalSchema; this.database = checkNotNull(database); this.tableName = checkNotNull(tableName); this.pdAddresses = checkNotNull(pdAddresses); + this.port = port; + this.username = username; + this.password = password; + this.serverTimeZone = serverTimeZone; + this.jdbcProperties = jdbcProperties; + this.hostName = hostName; + this.options = options; + + // incremental snapshot options + this.enableParallelRead = enableParallelRead; + this.splitSize = splitSize; + this.splitMetaGroupSize = splitMetaGroupSize; + this.fetchSize = fetchSize; + this.connectMaxRetries = connectMaxRetries; + this.connectionPoolSize = connectionPoolSize; + this.distributionFactorUpper = distributionFactorUpper; + this.distributionFactorLower = distributionFactorLower; + this.chunkKeyColumn = chunkKeyColumn; + this.chunkKeyColumns = chunkKeyColumns; + this.heartbeatInterval = heartbeatInterval; + this.jdbcDriver = jdbcDriver; + this.connectTimeout = connectTimeout; this.hostMapping = hostMapping; this.startupOptions = startupOptions; this.producedDataType = physicalSchema.toPhysicalRowDataType(); - this.options = options; this.metadataKeys = Collections.emptyList(); } @Override public ChangelogMode getChangelogMode() { - return ChangelogMode.newBuilder() - .addContainedKind(RowKind.INSERT) - .addContainedKind(RowKind.UPDATE_AFTER) - .addContainedKind(RowKind.DELETE) - .build(); + return ChangelogMode.all(); } @Override public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { + // TIDB source builder final TiConfiguration tiConf = - TDBSourceOptions.getTiConfiguration(pdAddresses, hostMapping, options); + TiDBSourceOptions.getTiConfiguration(pdAddresses, hostMapping, options); + RowType physicalDataType = (RowType) physicalSchema.toPhysicalRowDataType().getLogicalType(); + TypeInformation<RowData> typeInfo = scanContext.createTypeInformation(producedDataType); - TiKVMetadataConverter[] metadataConverters = getMetadataConverters(); + MetadataConverter[] metadataConverters = getMetadataConverters(); - RowDataTiKVSnapshotEventDeserializationSchema snapshotEventDeserializationSchema = - new RowDataTiKVSnapshotEventDeserializationSchema( - tiConf, - database, - tableName, - typeInfo, - metadataConverters, - physicalDataType); + DebeziumDeserializationSchema<RowData> deserializer = + RowDataDebeziumDeserializeSchema.newBuilder() + .setPhysicalRowType(physicalDataType) + .setMetadataConverters(metadataConverters) + .setResultTypeInfo(typeInfo) + .setServerTimeZone( + serverTimeZone == null + ? ZoneId.systemDefault() + : ZoneId.of(serverTimeZone)) + .setUserDefinedConverterFactory( + TiDBDeserializationConverterFactory.instance()) + .build(); - RowDataTiKVChangeEventDeserializationSchema changeEventDeserializationSchema = - new RowDataTiKVChangeEventDeserializationSchema( - tiConf, - database, - tableName, - typeInfo, - metadataConverters, - physicalDataType); - - TiDBSource.Builder<RowData> builder = - TiDBSource.<RowData>builder() - .database(database) - .tableName(tableName) + JdbcIncrementalSource<RowData> parallelSource = Review Comment: Would it make sense to either wire these accepted options into the runtime or reject them during validation? The source is always built as a `JdbcIncrementalSource`, so `scan.incremental.snapshot.enabled=false` does not change the execution path. Similarly, `heartbeatInterval` is stored but not passed to the builder, and arbitrary `tikv.*` options are accepted even though `TiDBSourceOptions.getTiConfiguration()` does not apply them. This may lead users to believe that snapshot mode, heartbeat, or TiKV timeout settings are active when they have no runtime effect. If these settings are not supported yet, failing validation with a clear message would make the contract explicit. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/TiDBScanFetchTask.java: ########## @@ -0,0 +1,362 @@ +/* + * 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.flink.cdc.connectors.tidb.source.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.connection.TiDBConnection; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetUtils; +import org.apache.flink.cdc.connectors.tidb.source.schema.TiDBDatabaseSchema; +import org.apache.flink.cdc.connectors.tidb.utils.TiDBUtils; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.pipeline.EventDispatcher; +import io.debezium.pipeline.source.AbstractSnapshotChangeEventSource; +import io.debezium.pipeline.source.spi.ChangeEventSource; +import io.debezium.pipeline.source.spi.SnapshotProgressListener; +import io.debezium.pipeline.spi.ChangeRecordEmitter; +import io.debezium.pipeline.spi.SnapshotResult; +import io.debezium.relational.RelationalSnapshotChangeEventSource; +import io.debezium.relational.SnapshotChangeRecordEmitter; +import io.debezium.relational.Table; +import io.debezium.relational.TableId; +import io.debezium.util.Clock; +import io.debezium.util.ColumnUtils; +import io.debezium.util.Strings; +import io.debezium.util.Threads; +import org.apache.kafka.connect.errors.ConnectException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.time.Duration; + +/** A wrapped task to fetch snapshot split of table. */ +public class TiDBScanFetchTask extends AbstractScanFetchTask { + private static final Logger LOG = LoggerFactory.getLogger(TiDBScanFetchTask.class); + private volatile EventSourceReader backfillEventSourceReader; + private volatile StoppableChangeEventSourceContext backfillSourceContext; + + public TiDBScanFetchTask(SnapshotSplit split) { + super(split); + } + + @Override + protected void executeBackfillTask(Context context, StreamSplit backfillStreamSplit) + throws Exception { + TiDBSourceFetchTaskContext ctx = (TiDBSourceFetchTaskContext) context; + final EventOffsetContext.Loader loader = + new EventOffsetContext.Loader(ctx.getDbzConnectorConfig()); + final EventOffsetContext cdcEventOffsetContext = + EventOffsetUtils.getEventOffsetContext( + loader, backfillStreamSplit.getStartingOffset()); + final EventSourceReader cdcEventSource = + new EventSourceReader( + ctx.getDbzConnectorConfig(), + ctx.getEventDispatcher(), + ctx.getErrorHandler(), + ctx.getTaskContext(), + backfillStreamSplit); + final StoppableChangeEventSourceContext changeEventSourceContext = + new StoppableChangeEventSourceContext(); + this.backfillEventSourceReader = cdcEventSource; + this.backfillSourceContext = changeEventSourceContext; + + LOG.info( + "Execute bounded backfill task for snapshot split {} with stream split {}", + snapshotSplit, + backfillStreamSplit); + try { + if (!taskRunning) { + return; + } + cdcEventSource.init(); + if (!taskRunning) { + return; + } + final boolean completed = + cdcEventSource.executeBackfill( + changeEventSourceContext, ctx.getPartition(), cdcEventOffsetContext); + if (completed) { + dispatchEndWaterMarkEvent( + context, backfillStreamSplit, backfillStreamSplit.getEndingOffset()); + } + } finally { + cdcEventSource.close(); + this.backfillEventSourceReader = null; + this.backfillSourceContext = null; + } + } + + @Override + public void close() { + super.close(); + StoppableChangeEventSourceContext sourceContext = backfillSourceContext; + if (sourceContext != null) { + sourceContext.stopChangeEventSource(); + } + EventSourceReader eventSourceReader = backfillEventSourceReader; + if (eventSourceReader != null) { + eventSourceReader.close(); + } + } + + /** + * Execute data snapshot task. + * + * @param context the task context + * @throws Exception exception + */ + @Override + protected void executeDataSnapshot(Context context) throws Exception { + TiDBSourceFetchTaskContext sourceFetchContext = (TiDBSourceFetchTaskContext) context; + TiDBSnapshotSplitReadTask tiDBSnapshotSplitReadTask = + new TiDBSnapshotSplitReadTask( + sourceFetchContext.getDbzConnectorConfig(), + sourceFetchContext.getOffsetContext(), + sourceFetchContext.getSnapshotChangeEventSourceMetrics(), + sourceFetchContext.getConnection(), + sourceFetchContext.getDatabaseSchema(), + sourceFetchContext.getEventDispatcher(), + snapshotSplit); + TiDBSnapshotSplitChangeEventSourceContext tiDBSnapshotSplitChangeEventSourceContext = + new TiDBSnapshotSplitChangeEventSourceContext(); + SnapshotResult<EventOffsetContext> snapshotResult = + tiDBSnapshotSplitReadTask.execute( + tiDBSnapshotSplitChangeEventSourceContext, + sourceFetchContext.getPartition(), + sourceFetchContext.getOffsetContext()); + + if (!snapshotResult.isCompletedOrSkipped()) { + taskRunning = false; + throw new IllegalStateException( + String.format("Read snapshot for tidb split %s fail", snapshotResult)); + } + } + + /** A wrapped task to fetch snapshot split of table. 负责从TiDB读取快照分片 */ + public static class TiDBSnapshotSplitReadTask + extends AbstractSnapshotChangeEventSource<TiDBPartition, EventOffsetContext> { + + private static final Logger LOG = LoggerFactory.getLogger(TiDBSnapshotSplitReadTask.class); + private static final Duration LOG_INTERVAL = Duration.ofMillis(10_000); + private final TiDBConnectorConfig connectorConfig; + private final TiDBDatabaseSchema databaseSchema; + + private final TiDBConnection jdbcConnection; + + private final JdbcSourceEventDispatcher<TiDBPartition> dispatcher; + private final Clock clock; + + private final SnapshotSplit snapshotSplit; + + private final EventOffsetContext offsetContext; + private final SnapshotProgressListener<TiDBPartition> snapshotProgressListener; + + public TiDBSnapshotSplitReadTask( + TiDBConnectorConfig connectorConfig, + EventOffsetContext previousOffset, + SnapshotProgressListener<TiDBPartition> snapshotProgressListener, + TiDBConnection jdbcConnection, + TiDBDatabaseSchema databaseSchema, + JdbcSourceEventDispatcher<TiDBPartition> dispatcher, + SnapshotSplit snapshotSplit) { + super(connectorConfig, snapshotProgressListener); + this.connectorConfig = connectorConfig; + this.databaseSchema = databaseSchema; + this.jdbcConnection = jdbcConnection; + this.dispatcher = dispatcher; + this.snapshotSplit = snapshotSplit; + this.offsetContext = previousOffset; + this.snapshotProgressListener = snapshotProgressListener; + this.clock = Clock.SYSTEM; + } + + @Override + public SnapshotResult<EventOffsetContext> execute( + ChangeEventSource.ChangeEventSourceContext context, + TiDBPartition partition, + EventOffsetContext previousOffset) + throws InterruptedException { + // todo 返回为null + AbstractSnapshotChangeEventSource.SnapshottingTask snapshottingTask = + getSnapshottingTask(partition, previousOffset); + final TiDBSnapshotContext ctx; + try { + ctx = prepare(partition); + } catch (Exception e) { + LOG.error("Failed to initialize snapshot context.", e); + throw new RuntimeException(e); + } + try { + return doExecute(context, previousOffset, ctx, snapshottingTask); + } catch (InterruptedException e) { + LOG.warn("Snapshot was interrupted before completion"); + throw e; + } catch (Exception e) { + LOG.warn("Snapshot was interrupted before completion"); + throw new RuntimeException(e); + } + } + + private static class TiDBSnapshotContext + extends RelationalSnapshotChangeEventSource.RelationalSnapshotContext< + TiDBPartition, EventOffsetContext> { + + public TiDBSnapshotContext(TiDBPartition partition) throws SQLException { + super(partition, ""); + } + } + + @Override + protected SnapshotResult<EventOffsetContext> doExecute( + ChangeEventSourceContext context, + EventOffsetContext previousOffset, + SnapshotContext snapshotContext, + SnapshottingTask snapshottingTask) // 没有调用这个参数 + throws Exception { + final TiDBSnapshotContext ctx = (TiDBSnapshotContext) snapshotContext; + ctx.offset = offsetContext; + createDataEvents(ctx, snapshotSplit.getTableId()); + + return SnapshotResult.completed(ctx.offset); + } + + private void createDataEvents(TiDBSnapshotContext snapshotContext, TableId tableId) + throws Exception { + EventDispatcher.SnapshotReceiver<TiDBPartition> snapshotReceiver = + dispatcher.getSnapshotChangeEventReceiver(); + LOG.debug("Snapshotting table {}", tableId); + createDataEventsForTable( + snapshotContext, snapshotReceiver, databaseSchema.tableFor(tableId)); + snapshotReceiver.completeSnapshot(); + } + + private void createDataEventsForTable( + TiDBSnapshotContext snapshotContext, + EventDispatcher.SnapshotReceiver<TiDBPartition> snapshotReceiver, + Table table) + throws InterruptedException { + + long exportStart = clock.currentTimeInMillis(); + LOG.info( + "Exporting data from split '{}' of table {}", + snapshotSplit.splitId(), + table.id()); + + final String selectSql = + TiDBUtils.buildSplitScanQuery( + snapshotSplit.getTableId(), + snapshotSplit.getSplitKeyType(), + snapshotSplit.getSplitStart() == null, + snapshotSplit.getSplitEnd() == null); + LOG.info( + "For split '{}' of table {} using select statement: '{}'", + snapshotSplit.splitId(), + table.id(), + selectSql); + + try (PreparedStatement selectStatement = + TiDBUtils.readTableSplitDataStatement( + jdbcConnection, + selectSql, + snapshotSplit.getSplitStart() == null, + snapshotSplit.getSplitEnd() == null, + snapshotSplit.getSplitStart(), + snapshotSplit.getSplitEnd(), + snapshotSplit.getSplitKeyType().getFieldCount(), + connectorConfig.getQueryFetchSize()); + ResultSet rs = selectStatement.executeQuery()) { + ColumnUtils.ColumnArray columnArray = ColumnUtils.toArray(rs, table); + long rows = 0; + Threads.Timer logTimer = getTableScanLogTimer(); + + while (rs.next()) { + rows++; + final Object[] row = + jdbcConnection.rowToArray(table, databaseSchema, rs, columnArray); + if (logTimer.expired()) { + long stop = clock.currentTimeInMillis(); + LOG.info( + "Exported {} records for split '{}' after {}", + rows, + snapshotSplit.splitId(), + Strings.duration(stop - exportStart)); + snapshotProgressListener.rowsScanned( + snapshotContext.partition, table.id(), rows); + logTimer = getTableScanLogTimer(); + } + dispatcher.dispatchSnapshotEvent( + snapshotContext.partition, + table.id(), + getChangeRecordEmitter(snapshotContext, table.id(), row), + snapshotReceiver); + } + LOG.info( + "Finished exporting {} records for split '{}', total duration '{}'", + rows, + snapshotSplit.splitId(), + Strings.duration(clock.currentTimeInMillis() - exportStart)); + } catch (SQLException e) { + throw new ConnectException("Snapshotting of table " + table.id() + " failed", e); + } + } + + protected ChangeRecordEmitter<TiDBPartition> getChangeRecordEmitter( + TiDBSnapshotContext snapshotContext, TableId tableId, Object[] row) { + snapshotContext.offset.event(tableId, clock.currentTime()); Review Comment: Could snapshot records keep an epoch-zero source time here? `TiDBReadableMetadata.OP_TS` documents the snapshot value as always `0`, and the previous implementation explicitly returned epoch zero. Using `clock.currentTime()` makes snapshot rows expose the local read time instead, which changes observable SQL metadata semantics and may affect consumers that distinguish snapshot records by `op_ts`. One option is to set `Instant.EPOCH` for snapshot records; a metadata test covering both snapshot and streaming phases would help preserve the contract. -- 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]
