flink jdbc support use limit statement for SQL Server query

Sleepy0521 · Sleepy0521 · commit 6ead5b878cbf · 2025-04-02T16:16:26.000+08:00
diff --git a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/source/JdbcDynamicTableSource.java b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/source/JdbcDynamicTableSource.java
@@ -179,7 +179,12 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
         }
 
         if (limit >= 0) {
-            query = String.format("%s %s", query, dialect.getLimitClause(limit));
+            if(dialect instanceof SqlServerDialect){
+                query = query.replace("SELECT",String.format("SELECT TOP %s ",limit));
+            }
+            else{
+                query = String.format("%s %s", query, dialect.getLimitClause(limit));
+            }
         }
 
         LOG.debug("Query generated for JDBC scan: " + query);

Original file line number	Diff line number	Diff line change
`@@ -179,7 +179,12 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon`
`179`	`179`	`}`
`180`	`180`
`181`	`181`	`if (limit >= 0) {`
`182`		`- query = String.format("%s %s", query, dialect.getLimitClause(limit));`
	`182`	`+ if(dialect instanceof SqlServerDialect){`
	`183`	`+ query = query.replace("SELECT",String.format("SELECT TOP %s ",limit));`
	`184`	`+ }`
	`185`	`+ else{`
	`186`	`+ query = String.format("%s %s", query, dialect.getLimitClause(limit));`
	`187`	`+ }`
`183`	`188`	`}`
`184`	`189`
`185`	`190`	`LOG.debug("Query generated for JDBC scan: " + query);`