flink jdbc support use limit statement for SQL Server query

Sleepy0521 · Sleepy0521 · commit c861211581ef · 2025-04-02T16:48:56.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,10 +179,9 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
         }
 
         if (limit >= 0) {
-            if(dialect instanceof SqlServerDialect){
-                query = query.replace("SELECT",String.format("SELECT TOP %s ",limit));
-            }
-            else{
+            if ("SqlServer".equals(dialectName)) {
+                query = query.replace("SELECT", String.format("SELECT TOP %s ", limit));
+            } else {
                 query = String.format("%s %s", query, dialect.getLimitClause(limit));
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -179,10 +179,9 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon`
`179`	`179`	`}`
`180`	`180`
`181`	`181`	`if (limit >= 0) {`
`182`		`- if(dialect instanceof SqlServerDialect){`
`183`		`- query = query.replace("SELECT",String.format("SELECT TOP %s ",limit));`
`184`		`- }`
`185`		`- else{`
	`182`	`+ if ("SqlServer".equals(dialectName)) {`
	`183`	`+ query = query.replace("SELECT", String.format("SELECT TOP %s ", limit));`
	`184`	`+ } else {`
`186`	`185`	`query = String.format("%s %s", query, dialect.getLimitClause(limit));`
`187`	`186`	`}`
`188`	`187`	`}`