You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "navis (via GitHub)" <gi...@apache.org> on 2026/03/27 07:58:02 UTC
[PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
navis opened a new pull request, #27840:
URL: https://github.com/apache/flink/pull/27840
The most of codes/descriptions are made by claude
## DatabaseMetaData
- getTables(catalog, schemaPattern, tableNamePattern, types) — Lists tables and views across catalogs/schemas using SHOW TABLES and SHOW VIEWS SQL statements. Supports filtering by table name pattern (SQL % and _ wildcards) and table type (TABLE, VIEW).
- getColumns(catalog, schemaPattern, tableNamePattern, columnNamePattern) — Retrieves column metadata (name, JDBC type, nullability, ordinal position) by executing SELECT * FROM <table> LIMIT 0 and inspecting the ResolvedSchema. Maps Flink logical types to JDBC java.sql.Types constants.
- getSchemas(catalog, schemaPattern) — Previously threw UnsupportedOperationException. Now properly implemented with catalog and schema pattern filtering via SHOW DATABASES IN <catalog>. The no-arg getSchemas() is refactored to delegate to this method.
- getPrimaryKeys(catalog, schema, table) — Returns primary key information from the table's ResolvedSchema when available.
- getTableTypes() — Returns the supported table types: TABLE and VIEW.
## FlinkResultSet
- ROW type — Converted to Map<String, Object> preserving field names and values.
- ARRAY type — Converted to List<Object> with proper element type conversion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27840:
URL: https://github.com/apache/flink/pull/27840#discussion_r3032090116
##########
flink-table/flink-sql-jdbc-driver/src/test/java/org/apache/flink/table/jdbc/FlinkDatabaseMetaDataTest.java:
##########
@@ -193,6 +195,156 @@ public void testCatalogSchemas() throws Exception {
}
}
+ @Test
+ public void testGetSchemasWithFilter() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL("CREATE DATABASE database_a", executor);
+ executeDDL("CREATE DATABASE database_b", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Filter by catalog
+ List<String> schemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", null));
+ assertThat(schemas).allMatch(s -> s.contains("default_catalog"));
+
+ // Filter by schema pattern
+ List<String> filteredSchemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", "database_%"));
+ assertThat(filteredSchemas)
+ .allMatch(s -> s.startsWith("database_a,") || s.startsWith("database_b,"));
+ }
+ }
+
+ @Test
+ public void testGetTables() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE test_table1 (id INT, name STRING) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL(
+ "CREATE TABLE test_table2 (id INT, val DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL("CREATE VIEW test_view1 AS SELECT id, name FROM test_table1", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Get all tables and views
+ ResultSet rs = metaData.getTables("default_catalog", "default_database", null, null);
+ List<String> allResults = resultSetToListAndClose(rs);
+ Set<String> tableNames = new HashSet<>();
+ for (String row : allResults) {
+ tableNames.add(row.split(",")[2]);
+ }
+ assertThat(tableNames).contains("test_table1", "test_table2", "test_view1");
+
+ // Filter by type TABLE only
+ rs =
+ metaData.getTables(
+ "default_catalog", "default_database", null, new String[] {"TABLE"});
+ List<String> tableOnly = resultSetToListAndClose(rs);
+ for (String row : tableOnly) {
+ assertThat(row).contains("TABLE");
+ }
+ Set<String> tableOnlyNames = new HashSet<>();
+ for (String row : tableOnly) {
+ tableOnlyNames.add(row.split(",")[2]);
+ }
+ assertThat(tableOnlyNames).contains("test_table1", "test_table2");
+ assertThat(tableOnlyNames).doesNotContain("test_view1");
+
+ // Filter by name pattern
+ rs = metaData.getTables("default_catalog", "default_database", "test_table%", null);
+ List<String> patternResults = resultSetToListAndClose(rs);
+ for (String row : patternResults) {
+ assertThat(row.split(",")[2]).startsWith("test_table");
+ }
+ }
+ }
+
+ @Test
+ public void testGetColumns() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE col_test (id INT NOT NULL, name STRING, score DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ ResultSet rs =
+ metaData.getColumns("default_catalog", "default_database", "col_test", null);
+
+ // Verify column count and names using COLUMN_NAME (column 4)
+ List<String> columnNames = new ArrayList<>();
+ while (rs.next()) {
+ columnNames.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(columnNames).containsExactly("id", "name", "score");
+
+ // Filter by column name pattern
+ rs = metaData.getColumns("default_catalog", "default_database", "col_test", "na%");
+ List<String> filtered = new ArrayList<>();
+ while (rs.next()) {
+ filtered.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(filtered).containsExactly("name");
+ }
+ }
+
+ @Test
+ public void testGetTableTypes() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ List<String> types = resultSetToListAndClose(metaData.getTableTypes());
+ assertThat(types).containsExactly("TABLE", "VIEW");
Review Comment:
are we sure about order or should use `inAnyOrder`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27840:
URL: https://github.com/apache/flink/pull/27840#discussion_r3032086746
##########
flink-table/flink-sql-jdbc-driver/src/test/java/org/apache/flink/table/jdbc/FlinkDatabaseMetaDataTest.java:
##########
@@ -193,6 +195,156 @@ public void testCatalogSchemas() throws Exception {
}
}
+ @Test
+ public void testGetSchemasWithFilter() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL("CREATE DATABASE database_a", executor);
+ executeDDL("CREATE DATABASE database_b", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Filter by catalog
+ List<String> schemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", null));
+ assertThat(schemas).allMatch(s -> s.contains("default_catalog"));
+
+ // Filter by schema pattern
+ List<String> filteredSchemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", "database_%"));
+ assertThat(filteredSchemas)
+ .allMatch(s -> s.startsWith("database_a,") || s.startsWith("database_b,"));
+ }
+ }
+
+ @Test
+ public void testGetTables() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE test_table1 (id INT, name STRING) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL(
+ "CREATE TABLE test_table2 (id INT, val DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL("CREATE VIEW test_view1 AS SELECT id, name FROM test_table1", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Get all tables and views
+ ResultSet rs = metaData.getTables("default_catalog", "default_database", null, null);
+ List<String> allResults = resultSetToListAndClose(rs);
+ Set<String> tableNames = new HashSet<>();
+ for (String row : allResults) {
+ tableNames.add(row.split(",")[2]);
+ }
+ assertThat(tableNames).contains("test_table1", "test_table2", "test_view1");
+
+ // Filter by type TABLE only
+ rs =
+ metaData.getTables(
+ "default_catalog", "default_database", null, new String[] {"TABLE"});
+ List<String> tableOnly = resultSetToListAndClose(rs);
+ for (String row : tableOnly) {
+ assertThat(row).contains("TABLE");
+ }
+ Set<String> tableOnlyNames = new HashSet<>();
+ for (String row : tableOnly) {
+ tableOnlyNames.add(row.split(",")[2]);
+ }
+ assertThat(tableOnlyNames).contains("test_table1", "test_table2");
+ assertThat(tableOnlyNames).doesNotContain("test_view1");
+
+ // Filter by name pattern
+ rs = metaData.getTables("default_catalog", "default_database", "test_table%", null);
+ List<String> patternResults = resultSetToListAndClose(rs);
+ for (String row : patternResults) {
+ assertThat(row.split(",")[2]).startsWith("test_table");
+ }
+ }
+ }
+
+ @Test
+ public void testGetColumns() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE col_test (id INT NOT NULL, name STRING, score DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ ResultSet rs =
+ metaData.getColumns("default_catalog", "default_database", "col_test", null);
+
+ // Verify column count and names using COLUMN_NAME (column 4)
+ List<String> columnNames = new ArrayList<>();
+ while (rs.next()) {
+ columnNames.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(columnNames).containsExactly("id", "name", "score");
+
+ // Filter by column name pattern
+ rs = metaData.getColumns("default_catalog", "default_database", "col_test", "na%");
+ List<String> filtered = new ArrayList<>();
+ while (rs.next()) {
+ filtered.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(filtered).containsExactly("name");
+ }
+ }
+
+ @Test
+ public void testGetTableTypes() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ List<String> types = resultSetToListAndClose(metaData.getTableTypes());
+ assertThat(types).containsExactly("TABLE", "VIEW");
+ }
+ }
+
+ @Test
+ public void testGetPrimaryKeys() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE pk_test (id INT NOT NULL, name STRING, PRIMARY KEY (id) NOT ENFORCED) WITH ('connector' = 'datagen')",
+ executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // getPrimaryKeys uses SELECT * LIMIT 0 which may not preserve PK info
+ // in the result schema, so we verify at least it does not throw
+ ResultSet rs =
+ metaData.getPrimaryKeys("default_catalog", "default_database", "pk_test");
+ List<String> pkColumns = new ArrayList<>();
+ while (rs.next()) {
+ pkColumns.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
Review Comment:
if exception happens before that, resource will not be closed, correct?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27840:
URL: https://github.com/apache/flink/pull/27840#issuecomment-4873780829
This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.
If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/
If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27840:
URL: https://github.com/apache/flink/pull/27840#issuecomment-5163484305
This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27840:
URL: https://github.com/apache/flink/pull/27840#discussion_r3032082815
##########
flink-table/flink-sql-jdbc-driver/src/test/java/org/apache/flink/table/jdbc/FlinkDatabaseMetaDataTest.java:
##########
@@ -193,6 +195,156 @@ public void testCatalogSchemas() throws Exception {
}
}
+ @Test
+ public void testGetSchemasWithFilter() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL("CREATE DATABASE database_a", executor);
+ executeDDL("CREATE DATABASE database_b", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Filter by catalog
+ List<String> schemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", null));
+ assertThat(schemas).allMatch(s -> s.contains("default_catalog"));
+
+ // Filter by schema pattern
+ List<String> filteredSchemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", "database_%"));
+ assertThat(filteredSchemas)
+ .allMatch(s -> s.startsWith("database_a,") || s.startsWith("database_b,"));
+ }
+ }
+
+ @Test
+ public void testGetTables() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE test_table1 (id INT, name STRING) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL(
+ "CREATE TABLE test_table2 (id INT, val DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL("CREATE VIEW test_view1 AS SELECT id, name FROM test_table1", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Get all tables and views
+ ResultSet rs = metaData.getTables("default_catalog", "default_database", null, null);
+ List<String> allResults = resultSetToListAndClose(rs);
+ Set<String> tableNames = new HashSet<>();
+ for (String row : allResults) {
+ tableNames.add(row.split(",")[2]);
+ }
+ assertThat(tableNames).contains("test_table1", "test_table2", "test_view1");
+
+ // Filter by type TABLE only
+ rs =
+ metaData.getTables(
+ "default_catalog", "default_database", null, new String[] {"TABLE"});
+ List<String> tableOnly = resultSetToListAndClose(rs);
+ for (String row : tableOnly) {
+ assertThat(row).contains("TABLE");
+ }
+ Set<String> tableOnlyNames = new HashSet<>();
+ for (String row : tableOnly) {
+ tableOnlyNames.add(row.split(",")[2]);
+ }
+ assertThat(tableOnlyNames).contains("test_table1", "test_table2");
+ assertThat(tableOnlyNames).doesNotContain("test_view1");
+
+ // Filter by name pattern
+ rs = metaData.getTables("default_catalog", "default_database", "test_table%", null);
+ List<String> patternResults = resultSetToListAndClose(rs);
+ for (String row : patternResults) {
+ assertThat(row.split(",")[2]).startsWith("test_table");
+ }
+ }
+ }
+
+ @Test
+ public void testGetColumns() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE col_test (id INT NOT NULL, name STRING, score DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ ResultSet rs =
+ metaData.getColumns("default_catalog", "default_database", "col_test", null);
+
+ // Verify column count and names using COLUMN_NAME (column 4)
+ List<String> columnNames = new ArrayList<>();
+ while (rs.next()) {
+ columnNames.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(columnNames).containsExactly("id", "name", "score");
+
+ // Filter by column name pattern
+ rs = metaData.getColumns("default_catalog", "default_database", "col_test", "na%");
+ List<String> filtered = new ArrayList<>();
+ while (rs.next()) {
+ filtered.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(filtered).containsExactly("name");
+ }
+ }
+
+ @Test
+ public void testGetTableTypes() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ List<String> types = resultSetToListAndClose(metaData.getTableTypes());
+ assertThat(types).containsExactly("TABLE", "VIEW");
+ }
+ }
+
+ @Test
+ public void testGetPrimaryKeys() throws Exception {
Review Comment:
```suggestion
void testGetPrimaryKeys() throws Exception {
```
do not use `public` for junit5, no need for that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27840:
URL: https://github.com/apache/flink/pull/27840#discussion_r3032085064
##########
flink-table/flink-sql-jdbc-driver/src/test/java/org/apache/flink/table/jdbc/FlinkDatabaseMetaDataTest.java:
##########
@@ -193,6 +195,156 @@ public void testCatalogSchemas() throws Exception {
}
}
+ @Test
+ public void testGetSchemasWithFilter() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL("CREATE DATABASE database_a", executor);
+ executeDDL("CREATE DATABASE database_b", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Filter by catalog
+ List<String> schemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", null));
+ assertThat(schemas).allMatch(s -> s.contains("default_catalog"));
+
+ // Filter by schema pattern
+ List<String> filteredSchemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", "database_%"));
+ assertThat(filteredSchemas)
+ .allMatch(s -> s.startsWith("database_a,") || s.startsWith("database_b,"));
+ }
+ }
+
+ @Test
+ public void testGetTables() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE test_table1 (id INT, name STRING) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL(
+ "CREATE TABLE test_table2 (id INT, val DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL("CREATE VIEW test_view1 AS SELECT id, name FROM test_table1", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Get all tables and views
+ ResultSet rs = metaData.getTables("default_catalog", "default_database", null, null);
+ List<String> allResults = resultSetToListAndClose(rs);
+ Set<String> tableNames = new HashSet<>();
+ for (String row : allResults) {
+ tableNames.add(row.split(",")[2]);
+ }
+ assertThat(tableNames).contains("test_table1", "test_table2", "test_view1");
+
+ // Filter by type TABLE only
+ rs =
+ metaData.getTables(
+ "default_catalog", "default_database", null, new String[] {"TABLE"});
+ List<String> tableOnly = resultSetToListAndClose(rs);
+ for (String row : tableOnly) {
+ assertThat(row).contains("TABLE");
+ }
+ Set<String> tableOnlyNames = new HashSet<>();
+ for (String row : tableOnly) {
+ tableOnlyNames.add(row.split(",")[2]);
+ }
+ assertThat(tableOnlyNames).contains("test_table1", "test_table2");
+ assertThat(tableOnlyNames).doesNotContain("test_view1");
+
+ // Filter by name pattern
+ rs = metaData.getTables("default_catalog", "default_database", "test_table%", null);
+ List<String> patternResults = resultSetToListAndClose(rs);
+ for (String row : patternResults) {
+ assertThat(row.split(",")[2]).startsWith("test_table");
+ }
+ }
+ }
+
+ @Test
+ public void testGetColumns() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE col_test (id INT NOT NULL, name STRING, score DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ ResultSet rs =
+ metaData.getColumns("default_catalog", "default_database", "col_test", null);
+
+ // Verify column count and names using COLUMN_NAME (column 4)
+ List<String> columnNames = new ArrayList<>();
+ while (rs.next()) {
+ columnNames.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(columnNames).containsExactly("id", "name", "score");
+
+ // Filter by column name pattern
+ rs = metaData.getColumns("default_catalog", "default_database", "col_test", "na%");
+ List<String> filtered = new ArrayList<>();
+ while (rs.next()) {
+ filtered.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(filtered).containsExactly("name");
+ }
+ }
+
+ @Test
+ public void testGetTableTypes() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ List<String> types = resultSetToListAndClose(metaData.getTableTypes());
+ assertThat(types).containsExactly("TABLE", "VIEW");
+ }
+ }
+
+ @Test
+ public void testGetPrimaryKeys() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE pk_test (id INT NOT NULL, name STRING, PRIMARY KEY (id) NOT ENFORCED) WITH ('connector' = 'datagen')",
+ executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // getPrimaryKeys uses SELECT * LIMIT 0 which may not preserve PK info
+ // in the result schema, so we verify at least it does not throw
+ ResultSet rs =
Review Comment:
why`ResultSet` is not closed or not in try with resources?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27840:
URL: https://github.com/apache/flink/pull/27840#issuecomment-4140871355
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d3bdd8bef585b7f7bb54d092e66a03910145aa78",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3bdd8bef585b7f7bb54d092e66a03910145aa78",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3bdd8bef585b7f7bb54d092e66a03910145aa78 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27840:
URL: https://github.com/apache/flink/pull/27840#discussion_r3032084199
##########
flink-table/flink-sql-jdbc-driver/src/test/java/org/apache/flink/table/jdbc/FlinkDatabaseMetaDataTest.java:
##########
@@ -193,6 +195,156 @@ public void testCatalogSchemas() throws Exception {
}
}
+ @Test
+ public void testGetSchemasWithFilter() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL("CREATE DATABASE database_a", executor);
+ executeDDL("CREATE DATABASE database_b", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Filter by catalog
+ List<String> schemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", null));
+ assertThat(schemas).allMatch(s -> s.contains("default_catalog"));
+
+ // Filter by schema pattern
+ List<String> filteredSchemas =
+ resultSetToListAndClose(metaData.getSchemas("default_catalog", "database_%"));
+ assertThat(filteredSchemas)
+ .allMatch(s -> s.startsWith("database_a,") || s.startsWith("database_b,"));
+ }
+ }
+
+ @Test
+ public void testGetTables() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE test_table1 (id INT, name STRING) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL(
+ "CREATE TABLE test_table2 (id INT, val DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+ executeDDL("CREATE VIEW test_view1 AS SELECT id, name FROM test_table1", executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ // Get all tables and views
+ ResultSet rs = metaData.getTables("default_catalog", "default_database", null, null);
+ List<String> allResults = resultSetToListAndClose(rs);
+ Set<String> tableNames = new HashSet<>();
+ for (String row : allResults) {
+ tableNames.add(row.split(",")[2]);
+ }
+ assertThat(tableNames).contains("test_table1", "test_table2", "test_view1");
+
+ // Filter by type TABLE only
+ rs =
+ metaData.getTables(
+ "default_catalog", "default_database", null, new String[] {"TABLE"});
+ List<String> tableOnly = resultSetToListAndClose(rs);
+ for (String row : tableOnly) {
+ assertThat(row).contains("TABLE");
+ }
+ Set<String> tableOnlyNames = new HashSet<>();
+ for (String row : tableOnly) {
+ tableOnlyNames.add(row.split(",")[2]);
+ }
+ assertThat(tableOnlyNames).contains("test_table1", "test_table2");
+ assertThat(tableOnlyNames).doesNotContain("test_view1");
+
+ // Filter by name pattern
+ rs = metaData.getTables("default_catalog", "default_database", "test_table%", null);
+ List<String> patternResults = resultSetToListAndClose(rs);
+ for (String row : patternResults) {
+ assertThat(row.split(",")[2]).startsWith("test_table");
+ }
+ }
+ }
+
+ @Test
+ public void testGetColumns() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
+ executeDDL(
+ "CREATE TABLE col_test (id INT NOT NULL, name STRING, score DOUBLE) WITH ('connector' = 'datagen')",
+ executor);
+
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ ResultSet rs =
+ metaData.getColumns("default_catalog", "default_database", "col_test", null);
+
+ // Verify column count and names using COLUMN_NAME (column 4)
+ List<String> columnNames = new ArrayList<>();
+ while (rs.next()) {
+ columnNames.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(columnNames).containsExactly("id", "name", "score");
+
+ // Filter by column name pattern
+ rs = metaData.getColumns("default_catalog", "default_database", "col_test", "na%");
+ List<String> filtered = new ArrayList<>();
+ while (rs.next()) {
+ filtered.add(rs.getString("COLUMN_NAME"));
+ }
+ rs.close();
+ assertThat(filtered).containsExactly("name");
+ }
+ }
+
+ @Test
+ public void testGetTableTypes() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ DatabaseMetaData metaData =
+ new FlinkDatabaseMetaData(
+ driverUri.getURL(), connection, new TestingStatement());
+
+ List<String> types = resultSetToListAndClose(metaData.getTableTypes());
+ assertThat(types).containsExactly("TABLE", "VIEW");
+ }
+ }
+
+ @Test
+ public void testGetPrimaryKeys() throws Exception {
+ DriverUri driverUri = getDriverUri();
+ try (FlinkConnection connection = new FlinkConnection(driverUri)) {
+ Executor executor = connection.getExecutor();
Review Comment:
why `Executor` is not closed or not in `try` with resources?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #27840: [FLINK-39341][jdbc-driver] Implement DatabaseMetaData methods for DBeaver
URL: https://github.com/apache/flink/pull/27840
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org