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 63da558652 [flink] Implement FLIP-314 LineageVertexProvider for
non-table source & sink APIs (#8001)
63da558652 is described below
commit 63da558652192e06624e1740c5b7f19bcb30bfc4
Author: jsingh-yelp <[email protected]>
AuthorDate: Wed Jul 1 02:09:53 2026 -0400
[flink] Implement FLIP-314 LineageVertexProvider for non-table source &
sink APIs (#8001)
---
.../apache/paimon/flink/lineage/LineageUtils.java | 74 ++++++-
.../paimon/flink/lineage/PaimonLineageDataset.java | 51 +++++
.../flink/sink/FlinkFormatTableDataStreamSink.java | 10 +-
.../org/apache/paimon/flink/sink/FlinkSink.java | 3 +-
.../paimon/flink/sink/PaimonDiscardingSink.java | 46 ++++
.../paimon/flink/source/FlinkSourceBuilder.java | 5 +-
.../flink/source/PaimonDataStreamSource.java | 89 ++++++++
.../flink/source/operator/MonitorSource.java | 44 +++-
.../paimon/flink/lineage/LineageUtilsTest.java | 238 +++++++++++++++------
.../sink/FlinkFormatTableDataStreamSinkTest.java | 73 +++++++
.../flink/sink/FlinkSinkBuilderLineageTest.java | 84 ++++++++
.../flink/source/FlinkSourceBuilderTest.java | 68 ++++++
12 files changed, 705 insertions(+), 80 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 e9c51d489a..df69031fcf 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,12 +19,19 @@
package org.apache.paimon.flink.lineage;
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.fs.Path;
+import org.apache.paimon.iceberg.IcebergOptions;
+import org.apache.paimon.jdbc.JdbcCatalogFactory;
+import org.apache.paimon.jdbc.JdbcCatalogOptions;
import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.StringUtils;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.streaming.api.lineage.LineageDataset;
@@ -37,9 +44,11 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* Lineage utilities for building {@link SourceLineageVertex} and {@link
LineageVertex} from a
@@ -57,8 +66,13 @@ public class LineageUtils {
new HashSet<>(
Arrays.asList(CatalogOptions.WAREHOUSE.key(),
CatalogOptions.METASTORE.key()));
+ private static final String DEFAULT_CATALOG_IDENTIFIER =
FlinkCatalogFactory.IDENTIFIER;
+
private static final Set<String> PAIMON_OPTION_KEYS =
- CoreOptions.getOptions().stream().map(opt ->
opt.key()).collect(Collectors.toSet());
+ Stream.of(CoreOptions.getOptions(), IcebergOptions.getOptions())
+ .flatMap(List::stream)
+ .map(opt -> opt.key())
+ .collect(Collectors.toSet());
/** Extracts the {@link CatalogContext} from a table, or null if not
available. */
@Nullable
@@ -73,9 +87,9 @@ public class LineageUtils {
}
/**
- * 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."}.
+ * Builds the config map for a dataset facet from a {@link Table}.
Includes {@link CoreOptions}
+ * and {@link IcebergOptions}, partition keys, primary keys, and a safe
subset of catalog-level
+ * options prefixed with {@code "catalog."}.
*/
private static Map<String, String> buildConfigMap(
Table table, @Nullable CatalogContext catalogContext) {
@@ -122,6 +136,27 @@ public class LineageUtils {
return DEFAULT_NAMESPACE;
}
+ @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))) {
+ String catalogKeyValue =
catalogOptions.get(JdbcCatalogOptions.CATALOG_KEY);
+ if (!StringUtils.isNullOrWhitespaceOnly(catalogKeyValue)) {
+ return catalogKeyValue + "." + table.fullName();
+ }
+ }
+ }
+ return DEFAULT_CATALOG_IDENTIFIER + "." + table.fullName();
+ }
+
/**
* Creates a {@link SourceLineageVertex} for a Paimon source table.
*
@@ -134,12 +169,26 @@ public class LineageUtils {
CatalogContext ctx = catalogContext(table);
LineageDataset dataset =
new PaimonLineageDataset(
- name, getNamespace(table, ctx), buildConfigMap(table,
ctx));
+ resolveNameByMetastore(table, name),
+ getNamespace(table, ctx),
+ buildConfigMap(table, ctx),
+ table.rowType());
Boundedness boundedness =
isBounded ? Boundedness.BOUNDED :
Boundedness.CONTINUOUS_UNBOUNDED;
return new PaimonSourceLineageVertex(boundedness,
Collections.singletonList(dataset));
}
+ /**
+ * Creates a {@link SourceLineageVertex} for a Paimon DataStream source
table. The table name is
+ * derived from the table's full name, prefixed with the {@code
catalog-key} if available.
+ *
+ * @param isBounded whether the source is bounded (batch) or unbounded
(streaming)
+ * @param table the Paimon table
+ */
+ public static SourceLineageVertex sourceLineageVertex(boolean isBounded,
Table table) {
+ return sourceLineageVertex(null, isBounded, table);
+ }
+
/**
* Creates a {@link LineageVertex} for a Paimon sink table.
*
@@ -150,7 +199,20 @@ public class LineageUtils {
CatalogContext ctx = catalogContext(table);
LineageDataset dataset =
new PaimonLineageDataset(
- name, getNamespace(table, ctx), buildConfigMap(table,
ctx));
+ resolveNameByMetastore(table, name),
+ getNamespace(table, ctx),
+ buildConfigMap(table, ctx),
+ table.rowType());
return new PaimonSinkLineageVertex(Collections.singletonList(dataset));
}
+
+ /**
+ * Creates a {@link LineageVertex} for a Paimon DataStream sink table. The
table name is derived
+ * from the table's full name, prefixed with the {@code catalog-key} if
available.
+ *
+ * @param table the Paimon table
+ */
+ public static LineageVertex sinkLineageVertex(Table table) {
+ return sinkLineageVertex(null, table);
+ }
}
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 cdc072623b..8f215f3902 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
@@ -18,11 +18,19 @@
package org.apache.paimon.flink.lineage;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+
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;
/**
@@ -34,11 +42,21 @@ public class PaimonLineageDataset implements LineageDataset
{
private final String name;
private final String namespace;
private final Map<String, String> tableOptions;
+ @Nullable private final RowType rowType;
public PaimonLineageDataset(String name, String namespace, Map<String,
String> tableOptions) {
+ this(name, namespace, tableOptions, null);
+ }
+
+ public PaimonLineageDataset(
+ String name,
+ String namespace,
+ Map<String, String> tableOptions,
+ @Nullable RowType rowType) {
this.name = name;
this.namespace = namespace;
this.tableOptions = tableOptions;
+ this.rowType = rowType;
}
@Override
@@ -67,6 +85,39 @@ public class PaimonLineageDataset implements LineageDataset {
return tableOptions;
}
});
+ if (rowType != null) {
+ 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 (DataField field : rowType.getFields()) {
+ String fieldName = field.name();
+ String fieldType = field.type().asSQLString();
+ result.put(
+ fieldName,
+ new DatasetSchemaField<String>() {
+ @Override
+ public String name() {
+ return fieldName;
+ }
+
+ @Override
+ public String type() {
+ return fieldType;
+ }
+ });
+ }
+ return result;
+ }
+ });
+ }
return facets;
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
index 3d133a9cea..a223ad5129 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkRowWrapper;
+import org.apache.paimon.flink.lineage.LineageUtils;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.format.FormatTableWrite;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -32,6 +33,8 @@ import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
import org.apache.flink.table.data.RowData;
import java.util.List;
@@ -55,7 +58,7 @@ public class FlinkFormatTableDataStreamSink {
return dataStream.sinkTo(new FormatTableSink(table, overwrite,
staticPartitions));
}
- private static class FormatTableSink implements Sink<RowData> {
+ private static class FormatTableSink implements Sink<RowData>,
LineageVertexProvider {
private final FormatTable table;
private final boolean overwrite;
@@ -68,6 +71,11 @@ public class FlinkFormatTableDataStreamSink {
this.staticPartitions = staticPartitions;
}
+ @Override
+ public LineageVertex getLineageVertex() {
+ return LineageUtils.sinkLineageVertex(table);
+ }
+
/**
* Do not annotate with <code>@override</code> here to maintain
compatibility with Flink
* 2.0+.
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index 959132ad58..9948863c54 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -40,7 +40,6 @@ import
org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
import javax.annotation.Nullable;
@@ -240,7 +239,7 @@ public abstract class FlinkSink<T> implements Serializable {
}
configureSlotSharingGroup(
committed, options.get(SINK_COMMITTER_CPU),
options.get(SINK_COMMITTER_MEMORY));
- return committed.sinkTo(new
DiscardingSink<>()).name("end").setParallelism(1);
+ return committed.sinkTo(new
PaimonDiscardingSink<>(table)).name("end").setParallelism(1);
}
public static void configureSlotSharingGroup(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PaimonDiscardingSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PaimonDiscardingSink.java
new file mode 100644
index 0000000000..84c5041669
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PaimonDiscardingSink.java
@@ -0,0 +1,46 @@
+/*
+ * 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.flink.sink;
+
+import org.apache.paimon.flink.lineage.LineageUtils;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
+
+/**
+ * A {@link DiscardingSink} that implements {@link LineageVertexProvider} so
Flink's lineage graph
+ * discovers the Paimon sink table when using the DataStream API.
+ */
+public class PaimonDiscardingSink<T> extends DiscardingSink<T> implements
LineageVertexProvider {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FileStoreTable table;
+
+ public PaimonDiscardingSink(FileStoreTable table) {
+ this.table = table;
+ }
+
+ @Override
+ public LineageVertex getLineageVertex() {
+ return LineageUtils.sinkLineageVertex(table);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index 2e9c8ae494..d00d2843ff 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -240,7 +240,7 @@ public class FlinkSourceBuilder {
private DataStream<RowData> toDataStream(Source<RowData, ?, ?> source) {
DataStreamSource<RowData> dataStream =
env.fromSource(
- source,
+ new PaimonDataStreamSource<>(source, table),
watermarkStrategy == null
? WatermarkStrategy.noWatermarks()
: watermarkStrategy,
@@ -360,7 +360,8 @@ public class FlinkSourceBuilder {
unordered,
outerProject(),
isBounded,
- limit);
+ limit,
+ table);
if (parallelism != null) {
dataStream.getTransformation().setParallelism(parallelism);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/PaimonDataStreamSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/PaimonDataStreamSource.java
new file mode 100644
index 0000000000..bfef8bc630
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/PaimonDataStreamSource.java
@@ -0,0 +1,89 @@
+/*
+ * 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.flink.source;
+
+import org.apache.paimon.flink.lineage.LineageUtils;
+import org.apache.paimon.table.Table;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
+
+/**
+ * A {@link Source} wrapper that preserves the wrapped source behavior and
exposes Paimon lineage
+ * for sources built through {@link FlinkSourceBuilder}.
+ */
+public class PaimonDataStreamSource<T, SplitT extends SourceSplit, CheckpointT>
+ implements Source<T, SplitT, CheckpointT>, LineageVertexProvider {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Source<T, SplitT, CheckpointT> source;
+ private final Table table;
+
+ public PaimonDataStreamSource(Source<T, SplitT, CheckpointT> source, Table
table) {
+ this.source = source;
+ this.table = table;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return source.getBoundedness();
+ }
+
+ @Override
+ public SourceReader<T, SplitT> createReader(SourceReaderContext
readerContext)
+ throws Exception {
+ return source.createReader(readerContext);
+ }
+
+ @Override
+ public SplitEnumerator<SplitT, CheckpointT> createEnumerator(
+ SplitEnumeratorContext<SplitT> enumContext) throws Exception {
+ return source.createEnumerator(enumContext);
+ }
+
+ @Override
+ public SplitEnumerator<SplitT, CheckpointT> restoreEnumerator(
+ SplitEnumeratorContext<SplitT> enumContext, CheckpointT
checkpoint) throws Exception {
+ return source.restoreEnumerator(enumContext, checkpoint);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<SplitT> getSplitSerializer() {
+ return source.getSplitSerializer();
+ }
+
+ @Override
+ public SimpleVersionedSerializer<CheckpointT>
getEnumeratorCheckpointSerializer() {
+ return source.getEnumeratorCheckpointSerializer();
+ }
+
+ @Override
+ public LineageVertex getLineageVertex() {
+ return LineageUtils.sourceLineageVertex(getBoundedness() ==
Boundedness.BOUNDED, table);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
index d9b7d054cf..b6de64472b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
@@ -21,9 +21,12 @@ package org.apache.paimon.flink.source.operator;
import org.apache.paimon.flink.NestedProjectedRowData;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.NoOpEnumState;
+import org.apache.paimon.flink.source.PaimonDataStreamSource;
import org.apache.paimon.flink.source.SimpleSourceSplit;
import org.apache.paimon.flink.source.SplitListState;
import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.ChannelComputer;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.EndOfScanException;
@@ -37,6 +40,7 @@ import
org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -242,13 +246,43 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
NestedProjectedRowData nestedProjectedRowData,
boolean isBounded,
@Nullable Long limit) {
+ return buildSource(
+ env,
+ name,
+ typeInfo,
+ readBuilder,
+ monitorInterval,
+ emitSnapshotWatermark,
+ shuffleBucketWithPartition,
+ unordered,
+ nestedProjectedRowData,
+ isBounded,
+ limit,
+ null);
+ }
+
+ public static DataStream<RowData> buildSource(
+ StreamExecutionEnvironment env,
+ String name,
+ TypeInformation<RowData> typeInfo,
+ ReadBuilder readBuilder,
+ long monitorInterval,
+ boolean emitSnapshotWatermark,
+ boolean shuffleBucketWithPartition,
+ boolean unordered,
+ NestedProjectedRowData nestedProjectedRowData,
+ boolean isBounded,
+ @Nullable Long limit,
+ @Nullable Table table) {
+ MonitorSource monitorSource =
+ new MonitorSource(readBuilder, monitorInterval,
emitSnapshotWatermark, isBounded);
+ Source<Split, SimpleSourceSplit, NoOpEnumState> source = monitorSource;
+ if (table != null) {
+ source = new PaimonDataStreamSource<>(monitorSource, table);
+ }
SingleOutputStreamOperator<Split> operator =
env.fromSource(
- new MonitorSource(
- readBuilder,
- monitorInterval,
- emitSnapshotWatermark,
- isBounded),
+ source,
WatermarkStrategy.noWatermarks(),
name + "-Monitor",
new JavaTypeInfo<>(Split.class))
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 ed06e45430..2686a9e404 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,17 +19,18 @@
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.flink.sink.PaimonDiscardingSink;
+import org.apache.paimon.flink.source.ContinuousFileStoreSource;
+import org.apache.paimon.flink.source.PaimonDataStreamSource;
+import org.apache.paimon.flink.source.operator.MonitorSource;
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.CatalogEnvironment;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.types.IntType;
@@ -38,6 +39,7 @@ import org.apache.paimon.types.VarCharType;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.streaming.api.lineage.DatasetConfigFacet;
+import org.apache.flink.streaming.api.lineage.DatasetSchemaFacet;
import org.apache.flink.streaming.api.lineage.LineageDataset;
import org.apache.flink.streaming.api.lineage.LineageDatasetFacet;
import org.apache.flink.streaming.api.lineage.LineageVertex;
@@ -59,14 +61,11 @@ class LineageUtilsTest {
@TempDir java.nio.file.Path temp;
- private Path warehouse;
private Path tablePath;
@BeforeEach
void setUp() {
- // mirror real Paimon layout: <warehouse>/<database>.db/<table>
- warehouse = new Path(temp.toUri().toString());
- tablePath = new Path(warehouse, "test_db.db/test_table");
+ tablePath = new Path(temp.toUri().toString());
}
private FileStoreTable createTable(
@@ -86,51 +85,50 @@ class LineageUtilsTest {
return FileStoreTableFactory.create(LocalFileIO.create(), tablePath);
}
- @Test
- void testGetNamespaceWithNullCatalogContext() throws Exception {
+ private FileStoreTable createTableWithCatalogOptions(Map<String, String>
catalogOptions)
+ throws Exception {
FileStoreTable table =
createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
- assertThat(LineageUtils.getNamespace(table,
null)).isEqualTo(table.options().get("path"));
+ CatalogEnvironment catalogEnvironment =
+ new CatalogEnvironment(
+ null,
+ null,
+ null,
+ null,
+ null,
+ CatalogContext.create(Options.fromMap(catalogOptions)),
+ false,
+ false);
+ return FileStoreTableFactory.create(
+ LocalFileIO.create(), tablePath, table.schema(),
catalogEnvironment);
}
@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);
+ void testGetNamespaceWithWarehouse() throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("warehouse", "s3://my-bucket/warehouse");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ String namespace =
+ LineageUtils.getNamespace(table,
table.catalogEnvironment().catalogContext());
- assertThat(LineageUtils.getNamespace(table,
ctx)).isEqualTo(warehouse.toString());
+ assertThat(namespace).isEqualTo("s3://my-bucket/warehouse");
}
@Test
- void testGetNamespaceWithCatalogContextNoWarehouse() throws Exception {
+ void testGetNamespaceFallsBackToPath() 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"));
+
+ String namespace = LineageUtils.getNamespace(table, null);
+
+ assertThat(namespace).contains(tablePath.toString());
}
@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"));
+ createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
SourceLineageVertex vertex =
LineageUtils.sourceLineageVertex("paimon.db.src", true, table);
@@ -140,8 +138,7 @@ class LineageUtilsTest {
LineageDataset dataset = vertex.datasets().get(0);
assertThat(dataset.name()).isEqualTo("paimon.db.src");
- assertThat(dataset.namespace()).isEqualTo(warehouse.toString());
- catalog.close();
+ assertThat(dataset.namespace()).contains(tablePath.toString());
}
@Test
@@ -155,25 +152,58 @@ class LineageUtilsTest {
assertThat(vertex.boundedness()).isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
}
+ @Test
+ void testSourceLineageVertexKeepsProvidedNameWhenCatalogKeyExists() throws
Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("catalog-key", "jdbc-warehouse");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ SourceLineageVertex vertex =
LineageUtils.sourceLineageVertex("paimon.db.src", true, table);
+
+ assertThat(vertex.datasets().get(0).name()).isEqualTo("paimon.db.src");
+ }
+
+ @Test
+ void testResolveNameByMetastoreUsesCatalogKeyForJdbcMetastore() throws
Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("metastore", "jdbc");
+ catalogOptions.put("catalog-key", "jdbc-warehouse");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ // catalog-key is only used when no explicit name is provided
(DataStream path)
+ assertThat(LineageUtils.resolveNameByMetastore(table, null))
+ .isEqualTo("jdbc-warehouse." + table.fullName());
+ // when explicit name is provided (Table API path), it takes precedence
+ assertThat(LineageUtils.resolveNameByMetastore(table,
"my_catalog.db.src"))
+ .isEqualTo("my_catalog.db.src");
+ }
+
+ @Test
+ void testResolveNameByMetastoreIgnoresCatalogKeyForNonJdbcMetastore()
throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("catalog-key", "jdbc-warehouse");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ assertThat(LineageUtils.resolveNameByMetastore(table, "paimon.db.src"))
+ .isEqualTo("paimon.db.src");
+ }
+
+ @Test
+ void testDataStreamSourceLineageVertexUsesCatalogKey() throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("metastore", "jdbc");
+ catalogOptions.put("catalog-key", "jdbc-warehouse");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ SourceLineageVertex vertex = LineageUtils.sourceLineageVertex(true,
table);
+
+
assertThat(vertex.datasets().get(0).name()).isEqualTo("jdbc-warehouse." +
table.fullName());
+ }
+
@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 =
- (FileStoreTable)
- catalog.getTable(
-
org.apache.paimon.catalog.Identifier.create("test_db", "sink"));
+ createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
LineageVertex vertex =
LineageUtils.sinkLineageVertex("paimon.db.sink", table);
@@ -182,8 +212,19 @@ class LineageUtilsTest {
LineageDataset dataset = vertex.datasets().get(0);
assertThat(dataset.name()).isEqualTo("paimon.db.sink");
- assertThat(dataset.namespace()).isEqualTo(warehouse.toString());
- catalog.close();
+ assertThat(dataset.namespace()).contains(tablePath.toString());
+ }
+
+ @Test
+ void testDataStreamSinkLineageVertexUsesCatalogKey() throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("metastore", "jdbc");
+ catalogOptions.put("catalog-key", "jdbc-warehouse");
+ FileStoreTable table = createTableWithCatalogOptions(catalogOptions);
+
+ LineageVertex vertex = LineageUtils.sinkLineageVertex(table);
+
+
assertThat(vertex.datasets().get(0).name()).isEqualTo("jdbc-warehouse." +
table.fullName());
}
@Test
@@ -199,7 +240,6 @@ 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");
}
@@ -219,6 +259,27 @@ class LineageUtilsTest {
assertThat(config).containsEntry(CoreOptions.CHANGELOG_PRODUCER.key(),
"lookup");
}
+ @Test
+ void testConfigFacetIncludesIcebergOptionsAndExcludesArbitrary() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put("metadata.iceberg.storage", "hadoop-catalog");
+ options.put("metadata.iceberg.uri", "s3://my-bucket/iceberg");
+ options.put("metadata.iceberg.rest.token", "should-not-appear");
+ options.put("some.arbitrary.secret", "should-not-appear");
+
+ FileStoreTable table = createTable(options, Collections.emptyList(),
Arrays.asList("f0"));
+
+ LineageVertex vertex = LineageUtils.sinkLineageVertex("paimon.db.t",
table);
+ LineageDataset dataset = vertex.datasets().get(0);
+
+ DatasetConfigFacet configFacet = (DatasetConfigFacet)
dataset.facets().get("config");
+ Map<String, String> config = configFacet.config();
+ assertThat(config).containsEntry("metadata.iceberg.storage",
"hadoop-catalog");
+ assertThat(config).containsEntry("metadata.iceberg.uri",
"s3://my-bucket/iceberg");
+ assertThat(config).doesNotContainKey("metadata.iceberg.rest.token");
+ assertThat(config).doesNotContainKey("some.arbitrary.secret");
+ }
+
@Test
void testConfigFacetWithEmptyKeys() throws Exception {
FileStoreTable table =
@@ -233,6 +294,24 @@ class LineageUtilsTest {
assertThat(config).containsEntry("primary-keys", "");
}
+ @Test
+ void testSchemaFacetContainsPaimonFields() throws Exception {
+ FileStoreTable table =
+ createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
+
+ LineageVertex vertex = LineageUtils.sinkLineageVertex("paimon.db.t",
table);
+ LineageDataset dataset = vertex.datasets().get(0);
+
+ Map<String, LineageDatasetFacet> facets = dataset.facets();
+ assertThat(facets).containsKey("schema");
+
+ DatasetSchemaFacet schemaFacet = (DatasetSchemaFacet)
facets.get("schema");
+ assertThat(schemaFacet.fields()).containsOnlyKeys("f0", "f1", "f2");
+ assertThat(schemaFacet.fields().get("f0").type()).isEqualTo("INT NOT
NULL");
+
assertThat(schemaFacet.fields().get("f1").type()).isEqualTo("VARCHAR(100)");
+ assertThat(schemaFacet.fields().get("f2").type()).isEqualTo("INT");
+ }
+
@Test
void testScanProviderImplementsLineageVertexProvider() throws Exception {
FileStoreTable table =
@@ -249,16 +328,47 @@ class LineageUtilsTest {
}
@Test
- void testSinkProviderImplementsLineageVertexProvider() throws Exception {
+ void testSinkLineageViaPaimonDiscardingSink() throws Exception {
FileStoreTable table =
createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
- PaimonDataStreamSinkProvider provider =
- new PaimonDataStreamSinkProvider(dataStream -> null,
"paimon.db.sink", table);
+ PaimonDiscardingSink<?> sink = new PaimonDiscardingSink<>(table);
- assertThat(provider).isInstanceOf(LineageVertexProvider.class);
- LineageVertex vertex = provider.getLineageVertex();
+ assertThat(sink).isInstanceOf(LineageVertexProvider.class);
+ LineageVertex vertex = sink.getLineageVertex();
+ assertThat(vertex.datasets()).hasSize(1);
+ }
+
+ @Test
+ void testPaimonDataStreamSourceWrapsMonitorSourceLineageVertex() throws
Exception {
+ FileStoreTable table =
+ createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
+
+ PaimonDataStreamSource<?, ?, ?> source =
+ new PaimonDataStreamSource<>(
+ new MonitorSource(table.newReadBuilder(), 10, false,
true), table);
+
+ assertThat(source).isInstanceOf(LineageVertexProvider.class);
+ SourceLineageVertex vertex = (SourceLineageVertex)
source.getLineageVertex();
+ assertThat(vertex.boundedness()).isEqualTo(Boundedness.BOUNDED);
+ assertThat(vertex.datasets()).hasSize(1);
+ assertThat(vertex.datasets().get(0).name()).isEqualTo("paimon." +
table.fullName());
+ }
+
+ @Test
+ void testPaimonDataStreamSourceWrapsFlinkSourceLineageVertex() throws
Exception {
+ FileStoreTable table =
+ createTable(new HashMap<>(), Collections.emptyList(),
Arrays.asList("f0"));
+
+ PaimonDataStreamSource<?, ?, ?> source =
+ new PaimonDataStreamSource<>(
+ new ContinuousFileStoreSource(
+ table.newReadBuilder(), table.options(), null),
+ table);
+
+ SourceLineageVertex vertex = (SourceLineageVertex)
source.getLineageVertex();
+
assertThat(vertex.boundedness()).isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
assertThat(vertex.datasets()).hasSize(1);
-
assertThat(vertex.datasets().get(0).name()).isEqualTo("paimon.db.sink");
+ assertThat(vertex.datasets().get(0).name()).isEqualTo("paimon." +
table.fullName());
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
new file mode 100644
index 0000000000..9a838bea68
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.sink;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Constructor;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FlinkFormatTableDataStreamSink}. */
+class FlinkFormatTableDataStreamSinkTest {
+
+ @TempDir java.nio.file.Path temp;
+
+ @Test
+ void testFormatTableSinkLineageVertex() throws Exception {
+ FormatTable table =
+ FormatTable.builder()
+ .fileIO(LocalFileIO.create())
+ .identifier(Identifier.create("test_db", "test_table"))
+ .rowType(RowType.of(new IntType()))
+ .partitionKeys(Collections.emptyList())
+ .location(new Path(temp.toUri().toString()).toString())
+ .format(FormatTable.Format.PARQUET)
+ .options(Collections.singletonMap("path",
temp.toUri().toString()))
+ .catalogContext(CatalogContext.create(new Options()))
+ .build();
+
+ Class<?> sinkClass =
+ Class.forName(
+
"org.apache.paimon.flink.sink.FlinkFormatTableDataStreamSink$FormatTableSink");
+ Constructor<?> constructor =
+ sinkClass.getDeclaredConstructor(FormatTable.class,
boolean.class, Map.class);
+ constructor.setAccessible(true);
+ Object sink = constructor.newInstance(table, false,
Collections.emptyMap());
+
+ assertThat(sink).isInstanceOf(LineageVertexProvider.class);
+ LineageVertex vertex = ((LineageVertexProvider)
sink).getLineageVertex();
+ assertThat(vertex.datasets()).hasSize(1);
+ assertThat(vertex.datasets().get(0).name()).isEqualTo("paimon." +
table.fullName());
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkBuilderLineageTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkBuilderLineageTest.java
new file mode 100644
index 0000000000..a7ba7f6879
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkBuilderLineageTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.flink.sink;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.transformations.SinkTransformation;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for sink lineage in {@link FlinkSinkBuilder}. */
+class FlinkSinkBuilderLineageTest {
+
+ @TempDir java.nio.file.Path temp;
+
+ @Test
+ void testFlinkSinkBuilderUsesPaimonDiscardingSinkForLineage() throws
Exception {
+ FileStoreTable table = createTable();
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<RowData> input =
+ env.fromCollection(
+ Collections.singletonList((RowData)
GenericRowData.of(1)),
+ InternalTypeInfo.of(toLogicalType(table.rowType())));
+
+ DataStreamSink<?> sink = new
FlinkSinkBuilder(table).forRowData(input).build();
+
+
assertThat(sink.getTransformation()).isInstanceOf(SinkTransformation.class);
+ SinkTransformation<?, ?> transformation =
+ (SinkTransformation<?, ?>) sink.getTransformation();
+
assertThat(transformation.getSink()).isInstanceOf(PaimonDiscardingSink.class);
+ }
+
+ private FileStoreTable createTable() throws Exception {
+ Path tablePath = new Path(temp.toUri().toString());
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ new SchemaManager(LocalFileIO.create(), tablePath)
+ .createTable(
+ new Schema(
+ RowType.of(new IntType()).getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options,
+ ""));
+ return FileStoreTableFactory.create(LocalFileIO.create(), tablePath);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
index bc2ccb0fed..9f6c46c2a7 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
@@ -22,16 +22,25 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.source.operator.MonitorSource;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.Table;
import org.apache.paimon.types.DataTypes;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Path;
+import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -114,4 +123,63 @@ public class FlinkSourceBuilderTest {
builder = new FlinkSourceBuilder(table);
assertTrue(builder.isUnordered());
}
+
+ @Test
+ public void testBuildWrapsStaticSourceWithPaimonDataStreamSource() throws
Exception {
+ Table table = createTable("static_source", false, -1, true);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<RowData> dataStream =
+ new
FlinkSourceBuilder(table).env(env).sourceBounded(true).build();
+
+
assertThat(dataStream.getTransformation()).isInstanceOf(SourceTransformation.class);
+ SourceTransformation<?, ?, ?> transformation =
+ (SourceTransformation<?, ?, ?>) dataStream.getTransformation();
+
assertThat(transformation.getSource()).isInstanceOf(PaimonDataStreamSource.class);
+ }
+
+ @Test
+ public void testBuildWrapsContinuousSourceWithPaimonDataStreamSource()
throws Exception {
+ Table table = createTable("continuous_source", false, -1, true);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<RowData> dataStream =
+ new
FlinkSourceBuilder(table).env(env).sourceBounded(false).build();
+
+
assertThat(dataStream.getTransformation()).isInstanceOf(SourceTransformation.class);
+ SourceTransformation<?, ?, ?> transformation =
+ (SourceTransformation<?, ?, ?>) dataStream.getTransformation();
+
assertThat(transformation.getSource()).isInstanceOf(PaimonDataStreamSource.class);
+ }
+
+ @Test
+ public void testMonitorSourceBuildSourceWrapsWithPaimonDataStreamSource()
throws Exception {
+ Table table = createTable("monitor_source", false, -1, true);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<RowData> dataStream =
+ MonitorSource.buildSource(
+ env,
+ "source",
+ InternalTypeInfo.of(toLogicalType(table.rowType())),
+ table.newReadBuilder(),
+ 10,
+ false,
+ false,
+ false,
+ null,
+ true,
+ null,
+ table);
+
+ assertThat(dataStream.getTransformation().getTransitivePredecessors())
+ .filteredOn(Transformation.class::isInstance)
+ .filteredOn(transformation -> transformation instanceof
SourceTransformation)
+ .anySatisfy(
+ transformation ->
+ assertThat(
+ ((SourceTransformation<?, ?,
?>) transformation)
+ .getSource())
+
.isInstanceOf(PaimonDataStreamSource.class));
+ }
}
