constrain upsert behavior for auto increment column

Author: xx789633

fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java

@@ -61,7 +61,11 @@ class UpsertWriterImpl extends AbstractTableWriter implements UpsertWriter {
             WriterClient writerClient) {
         super(tablePath, tableInfo, writerClient);
         RowType rowType = tableInfo.getRowType();
-        sanityCheck(rowType, tableInfo.getPrimaryKeys(), partialUpdateColumns);
+        sanityCheck(
+                rowType,
+                tableInfo.getPrimaryKeys(),
+                tableInfo.getSchema().getAutoIncrementColumnNames(),
+                partialUpdateColumns);
 
         this.targetColumns = partialUpdateColumns;
         DataLakeFormat lakeFormat = tableInfo.getTableConfig().getDataLakeFormat().orElse(null);
@@ -80,9 +84,20 @@ class UpsertWriterImpl extends AbstractTableWriter implements UpsertWriter {
     }
 
     private static void sanityCheck(
-            RowType rowType, List<String> primaryKeys, @Nullable int[] targetColumns) {
+            RowType rowType,
+            List<String> primaryKeys,
+            List<String> autoIncrementColumnNames,
+            @Nullable int[] targetColumns) {
         // skip check when target columns is null
         if (targetColumns == null) {
+            if (!autoIncrementColumnNames.isEmpty()) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "This table has auto increment column %s."
+                                        + "Explicitly specifying values for an auto increment column is not allowed."
+                                        + "Please specify non-auto-increment columns as target columns using partialUpdate first.",
+                                autoIncrementColumnNames));
+            }
             return;
         }
         BitSet targetColumnsSet = new BitSet();
@@ -116,6 +131,16 @@ private static void sanityCheck(
                 }
             }
         }
+
+        // explicitly specifying values for an auto increment column is not allowed
+        for (String autoIncrementColumnName : autoIncrementColumnNames) {
+            if (targetColumnsSet.get(rowType.getFieldIndex(autoIncrementColumnName))) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Explicitly specifying values for the auto increment column %s is not allowed.",
+                                autoIncrementColumnName));
+            }
+        }
     }
 
     /**

fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java

@@ -662,6 +662,28 @@ void testInvalidPartialUpdate() throws Exception {
                     .hasMessage(
                             "Invalid target column index: 3 for table test_db_1.test_pk_table_1. The table only has 3 columns.");
         }
+
+        // test invalid auto increment column upsert
+        schema =
+                Schema.newBuilder()
+                        .column("a", DataTypes.INT())
+                        .column("b", DataTypes.INT())
+                        .primaryKey("a")
+                        .enableAutoIncrement("b")
+                        .build();
+        tableDescriptor = TableDescriptor.builder().schema(schema).distributedBy(3, "a").build();
+        TablePath tablePath =
+                TablePath.of("test_db_1", "test_invalid_auto_increment_column_upsert");
+        createTable(tablePath, tableDescriptor, true);
+        try (Table table = conn.getTable(tablePath)) {
+            assertThatThrownBy(() -> table.newUpsert().createWriter())
+                    .hasMessage(
+                            "This table has auto increment column [b]. Explicitly specifying values for an auto increment column is not allowed. Please specify non-auto-increment columns as target columns using partialUpdate first.");
+
+            assertThatThrownBy(() -> table.newUpsert().partialUpdate("a", "b").createWriter())
+                    .hasMessage(
+                            "Explicitly specifying values for the auto increment column b is not allowed.");
+        }
     }
 
     @Test