[hotfix] Introduce SinkUpsertMaterializer factory method

Author: rkhachatryan

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializer.java

@@ -31,6 +31,8 @@
 import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
 import org.apache.flink.table.runtime.generated.RecordEqualiser;
 import org.apache.flink.table.runtime.operators.TableStreamOperator;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Preconditions;
 
@@ -222,4 +224,18 @@ private boolean equalsIgnoreRowKind(RowData newRow, RowData oldRow) {
         }
         return equaliser.equals(newRow, oldRow);
     }
+
+    public static SinkUpsertMaterializer create(
+            StateTtlConfig ttlConfig,
+            RowType physicalRowType,
+            GeneratedRecordEqualiser rowEqualiser,
+            GeneratedRecordEqualiser upsertKeyEqualiser,
+            int[] inputUpsertKey) {
+        return new SinkUpsertMaterializer(
+                ttlConfig,
+                InternalSerializers.create(physicalRowType),
+                rowEqualiser,
+                upsertKeyEqualiser,
+                inputUpsertKey);
+    }
 }