Skip to content

[FLINK-37809][Connector/JDBC] sqlserver limit statement support #160

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

[FLINK-37809][Connector/JDBC] sqlserver limit statement support #160

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a322f2f
[hotfix] sqlserver limit statement support
Sleepy0521 Apr 8, 2025
3954deb
[hotfix] sqlserver limit statement support
Sleepy0521 Apr 18, 2025
7789360
[hotfix] sqlserver limit statement support
Sleepy0521 Apr 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions .../src/main/java/org/apache/flink/connector/jdbc/core/database/dialect/AbstractDialect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,4 +265,14 @@ private Range(int min, int max) {
this.max = max;
}
}

/**
* The default way of append by origin sql end.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do not understand this sentence.

* @param query origin query sql
* @param limit number of row to emit. The value of the parameter should be non-negative.
* @return the entire sql after adding limit clause.
*/
public String addLimitClause(String query, long limit) {
return String.format("%s %s", query, getLimitClause(limit));
}
}
9 changes: 9 additions & 0 deletions ...core/src/main/java/org/apache/flink/connector/jdbc/core/database/dialect/JdbcDialect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,14 @@ public interface JdbcDialect extends Serializable {
*/
String getLimitClause(long limit);

/**
* Get the way of add limit clause.
* @param query origin query sql
* @param limit number of row to emit. The value of the parameter should be non-negative.
* @return the entire sql after adding limit clause.
*/
String addLimitClause(String query, long limit);

/**
* Check if this dialect instance support a specific data type in table schema.
*
Expand Down Expand Up @@ -154,4 +162,5 @@ String getSelectFromStatement(
default String appendDefaultUrlProperties(String url) {
return url;
}

}
2 changes: 1 addition & 1 deletion ...c/main/java/org/apache/flink/connector/jdbc/core/table/source/JdbcDynamicTableSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
}

if (limit >= 0) {
query = String.format("%s %s", query, dialect.getLimitClause(limit));
query = dialect.addLimitClause(query, limit);
Copy link
Contributor

@davidradl davidradl May 7, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I find it confusing that we have a getLimitClause and an addLimitClause. They seem to be doing the same thing.

Why did we not just extend the getLimitClause to include the query and limit as parameters. Then the dialect can return the limit clause as it likes . And change this calling line to:
query = dialect.getLimitClause(query, limit));

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The initial plan would be to deprecate the getLimitClause method in the interface and keep it only in the AbstractDialect. This way, we don't have to update all dialects or duplicate code between them.

Currently, getLimitClause is responsible for returning the part of the query that handles the limit. Previously, adding this to the query was done in JdbcDynamicTableSource, but now it has been moved to addLimitClause in AbstractDialect, which always adds the limit at the end of the query. However, in this case, the limit needs to be added at the beginning/middle of the query.

If the problem is with the method name, I don't see an issue with keeping it the same as before. However, it does seem a bit odd to have a get method that modifies a parameter passed to it.

Would you like to revisit the method naming or reconsider the overall approach?

}

LOG.debug("Query generated for JDBC scan: " + query);
Expand Down
7 changes: 6 additions & 1 deletion ...ain/java/org/apache/flink/connector/jdbc/sqlserver/database/dialect/SqlServerDialect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,12 @@ public JdbcDialectConverter getRowConverter(RowType rowType) {

@Override
public String getLimitClause(long limit) {
throw new IllegalArgumentException("SqlServerDialect does not support limit clause");
return String.format("SELECT TOP %s", limit);
}

@Override
public String addLimitClause(String query, long limit) {
return query.replace("SELECT", getLimitClause(limit));
}

@Override
Expand Down
7 changes: 7 additions & 0 deletions ...pache/flink/connector/jdbc/sqlserver/database/dialect/SqlServerPreparedStatementTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,4 +91,11 @@ void testSelectStatement() {
"SELECT id, name, email, ts, field1, field_2, __field_3__ FROM tbl "
+ "WHERE id = :id AND __field_3__ = :__field_3__");
}

@Test
void testLimitStatement() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should have a limit test for all dialects using the JdbcITCase.

String selectStmt = dialect.addLimitClause("SELECT * FROM TBL", 10);
assertThat(selectStmt)
.isEqualTo("SELECT TOP 10 * FROM TBL");
}
}