fix review comment

Author: beryllw

fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncProcessor.java

@@ -46,22 +46,17 @@ static AutoIncProcessor create(
             Schema schema,
             ZooKeeperClient zkClient) {
         int[] autoIncColumnIndexes = schema.getAutoIncColumnIndexes();
-        if (autoIncColumnIndexes.length > 1) {
-            throw new IllegalArgumentException(
-                    "Multiple auto increment columns are not supported yet.");
-        }
         if (autoIncColumnIndexes.length > 0) {
             IncIDGenerator[] incIDGenerators = new IncIDGenerator[autoIncColumnIndexes.length];
             for (int i = 0; i < autoIncColumnIndexes.length; i++) {
                 ZkSequenceIDCounter zkSequenceIDCounter =
                         new ZkSequenceIDCounter(
                                 zkClient.getCuratorClient(),
                                 ZkData.AutoIncrementColumnZNode.path(
-                                        tablePath, schemaId, autoIncColumnIndexes[i]));
+                                        tablePath, autoIncColumnIndexes[i]));
                 incIDGenerators[i] =
                         new SegmentIncIDGenerator(
                                 tablePath,
-                                schemaId,
                                 autoIncColumnIndexes[i],
                                 schema.getColumnName(autoIncColumnIndexes[i]),
                                 zkSequenceIDCounter,

fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/SegmentIncIDGenerator.java

@@ -36,7 +36,6 @@ public class SegmentIncIDGenerator implements IncIDGenerator {
 
     private final SequenceIDCounter sequenceIDCounter;
     private final TablePath tablePath;
-    private final int schemaId;
     private final int columnIdx;
     private final String columnName;
 
@@ -46,15 +45,13 @@ public class SegmentIncIDGenerator implements IncIDGenerator {
 
     public SegmentIncIDGenerator(
             TablePath tablePath,
-            int schemaId,
             int columnIdx,
             String columnName,
             SequenceIDCounter sequenceIDCounter,
             Configuration properties) {
         batchSize = properties.getLong(ConfigOptions.TABLE_AUTO_INC_BATCH_SIZE);
         this.columnName = columnName;
         this.tablePath = tablePath;
-        this.schemaId = schemaId;
         this.columnIdx = columnIdx;
         this.sequenceIDCounter = sequenceIDCounter;
     }
@@ -63,19 +60,18 @@ private void fetchSegment() {
         try {
             long start = sequenceIDCounter.getAndAdd(batchSize);
             LOG.info(
-                    "Successfully fetch auto-increment values range [{}, {}), table_path={}, schema_id={}, column_idx={}, column_name={}.",
+                    "Successfully fetch auto-increment values range [{}, {}), table_path={}, column_idx={}, column_name={}.",
                     start,
                     start + batchSize,
                     tablePath,
-                    schemaId,
                     columnIdx,
                     columnName);
             segment = new AutoIncIdSegment(start, batchSize);
         } catch (Exception e) {
             throw new FlussRuntimeException(
                     String.format(
-                            "Failed to fetch auto-increment values, table_path=%s, schema_id=%d, column_idx=%d, column_name=%s.",
-                            tablePath, schemaId, columnIdx, columnName),
+                            "Failed to fetch auto-increment values, table_path=%s, column_idx=%d, column_name=%s.",
+                            tablePath, columnIdx, columnName),
                     e);
         }
     }

fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java

@@ -253,37 +253,26 @@ public static String path() {
     }
 
     /**
-     * The znode for columns of a table. The znode path is:
+     * The znode for auto increment columns of a table. The znode path is:
      *
-     * <p>/metadata/databases/[databaseName]/tables/[tableName]/schemas/[schemaId]/columns
+     * <p>/metadata/databases/[databaseName]/tables/[tableName]/auto_inc
      */
-    public static final class ColumnsZNode {
-        public static String path(TablePath tablePath, int schemaId) {
-            return SchemaZNode.path(tablePath, schemaId) + "/columns";
-        }
-    }
-
-    /**
-     * The znode for schemas of a table. The znode path is:
-     *
-     * <p>/metadata/databases/[databaseName]/tables/[tableName]/schemas/[schemaId]/columns/[columnIdx]
-     */
-    public static final class ColumnZNode {
-        public static String path(TablePath tablePath, int schemaId, int columnIdx) {
-            return ColumnsZNode.path(tablePath, schemaId) + "/" + columnIdx;
+    public static final class AutoIncrementColumnsZNode {
+        public static String path(TablePath tablePath) {
+            return TableZNode.path(tablePath) + "/auto_inc";
         }
     }
 
     /**
      * The znode for auto increment column. The znode path is:
      *
-     * <p>/metadata/databases/[databaseName]/tables/[tableName]/schemas/[schemaId]/columns/[columnIdx]/auto_increment_counter
+     * <p>/metadata/databases/[databaseName]/tables/[tableName]/auto_inc/col_[columnIdx]
      */
     public static final class AutoIncrementColumnZNode {
-        public static String path(TablePath tablePath, int schemaId, int columnIdx) {
-            return ColumnZNode.path(tablePath, schemaId, columnIdx)
+        public static String path(TablePath tablePath, int columnIdx) {
+            return AutoIncrementColumnsZNode.path(tablePath)
                     + "/"
-                    + "auto-increment-counter";
+                    + String.format("col_%d", columnIdx);
         }
     }
 

fluss-server/src/test/java/org/apache/fluss/server/kv/autoinc/SegmentIncIDGeneratorTest.java

@@ -39,7 +39,6 @@
 class SegmentIncIDGeneratorTest {
 
     private static final TablePath TABLE_PATH = new TablePath("test_db", "test_table");
-    private static final int SCHEMA_ID = 1;
     private static final int COLUMN_IDX = 0;
     private static final String COLUMN_NAME = "id";
     private static final long BATCH_SIZE = 100;
@@ -60,7 +59,6 @@ void testNextValBasicContinuousId() {
         SegmentIncIDGenerator generator =
                 new SegmentIncIDGenerator(
                         TABLE_PATH,
-                        SCHEMA_ID,
                         COLUMN_IDX,
                         COLUMN_NAME,
                         new TestingSnapshotIDCounter(snapshotIdGenerator),
@@ -84,7 +82,6 @@ void testMultiGenerator() {
                                 SegmentIncIDGenerator generator =
                                         new SegmentIncIDGenerator(
                                                 new TablePath("test_db", "table1"),
-                                                SCHEMA_ID,
                                                 COLUMN_IDX,
                                                 COLUMN_NAME + "_table1",
                                                 new TestingSnapshotIDCounter(snapshotIdGenerator),
@@ -105,7 +102,6 @@ void testFetchFailed() {
         SegmentIncIDGenerator generator =
                 new SegmentIncIDGenerator(
                         new TablePath("test_db", "table1"),
-                        SCHEMA_ID,
                         COLUMN_IDX,
                         COLUMN_NAME + "_table1",
                         new TestingSnapshotIDCounter(snapshotIdGenerator, 2),
@@ -117,8 +113,8 @@ void testFetchFailed() {
                 .isInstanceOf(FlussRuntimeException.class)
                 .hasMessage(
                         String.format(
-                                "Failed to fetch auto-increment values, table_path=%s, schema_id=%d, column_idx=%d, column_name=%s.",
-                                "test_db.table1", SCHEMA_ID, COLUMN_IDX, COLUMN_NAME + "_table1"));
+                                "Failed to fetch auto-increment values, table_path=%s, column_idx=%d, column_name=%s.",
+                                "test_db.table1", COLUMN_IDX, COLUMN_NAME + "_table1"));
     }
 
     private static class TestingSnapshotIDCounter implements SequenceIDCounter {