Stefano Franco created FLINK-40650:
--------------------------------------
Summary: Improve SqlServerDialect.discoverDataCollectionSchemas to
reduce snapshot start time with many tables
Key: FLINK-40650
URL: https://issues.apache.org/jira/browse/FLINK-40650
Project: Flink
Issue Type: Improvement
Components: Flink CDC
Affects Versions: 2.2.1
Environment: flink-connector-sqlserver-cdc 3.6.0, Flink 2.2.1
Reporter: Stefano Franco
{{SqlServerDialect.discoverDataCollectionSchemas}} still fetches each table's
schema individually:
{noformat}
final List<TableId> capturedTableIds =
discoverDataCollections(sourceConfig);
try (SqlServerConnection jdbc = createSqlServerConnection(...)) {
Map<TableId, TableChange> tableSchemas = new HashMap<>();
for (TableId tableId : capturedTableIds) {
TableChange tableSchema = queryTableSchema(jdbc, tableId);
tableSchemas.put(tableId, tableSchema);
}
return tableSchemas;
}
{noformat}
Additionally, {{IncrementalSourceReader }}(flink-cdc-base) calls
{{dialect.discoverDataCollectionSchemas(sourceConfig)}} on every new split
assignment it receives (lines ~340-342), not just once at startup:
{code:java}
Map<TableId, TableChanges.TableChange> existTableSchemas =
split.getTableSchemas();
tableSchemas = dialect.discoverDataCollectionSchemas(sourceConfig);
tableSchemas.putAll(existTableSchemas);
{code}
Combined, this means every single split assignment re-runs a full
{{INFORMATION_SCHEMA.TABLES}} scan plus one JDBC round-trip per captured table,
discarding almost all of that work immediately (only the not-yet-known table's
schema is actually needed).
{*}Reproduction{*}: A SQL Server CDC source with 812 captured tables. DEBUG
logging on io.debezium.jdbc.JdbcConnection shows ~804 "Retrieving columns of
table ..." calls recurring every ~82-88 seconds throughout the snapshot phase,
correlating exactly with each table-to-table transition. Per-table
SnapshotSplitAssigner "Split table X into N chunks" timing is a near-constant
~82s regardless of table size (a 3-row table and a 3.8M-row/474-chunk table
both take ~82s), which is inconsistent with query cost and consistent with this
fixed per-cycle full-schema-rescan cost.
{*}Suggested fix{*}: Apply the same approach as FLINK-36618
(https://issues.apache.org/jira/browse/FLINK-36618) / PR #3672
([https://github.com/apache/flink-cdc/pull/3672]) (already implemented for
PostgresDialect/CustomPostgresSchema): batch the schema read into a single JDBC
call for all requested table IDs (jdbcConnection.readSchema(tables, database,
null, tableFilter, null, false) covering all tables at once), split results
back out per-TableId locally, and cache by TableId so already-known tables are
skipped on subsequent calls.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
