address comment2

Author: luoyuxia

fluss-connectors/fluss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/catalog/FlinkTableFactory.java

@@ -57,6 +57,11 @@
 /** Factory to create table source and table sink for Fluss. */
 public class FlinkTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
 
+    private static final ConfigOption<ConfigOptions.MergeEngine> MERGE_ENGINE_OPTION =
+            key(ConfigOptions.TABLE_MERGE_ENGINE.key())
+                    .enumType(ConfigOptions.MergeEngine.class)
+                    .noDefaultValue();
+
     private volatile LakeTableFactory lakeTableFactory;
 
     @Override
@@ -146,20 +151,14 @@ public DynamicTableSink createDynamicTableSink(Context context) {
                         == RuntimeExecutionMode.STREAMING;
 
         RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType();
-        ConfigOptions.MergeEngine mergeEngine =
-                helper.getOptions()
-                        .get(
-                                key(ConfigOptions.TABLE_MERGE_ENGINE.key())
-                                        .enumType(ConfigOptions.MergeEngine.class)
-                                        .noDefaultValue());
 
         return new FlinkTableSink(
                 toFlussTablePath(context.getObjectIdentifier()),
                 toFlussClientConfig(helper.getOptions(), context.getConfiguration()),
                 rowType,
                 context.getPrimaryKeyIndexes(),
                 isStreamingMode,
-                mergeEngine);
+                helper.getOptions().get(MERGE_ENGINE_OPTION));
     }
 
     @Override

fluss-connectors/fluss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/sink/FlinkTableSink.java

@@ -116,12 +116,20 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
                 // is 0, when no column specified, it's not partial update
                 // see FLINK-36000
                 && context.getTargetColumns().get().length != 0) {
-            // check partial update
-            if (primaryKeyIndexes.length == 0
-                    && context.getTargetColumns().get().length != tableRowType.getFieldCount()) {
-                throw new ValidationException(
-                        "Fluss table sink does not support partial updates for table without primary key. Please make sure the "
-                                + "number of specified columns in INSERT INTO matches columns of the Fluss table.");
+
+            // is partial update, check whether partial update is supported or not
+            if (context.getTargetColumns().get().length != tableRowType.getFieldCount()) {
+                if (primaryKeyIndexes.length == 0) {
+                    throw new ValidationException(
+                            "Fluss table sink does not support partial updates for table without primary key. Please make sure the "
+                                    + "number of specified columns in INSERT INTO matches columns of the Fluss table.");
+                } else if (mergeEngine == ConfigOptions.MergeEngine.FIRST_ROW) {
+                    throw new ValidationException(
+                            String.format(
+                                    "Table %s uses the '%s' merge engine which does not support partial updates. Please make sure the "
+                                            + "number of specified columns in INSERT INTO matches columns of the Fluss table.",
+                                    tablePath, ConfigOptions.MergeEngine.FIRST_ROW));
+                }
             }
             int[][] targetColumns = context.getTargetColumns().get();
             targetColumnIndexes = new int[targetColumns.length];
@@ -291,13 +299,11 @@ private void validateUpdatableAndDeletable() {
                             tablePath));
         }
 
-        if (mergeEngine != null && mergeEngine == ConfigOptions.MergeEngine.FIRST_ROW) {
+        if (mergeEngine == ConfigOptions.MergeEngine.FIRST_ROW) {
             throw new UnsupportedOperationException(
                     String.format(
-                            "Table %s is with merge engine '%s'. Table with '%s' merge engine doesn't support DELETE and UPDATE statements.",
-                            tablePath,
-                            ConfigOptions.MergeEngine.FIRST_ROW,
-                            ConfigOptions.MergeEngine.FIRST_ROW));
+                            "Table %s uses the '%s' merge engine which does not support DELETE or UPDATE statements.",
+                            tablePath, ConfigOptions.MergeEngine.FIRST_ROW));
         }
     }
 

fluss-connectors/fluss-connector-flink/src/test/java/com/alibaba/fluss/connector/flink/sink/FlinkTableSinkITCase.java

@@ -37,6 +37,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.table.api.config.ExecutionConfigOptions;
 import org.apache.flink.types.Row;
@@ -746,21 +747,22 @@ void testUnsupportedDeleteAndUpdateStmtOnPartialPK() {
     }
 
     @Test
-    void testUnsupportedDeleteAndUpdateStmtOnFirstRowMergeEngine() {
+    void testUnsupportedStmtOnFirstRowMergeEngine() {
         String t1 = "firstRowMergeEngineTable";
         TablePath tablePath = TablePath.of(DEFAULT_DB, t1);
         tBatchEnv.executeSql(
                 String.format(
                         "create table %s ("
                                 + " a int not null,"
-                                + " b bigint not null, "
+                                + " b bigint null, "
+                                + " c string null, "
                                 + " primary key (a) not enforced"
                                 + ") with ('table.merge-engine' = 'first_row')",
                         t1));
         assertThatThrownBy(() -> tBatchEnv.executeSql("DELETE FROM " + t1 + " WHERE a = 1").await())
                 .isInstanceOf(UnsupportedOperationException.class)
                 .hasMessage(
-                        "Table %s is with merge engine 'first_row'. Table with 'first_row' merge engine doesn't support DELETE and UPDATE statements.",
+                        "Table %s uses the 'first_row' merge engine which does not support DELETE or UPDATE statements.",
                         tablePath);
 
         assertThatThrownBy(
@@ -770,7 +772,18 @@ void testUnsupportedDeleteAndUpdateStmtOnFirstRowMergeEngine() {
                                         .await())
                 .isInstanceOf(UnsupportedOperationException.class)
                 .hasMessage(
-                        "Table %s is with merge engine 'first_row'. Table with 'first_row' merge engine doesn't support DELETE and UPDATE statements.",
+                        "Table %s uses the 'first_row' merge engine which does not support DELETE or UPDATE statements.",
+                        tablePath);
+
+        assertThatThrownBy(
+                        () ->
+                                tBatchEnv
+                                        .executeSql("INSERT INTO " + t1 + "(a, c) VALUES(1, 'c1')")
+                                        .await())
+                .isInstanceOf(ValidationException.class)
+                .hasMessage(
+                        "Table %s uses the 'first_row' merge engine which does not support partial updates."
+                                + " Please make sure the number of specified columns in INSERT INTO matches columns of the Fluss table.",
                         tablePath);
     }