[jira] [Created] (FLINK-40650) Improve SqlServerDialect.discoverDataCollectionSchemas to reduce snapshot start time with many tables

["Stefano Franco (Jira)" <ji...@apache.org>]

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)