platinumhamburg commented on code in PR #2161: URL: https://github.com/apache/fluss/pull/2161#discussion_r2654744158
########## fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncColumnProcessor.java: ########## @@ -0,0 +1,91 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.kv.autoinc; + +import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.record.BinaryValue; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.types.DataType; + +import javax.annotation.Nullable; +import javax.annotation.concurrent.NotThreadSafe; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** A updater to auto increment column . */ +@NotThreadSafe +public class AutoIncColumnProcessor implements AutoIncProcessor { + + private final InternalRow.FieldGetter[] flussFieldGetters; + + private final RowEncoder rowEncoder; + + private final DataType[] fieldDataTypes; + private final Map<Integer, IncIDGenerator> idGeneratorMap; + private final short schemaId; + + public AutoIncColumnProcessor( + KvFormat kvFormat, + short schemaId, + Schema schema, + int[] targetColumnIds, + IncIDGenerator[] incIDGenerator) { + this.idGeneratorMap = new HashMap<>(); + List<Integer> columnIds = schema.getColumnIds(); + int[] targetColumnIdx = Arrays.stream(targetColumnIds).map(columnIds::indexOf).toArray(); + for (int i = 0; i < targetColumnIds.length; i++) { + idGeneratorMap.put(targetColumnIdx[i], incIDGenerator[i]); + } + this.fieldDataTypes = schema.getRowType().getChildren().toArray(new DataType[0]); + + // getter for the fields in row + flussFieldGetters = new InternalRow.FieldGetter[fieldDataTypes.length]; + for (int i = 0; i < fieldDataTypes.length; i++) { + flussFieldGetters[i] = InternalRow.createFieldGetter(fieldDataTypes[i], i); + } + this.rowEncoder = RowEncoder.create(kvFormat, fieldDataTypes); + this.schemaId = schemaId; + } + + @Nullable + @Override + public BinaryValue processAutoInc(BinaryValue oldValue) { Review Comment: It would be great to add test cases that validate the system’s behavior under schema evolution, particularly when handling rows from different schema versions where the column set and/or column order may differ. ########## fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncProcessor.java: ########## @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.kv.autoinc; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.TableConfig; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaGetter; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.BinaryValue; +import org.apache.fluss.server.zk.ZkSequenceIDCounter; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.ZkData; + +/** AutoIncProcessor is used to process auto increment column. */ +public interface AutoIncProcessor { Review Comment: Perhaps the AutoIncColumnProcessor could be renamed to AutoIncProcessor, and this static factory class could be renamed to AutoIncProcessors. ########## fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java: ########## @@ -468,8 +474,9 @@ private long applyInsert( PaddingRow latestSchemaRow, long logOffset) throws Exception { - walBuilder.append(ChangeType.INSERT, latestSchemaRow.replaceRow(currentValue.row)); - kvPreWriteBuffer.put(key, currentValue.encodeValue(), logOffset); + BinaryValue newValue = autoIncProcessor.processAutoInc(currentValue); Review Comment: I agree with this perspective. One point worth noting is that when changelogImage == ChangelogImage.WAL, the applyInsert method might not be invoked. It would be helpful to clarify the intended design and expected behavior in this case. Additionally, since PartialUpdater is currently used only by DefaultRowMerger and isn’t a general-purpose class, we might defer considering its consolidation for now. ########## fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/IncIDGenerator.java: ########## @@ -16,24 +16,17 @@ * limitations under the License. */ -package org.apache.fluss.row.columnar; +package org.apache.fluss.server.kv.autoinc; -import org.apache.fluss.annotation.Internal; -import org.apache.fluss.row.InternalRow; - -/** - * Row {@link ColumnVector}. - * - * @since 0.9 - */ -@Internal -public interface RowColumnVector extends ColumnVector { +/** IncIDGenerator is used to generate auto increment column ID. */ +public interface IncIDGenerator { Review Comment: Please rename to SequenceGenerator ########## fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/SegmentIncIDGenerator.java: ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.kv.autoinc; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.server.SequenceIDCounter; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.concurrent.NotThreadSafe; + +/** Segment ID generator, fetch ID with a batch size. */ +@NotThreadSafe +public class SegmentIncIDGenerator implements IncIDGenerator { Review Comment: ditto -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
