This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 911d9efa [FLINK-40743][jdbc] Fix the Table sink's empty lineage
dataset name
911d9efa is described below
commit 911d9efac738169fad1ca9f9726676292a152848
Author: Martijn Visser <[email protected]>
AuthorDate: Sun Sep 20 20:01:51 2026 +0200
[FLINK-40743][jdbc] Fix the Table sink's empty lineage dataset name
- JdbcOutputFormatBuilder hands the table name to JdbcOutputFormat as the
lineage dataset name
- JdbcOutputFormat.getLineageVertex() reports that name instead of parsing
the executor's statement, which only exists after open()
Generated-by: Claude Code (Claude Fable 5.1, Claude Opus 5.5)
---
.../core/table/sink/JdbcOutputFormatBuilder.java | 9 ++++++--
.../connector/jdbc/internal/JdbcOutputFormat.java | 23 ++++++++++++++------
.../table/sink/JdbcDynamicTableSinkITCase.java | 25 ++++++++++++++++++++++
.../jdbc/core/table/sink/JdbcOutputFormatTest.java | 8 ++++++-
4 files changed, 55 insertions(+), 10 deletions(-)
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatBuilder.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatBuilder.java
index 493cc8d2..484f00c4 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatBuilder.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatBuilder.java
@@ -86,12 +86,16 @@ public class JdbcOutputFormatBuilder implements
Serializable {
Arrays.stream(fieldDataTypes)
.map(DataType::getLogicalType)
.toArray(LogicalType[]::new);
+ // The planner reads lineage before open(). Parsing the name from a
statement would load
+ // OpenLineage's native SQL parser at plan time.
+ final String lineageTableName = dmlOptions.getTableName();
if (dmlOptions.getKeyFields().isPresent() &&
dmlOptions.getKeyFields().get().length > 0) {
// upsert query
return new JdbcOutputFormat<>(
new SimpleJdbcConnectionProvider(jdbcOptions),
executionOptions,
- () -> createBufferReduceExecutor(dmlOptions,
logicalTypes));
+ () -> createBufferReduceExecutor(dmlOptions, logicalTypes),
+ lineageTableName);
} else {
// append only query
final String sql =
@@ -107,7 +111,8 @@ public class JdbcOutputFormatBuilder implements
Serializable {
dmlOptions.getDialect(),
dmlOptions.getFieldNames(),
logicalTypes,
- sql));
+ sql),
+ lineageTableName);
}
}
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java
index e02657bf..221caba2 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java
@@ -34,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.io.Flushable;
import java.io.IOException;
@@ -41,7 +42,6 @@ import java.io.Serializable;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collections;
-import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -71,6 +71,7 @@ public class JdbcOutputFormat<In, JdbcIn, JdbcExec extends
JdbcBatchStatementExe
private final JdbcExecutionOptions executionOptions;
private final StatementExecutorFactory<JdbcExec> statementExecutorFactory;
+ @Nullable private final String lineageTableName;
@SuppressWarnings("unchecked")
protected Function<In, JdbcIn> getExtractor() {
@@ -90,9 +91,21 @@ public class JdbcOutputFormat<In, JdbcIn, JdbcExec extends
JdbcBatchStatementExe
@Nonnull JdbcConnectionProvider connectionProvider,
@Nonnull JdbcExecutionOptions executionOptions,
@Nonnull StatementExecutorFactory<JdbcExec>
statementExecutorFactory) {
+ this(connectionProvider, executionOptions, statementExecutorFactory,
null);
+ }
+
+ /**
+ * @param lineageTableName the lineage dataset name, or null for an empty
name
+ */
+ public JdbcOutputFormat(
+ @Nonnull JdbcConnectionProvider connectionProvider,
+ @Nonnull JdbcExecutionOptions executionOptions,
+ @Nonnull StatementExecutorFactory<JdbcExec>
statementExecutorFactory,
+ @Nullable String lineageTableName) {
this.connectionProvider = checkNotNull(connectionProvider);
this.executionOptions = checkNotNull(executionOptions);
this.statementExecutorFactory = checkNotNull(statementExecutorFactory);
+ this.lineageTableName = lineageTableName;
}
/** Connects to the target database and initializes the prepared
statement. */
@@ -256,13 +269,9 @@ public class JdbcOutputFormat<In, JdbcIn, JdbcExec extends
JdbcBatchStatementExe
@Override
public LineageVertex getLineageVertex() {
- Optional<String> nameOpt =
- jdbcStatementExecutor == null
- ? Optional.empty()
- :
LineageUtils.tableNameOf(jdbcStatementExecutor.insertSql(), false);
+ String name = lineageTableName == null ? "" : lineageTableName;
String namespace = LineageUtils.namespaceOf(connectionProvider);
- LineageDataset dataset =
- LineageUtils.datasetOf(nameOpt.orElse(""), namespace,
Collections.emptyList());
+ LineageDataset dataset = LineageUtils.datasetOf(name, namespace,
Collections.emptyList());
return LineageUtils.lineageVertexOf(Collections.singleton(dataset));
}
}
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
index d5a294ba..029dae9f 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
@@ -36,6 +36,8 @@ import
org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
@@ -544,6 +546,29 @@ public abstract class JdbcDynamicTableSinkITCase extends
AbstractTestBase implem
sinkFunction.close();
}
+ /**
+ * {@code TableLineageUtils} asks the object the provider returns for
lineage at plan time,
+ * before {@code open()}. The dataset must already carry the physical
table name then.
+ */
+ @Test
+ void testLineageDatasetNameIsKnownAtPlanTime() {
+ // checkpointOutputTable has no primary key, userOutputTable has one
+ for (TableRow table : Arrays.asList(checkpointOutputTable,
userOutputTable)) {
+ Map<String, String> options = getOptions();
+ options.put("table-name", table.getTableName());
+ DynamicTableSink tableSink =
createTableSink(table.getTableResolvedSchema(), options);
+ SinkFunctionProvider sinkProvider =
+ (SinkFunctionProvider)
+ tableSink.getSinkRuntimeProvider(new
SinkRuntimeProviderContext(false));
+
+ LineageVertex lineageVertex =
+ ((LineageVertexProvider)
sinkProvider.createSinkFunction()).getLineageVertex();
+
+ assertThat(lineageVertex.datasets()).hasSize(1);
+
assertThat(lineageVertex.datasets().get(0).name()).isEqualTo(table.getTableName());
+ }
+ }
+
/** The help class to assure the test cases run compatibility. */
static class TestingStreamingRuntimeContext extends
StreamingRuntimeContext {
private final boolean isCheckpointingEnabledUselessPlaceholder;
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatTest.java
index 87d8e388..4286763f 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatTest.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcOutputFormatTest.java
@@ -430,11 +430,17 @@ class JdbcOutputFormatTest extends JdbcDataTestBase {
.setJdbcExecutionOptions(JdbcExecutionOptions.builder().build())
.build();
+ // the planner reads lineage at plan time, before open()
+ assertLineageDataset(outputFormat.getLineageVertex());
+
JdbcOutputSerializer<RowData> serializer =
JdbcOutputSerializer.of(getSerializer(TypeInformation.of(RowData.class), true));
outputFormat.open(serializer);
- LineageVertex lineageVertex = outputFormat.getLineageVertex();
+ assertLineageDataset(outputFormat.getLineageVertex());
+ }
+
+ private static void assertLineageDataset(LineageVertex lineageVertex) {
assertThat(lineageVertex.datasets().size()).isEqualTo(1);
assertThat(lineageVertex.datasets().get(0).name()).isEqualTo("newbooks");
assertThat(lineageVertex.datasets().get(0).namespace()).isEqualTo("derby:memory:test");
