auto increment column should not be nullable

Author: xx789633

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

@@ -118,10 +118,23 @@ private static void sanityCheck(
             pkColumnSet.set(pkIndex);
         }
 
+        BitSet autoIncrementColumnSet = new BitSet();
+        // explicitly specifying values for an auto increment column is not allowed
+        for (String autoIncrementColumnName : autoIncrementColumnNames) {
+            int autoIncrementColumnIndex = rowType.getFieldIndex(autoIncrementColumnName);
+            if (targetColumnsSet.get(autoIncrementColumnIndex)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Explicitly specifying values for the auto increment column %s is not allowed.",
+                                autoIncrementColumnName));
+            }
+            autoIncrementColumnSet.set(autoIncrementColumnIndex);
+        }
+
         // check the columns not in targetColumns should be nullable
         for (int i = 0; i < rowType.getFieldCount(); i++) {
-            // column not in primary key
-            if (!pkColumnSet.get(i)) {
+            // column not in primary key and not in auto increment column
+            if (!pkColumnSet.get(i) && !autoIncrementColumnSet.get(i)) {
                 // the column should be nullable
                 if (!rowType.getTypeAt(i).isNullable()) {
                     throw new IllegalArgumentException(
@@ -131,16 +144,6 @@ 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

@@ -668,8 +668,9 @@ void testInvalidPartialUpdate() throws Exception {
                 Schema.newBuilder()
                         .column("a", DataTypes.INT())
                         .column("b", DataTypes.INT())
+                        .column("c", DataTypes.INT())
                         .primaryKey("a")
-                        .enableAutoIncrement("b")
+                        .enableAutoIncrement("c")
                         .build();
         tableDescriptor = TableDescriptor.builder().schema(schema).distributedBy(3, "a").build();
         TablePath tablePath =
@@ -678,11 +679,11 @@ void testInvalidPartialUpdate() throws Exception {
         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.");
+                            "This table has auto increment column [c]. 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())
+            assertThatThrownBy(() -> table.newUpsert().partialUpdate("a", "c").createWriter())
                     .hasMessage(
-                            "Explicitly specifying values for the auto increment column b is not allowed.");
+                            "Explicitly specifying values for the auto increment column c is not allowed.");
         }
     }
 

fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java

@@ -607,8 +607,10 @@ private static List<Column> normalizeColumns(
                         "The data type of auto increment column must be INT or BIGINT.");
             }
 
-            // primary key should not nullable
-            if (pkSet.contains(column.getName()) && column.getDataType().isNullable()) {
+            // primary key and auto increment column should not nullable
+            if ((pkSet.contains(column.getName())
+                            || autoIncrementColumnNames.contains(column.getName()))
+                    && column.getDataType().isNullable()) {
                 newColumns.add(
                         new Column(
                                 column.getName(),