From 00b416a70a6a0163c338720585e5ec0fea15e0ff Mon Sep 17 00:00:00 2001 From: Vadzim Ramanenka Date: Wed, 22 Jan 2025 13:05:21 +0100 Subject: [PATCH] CXP-3424: Use default CDC index for SqlServer Connector Adding `__$command_id ASC` to the order clause of the query allows for the usage of the clustered index. --- .../connector/sqlserver/SqlServerConnection.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java b/debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java index 17145f304ba..0557bca81f3 100644 --- a/debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java +++ b/debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java @@ -80,7 +80,8 @@ public class SqlServerConnection extends JdbcConnection { + LSN_TIMESTAMP_SELECT_STATEMENT; private static final String GET_ALL_CHANGES_FOR_TABLE_FROM_FUNCTION = "FROM [#db].cdc.[fn_cdc_get_all_changes_#table](?, ?, N'all update old')"; private static final String GET_ALL_CHANGES_FOR_TABLE_FROM_DIRECT = "FROM [#db].cdc.[#table]"; - private static final String GET_ALL_CHANGES_FOR_TABLE_ORDER_BY = "ORDER BY [__$start_lsn] ASC, [__$seqval] ASC, [__$operation] ASC"; + private static final String GET_ALL_CHANGES_FOR_TABLE_FROM_FUNCTION_ORDER_BY = "ORDER BY [__$start_lsn] ASC, [__$seqval] ASC, [__$operation] ASC"; + private static final String GET_ALL_CHANGES_FOR_TABLE_FROM_DIRECT_ORDER_BY = "ORDER BY [__$start_lsn] ASC, [__$command_id] ASC, [__$seqval] ASC, [__$operation] ASC"; /** * Queries the list of captured column names and their change table identifiers in the given database. @@ -212,7 +213,14 @@ private String buildGetAllChangesForTableQuery(SqlServerConnectorConfig.DataQuer result += " WHERE " + String.join(" AND ", where) + " "; } - result += GET_ALL_CHANGES_FOR_TABLE_ORDER_BY; + switch (dataQueryMode) { + case FUNCTION: + result += GET_ALL_CHANGES_FOR_TABLE_FROM_FUNCTION_ORDER_BY; + break; + case DIRECT: + result += GET_ALL_CHANGES_FOR_TABLE_FROM_DIRECT_ORDER_BY; + break; + } return result; }