[FLINK-39765][cdc-pipeline/mysql] Implement lineage (FLIP-314) support for mysql source connectors

Author: jsingh-yelp

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/lineage/CDCLineageDataset.java

@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.lineage;
+
+import org.apache.flink.streaming.api.lineage.DatasetConfigFacet;
+import org.apache.flink.streaming.api.lineage.DatasetSchemaFacet;
+import org.apache.flink.streaming.api.lineage.DatasetSchemaField;
+import org.apache.flink.streaming.api.lineage.LineageDataset;
+import org.apache.flink.streaming.api.lineage.LineageDatasetFacet;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** A {@link LineageDataset} implementation for CDC source tables. */
+public class CDCLineageDataset implements LineageDataset {
+
+    private final String name;
+    private final String namespace;
+    private final Map<String, String> config;
+    @Nullable private final LinkedHashMap<String, String> schema;
+
+    /**
+     * Creates a CDC lineage dataset without a schema facet.
+     *
+     * @param name dataset name, usually a concrete table name
+     * @param namespace dataset namespace identifying the source system
+     * @param config config facet values for the dataset
+     */
+    public CDCLineageDataset(String name, String namespace, Map<String, String> config) {
+        this(name, namespace, config, null);
+    }
+
+    /**
+     * Creates a CDC lineage dataset with optional schema metadata.
+     *
+     * @param name dataset name, usually a concrete table name
+     * @param namespace dataset namespace identifying the source system
+     * @param config config facet values for the dataset
+     * @param schema optional ordered map from field name to field type
+     */
+    public CDCLineageDataset(
+            String name,
+            String namespace,
+            Map<String, String> config,
+            @Nullable LinkedHashMap<String, String> schema) {
+        this.name = name;
+        this.namespace = namespace;
+        this.config = config;
+        this.schema = schema;
+    }
+
+    /**
+     * Returns the dataset name.
+     *
+     * @return dataset name, usually a concrete table name
+     */
+    @Override
+    public String name() {
+        return name;
+    }
+
+    /**
+     * Returns the namespace identifying the source system.
+     *
+     * @return dataset namespace
+     */
+    @Override
+    public String namespace() {
+        return namespace;
+    }
+
+    /**
+     * Returns the dataset facets.
+     *
+     * <p>Every dataset contains a {@code config} facet. Datasets with non-empty schema metadata
+     * also contain a {@code schema} facet.
+     *
+     * @return map of facet name to lineage dataset facet
+     */
+    @Override
+    public Map<String, LineageDatasetFacet> facets() {
+        Map<String, LineageDatasetFacet> facets = new HashMap<>();
+        facets.put(
+                "config",
+                new DatasetConfigFacet() {
+                    @Override
+                    public String name() {
+                        return "config";
+                    }
+
+                    @Override
+                    public Map<String, String> config() {
+                        return config;
+                    }
+                });
+        if (schema != null && !schema.isEmpty()) {
+            facets.put(
+                    "schema",
+                    new DatasetSchemaFacet() {
+                        @Override
+                        public String name() {
+                            return "schema";
+                        }
+
+                        @Override
+                        public Map<String, DatasetSchemaField<String>> fields() {
+                            Map<String, DatasetSchemaField<String>> result = new LinkedHashMap<>();
+                            for (Map.Entry<String, String> entry : schema.entrySet()) {
+                                String fieldName = entry.getKey();
+                                String fieldType = entry.getValue();
+                                result.put(
+                                        fieldName,
+                                        new DatasetSchemaField<String>() {
+                                            @Override
+                                            public String name() {
+                                                return fieldName;
+                                            }
+
+                                            @Override
+                                            public String type() {
+                                                return fieldType;
+                                            }
+                                        });
+                            }
+                            return result;
+                        }
+                    });
+        }
+        return facets;
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/lineage/CDCSourceLineageVertex.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.lineage;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.streaming.api.lineage.LineageDataset;
+import org.apache.flink.streaming.api.lineage.SourceLineageVertex;
+
+import java.util.List;
+
+/** A {@link SourceLineageVertex} implementation for CDC sources. */
+public class CDCSourceLineageVertex implements SourceLineageVertex {
+
+    private final Boundedness boundedness;
+    private final List<LineageDataset> datasets;
+
+    /**
+     * Creates a CDC source lineage vertex.
+     *
+     * @param boundedness source boundedness
+     * @param datasets datasets produced by the source
+     */
+    public CDCSourceLineageVertex(Boundedness boundedness, List<LineageDataset> datasets) {
+        this.boundedness = boundedness;
+        this.datasets = datasets;
+    }
+
+    /**
+     * Returns whether the source is bounded or continuously unbounded.
+     *
+     * @return source boundedness
+     */
+    @Override
+    public Boundedness boundedness() {
+        return boundedness;
+    }
+
+    /**
+     * Returns the datasets produced by the source.
+     *
+     * @return source lineage datasets
+     */
+    @Override
+    public List<LineageDataset> datasets() {
+        return datasets;
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/lineage/LineageUtils.java

@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.lineage;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.streaming.api.lineage.LineageDataset;
+import org.apache.flink.streaming.api.lineage.SourceLineageVertex;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Utilities for building lineage vertices from CDC source metadata. */
+public class LineageUtils {
+
+    /**
+     * Builds the dataset namespace for a CDC source.
+     *
+     * <p>The namespace identifies the source system instance, for example {@code
+     * mysql://localhost:3306}.
+     *
+     * @param type CDC source type, such as {@code mysql}
+     * @param hostname source host name
+     * @param port source port
+     * @return namespace used by lineage datasets
+     */
+    public static String getNamespace(String type, String hostname, int port) {
+        return type + "://" + hostname + ":" + port;
+    }
+
+    /**
+     * Builds the config facet values shared by all datasets for a CDC source.
+     *
+     * @param type CDC source type, such as {@code mysql}
+     * @return config facet map for lineage datasets
+     */
+    private static Map<String, String> buildConfigMap(String type) {
+        Map<String, String> config = new HashMap<>();
+        config.put("type", type + "-cdc");
+        return config;
+    }
+
+    /**
+     * Builds a source lineage vertex without schema facets.
+     *
+     * @param type CDC source type, such as {@code mysql}
+     * @param hostname source host name
+     * @param port source port
+     * @param isBounded whether the source is bounded
+     * @param tableList concrete source table names; if empty, a connector-level dataset is used
+     * @return source lineage vertex for the CDC source
+     */
+    public static SourceLineageVertex sourceLineageVertex(
+            String type, String hostname, int port, boolean isBounded, List<String> tableList) {
+        return sourceLineageVertex(type, hostname, port, isBounded, tableList, null);
+    }
+
+    /**
+     * Builds a source lineage vertex for a CDC source.
+     *
+     * <p>When {@code tableList} is not empty, each table becomes one lineage dataset. When it is
+     * empty, a single connector-level dataset is returned. If schemas are provided for a table,
+     * they are attached as schema facets on that table dataset.
+     *
+     * @param type CDC source type, such as {@code mysql}
+     * @param hostname source host name
+     * @param port source port
+     * @param isBounded whether the source is bounded
+     * @param tableList concrete source table names
+     * @param tableSchemas optional map from table name to ordered field name/type pairs
+     * @return source lineage vertex for the CDC source
+     */
+    public static SourceLineageVertex sourceLineageVertex(
+            String type,
+            String hostname,
+            int port,
+            boolean isBounded,
+            List<String> tableList,
+            Map<String, LinkedHashMap<String, String>> tableSchemas) {
+        String namespace = getNamespace(type, hostname, port);
+        Boundedness boundedness =
+                isBounded ? Boundedness.BOUNDED : Boundedness.CONTINUOUS_UNBOUNDED;
+
+        Map<String, String> config = buildConfigMap(type);
+
+        List<LineageDataset> datasets;
+        if (tableList != null && !tableList.isEmpty()) {
+            datasets =
+                    tableList.stream()
+                            .map(
+                                    table -> {
+                                        LinkedHashMap<String, String> schema =
+                                                tableSchemas != null
+                                                        ? tableSchemas.get(table)
+                                                        : null;
+                                        return (LineageDataset)
+                                                new CDCLineageDataset(
+                                                        table, namespace, config, schema);
+                                    })
+                            .collect(Collectors.toList());
+        } else {
+            datasets = Collections.singletonList(new CDCLineageDataset(type, namespace, config));
+        }
+
+        return new CDCSourceLineageVertex(boundedness, datasets);
+    }
+}