add validation

liyubin117 · Nov 13, 2024 · 30f08dd · 30f08dd
1 parent eaeb53a
commit 30f08dd
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 8 deletions.
diff --git a/...on-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java b/...on-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -291,7 +291,7 @@ private static class Factory implements MergeFunctionFactory<KeyValue> {
 
         private final String removeRecordOnSequenceGroup;
 
-        private final Set<Integer> sequenceGroupPartialDelete;
+        private Set<Integer> sequenceGroupPartialDelete;
 
         private Factory(Options options, RowType rowType, List<String> primaryKeys) {
             this.ignoreDelete = options.get(CoreOptions.IGNORE_DELETE);
@@ -303,6 +303,7 @@ private Factory(Options options, RowType rowType, List<String> primaryKeys) {
 
             List<String> fieldNames = rowType.getFieldNames();
             this.fieldSeqComparators = new HashMap<>();
+            Map<String, Integer> sequenceGroupMap = new HashMap<>();
             for (Map.Entry<String, String> entry : options.toMap().entrySet()) {
                 String k = entry.getKey();
                 String v = entry.getValue();
@@ -341,13 +342,7 @@ private Factory(Options options, RowType rowType, List<String> primaryKeys) {
                             fieldName -> {
                                 int index = fieldNames.indexOf(fieldName);
                                 fieldSeqComparators.put(index, userDefinedSeqComparator);
-                                if (removeRecordOnSequenceGroup != null
-                                        && Arrays.asList(
-                                                        removeRecordOnSequenceGroup.split(
-                                                                FIELDS_SEPARATOR))
-                                                .contains(fieldName)) {
-                                    sequenceGroupPartialDelete.add(index);
-                                }
+                                sequenceGroupMap.put(fieldName, index);
                             });
                 }
             }
@@ -370,6 +365,21 @@ private Factory(Options options, RowType rowType, List<String> primaryKeys) {
                     String.format(
                             "sequence group and %s have conflicting behavior so should not be enabled at the same time.",
                             PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE));
+
+            if (removeRecordOnSequenceGroup != null) {
+                String[] sequenceGroupArr = removeRecordOnSequenceGroup.split(FIELDS_SEPARATOR);
+                Preconditions.checkState(
+                        sequenceGroupMap.keySet().containsAll(Arrays.asList(sequenceGroupArr)),
+                        String.format(
+                                "field '%s' defined in '%s' option must be part of sequence groups",
+                                removeRecordOnSequenceGroup,
+                                PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP.key()));
+                sequenceGroupPartialDelete =
+                        Arrays.stream(sequenceGroupArr)
+                                .filter(sequenceGroupMap::containsKey)
+                                .map(sequenceGroupMap::get)
+                                .collect(Collectors.toSet());
+            }
         }
 
         @Override

diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
@@ -1236,7 +1236,21 @@ public void testPartialUpdateRemoveRecordOnSequenceGroup() throws Exception {
                             options.set("partial-update.remove-record-on-sequence-group", "seq2");
                         },
                         rowType);
+        FileStoreTable wrongTable =
+                createFileStoreTable(
+                        options -> {
+                            options.set("merge-engine", "partial-update");
+                            options.set("fields.seq1.sequence-group", "b");
+                            options.set("fields.seq2.sequence-group", "c,d");
+                            options.set("partial-update.remove-record-on-sequence-group", "b");
+                        },
+                        rowType);
         Function<InternalRow, String> rowToString = row -> internalRowToString(row, rowType);
+
+        assertThatThrownBy(() -> wrongTable.newWrite(""))
+                .hasMessageContaining(
+                        "field 'b' defined in 'partial-update.remove-record-on-sequence-group' option must be part of sequence groups");
+
         SnapshotReader snapshotReader = table.newSnapshotReader();
         TableRead read = table.newRead();
         StreamTableWrite write = table.newWrite("");