This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e70d047dd0 [flink] add lineage-use-catalog-key-as-identifier option
for table apis for jdbc metastore (#8473)
e70d047dd0 is described below
commit e70d047dd09b17509a7e761a08214e21e77ec185
Author: jsingh-yelp <[email protected]>
AuthorDate: Mon Jul 6 08:46:23 2026 -0400
[flink] add lineage-use-catalog-key-as-identifier option for table apis for
jdbc metastore (#8473)
---
docs/generated/flink_catalog_configuration.html | 6 ++++++
.../apache/paimon/flink/FlinkCatalogOptions.java | 9 ++++++++
.../apache/paimon/flink/lineage/LineageUtils.java | 20 ++++++++++-------
.../paimon/flink/lineage/LineageUtilsTest.java | 25 ++++++++++++++++++++++
4 files changed, 52 insertions(+), 8 deletions(-)
diff --git a/docs/generated/flink_catalog_configuration.html
b/docs/generated/flink_catalog_configuration.html
index 6241c224ea..b1ce3ac150 100644
--- a/docs/generated/flink_catalog_configuration.html
+++ b/docs/generated/flink_catalog_configuration.html
@@ -38,5 +38,11 @@ under the License.
<td>Boolean</td>
<td>If true, creating table in default database is not allowed.
Default is false.</td>
</tr>
+ <tr>
+ <td><h5>lineage-use-catalog-key-as-identifier</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to use the 'catalog-key' as the catalog identifier in
the lineage dataset name (i.e., catalog_key.db_name.table_name) when using a
JDBC metastore. If false, the internal Flink catalog name is used instead.</td>
+ </tr>
</tbody>
</table>
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
index 8f6214ee08..206c0984e4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
@@ -36,4 +36,13 @@ public class FlinkCatalogOptions {
.defaultValue(false)
.withDescription(
"If true, creating table in default database is
not allowed. Default is false.");
+
+ public static final ConfigOption<Boolean> LINEAGE_USE_CATALOG_KEY =
+ ConfigOptions.key("lineage-use-catalog-key-as-identifier")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to use the 'catalog-key' as the catalog
identifier in the lineage dataset name "
+ + "(i.e., catalog_key.db_name.table_name)
when using a JDBC metastore. "
+ + "If false, the internal Flink catalog
name is used instead.");
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/LineageUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/LineageUtils.java
index df69031fcf..e7a13ceb9a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/LineageUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/LineageUtils.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.FlinkCatalogOptions;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.jdbc.JdbcCatalogFactory;
@@ -138,23 +139,26 @@ public class LineageUtils {
@VisibleForTesting
static String resolveNameByMetastore(Table table, @Nullable String
defaultName) {
- if (defaultName != null) {
- return defaultName;
- }
-
CatalogContext ctx = catalogContext(table);
if (ctx != null) {
Options catalogOptions = ctx.options();
- // If jdbc metastore is used, use catalog-key as the catalog
identifier.
- if (JdbcCatalogFactory.IDENTIFIER.equals(
- catalogOptions.get(CatalogOptions.METASTORE))) {
+ // Use catalog-key as the identifier when explicitly opted-in via
config,
+ // or when no explicit name is provided (DataStream API path).
+ boolean useCatalogKey =
+ defaultName == null
+ ||
catalogOptions.get(FlinkCatalogOptions.LINEAGE_USE_CATALOG_KEY);
+ if (useCatalogKey
+ && JdbcCatalogFactory.IDENTIFIER.equals(
+ catalogOptions.get(CatalogOptions.METASTORE))) {
String catalogKeyValue =
catalogOptions.get(JdbcCatalogOptions.CATALOG_KEY);
if (!StringUtils.isNullOrWhitespaceOnly(catalogKeyValue)) {
return catalogKeyValue + "." + table.fullName();
}
}
}
- return DEFAULT_CATALOG_IDENTIFIER + "." + table.fullName();
+ return defaultName != null
+ ? defaultName
+ : DEFAULT_CATALOG_IDENTIFIER + "." + table.fullName();
}
/**
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lineage/LineageUtilsTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lineage/LineageUtilsTest.java
index 2686a9e404..b8ab16ec81 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lineage/LineageUtilsTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lineage/LineageUtilsTest.java
@@ -188,6 +188,31 @@ class LineageUtilsTest {
.isEqualTo("paimon.db.src");
}
+ @Test
+ void testResolveNameByMetastoreUsesCatalogKeyWhenFlagEnabled() throws
Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("metastore", "jdbc");
+ catalogOptions.put("catalog-key", "jdbc-warehouse");
+ catalogOptions.put("lineage-use-catalog-key-as-identifier", "true");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ // With flag enabled, catalog-key overrides even when explicit name is
provided
+ assertThat(LineageUtils.resolveNameByMetastore(table,
"my_catalog.db.src"))
+ .isEqualTo("jdbc-warehouse." + table.fullName());
+ }
+
+ @Test
+ void testResolveNameByMetastoreKeepsCatalogNameWhenFlagDisabled() throws
Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("metastore", "jdbc");
+ catalogOptions.put("catalog-key", "jdbc-warehouse");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ // With flag disabled (default), Flink catalog name is preserved
+ assertThat(LineageUtils.resolveNameByMetastore(table,
"my_catalog.db.src"))
+ .isEqualTo("my_catalog.db.src");
+ }
+
@Test
void testDataStreamSourceLineageVertexUsesCatalogKey() throws Exception {
Map<String, String> catalogOptions = new HashMap<>();
