address comment

Author: xuyangzhong

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeltaJoin.java

@@ -142,6 +142,7 @@ public class StreamExecDeltaJoin extends ExecNodeBase<RowData>
 
     @JsonProperty(FIELD_NAME_LEFT_UPSERT_KEY)
     @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
     private final int[] leftUpsertKeys;
 
     // left (streaming) side join right (lookup) side
@@ -155,6 +156,7 @@ public class StreamExecDeltaJoin extends ExecNodeBase<RowData>
 
     @JsonProperty(FIELD_NAME_RIGHT_UPSERT_KEY)
     @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
     private final int[] rightUpsertKeys;
 
     // right (streaming) side join left (lookup) side

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/DeltaJoinUtil.java

@@ -252,7 +252,8 @@ private static boolean areJoinConditionsSupported(StreamPhysicalJoin join) {
             return false;
         }
 
-        // if this join output cdc records, the non-equiv condition must be applied on upsert key
+        // if this join outputs cdc records and has non-equiv condition, the reference columns in
+        // the non-equiv condition must come from the same set of upsert keys
         ChangelogMode changelogMode = getChangelogMode(join);
         if (changelogMode.containsOnly(RowKind.INSERT)) {
             return true;