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 64380bbf8a [flink] fix namespace for lineage from table path to
warehouse path (#7576)
64380bbf8a is described below
commit 64380bbf8a6303426d185f8704606f0a0bfa4cea
Author: jsingh-yelp <[email protected]>
AuthorDate: Fri Jun 19 10:12:17 2026 -0400
[flink] fix namespace for lineage from table path to warehouse path (#7576)
---
.../apache/paimon/flink/lineage/LineageUtils.java | 86 ++++++++++++++++++----
.../paimon/flink/lineage/PaimonLineageDataset.java | 2 +-
.../paimon/flink/lineage/LineageUtilsTest.java | 74 ++++++++++++++++---
3 files changed, 138 insertions(+), 24 deletions(-)
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 110365c76e..e9c51d489a 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
@@ -19,6 +19,11 @@
package org.apache.paimon.flink.lineage;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.Table;
import org.apache.flink.api.connector.source.Boundedness;
@@ -26,28 +31,54 @@ import
org.apache.flink.streaming.api.lineage.LineageDataset;
import org.apache.flink.streaming.api.lineage.LineageVertex;
import org.apache.flink.streaming.api.lineage.SourceLineageVertex;
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
/**
* Lineage utilities for building {@link SourceLineageVertex} and {@link
LineageVertex} from a
- * Paimon table name and its physical warehouse path (namespace).
+ * Paimon table name and catalog options.
*/
public class LineageUtils {
- private static final String PAIMON_DATASET_PREFIX = "paimon://";
+ /** Default namespace when the catalog warehouse path is not available. */
+ private static final String DEFAULT_NAMESPACE = "paimon";
+
+ private static final String CATALOG_PREFIX = "catalog.";
+
+ /** Catalog option keys safe to include in lineage facets (no credentials
or secrets). */
+ private static final Set<String> CATALOG_OPTION_ALLOWLIST =
+ new HashSet<>(
+ Arrays.asList(CatalogOptions.WAREHOUSE.key(),
CatalogOptions.METASTORE.key()));
private static final Set<String> PAIMON_OPTION_KEYS =
CoreOptions.getOptions().stream().map(opt ->
opt.key()).collect(Collectors.toSet());
+ /** Extracts the {@link CatalogContext} from a table, or null if not
available. */
+ @Nullable
+ private static CatalogContext catalogContext(Table table) {
+ if (table instanceof FileStoreTable) {
+ return ((FileStoreTable)
table).catalogEnvironment().catalogContext();
+ }
+ if (table instanceof FormatTable) {
+ return ((FormatTable) table).catalogContext();
+ }
+ return null;
+ }
+
/**
- * Builds the config map for a dataset facet from a {@link Table}.
Includes filtered Paimon
- * {@link CoreOptions}, partition keys, primary keys, and the table
comment (if present).
+ * Builds the config map for a dataset facet. Includes filtered Paimon
{@link CoreOptions},
+ * partition keys, primary keys, and a safe subset of catalog-level
options (warehouse,
+ * metastore) prefixed with {@code "catalog."}.
*/
- private static Map<String, String> buildConfigMap(Table table) {
+ private static Map<String, String> buildConfigMap(
+ Table table, @Nullable CatalogContext catalogContext) {
Map<String, String> config = new HashMap<>();
config.put("partition-keys", String.join(",", table.partitionKeys()));
config.put("primary-keys", String.join(",", table.primaryKeys()));
@@ -56,16 +87,39 @@ public class LineageUtils {
.filter(e -> PAIMON_OPTION_KEYS.contains(e.getKey()))
.forEach(e -> config.put(e.getKey(), e.getValue()));
+ config.put("type", "paimon");
+
+ if (catalogContext != null) {
+ catalogContext
+ .options()
+ .toMap()
+ .forEach(
+ (k, v) -> {
+ if (CATALOG_OPTION_ALLOWLIST.contains(k)) {
+ config.put(CATALOG_PREFIX + k, v);
+ }
+ });
+ }
+
return config;
}
/**
- * Returns the lineage namespace for a Paimon table. The namespace uses
the {@code paimon://}
- * scheme followed by the table's physical warehouse path, e.g. {@code
- * "paimon://s3://my-bucket/warehouse/mydb.db/mytable"}.
+ * Returns the catalog warehouse path as the lineage namespace. Falls back
to the table path
+ * when no warehouse is configured, or {@code "paimon"} as a last resort.
*/
- public static String getNamespace(Table table) {
- return PAIMON_DATASET_PREFIX + CoreOptions.path(table.options());
+ public static String getNamespace(Table table, @Nullable CatalogContext
catalogContext) {
+ if (catalogContext != null) {
+ String warehouse =
catalogContext.options().get(CatalogOptions.WAREHOUSE);
+ if (warehouse != null) {
+ return warehouse;
+ }
+ }
+ Path path = CoreOptions.path(table.options());
+ if (path != null) {
+ return path.toString();
+ }
+ return DEFAULT_NAMESPACE;
}
/**
@@ -73,12 +127,14 @@ public class LineageUtils {
*
* @param name fully qualified table name, e.g. {@code
"paimon.mydb.mytable"}
* @param isBounded whether the source is bounded (batch) or unbounded
(streaming)
- * @param table the Paimon table (namespace is derived from its {@code
path} option)
+ * @param table the Paimon table
*/
public static SourceLineageVertex sourceLineageVertex(
String name, boolean isBounded, Table table) {
+ CatalogContext ctx = catalogContext(table);
LineageDataset dataset =
- new PaimonLineageDataset(name, getNamespace(table),
buildConfigMap(table));
+ new PaimonLineageDataset(
+ name, getNamespace(table, ctx), buildConfigMap(table,
ctx));
Boundedness boundedness =
isBounded ? Boundedness.BOUNDED :
Boundedness.CONTINUOUS_UNBOUNDED;
return new PaimonSourceLineageVertex(boundedness,
Collections.singletonList(dataset));
@@ -88,11 +144,13 @@ public class LineageUtils {
* Creates a {@link LineageVertex} for a Paimon sink table.
*
* @param name fully qualified table name, e.g. {@code
"paimon.mydb.mytable"}
- * @param table the Paimon table (namespace is derived from its {@code
path} option)
+ * @param table the Paimon table
*/
public static LineageVertex sinkLineageVertex(String name, Table table) {
+ CatalogContext ctx = catalogContext(table);
LineageDataset dataset =
- new PaimonLineageDataset(name, getNamespace(table),
buildConfigMap(table));
+ new PaimonLineageDataset(
+ name, getNamespace(table, ctx), buildConfigMap(table,
ctx));
return new PaimonSinkLineageVertex(Collections.singletonList(dataset));
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/PaimonLineageDataset.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/PaimonLineageDataset.java
index 5e99df0b2d..cdc072623b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/PaimonLineageDataset.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/PaimonLineageDataset.java
@@ -27,7 +27,7 @@ import java.util.Map;
/**
* A {@link LineageDataset} representing a Paimon table, identified by its
fully qualified name and
- * physical warehouse path as the namespace.
+ * catalog warehouse option as the namespace.
*/
public class PaimonLineageDataset implements LineageDataset {
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 62d601ec1b..ed06e45430 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
@@ -19,10 +19,15 @@
package org.apache.paimon.flink.lineage;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.flink.PaimonDataStreamScanProvider;
import org.apache.paimon.flink.PaimonDataStreamSinkProvider;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
@@ -54,11 +59,14 @@ class LineageUtilsTest {
@TempDir java.nio.file.Path temp;
+ private Path warehouse;
private Path tablePath;
@BeforeEach
void setUp() {
- tablePath = new Path(temp.toUri().toString());
+ // mirror real Paimon layout: <warehouse>/<database>.db/<table>
+ warehouse = new Path(temp.toUri().toString());
+ tablePath = new Path(warehouse, "test_db.db/test_table");
}
private FileStoreTable createTable(
@@ -79,20 +87,50 @@ class LineageUtilsTest {
}
@Test
- void testGetNamespace() throws Exception {
+ void testGetNamespaceWithNullCatalogContext() throws Exception {
FileStoreTable table =
createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
+ assertThat(LineageUtils.getNamespace(table,
null)).isEqualTo(table.options().get("path"));
+ }
- String namespace = LineageUtils.getNamespace(table);
+ @Test
+ void testGetNamespaceWithCatalogContext() throws Exception {
+ FileStoreTable table =
+ createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
+ Options options = new Options();
+ options.set(CatalogOptions.WAREHOUSE, warehouse.toString());
+ CatalogContext ctx = CatalogContext.create(options);
- assertThat(namespace).startsWith("paimon://");
- assertThat(namespace).contains(tablePath.toString());
+ assertThat(LineageUtils.getNamespace(table,
ctx)).isEqualTo(warehouse.toString());
}
@Test
- void testSourceLineageVertexBounded() throws Exception {
+ void testGetNamespaceWithCatalogContextNoWarehouse() throws Exception {
FileStoreTable table =
createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
+ CatalogContext ctx = CatalogContext.create(new Options());
+ assertThat(LineageUtils.getNamespace(table,
ctx)).isEqualTo(table.options().get("path"));
+ }
+
+ @Test
+ void testSourceLineageVertexBounded() throws Exception {
+ Options catalogOptions = new Options();
+ catalogOptions.set(CatalogOptions.WAREHOUSE, warehouse.toString());
+ Catalog catalog =
CatalogFactory.createCatalog(CatalogContext.create(catalogOptions));
+ catalog.createDatabase("test_db", true);
+ catalog.createTable(
+ org.apache.paimon.catalog.Identifier.create("test_db", "src"),
+ new Schema(
+ RowType.of(new IntType(), new VarCharType(100), new
IntType()).getFields(),
+ Collections.emptyList(),
+ Arrays.asList("f0"),
+ new HashMap<>(),
+ ""),
+ false);
+ FileStoreTable table =
+ (FileStoreTable)
+ catalog.getTable(
+
org.apache.paimon.catalog.Identifier.create("test_db", "src"));
SourceLineageVertex vertex =
LineageUtils.sourceLineageVertex("paimon.db.src", true, table);
@@ -102,7 +140,8 @@ class LineageUtilsTest {
LineageDataset dataset = vertex.datasets().get(0);
assertThat(dataset.name()).isEqualTo("paimon.db.src");
- assertThat(dataset.namespace()).startsWith("paimon://");
+ assertThat(dataset.namespace()).isEqualTo(warehouse.toString());
+ catalog.close();
}
@Test
@@ -118,8 +157,23 @@ class LineageUtilsTest {
@Test
void testSinkLineageVertex() throws Exception {
+ Options catalogOptions = new Options();
+ catalogOptions.set(CatalogOptions.WAREHOUSE, warehouse.toString());
+ Catalog catalog =
CatalogFactory.createCatalog(CatalogContext.create(catalogOptions));
+ catalog.createDatabase("test_db", true);
+ catalog.createTable(
+ org.apache.paimon.catalog.Identifier.create("test_db", "sink"),
+ new Schema(
+ RowType.of(new IntType(), new VarCharType(100), new
IntType()).getFields(),
+ Collections.emptyList(),
+ Arrays.asList("f0"),
+ new HashMap<>(),
+ ""),
+ false);
FileStoreTable table =
- createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
+ (FileStoreTable)
+ catalog.getTable(
+
org.apache.paimon.catalog.Identifier.create("test_db", "sink"));
LineageVertex vertex =
LineageUtils.sinkLineageVertex("paimon.db.sink", table);
@@ -128,7 +182,8 @@ class LineageUtilsTest {
LineageDataset dataset = vertex.datasets().get(0);
assertThat(dataset.name()).isEqualTo("paimon.db.sink");
- assertThat(dataset.namespace()).startsWith("paimon://");
+ assertThat(dataset.namespace()).isEqualTo(warehouse.toString());
+ catalog.close();
}
@Test
@@ -144,6 +199,7 @@ class LineageUtilsTest {
DatasetConfigFacet configFacet = (DatasetConfigFacet)
facets.get("config");
Map<String, String> config = configFacet.config();
+ assertThat(config).containsEntry("type", "paimon");
assertThat(config).containsEntry("partition-keys", "f2");
assertThat(config).containsEntry("primary-keys", "f0,f2");
}
