JingsongLi commented on code in PR #8372:
URL: https://github.com/apache/paimon/pull/8372#discussion_r3519491432
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java:
##########
@@ -132,4 +136,17 @@ public DataStream<Committable> doWrite(
return written;
}
+
+ @Override
+ public DataStreamSink<?> doCommit(DataStream<Committable> written, String
commitUser) {
+ Options options = Options.fromMap(table.options());
+ if (options.get(SINK_COMMITTER_COORDINATOR_OPERATOR_ENABLED)
+ && !options.get(PRECOMMIT_COMPACT)) {
+ return written.sinkTo(new DiscardingSink<>())
Review Comment:
This bypasses the normal wrapping, so table options that are implemented by
wrapping the committer operator are silently ignored when PWC is enabled. For
example and batch tag creation ( in batch runtime) are applied in , but this
branch replaces the committer with a bare and the coordinator does not install
equivalent tag handling. Please either preserve those wrappers for the
coordinator path or reject these option combinations explicitly; otherwise
enabling this option changes existing sink semantics beyond just moving the
commit into the coordinator.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java:
##########
@@ -132,4 +136,17 @@ public DataStream<Committable> doWrite(
return written;
}
+
+ @Override
+ public DataStreamSink<?> doCommit(DataStream<Committable> written, String
commitUser) {
+ Options options = Options.fromMap(table.options());
+ if (options.get(SINK_COMMITTER_COORDINATOR_OPERATOR_ENABLED)
+ && !options.get(PRECOMMIT_COMPACT)) {
+ return written.sinkTo(new DiscardingSink<>())
Review Comment:
This bypasses the normal `FlinkSink.doCommit` wrapping, so table options
that are implemented by wrapping the committer operator are silently ignored
when PWC is enabled. For example `sink.auto-tag-for-savepoint` and batch tag
creation (`tag.creation-mode=BATCH` in batch runtime) are applied in
`FlinkSink.doCommit`, but this branch replaces the committer with a bare
`DiscardingSink` and the coordinator does not install equivalent tag handling.
Please either preserve those wrappers for the coordinator path or reject these
option combinations explicitly; otherwise enabling this option changes existing
sink semantics beyond just moving the commit into the coordinator.
--
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]
