[core]Add system table for source and sink data lineage meta

paimon-common/src/main/java/org/apache/paimon/lineage/DataLineageEntityImpl.java

@@ -0,0 +1,77 @@
+/*
+ * 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.paimon.lineage;
+
+import org.apache.paimon.data.Timestamp;
+
+/** Default implementation for {@link DataLineageEntity}. */
+public class DataLineageEntityImpl implements DataLineageEntity {
+
+    private final String database;
+    private final String table;
+    private final String job;
+    private final long barrierId;
+    private final long snapshotId;
+    private final Timestamp timestamp;
+
+    public DataLineageEntityImpl(
+            String database,
+            String table,
+            String job,
+            long barrierId,
+            long snapshotId,
+            Timestamp timestamp) {
+        this.database = database;
+        this.table = table;
+        this.job = job;
+        this.barrierId = barrierId;
+        this.snapshotId = snapshotId;
+        this.timestamp = timestamp;
+    }
+
+    @Override
+    public String getDatabase() {
+        return database;
+    }
+
+    @Override
+    public String getTable() {
+        return table;
+    }
+
+    @Override
+    public String getJob() {
+        return job;
+    }
+
+    @Override
+    public Timestamp getCreateTime() {
+        return timestamp;
+    }
+
+    @Override
+    public long getBarrierId() {
+        return barrierId;
+    }
+
+    @Override
+    public long getSnapshotId() {
+        return snapshotId;
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/system/DataLineageTable.java

@@ -0,0 +1,168 @@
+/*
+ * 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.paimon.table.system;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.lineage.DataLineageEntity;
+import org.apache.paimon.lineage.LineageMeta;
+import org.apache.paimon.lineage.LineageMetaFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.ReadonlyTable;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.ProjectedRow;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.BiFunction;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Base lineage table for source and sink data lineage. */
+public abstract class DataLineageTable implements ReadonlyTable {
+
+    protected final LineageMetaFactory lineageMetaFactory;
+    protected final Options options;
+
+    protected DataLineageTable(LineageMetaFactory lineageMetaFactory, Options options) {
+        this.lineageMetaFactory = lineageMetaFactory;
+        this.options = options;
+    }
+
+    @Override
+    public InnerTableScan newScan() {
+        return new ReadOnceTableScan() {
+            @Override
+            public InnerTableScan withFilter(Predicate predicate) {
+                return this;
+            }
+
+            @Override
+            protected Plan innerPlan() {
+                /// TODO get the real row count for plan.
+                return () -> Collections.singletonList((Split) () -> 1L);
+            }
+        };
+    }
+
+    @Override
+    public RowType rowType() {
+        List<DataField> fields = new ArrayList<>();
+        fields.add(new DataField(0, "database_name", new VarCharType(VarCharType.MAX_LENGTH)));
+        fields.add(new DataField(1, "table_name", new VarCharType(VarCharType.MAX_LENGTH)));
+        fields.add(new DataField(2, "job_name", new VarCharType(VarCharType.MAX_LENGTH)));
+        fields.add(new DataField(3, "barrier_id", new BigIntType()));
+        fields.add(new DataField(4, "snapshot_id", new BigIntType()));
+        fields.add(new DataField(5, "create_time", new TimestampType()));
+        return new RowType(fields);
+    }
+
+    @Override
+    public List<String> primaryKeys() {
+        return Arrays.asList("database_name", "table_name", "job_name", "barrier_id");
+    }
+
+    /** Data lineage read with lineage meta query. */
+    protected static class DataLineageRead implements InnerTableRead {
+        private final LineageMetaFactory lineageMetaFactory;
+        private final Options options;
+        private final BiFunction<LineageMeta, Predicate, Iterator<DataLineageEntity>>
+                dataLineageQuery;
+        @Nullable private Predicate predicate;
+        private int[][] projection;
+
+        protected DataLineageRead(
+                LineageMetaFactory lineageMetaFactory,
+                Options options,
+                BiFunction<LineageMeta, Predicate, Iterator<DataLineageEntity>> dataLineageQuery) {
+            this.lineageMetaFactory = lineageMetaFactory;
+            this.options = options;
+            this.dataLineageQuery = dataLineageQuery;
+            this.predicate = null;
+        }
+
+        @Override
+        public InnerTableRead withFilter(Predicate predicate) {
+            this.predicate = predicate;
+            return this;
+        }
+
+        @Override
+        public InnerTableRead withProjection(int[][] projection) {
+            this.projection = projection;
+            return this;
+        }
+
+        @Override
+        public TableRead withIOManager(IOManager ioManager) {
+            return this;
+        }
+
+        @Override
+        public RecordReader<InternalRow> createReader(Split split) throws IOException {
+            try (LineageMeta lineageMeta = lineageMetaFactory.create(() -> options)) {
+                Iterator<DataLineageEntity> dataLineages =
+                        dataLineageQuery.apply(lineageMeta, predicate);
+                return new IteratorRecordReader<>(
+                        Iterators.transform(
+                                dataLineages,
+                                entity -> {
+                                    checkNotNull(entity);
+                                    GenericRow row =
+                                            GenericRow.of(
+                                                    BinaryString.fromString(entity.getDatabase()),
+                                                    BinaryString.fromString(entity.getTable()),
+                                                    BinaryString.fromString(entity.getJob()),
+                                                    entity.getBarrierId(),
+                                                    entity.getSnapshotId(),
+                                                    entity.getCreateTime());
+                                    if (projection != null) {
+                                        return ProjectedRow.from(projection).replaceRow(row);
+                                    } else {
+                                        return row;
+                                    }
+                                }));
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/system/SinkDataLineageTable.java

@@ -0,0 +1,65 @@
+/*
+ * 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.paimon.table.system;
+
+import org.apache.paimon.lineage.LineageMeta;
+import org.apache.paimon.lineage.LineageMetaFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.InnerTableRead;
+
+import java.util.Map;
+
+/**
+ * This is a system table to display all the sink data lineages.
+ *
+ * <pre>
+ *  For example:
+ *     If we select * from sys.sink_data_lineage, we will get
+ *  database_name  table_name  barrier_id  snapshot_id   job_name   create_time
+ *    default      test0        1L           2L          job1   2023-10-22 20:35:12
+ *   database      test1        1L           3L          job1   2023-10-28 21:35:52
+ *     ...         ...         ...          ...          ...        ...
+ *     We can write sql to fetch the information we need.
+ * </pre>
+ */
+public class SinkDataLineageTable extends DataLineageTable {
+
+    public static final String SINK_DATA_LINEAGE = "sink_data_lineage";
+
+    protected SinkDataLineageTable(LineageMetaFactory lineageMetaFactory, Options options) {
+        super(lineageMetaFactory, options);
+    }
+
+    @Override
+    public InnerTableRead newRead() {
+        return new DataLineageTable.DataLineageRead(
+                lineageMetaFactory, options, LineageMeta::sinkDataLineages);
+    }
+
+    @Override
+    public String name() {
+        return SINK_DATA_LINEAGE;
+    }
+
+    @Override
+    public Table copy(Map<String, String> dynamicOptions) {
+        return new SinkDataLineageTable(lineageMetaFactory, options);
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/system/SourceDataLineageTable.java

@@ -0,0 +1,65 @@
+/*
+ * 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.paimon.table.system;
+
+import org.apache.paimon.lineage.LineageMeta;
+import org.apache.paimon.lineage.LineageMetaFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.InnerTableRead;
+
+import java.util.Map;
+
+/**
+ * This is a system table to display all the sink data lineages.
+ *
+ * <pre>
+ *  For example:
+ *     If we select * from sys.source_data_lineage, we will get
+ *  database_name  table_name  barrier_id  snapshot_id   job_name   create_time
+ *    default      test0        1L           2L          job1   2023-10-22 20:35:12
+ *   database      test1        1L           3L          job1   2023-10-28 21:35:52
+ *     ...         ...         ...          ...          ...        ...
+ *     We can write sql to fetch the information we need.
+ * </pre>
+ */
+public class SourceDataLineageTable extends DataLineageTable {
+
+    public static final String SOURCE_DATA_LINEAGE = "source_data_lineage";
+
+    protected SourceDataLineageTable(LineageMetaFactory lineageMetaFactory, Options options) {
+        super(lineageMetaFactory, options);
+    }
+
+    @Override
+    public InnerTableRead newRead() {
+        return new DataLineageTable.DataLineageRead(
+                lineageMetaFactory, options, LineageMeta::sourceDataLineages);
+    }
+
+    @Override
+    public String name() {
+        return SOURCE_DATA_LINEAGE;
+    }
+
+    @Override
+    public Table copy(Map<String, String> dynamicOptions) {
+        return new SourceDataLineageTable(lineageMetaFactory, options);
+    }
+}