[#12929] fix(flink-connector): Use catalog database instead of schema for PostgreSQL JDBC table scans - #12930
Open
diqiu50 wants to merge 2 commits into
Open
Conversation
…schema for PostgreSQL JDBC table scans For a jdbc-postgresql catalog, JdbcPropertiesConverter#toFlinkTableProperties appended the Flink "database" (the PostgreSQL schema) to the base JDBC URL instead of the catalog's real database, so scans failed with "FATAL: database \"public\" does not exist" even though catalog browsing worked fine. PostgresqlPropertiesConverter now builds the scan connection URL from the catalog's database (via the Flink default-database option, defaulted from jdbc-database when not explicitly set), and carries the schema through a schema-qualified table-name instead. MySQL behavior is unchanged. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Kqa6kk4Wkdt9VC6s5Znipo
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
The new PostgreSQL unit tests currently use a MySQL base JDBC URL from the shared test suite, which can allow the tests to pass without actually exercising PostgreSQL URL handling.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Fixes Flink JDBC table scan property generation for PostgreSQL catalogs so scans connect to the configured PostgreSQL database (catalog jdbc-database → Flink default-database) while carrying the schema via a schema-qualified table-name, preventing failures like FATAL: database "public" does not exist.
Changes:
- Default Flink
default-databasefrom Gravitinojdbc-databaseduring catalog property conversion so it’s available for per-table scan properties. - Add PostgreSQL-specific overrides to build the scan connection database from
default-databaseand emitschema.tablefortable-name. - Add unit tests for PostgreSQL/MySQL behavior and update JDBC catalog documentation to include PostgreSQL notes.
File summaries
| File | Description |
|---|---|
| flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConverter.java | Defaults Flink default-database from jdbc-database and factors per-table URL/table-name generation into overridable hooks. |
| flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/postgresql/PostgresqlPropertiesConverter.java | Overrides connection database selection and table-name formatting for PostgreSQL scans. |
| flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestPostgresqlPropertiesConverter.java | Adds PostgreSQL-focused unit tests for correct scan URL DB selection and schema-qualified table naming. |
| flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestMysqlPropertiesConverter.java | Adds a MySQL unit test asserting existing behavior remains unchanged. |
| docs/flink-connector/flink-catalog-jdbc.md | Documents PostgreSQL support and clarifies default-database/jdbc-database behavior. |
Review details
Suppressed comments (1)
flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestPostgresqlPropertiesConverter.java:90
- This PostgreSQL converter test also asserts against
flinkUrl(a MySQL URL from the shared test suite). Setting the inputjdbc-urlto a PostgreSQL URL and asserting againstflinkUrlWithDomainmakes the test actually exercise the PostgreSQL path and avoids false confidence.
Map<String, String> flinkCatalogProperties =
getConverter(catalogPropertiesWithBoth).toFlinkCatalogProperties(catalogPropertiesWithBoth);
Map<String, String> tableProperties =
getConverter(catalogPropertiesWithBoth)
.toFlinkTableProperties(
flinkCatalogProperties, ImmutableMap.of(), new ObjectPath("public", "t"));
Assertions.assertEquals(
flinkUrl + explicitDefaultDatabase,
tableProperties.get(JdbcPropertiesConstants.FLINK_JDBC_TABLE_DATABASE_URL));
- Files reviewed: 5/5 changed files
- Comments generated: 1
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+44
to
+66
| // No 'flink.bypass.default-database' is set, so the connection database must fall back to | ||
| // the catalog's jdbc-database. | ||
| Map<String, String> catalogPropertiesWithDatabase = new HashMap<>(catalogProperties); | ||
| catalogPropertiesWithDatabase.remove(FLINK_BYPASS_DEFAULT_DATABASE); | ||
| catalogPropertiesWithDatabase.put( | ||
| JdbcPropertiesConstants.GRAVITINO_JDBC_DATABASE, jdbcDatabase); | ||
|
|
||
| // Mirrors the production call in BaseCatalog#toFlinkTable: the first argument is the Flink | ||
| // catalog properties (as produced by toFlinkCatalogProperties), the second is the Gravitino | ||
| // table's own properties, which do not carry catalog-level properties like jdbc-database. | ||
| Map<String, String> flinkCatalogProperties = | ||
| getConverter(catalogPropertiesWithDatabase) | ||
| .toFlinkCatalogProperties(catalogPropertiesWithDatabase); | ||
| Map<String, String> tableProperties = | ||
| getConverter(catalogPropertiesWithDatabase) | ||
| .toFlinkTableProperties( | ||
| flinkCatalogProperties, ImmutableMap.of(), new ObjectPath(schema, tableName)); | ||
|
|
||
| // The connection URL must target the PostgreSQL database (jdbc-database), not the schema. | ||
| Assertions.assertEquals( | ||
| flinkUrl + jdbcDatabase, | ||
| tableProperties.get(JdbcPropertiesConstants.FLINK_JDBC_TABLE_DATABASE_URL)); | ||
| // The schema must be carried via the schema-qualified table name instead. |
Code Coverage Report
Files
|
…tiesConverter tests Address Copilot review comment on apache#12930: the assertions were checking against the shared MySQL flinkUrl fixture, so the tests could pass even if PostgreSQL URL handling regressed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
For PostgreSQL catalogs,
JdbcPropertiesConverter#toFlinkTablePropertiesused the Flink "database" (= Postgres schema) as the JDBC scan connection database instead of the realjdbc-database. This PR makesPostgresqlPropertiesConverteruse the catalog's actual database (viadefault-database, defaulted fromjdbc-database) for the connection, and carries the schema via a schema-qualifiedtable-nameinstead. MySQL behavior is unchanged.Why are the changes needed?
Flink SQL
SELECTon a Postgres catalog table fails withFATAL: database "public" does not exist.Fix: #12929
Does this PR introduce any user-facing change?
No. PostgreSQL Flink table scans now work; previously always failed.
How was this patch tested?
Added unit tests in
TestPostgresqlPropertiesConverter/TestMysqlPropertiesConverter; ranFlinkJdbcMysqlCatalogIT120(passes, no regression).🤖 Generated with Claude Code
https://claude.ai/code/session_01Kqa6kk4Wkdt9VC6s5Znipo