gustavodemorais commented on code in PR #28918:
URL: https://github.com/apache/flink/pull/28918#discussion_r3714666643
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -1197,41 +1198,50 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
return false
}
- // For a DEDUPLICATE strategy and INSERT only input, we simply let
the inserts be handled
- // as UPSERT_AFTER and overwrite previous value
- if (inputIsAppend && sink.isDeduplicateConflictStrategy) {
- return false
- }
-
// if input has updates and primary key != upsert key we should
enable upsertMaterialize.
//
// An optimize is: do not enable upsertMaterialize when sink pk(s)
contains input
// changeLogUpsertKeys
val upsertKeyDiffersFromPk = !sink.primaryKeysContainsUpsertKey
+ validateOnConflictSpecifiedIfRequired(sink, tableConfig,
upsertKeyDiffersFromPk)
- // Validate that ON CONFLICT is specified when upsert key differs
from primary key
- val requireOnConflict =
-
tableConfig.get(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT)
- if (requireOnConflict && upsertKeyDiffersFromPk &&
sink.conflictStrategy == null) {
- val pkNames = sink.getPrimaryKeyNames
- val upsertKeyNames = sink.getUpsertKeyNames
- throw new ValidationException(
- "The query has an upsert key that differs from the primary key
of the sink table " +
-
s"'${sink.contextResolvedTable.getIdentifier.asSummaryString}'. " +
- s"Primary key: $pkNames, upsert key: $upsertKeyNames. " +
- "This can lead to non-deterministic results when multiple
records with different " +
- "upsert keys map to the same primary key. " +
- "Please specify an ON CONFLICT clause to define how conflicts
should be handled: " +
- "ON CONFLICT DO DEDUPLICATE (update to the latest record,
state intensive, since we" +
- " need to keep the entire history), or " +
- "ON CONFLICT DO ERROR (fail on conflict), or " +
- "ON CONFLICT DO NOTHING (keep first record).")
+ // Once enforcement above has passed, an absent clause leaves
DEDUPLICATE as the strategy.
+ val deduplicatesOnConflict =
+ sink.conflictStrategy == null || sink.isDeduplicateConflictStrategy
+
+ // For a DEDUPLICATE strategy and INSERT only input, we simply let
the inserts be handled
+ // as UPSERT_AFTER and overwrite previous value
Review Comment:
Ah, old comment has this and we just copied. Thanks, fixed
--
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]
