fixed

alibaba · Jan 20, 2025 · 10c74af · 10c74af
1 parent ad7909b
commit 10c74af
Show file tree

Hide file tree

Showing 6 changed files with 7 additions and 2 deletions.
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorService.java
@@ -203,14 +203,14 @@ private void checkMergeEngine(MergeEngine mergeEngine, Schema schema) {
             RowType rowType = schema.toRowType();
             int fieldIndex = rowType.getFieldIndex(column);
             if (fieldIndex == -1) {
-                throw new IllegalArgumentException(
+                throw new InvalidTableException(
                         String.format(
                                 "The version merge engine column %s does not exist.", column));
             }
             DataType dataType = rowType.getTypeAt(fieldIndex);
             if (!VersionRowMergeEngine.VERSION_MERGE_ENGINE_SUPPORTED_DATA_TYPES.contains(
                     dataType.getClass().getName())) {
-                throw new IllegalArgumentException(
+                throw new InvalidTableException(
                         String.format(
                                 "The version merge engine column does not support type %s .",
                                 dataType));

diff --git a/...rver/src/main/java/com/alibaba/fluss/server/kv/mergeengine/DeduplicateRowMergeEngine.java b/...rver/src/main/java/com/alibaba/fluss/server/kv/mergeengine/DeduplicateRowMergeEngine.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.alibaba.fluss.server.kv.mergeengine;
 
 import com.alibaba.fluss.row.BinaryRow;

diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/kv/mergeengine/FirstRowMergeEngine.java b/fluss-server/src/main/java/com/alibaba/fluss/server/kv/mergeengine/FirstRowMergeEngine.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.alibaba.fluss.server.kv.mergeengine;
 
 import com.alibaba.fluss.row.BinaryRow;

diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/kv/mergeengine/RowMergeEngine.java b/fluss-server/src/main/java/com/alibaba/fluss/server/kv/mergeengine/RowMergeEngine.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.alibaba.fluss.server.kv.mergeengine;
 
 import com.alibaba.fluss.row.BinaryRow;

diff --git a/...s-server/src/main/java/com/alibaba/fluss/server/kv/mergeengine/VersionRowMergeEngine.java b/...s-server/src/main/java/com/alibaba/fluss/server/kv/mergeengine/VersionRowMergeEngine.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.alibaba.fluss.server.kv.mergeengine;
 
 import com.alibaba.fluss.metadata.MergeEngine;

diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/kv/KvTabletTest.java
@@ -658,6 +658,7 @@ void testVersionRowMergeEngine(@TempDir File tempLogDir, @TempDir File tmpKvDir)
                                 "k2".getBytes(), new Object[] {2, 1000L}), // not update
                         kvRecordFactory.ofRecord(
                                 "k1".getBytes(), new Object[] {1, 1001L}), // -U , +U
+                        kvRecordFactory.ofRecord("k1".getBytes(), null), // not update
                         kvRecordFactory.ofRecord("k3".getBytes(), new Object[] {3, 1000L})); // +I
         KvRecordBatch kvRecordBatch2 = kvRecordBatchFactory.ofRecords(kvData2);
         kvTablet.putAsLeader(kvRecordBatch2, null, DATA3_SCHEMA_PK);