jsingh-yelp commented on code in PR #7576:
URL: https://github.com/apache/paimon/pull/7576#discussion_r3426729499
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lineage/LineageUtils.java:
##########
@@ -19,66 +19,115 @@
package org.apache.paimon.flink.lineage;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+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;
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("type", "paimon");
config.put("partition-keys", String.join(",", table.partitionKeys()));
config.put("primary-keys", String.join(",", table.primaryKeys()));
table.options().entrySet().stream()
.filter(e -> PAIMON_OPTION_KEYS.contains(e.getKey()))
.forEach(e -> config.put(e.getKey(), e.getValue()));
+ 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, or {@code
"paimon"} when the
+ * warehouse is not available.
*/
- public static String getNamespace(Table table) {
- return PAIMON_DATASET_PREFIX + CoreOptions.path(table.options());
+ public static String getNamespace(@Nullable CatalogContext catalogContext)
{
+ if (catalogContext != null) {
+ String warehouse =
catalogContext.options().get(CatalogOptions.WAREHOUSE);
+ if (warehouse != null) {
+ return warehouse;
+ }
+ }
+ return DEFAULT_NAMESPACE;
Review Comment:
@XiaoHongbo-Hope Yes, I have fixed this as you suggested.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
