[
https://issues.apache.org/jira/browse/FLINK-40650?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Stefano Franco updated FLINK-40650:
-----------------------------------
Description:
{{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.
was:
{{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.
> 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
> Priority: Major
>
> {{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)
