[#6233] feat(flink): flink jdbc catalog #6543

hdygxsj · 2025-02-26T17:19:04Z

What changes were proposed in this pull request?

Provide flink jdbc catalog

Why are the changes needed?

Does this PR introduce any user-facing change?

None

How was this patch tested?

org.apache.gravitino.flink.connector.integration.test.jdbc.FlinkJdbcMysqlCatalogIT
org.apache.gravitino.flink.connector.jdbc.TestJdbcPropertiesConverter

tengqm

Overall looks good.

...ctor/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java

FANNG1 · 2025-02-27T06:58:33Z

Thanks for the PR, could you add a related document? And besides Mysql, does Flink JDBC connector support other RDMS?

flink-connector/flink/build.gradle.kts

...-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java

...ctor/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java

...java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT.java

FANNG1 · 2025-02-27T06:59:51Z

...r/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConstants.java

+  public static final String GRAVITINO_JDBC_USER = "jdbc-user";
+  public static final String GRAVITINO_JDBC_PASSWORD = "jdbc-password";
+  public static final String GRAVITINO_JDBC_URL = "jdbc-url";
+  public static final String GRAVITINO_JDBC_DRIVER = "jdbc-driver";


Does flink JDBC connector supports setting JDBC driver?

yes, it is the document about how to create FlinkJdbcCatalog

seems missing the logic about tranforming from GRAVITINO_JDBC_DRIVER to Flink_JDBC_DRIVER?

the document seems not the latest, seems Jdbc connector supports setting driver

public static final ConfigOption<String> DRIVER = ConfigOptions.key("driver") .stringType() .noDefaultValue() .withDescription( "The class name of the JDBC driver to use to connect to this URL. " + "If not set, it will automatically be derived from the URL.");

https://github.com/apache/flink-connector-jdbc/blob/v3.2/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcConnectorOptions.java#L57-L63

Flink jdbc catalog doesn't support setting driver in catalog options but support in table options, the above configuration is for table not catalog.

hdygxsj · 2025-02-27T14:54:14Z

Thanks for the PR, could you add a related document? And besides Mysql, does Flink JDBC connector support other RDMS?

The FlinkJdbcCatalog supports only MySQL and PostgreSQL databases.

tengqm

Overall looks pretty good.
Left some picky comments for consideration.

flink-connector/flink/build.gradle.kts

...ctor/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java

...ink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java

Bumps [axios](https://github.com/axios/axios) from 1.7.2 to 1.7.4. <details> <summary>Release notes</summary> <p><em>Sourced from <a href="https://github.com/axios/axios/releases">axios's releases</a>.</em></p> <blockquote> <h2>Release v1.7.4</h2> <h2>Release notes:</h2> <h3>Bug Fixes</h3> <ul> <li><strong>sec:</strong> CVE-2024-39338 (<a href="https://redirect.github.com/axios/axios/issues/6539">#6539</a>) (<a href="https://redirect.github.com/axios/axios/issues/6543">#6543</a>) (<a href="https://github.com/axios/axios/commit/6b6b605eaf73852fb2dae033f1e786155959de3a">6b6b605</a>)</li> <li><strong>sec:</strong> disregard protocol-relative URL to remediate SSRF (<a href="https://redirect.github.com/axios/axios/issues/6539">#6539</a>) (<a href="https://github.com/axios/axios/commit/07a661a2a6b9092c4aa640dcc7f724ec5e65bdda">07a661a</a>)</li> </ul> <h3>Contributors to this release</h3> <ul> <li> <a href="https://github.com/levpachmanov" title="+47/-11 ([apache#6543](axios/axios#6543) )">Lev Pachmanov</a></li> <li> <a href="https://github.com/hainenber" title="+49/-4 ([apache#6539](axios/axios#6539) )">Đỗ Trọng Hải</a></li> </ul> <h2>Release v1.7.3</h2> <h2>Release notes:</h2> <h3>Bug Fixes</h3> <ul> <li><strong>adapter:</strong> fix progress event emitting; (<a href="https://redirect.github.com/axios/axios/issues/6518">#6518</a>) (<a href="https://github.com/axios/axios/commit/e3c76fc9bdd03aa4d98afaf211df943e2031453f">e3c76fc</a>)</li> <li><strong>fetch:</strong> fix withCredentials request config (<a href="https://redirect.github.com/axios/axios/issues/6505">#6505</a>) (<a href="https://github.com/axios/axios/commit/85d4d0ea0aae91082f04e303dec46510d1b4e787">85d4d0e</a>)</li> <li><strong>xhr:</strong> return original config on errors from XHR adapter (<a href="https://redirect.github.com/axios/axios/issues/6515">#6515</a>) (<a href="https://github.com/axios/axios/commit/8966ee7ea62ecbd6cfb39a905939bcdab5cf6388">8966ee7</a>)</li> </ul> <h3>Contributors to this release</h3> <ul> <li> <a href="https://github.com/DigitalBrainJS" title="+211/-159 ([apache#6518](axios/axios#6518) [apache#6519](axios/axios#6519) )">Dmitriy Mozgovoy</a></li> <li> <a href="https://github.com/ValeraS" title="+3/-3 ([apache#6515](axios/axios#6515) )">Valerii Sidorenko</a></li> <li> <a href="https://github.com/prianyu" title="+2/-2 ([apache#6505](axios/axios#6505) )">prianYu</a></li> </ul> </blockquote> </details> <details> <summary>Changelog</summary> <p><em>Sourced from <a href="https://github.com/axios/axios/blob/v1.x/CHANGELOG.md">axios's changelog</a>.</em></p> <blockquote> <h2><a href="https://github.com/axios/axios/compare/v1.7.3...v1.7.4">1.7.4</a> (2024-08-13)</h2> <h3>Bug Fixes</h3> <ul> <li><strong>sec:</strong> CVE-2024-39338 (<a href="https://redirect.github.com/axios/axios/issues/6539">#6539</a>) (<a href="https://redirect.github.com/axios/axios/issues/6543">#6543</a>) (<a href="https://github.com/axios/axios/commit/6b6b605eaf73852fb2dae033f1e786155959de3a">6b6b605</a>)</li> <li><strong>sec:</strong> disregard protocol-relative URL to remediate SSRF (<a href="https://redirect.github.com/axios/axios/issues/6539">#6539</a>) (<a href="https://github.com/axios/axios/commit/07a661a2a6b9092c4aa640dcc7f724ec5e65bdda">07a661a</a>)</li> </ul> <h3>Contributors to this release</h3> <ul> <li> <a href="https://github.com/levpachmanov" title="+47/-11 ([apache#6543](axios/axios#6543) )">Lev Pachmanov</a></li> <li> <a href="https://github.com/hainenber" title="+49/-4 ([apache#6539](axios/axios#6539) )">Đỗ Trọng Hải</a></li> </ul> <h2><a href="https://github.com/axios/axios/compare/v1.7.2...v1.7.3">1.7.3</a> (2024-08-01)</h2> <h3>Bug Fixes</h3> <ul> <li><strong>adapter:</strong> fix progress event emitting; (<a href="https://redirect.github.com/axios/axios/issues/6518">#6518</a>) (<a href="https://github.com/axios/axios/commit/e3c76fc9bdd03aa4d98afaf211df943e2031453f">e3c76fc</a>)</li> <li><strong>fetch:</strong> fix withCredentials request config (<a href="https://redirect.github.com/axios/axios/issues/6505">#6505</a>) (<a href="https://github.com/axios/axios/commit/85d4d0ea0aae91082f04e303dec46510d1b4e787">85d4d0e</a>)</li> <li><strong>xhr:</strong> return original config on errors from XHR adapter (<a href="https://redirect.github.com/axios/axios/issues/6515">#6515</a>) (<a href="https://github.com/axios/axios/commit/8966ee7ea62ecbd6cfb39a905939bcdab5cf6388">8966ee7</a>)</li> </ul> <h3>Contributors to this release</h3> <ul> <li> <a href="https://github.com/DigitalBrainJS" title="+211/-159 ([apache#6518](axios/axios#6518) [apache#6519](axios/axios#6519) )">Dmitriy Mozgovoy</a></li> <li> <a href="https://github.com/ValeraS" title="+3/-3 ([apache#6515](axios/axios#6515) )">Valerii Sidorenko</a></li> <li> <a href="https://github.com/prianyu" title="+2/-2 ([apache#6505](axios/axios#6505) )">prianYu</a></li> </ul> </blockquote> </details> <details> <summary>Commits</summary> <ul> <li><a href="https://github.com/axios/axios/commit/abd24a7367726616e60dfc04cb394b4be37cf597"><code>abd24a7</code></a> chore(release): v1.7.4 (<a href="https://redirect.github.com/axios/axios/issues/6544">#6544</a>)</li> <li><a href="https://github.com/axios/axios/commit/6b6b605eaf73852fb2dae033f1e786155959de3a"><code>6b6b605</code></a> fix(sec): CVE-2024-39338 (<a href="https://redirect.github.com/axios/axios/issues/6539">#6539</a>) (<a href="https://redirect.github.com/axios/axios/issues/6543">#6543</a>)</li> <li><a href="https://github.com/axios/axios/commit/07a661a2a6b9092c4aa640dcc7f724ec5e65bdda"><code>07a661a</code></a> fix(sec): disregard protocol-relative URL to remediate SSRF (<a href="https://redirect.github.com/axios/axios/issues/6539">#6539</a>)</li> <li><a href="https://github.com/axios/axios/commit/c6cce43cd94489f655f4488c5a50ecaf781c94f2"><code>c6cce43</code></a> chore(release): v1.7.3 (<a href="https://redirect.github.com/axios/axios/issues/6521">#6521</a>)</li> <li><a href="https://github.com/axios/axios/commit/e3c76fc9bdd03aa4d98afaf211df943e2031453f"><code>e3c76fc</code></a> fix(adapter): fix progress event emitting; (<a href="https://redirect.github.com/axios/axios/issues/6518">#6518</a>)</li> <li><a href="https://github.com/axios/axios/commit/85d4d0ea0aae91082f04e303dec46510d1b4e787"><code>85d4d0e</code></a> fix(fetch): fix withCredentials request config (<a href="https://redirect.github.com/axios/axios/issues/6505">#6505</a>)</li> <li><a href="https://github.com/axios/axios/commit/92cd8ed94362f929d3d0ed85ca84296c0ac8fd6d"><code>92cd8ed</code></a> chore(github): update ISSUE_TEMPLATE.md (<a href="https://redirect.github.com/axios/axios/issues/6519">#6519</a>)</li> <li><a href="https://github.com/axios/axios/commit/8966ee7ea62ecbd6cfb39a905939bcdab5cf6388"><code>8966ee7</code></a> fix(xhr): return original config on errors from XHR adapter (<a href="https://redirect.github.com/axios/axios/issues/6515">#6515</a>)</li> <li>See full diff in <a href="https://github.com/axios/axios/compare/v1.7.2...v1.7.4">compare view</a></li> </ul> </details> <br /> [![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=axios&package-manager=npm_and_yarn&previous-version=1.7.2&new-version=1.7.4)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores) Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`. [//]: # (dependabot-automerge-start) [//]: # (dependabot-automerge-end) --- <details> <summary>Dependabot commands and options</summary> <br /> You can trigger Dependabot actions by commenting on this PR: - `@dependabot rebase` will rebase this PR - `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it - `@dependabot merge` will merge this PR after your CI passes on it - `@dependabot squash and merge` will squash and merge this PR after your CI passes on it - `@dependabot cancel merge` will cancel a previously requested merge and block automerging - `@dependabot reopen` will reopen this PR if it is closed - `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually - `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency - `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself) - `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself) - `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself) You can disable automated security fix PRs for this repo from the [Security Alerts page](https://github.com/apache/gravitino/network/alerts). </details> Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

flink-connector/flink/build.gradle.kts

FANNG1 · 2025-03-05T01:36:17Z

...r/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConstants.java

+  public static final String GRAVITINO_JDBC_USER = "jdbc-user";
+  public static final String GRAVITINO_JDBC_PASSWORD = "jdbc-password";
+  public static final String GRAVITINO_JDBC_URL = "jdbc-url";
+  public static final String GRAVITINO_JDBC_DRIVER = "jdbc-driver";


seems missing the logic about tranforming from GRAVITINO_JDBC_DRIVER to Flink_JDBC_DRIVER?

...ink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java

...ctor/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java

...-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java

FANNG1 · 2025-03-11T14:11:59Z

...ink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java

+  }
+
+  @Override
+  public Set<ConfigOption<?>> requiredOptions() {


Should we place the following properties in requiredOptions?

public static final String FLINK_JDBC_URL = "base-url"; public static final String FLINK_JDBC_USER = "username"; public static final String FLINK_JDBC_PASSWORD = "password"; public static final String FLINK_JDBC_DEFAULT_DATABASE = "default-database";

They are already in the flink JdbcCatalogFactory.

so we should use the requireOptions in JdbcCatalogFactory for GravitinoJdbcCatalogFactory?

This would cause these options to be validated twice, both in JdbcCatalogFactory and GravitinoJdbcCatalogFactory. Additionally, since these options are intended for use by JdbcCatalogFactory only, and not needed within GravitinoJdbcCatalogFactory, I think it's unnecessary to add them to the requireOptions method in GravitinoJdbcCatalogFactory.

could you provide the code path when is's validate in JdbcCatalogFactory?

org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory#requiredOptions()

FANNG1 · 2025-03-11T14:14:20Z

.../flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/MysqlPropertiesConverter.java

+
+  @Override
+  public String driverName() {
+    return "com.mysql.jdbc.Driver";


There may multi Mysql drivers with different names, seems we need a more general solution to handle jdbc-driver

To ensure compatibility with MySQL 5.x, the Flink JDBC catalog uses com.mysql.jdbc.Driver as the JDBC driver.

...ink/src/test/java/org/apache/gravitino/flink/connector/integration/test/utils/TestUtils.java

...r/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConstants.java

FANNG1 · 2025-03-13T03:06:27Z

...-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java

@@ -136,7 +137,8 @@ default Map<String, String> toFlinkDatabaseProperties(Map<String, String> gravit
   * @param gravitinoProperties The table properties provided by Gravitino.
   * @return The table properties for the Flink connector.
   */
-  default Map<String, String> toFlinkTableProperties(Map<String, String> gravitinoProperties) {
+  default Map<String, String> toFlinkTableProperties(


could you add tablePath to Java doc and explain why add this parameters?

FANNG1 · 2025-03-13T03:07:43Z

...tor/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalogFactory.java

@@ -45,7 +46,7 @@ public interface BaseCatalogFactory extends CatalogFactory {
   *
   * @return The requested property converter.
   */
-  PropertiesConverter propertiesConverter();
+  PropertiesConverter propertiesConverter(Map<String, String> catalogOptions);


please add Java doc and explain what is it and why add this parameters.

FANNG1 · 2025-03-13T03:27:40Z

...r/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConstants.java

+  public static final String GRAVITINO_JDBC_USER = "jdbc-user";
+  public static final String GRAVITINO_JDBC_PASSWORD = "jdbc-password";
+  public static final String GRAVITINO_JDBC_URL = "jdbc-url";
+  public static final String GRAVITINO_JDBC_DRIVER = "jdbc-driver";


Flink jdbc catalog doesn't support setting driver in catalog options but support in table options, the above configuration is for table not catalog.

FANNG1 · 2025-03-13T03:28:49Z

...r/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConstants.java

+    flinkToGravitinoMap.put(FLINK_JDBC_URL, GRAVITINO_JDBC_URL);
+    flinkToGravitinoMap.put(FLINK_JDBC_USER, GRAVITINO_JDBC_USER);
+    flinkToGravitinoMap.put(FLINK_JDBC_PASSWORD, GRAVITINO_JDBC_PASSWORD);
+    flinkToGravitinoMap.put(FLINK_JDBC_DEFAULT_DATABASE, GRAVITINO_JDBC_DEFAULT_DATABASE);


no need to transform FLINK_JDBC_DEFAULT_DATABASE between GRAVITINO_JDBC_DEFAULT_DATABASE, Flink properties converter will do it automaticly.

hdygxsj added 3 commits February 27, 2025 01:14

jdbc

55424fc

jdbc

5a43410

license

555d014

tengqm reviewed Feb 26, 2025

View reviewed changes

...ctor/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java Show resolved Hide resolved

FANNG1 reviewed Feb 27, 2025

View reviewed changes

hdygxsj closed this Feb 27, 2025

license

f24b924

hdygxsj reopened this Mar 3, 2025

support jdbc catalog

d30d191

tengqm reviewed Mar 4, 2025

View reviewed changes

hdygxsj added 2 commits March 4, 2025 20:48

Merge remote-tracking branch 'upstream/main' into jdbc-1

a393de9

check not null

d8d55d7

FANNG1 reviewed Mar 5, 2025

View reviewed changes

hdygxsj added 3 commits March 5, 2025 22:46

check not null

f7eee39

Merge remote-tracking branch 'upstream/main' into jdbc-1

7dd5452

remove open and close

86039e4

FANNG1 reviewed Mar 11, 2025

View reviewed changes

hdygxsj and others added 2 commits March 12, 2025 00:20

fix

d529f07

fixci

67d731c

FANNG1 reviewed Mar 12, 2025

View reviewed changes

...r/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConstants.java Outdated Show resolved Hide resolved

hdygxsj added 5 commits March 12, 2025 19:27

fix

9bb4078

Optimize the logic of jdbcCatalog's getTable method.

deac3d4

Optimize the logic of jdbcCatalog's getTable method.

0e48086

Optimize the logic of jdbcCatalog's getTable method.

16713c1

fixci

e4fad6c

FANNG1 reviewed Mar 13, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#6233] feat(flink): flink jdbc catalog #6543

[#6233] feat(flink): flink jdbc catalog #6543

hdygxsj commented Feb 26, 2025

tengqm left a comment

FANNG1 commented Feb 27, 2025

FANNG1 Feb 27, 2025

hdygxsj Feb 27, 2025

FANNG1 Mar 5, 2025

FANNG1 Mar 12, 2025

FANNG1 Mar 13, 2025

hdygxsj commented Feb 27, 2025

tengqm left a comment

FANNG1 Mar 5, 2025

FANNG1 Mar 11, 2025

hdygxsj Mar 11, 2025

FANNG1 Mar 12, 2025

hdygxsj Mar 12, 2025

FANNG1 Mar 12, 2025

hdygxsj Mar 12, 2025

FANNG1 Mar 11, 2025

hdygxsj Mar 11, 2025 •

edited

Loading

FANNG1 Mar 13, 2025

FANNG1 Mar 13, 2025

FANNG1 Mar 13, 2025

FANNG1 Mar 13, 2025

[#6233] feat(flink): flink jdbc catalog #6543

Are you sure you want to change the base?

[#6233] feat(flink): flink jdbc catalog #6543

Conversation

hdygxsj commented Feb 26, 2025

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

tengqm left a comment

Choose a reason for hiding this comment

FANNG1 commented Feb 27, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hdygxsj commented Feb 27, 2025

tengqm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hdygxsj Mar 11, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hdygxsj Mar 11, 2025 •

edited

Loading