You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/02/03 17:46:41 UTC
(fluss) 02/02: [kv] Improve validation for target columns on auto-increment tables
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 84b204421d7af24db28b9582626c6e06423ffd48
Author: Jark Wu <ja...@apache.org>
AuthorDate: Tue Feb 3 17:21:56 2026 +0800
[kv] Improve validation for target columns on auto-increment tables
---
.../java/org/apache/fluss/metadata/Schema.java | 11 ---
.../row/decode/paimon/PaimonKeyDecoderTest.java | 3 +-
.../fluss/flink/sink/FlinkTableSinkITCase.java | 93 ++++++++++++++++++++++
.../java/org/apache/fluss/server/kv/KvTablet.java | 17 +---
.../server/kv/autoinc/AutoIncrementManager.java | 20 +++--
.../server/kv/autoinc/AutoIncrementUpdater.java | 14 ++++
.../kv/autoinc/PerSchemaAutoIncrementUpdater.java | 33 ++++++++
.../server/kv/partialupdate/PartialUpdater.java | 7 ++
.../fluss/server/replica/ReplicaManager.java | 32 ++++----
.../org/apache/fluss/server/kv/KvTabletTest.java | 60 ++++++++++----
.../fluss/server/replica/ReplicaManagerTest.java | 50 +++---------
11 files changed, 239 insertions(+), 101 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
index a97f09ee9..92b675c42 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
@@ -151,17 +151,6 @@ public final class Schema implements Serializable {
.orElseGet(() -> new int[0]);
}
- /**
- * Returns column indexes excluding auto-increment columns, used for partial update operations.
- */
- public @Nullable int[] getNonAutoIncrementColumnIndexes() {
- if (autoIncrementColumnNames.isEmpty()) {
- return null;
- }
- int autoIncIdx = getColumnIndexes(autoIncrementColumnNames)[0];
- return IntStream.range(0, columns.size()).filter(i -> i != autoIncIdx).toArray();
- }
-
/** Returns the auto-increment columnIds, if any, otherwise returns an empty array. */
public int[] getAutoIncrementColumnIds() {
if (autoIncrementColumnNames.isEmpty()) {
diff --git a/fluss-common/src/test/java/org/apache/fluss/row/decode/paimon/PaimonKeyDecoderTest.java b/fluss-common/src/test/java/org/apache/fluss/row/decode/paimon/PaimonKeyDecoderTest.java
index 6066d827f..76c8065ea 100644
--- a/fluss-common/src/test/java/org/apache/fluss/row/decode/paimon/PaimonKeyDecoderTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/row/decode/paimon/PaimonKeyDecoderTest.java
@@ -141,7 +141,8 @@ class PaimonKeyDecoderTest {
PaimonKeyDecoder decoder = new PaimonKeyDecoder(rowType, rowType.getFieldNames());
// Test short strings (≤7 bytes with 0x80 marker) and long strings (>7 bytes)
- for (String testStr : Arrays.asList("", "a", "1234567", "12345678", "Hello, Paimon!")) {
+ for (String testStr :
+ Arrays.asList("", "a", "1234567", "12345678", "Hello, Paimon!", "你好,世界！")) {
InternalRow original = createStringRow(testStr);
byte[] encoded = encoder.encodeKey(original);
InternalRow decoded = decoder.decodeKey(encoded);
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
index f342379b1..73324a6ab 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
@@ -1670,6 +1670,99 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
assertQueryResultExactOrder(tEnv, aggQuery, expectedAggResults);
}
+ @Test
+ void testAutoIncrementWithTargetColumns() throws Exception {
+ // use single parallelism to make result ordering stable
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+
+ String tableName = "auto_increment_pk_table";
+ // Create a table with auto increment column
+ tEnv.executeSql(
+ String.format(
+ "create table %s ("
+ + " id int not null,"
+ + " auto_increment_id bigint,"
+ + " amount bigint,"
+ + " primary key (id) not enforced"
+ + ") with ('auto-increment.fields'='auto_increment_id')",
+ tableName));
+
+ // Insert initial data specifying only id and amount
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO %s (id, amount) VALUES "
+ + "(1, 100), "
+ + "(2, 200), "
+ + "(3, 150), "
+ + "(4, 250)",
+ tableName))
+ .await();
+
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO %s (id, amount) VALUES " + "(3, 350), " + "(5, 500)",
+ tableName))
+ .await();
+
+ List<String> expectedResults =
+ Arrays.asList(
+ "+I[1, 1, 100]",
+ "+I[2, 2, 200]",
+ "+I[3, 3, 150]",
+ "+I[4, 4, 250]",
+ "-U[3, 3, 150]",
+ "+U[3, 3, 350]",
+ "+I[5, 5, 500]");
+
+ // Collect results with timeout
+ assertQueryResultExactOrder(
+ tEnv,
+ String.format(
+ "SELECT id, auto_increment_id, amount FROM %s /*+ OPTIONS('scan.startup.mode' = 'earliest') */",
+ tableName),
+ expectedResults);
+
+ // verify invalid target columns that include auto-increment column
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO %s (id, auto_increment_id) VALUES "
+ + "(6, 10)",
+ tableName))
+ .await())
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .hasRootCauseMessage(
+ "Explicitly specifying values for the auto increment column auto_increment_id is not allowed.");
+
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO %s VALUES " + "(6, 10, 600)",
+ tableName))
+ .await())
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .hasRootCauseMessage(
+ "This table has auto increment column [auto_increment_id]. "
+ + "Explicitly specifying values for an auto increment column is not allowed. "
+ + "Please specify non-auto-increment columns as target columns using partialUpdate first.");
+
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO %s (id, amount, auto_increment_id) VALUES "
+ + "(6, 600, 10)",
+ tableName))
+ .await())
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .hasRootCauseMessage(
+ "This table has auto increment column [auto_increment_id]. "
+ + "Explicitly specifying values for an auto increment column is not allowed. "
+ + "Please specify non-auto-increment columns as target columns using partialUpdate first.");
+ }
+
@Test
void testWalModeWithAutoIncrement() throws Exception {
// use single parallelism to make result ordering stable
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
index 0c711f4f3..8d279555a 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
@@ -332,7 +332,7 @@ public final class KvTablet {
autoIncrementManager.getUpdaterForSchema(kvFormat, latestSchemaId);
// Validate targetColumns doesn't contain auto-increment column
- validateTargetColumns(targetColumns, currentAutoIncrementUpdater, latestSchema);
+ currentAutoIncrementUpdater.validateTargetColumns(targetColumns);
// Determine the row merger based on mergeMode:
// - DEFAULT: Use the configured merge engine (rowMerger)
@@ -411,21 +411,6 @@ public final class KvTablet {
}
}
- private void validateTargetColumns(
- int[] targetColumns, AutoIncrementUpdater autoIncrementUpdater, Schema schema) {
- if (!autoIncrementUpdater.hasAutoIncrement() || targetColumns == null) {
- return;
- }
- List<String> autoIncrementColumnNames = schema.getAutoIncrementColumnNames();
- for (int colIdx : targetColumns) {
- if (autoIncrementColumnNames.contains(schema.getColumnName(colIdx))) {
- throw new IllegalArgumentException(
- "targetColumns must not include auto-increment column name: "
- + schema.getColumnName(colIdx));
- }
- }
- }
-
private void processKvRecords(
KvRecordBatch kvRecords,
short schemaIdOfNewData,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java
index 971ca4fa8..fca383c8f 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementManager.java
@@ -53,11 +53,6 @@ public class AutoIncrementManager {
TablePath tablePath,
TableConfig tableConf,
SequenceGeneratorFactory seqGeneratorFactory) {
- this.autoIncrementUpdaterCache =
- Caffeine.newBuilder()
- .maximumSize(5)
- .expireAfterAccess(Duration.ofMinutes(5))
- .build();
this.schemaGetter = schemaGetter;
int schemaId = schemaGetter.getLatestSchemaInfo().getSchemaId();
Schema schema = schemaGetter.getSchema(schemaId);
@@ -74,17 +69,28 @@ public class AutoIncrementManager {
sequenceGenerator =
seqGeneratorFactory.createSequenceGenerator(
tablePath, autoIncrementColumn, tableConf.getAutoIncrementCacheSize());
+ autoIncrementUpdaterCache =
+ Caffeine.newBuilder()
+ .maximumSize(5)
+ .expireAfterAccess(Duration.ofMinutes(5))
+ .build();
} else {
autoIncrementColumnId = -1;
sequenceGenerator = null;
+ autoIncrementUpdaterCache = null;
}
}
// Supports removing or reordering columns; does NOT support adding an auto-increment column to
// an existing table.
public AutoIncrementUpdater getUpdaterForSchema(KvFormat kvFormat, int latestSchemaId) {
- return autoIncrementUpdaterCache.get(
- latestSchemaId, k -> createAutoIncrementUpdater(kvFormat, k));
+ if (autoIncrementColumnId == -1) {
+ // return no-op updater directly if there is no auto-increment column
+ return NO_OP_UPDATER;
+ } else {
+ return autoIncrementUpdaterCache.get(
+ latestSchemaId, k -> createAutoIncrementUpdater(kvFormat, k));
+ }
}
private AutoIncrementUpdater createAutoIncrementUpdater(KvFormat kvFormat, int schemaId) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementUpdater.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementUpdater.java
index 4451cc0f0..aa555661a 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementUpdater.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncrementUpdater.java
@@ -18,8 +18,11 @@
package org.apache.fluss.server.kv.autoinc;
+import org.apache.fluss.exception.InvalidTargetColumnException;
import org.apache.fluss.record.BinaryValue;
+import javax.annotation.Nullable;
+
/** A updater to auto increment column . */
public interface AutoIncrementUpdater {
@@ -35,6 +38,17 @@ public interface AutoIncrementUpdater {
*/
BinaryValue updateAutoIncrementColumns(BinaryValue rowValue);
+ /**
+ * Validates the target column indexes for auto-increment updates.
+ *
+ * @param targetColumnIndexes the indexes of the target columns to be validated, may be {@code
+ * null}
+ * @throws InvalidTargetColumnException if the target columns are invalid
+ */
+ default void validateTargetColumns(@Nullable int[] targetColumnIndexes) {
+ // no op
+ }
+
/**
* Returns whether this updater actually performs auto-increment logic.
*
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/PerSchemaAutoIncrementUpdater.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/PerSchemaAutoIncrementUpdater.java
index b3e98fb9d..a07698a65 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/PerSchemaAutoIncrementUpdater.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/PerSchemaAutoIncrementUpdater.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server.kv.autoinc;
+import org.apache.fluss.exception.InvalidTargetColumnException;
import org.apache.fluss.metadata.KvFormat;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.record.BinaryValue;
@@ -26,6 +27,7 @@ import org.apache.fluss.row.encode.RowEncoder;
import org.apache.fluss.types.DataType;
import org.apache.fluss.types.DataTypeRoot;
+import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
/**
@@ -42,6 +44,7 @@ public class PerSchemaAutoIncrementUpdater implements AutoIncrementUpdater {
private final RowEncoder rowEncoder;
private final int fieldLength;
private final int targetColumnIdx;
+ private final String autoIncrementColumnName;
private final SequenceGenerator sequenceGenerator;
private final short schemaId;
private final boolean requireInteger;
@@ -69,11 +72,13 @@ public class PerSchemaAutoIncrementUpdater implements AutoIncrementUpdater {
"Auto-increment column ID %d not found in schema columns: %s",
autoIncrementColumnId, schema.getColumnIds()));
}
+ this.autoIncrementColumnName = schema.getAutoIncrementColumnNames().get(0);
this.requireInteger = fieldDataTypes[targetColumnIdx].is(DataTypeRoot.INTEGER);
this.rowEncoder = RowEncoder.create(kvFormat, fieldDataTypes);
this.flussFieldGetters = flussFieldGetters;
}
+ @Override
public BinaryValue updateAutoIncrementColumns(BinaryValue rowValue) {
rowEncoder.startNewRow();
for (int i = 0; i < fieldLength; i++) {
@@ -93,6 +98,34 @@ public class PerSchemaAutoIncrementUpdater implements AutoIncrementUpdater {
return new BinaryValue(schemaId, rowEncoder.finishRow());
}
+ @Override
+ public void validateTargetColumns(@Nullable int[] targetColumnIndexes) {
+ if (targetColumnIndexes != null) {
+ boolean found = false;
+ for (int idx : targetColumnIndexes) {
+ if (idx == targetColumnIdx) {
+ found = true;
+ break;
+ }
+ }
+ if (found) {
+ throw new InvalidTargetColumnException(
+ String.format(
+ "Auto-increment column [%s] at index %d must not be included in target columns.",
+ autoIncrementColumnName, targetColumnIdx));
+ }
+ } else {
+ // The current table contains an auto-increment column, but no target columns have been
+ // specified, which implies all columns (including the auto-increment one) would be
+ // updated. This is not allowed.
+ throw new InvalidTargetColumnException(
+ String.format(
+ "The table contains an auto-increment column [%s], but update target columns are not explicitly specified. "
+ + "Please specify the update target columns and exclude the auto-increment column from them.",
+ autoIncrementColumnName));
+ }
+ }
+
@Override
public boolean hasAutoIncrement() {
return true;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java
index d8741e010..426bd37e6 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java
@@ -41,6 +41,7 @@ public class PartialUpdater {
private final BitSet partialUpdateCols = new BitSet();
private final BitSet primaryKeyCols = new BitSet();
+ private final boolean updatePrimaryKeyOnly;
private final DataType[] fieldDataTypes;
public PartialUpdater(KvFormat kvFormat, short schemaId, Schema schema, int[] targetColumns) {
@@ -60,6 +61,7 @@ public class PartialUpdater {
flussFieldGetters[i] = InternalRow.createFieldGetter(fieldDataTypes[i], i);
}
this.rowEncoder = RowEncoder.create(kvFormat, fieldDataTypes);
+ this.updatePrimaryKeyOnly = partialUpdateCols.equals(primaryKeyCols);
}
private void sanityCheck(Schema schema, int[] targetColumns) {
@@ -100,6 +102,11 @@ public class PartialUpdater {
* @return the updated value (schema id + row bytes)
*/
public BinaryValue updateRow(@Nullable BinaryValue oldValue, BinaryValue partialValue) {
+ if (updatePrimaryKeyOnly && oldValue != null) {
+ // only primary key columns are updated, return the old value directly
+ return oldValue;
+ }
+
rowEncoder.startNewRow();
// write each field
for (int i = 0; i < fieldDataTypes.length; i++) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index d1010918e..dda8cef8a 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -142,6 +142,7 @@ import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
import static org.apache.fluss.server.TabletManagerBase.getTableInfo;
import static org.apache.fluss.utils.FileUtils.isDirectoryEmpty;
import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
import static org.apache.fluss.utils.Preconditions.checkState;
import static org.apache.fluss.utils.concurrent.LockUtils.inLock;
@@ -597,13 +598,14 @@ public class ReplicaManager {
/**
* Collect missing keys from lookup results for insertion, populating both context and batch
- * maps.
+ * maps. And returns the schema of the table, may return null if there is no missing keys.
*/
- private void collectMissingKeysForInsert(
+ private Schema collectMissingKeysForInsert(
Map<TableBucket, List<byte[]>> entriesPerBucket,
Map<TableBucket, LookupResultForBucket> lookupResults,
Map<TableBucket, MissingKeysContext> missingKeysContextMap,
Map<TableBucket, KvRecordBatch> kvRecordBatchMap) {
+ Schema schema = null;
for (Map.Entry<TableBucket, List<byte[]>> entry : entriesPerBucket.entrySet()) {
TableBucket tb = entry.getKey();
LookupResultForBucket lookupResult = lookupResults.get(tb);
@@ -622,13 +624,14 @@ public class ReplicaManager {
}
}
if (!missingKeys.isEmpty()) {
+ TableInfo tableInfo = getReplicaOrException(tb).getTableInfo();
missingKeysContextMap.put(tb, new MissingKeysContext(missingIndexes, missingKeys));
- kvRecordBatchMap.put(
- tb,
- KeyRecordBatch.create(
- missingKeys, getReplicaOrException(tb).getTableInfo()));
+ kvRecordBatchMap.put(tb, KeyRecordBatch.create(missingKeys, tableInfo));
+ schema = tableInfo.getSchema();
}
}
+ // this assumes all table buckets belong to the same table
+ return schema;
}
/** Lookup a single key value. */
@@ -706,19 +709,18 @@ public class ReplicaManager {
checkArgument(
timeoutMs != null && requiredAcks != null,
"timeoutMs and requiredAcks must be set");
+
Map<TableBucket, MissingKeysContext> entriesPerBucketToInsert = new HashMap<>();
Map<TableBucket, KvRecordBatch> produceEntryData = new HashMap<>();
- collectMissingKeysForInsert(
- entriesPerBucket,
- lookupResultForBucketMap,
- entriesPerBucketToInsert,
- produceEntryData);
+ Schema schema =
+ collectMissingKeysForInsert(
+ entriesPerBucket,
+ lookupResultForBucketMap,
+ entriesPerBucketToInsert,
+ produceEntryData);
if (!produceEntryData.isEmpty()) {
- // Compute target columns: exclude auto-increment to prevent overwriting
- TableBucket firstBucket = produceEntryData.keySet().iterator().next();
- Schema schema = getReplicaOrException(firstBucket).getTableInfo().getSchema();
-
+ checkNotNull(schema, "Schema must be available for insert-if-not-exists");
// TODO: Performance optimization: during lookup-with-insert-if-not-exists flow,
// the original key bytes are wrapped in KeyRecordBatch, then during putRecordsToKv
// they are decoded to rows and immediately re-encoded back to key bytes, causing
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
index 582d27ece..1d59af396 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -290,17 +290,11 @@ class KvTabletTest {
0,
Arrays.asList(
// -- for batch 1
- ChangeType.INSERT,
- ChangeType.INSERT,
- ChangeType.UPDATE_BEFORE,
- ChangeType.UPDATE_AFTER),
+ ChangeType.INSERT, ChangeType.INSERT),
Arrays.asList(
// for k1
new Object[] {1, null, null},
- // for k2: +I
- new Object[] {2, null, null},
- // for k2: -U, +U
- new Object[] {2, null, null},
+ // for k2: +I, the second K2 update is skipped
new Object[] {2, null, null})));
LogRecords actualLogRecords = readLogRecords();
@@ -618,7 +612,8 @@ class KvTabletTest {
.build();
initLogTabletAndKvTablet(schema, new HashMap<>());
KvRecordTestUtils.KvRecordFactory recordFactory =
- KvRecordTestUtils.KvRecordFactory.of(schema.getRowType());
+ KvRecordTestUtils.KvRecordFactory.of(
+ schema.getRowType().project(Collections.singletonList("user_name")));
// start threads to put records
List<Future<LogAppendInfo>> putFutures = new ArrayList<>();
@@ -629,12 +624,12 @@ class KvTabletTest {
KvRecordBatch kvRecordBatch1 =
kvRecordBatchFactory.ofRecords(
Arrays.asList(
- recordFactory.ofRecord(k1.getBytes(), new Object[] {k1, null}),
- recordFactory.ofRecord(k2.getBytes(), new Object[] {k2, null}),
- recordFactory.ofRecord(
- k3.getBytes(), new Object[] {k3, null})));
+ recordFactory.ofRecord(k1.getBytes(), new Object[] {k1}),
+ recordFactory.ofRecord(k2.getBytes(), new Object[] {k2}),
+ recordFactory.ofRecord(k3.getBytes(), new Object[] {k3})));
// test concurrent putting to test thread-safety of AutoIncrementManager
- putFutures.add(executor.submit(() -> kvTablet.putAsLeader(kvRecordBatch1, null)));
+ putFutures.add(
+ executor.submit(() -> kvTablet.putAsLeader(kvRecordBatch1, new int[] {0})));
}
// wait for all putting finished
@@ -662,6 +657,43 @@ class KvTabletTest {
assertThat(actualUids).isEqualTo(expectedUids);
}
+ @Test
+ void testAutoIncrementWithInvalidTargetColumns() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("user_name", DataTypes.STRING())
+ .column("uid", DataTypes.INT())
+ .column("age", DataTypes.INT())
+ .primaryKey("user_name")
+ .enableAutoIncrement("uid")
+ .build();
+ initLogTabletAndKvTablet(schema, new HashMap<>());
+ KvRecordTestUtils.KvRecordFactory recordFactory =
+ KvRecordTestUtils.KvRecordFactory.of(schema.getRowType());
+
+ KvRecordBatch kvRecordBatch =
+ kvRecordBatchFactory.ofRecords(
+ Arrays.asList(
+ recordFactory.ofRecord(
+ "k1".getBytes(), new Object[] {"k1", null, null}),
+ recordFactory.ofRecord(
+ "k2".getBytes(), new Object[] {"k2", null, null})));
+ // target columns contain auto-increment column
+ assertThatThrownBy(() -> kvTablet.putAsLeader(kvRecordBatch, new int[] {0, 1}))
+ .isInstanceOf(InvalidTargetColumnException.class)
+ .hasMessageContaining(
+ "Auto-increment column [uid] at index 1 must not be included in target columns.");
+
+ // no specify target columns, which is also invalid for auto-increment
+ assertThatThrownBy(() -> kvTablet.putAsLeader(kvRecordBatch, null))
+ .isInstanceOf(InvalidTargetColumnException.class)
+ .hasMessageContaining(
+ "The table contains an auto-increment column [uid], but update target columns are not explicitly specified.");
+
+ // valid case: target columns don't contain auto-increment column
+ kvTablet.putAsLeader(kvRecordBatch, new int[] {0, 2});
+ }
+
@Test
void testPutAsLeaderWithOutOfOrderSequenceException() throws Exception {
initLogTabletAndKvTablet(DATA1_SCHEMA_PK, new HashMap<>());
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
index 8472964ab..2fd268d5e 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
@@ -906,7 +906,6 @@ class ReplicaManagerTest extends ReplicaTestBase {
assertLogRecordsEquals(DATA3_ROW_TYPE, records, expected, ChangeType.INSERT, schemaGetter2);
}
- // Fixme
@Test
void testConcurrentLookupWithInsertIfNotExistsAutoIncrement() throws Exception {
TableBucket tb = new TableBucket(DATA3_TABLE_ID_PK_AUTO_INC, 1);
@@ -992,45 +991,22 @@ class ReplicaManagerTest extends ReplicaTestBase {
// Values should be 1, 2, 3 (in any order due to concurrency)
assertThat(autoIncrementValues).containsExactlyInAnyOrder(1L, 2L, 3L);
- // Verify WAL: 3 INSERTs minimum, up to 15 if all concurrent updates execute
+ // Verify exactly 3 changelog entries were written (one per unique key)
FetchLogResultForBucket logResult = fetchLog(tb, 0L);
+ // Only the first upsert for a given primary key generates changelog records. Subsequent
+ // upserts on the same primary key produce empty batches, but still increment the log
+ // offset. As a result, the High Watermark may exceed 3 due to these empty batches advancing
+ // the offset.
assertThat(logResult.getHighWatermark()).isBetween(3L, 15L);
-
- // Decode all WAL records
+ // Verify log records contain the expected keys with INSERT change type
LogRecords records = logResult.records();
- LogRecordReadContext readContext =
- LogRecordReadContext.createArrowReadContext(
- DATA3_ROW_TYPE,
- DEFAULT_SCHEMA_ID,
- new TestingSchemaGetter(DEFAULT_SCHEMA_ID, DATA3_SCHEMA_PK_AUTO_INC));
-
- List<Tuple2<ChangeType, Long>> recordsWithAutoInc = new ArrayList<>();
- for (LogRecordBatch batch : records.batches()) {
- try (CloseableIterator<LogRecord> iterator = batch.records(readContext)) {
- while (iterator.hasNext()) {
- LogRecord record = iterator.next();
- recordsWithAutoInc.add(
- Tuple2.of(record.getChangeType(), record.getRow().getLong(2)));
- }
- }
- }
-
- // First 3 must be INSERTs with unique auto-increment values (1, 2, 3)
- assertThat(recordsWithAutoInc).hasSizeGreaterThanOrEqualTo(3);
- Set<Long> insertAutoIncs = new HashSet<>();
- for (int i = 0; i < 3; i++) {
- assertThat(recordsWithAutoInc.get(i).f0).isEqualTo(ChangeType.INSERT);
- insertAutoIncs.add(recordsWithAutoInc.get(i).f1);
- }
- assertThat(insertAutoIncs).containsExactlyInAnyOrder(1L, 2L, 3L);
-
- // Remaining 12 must be UPDATE pairs (UPDATE_BEFORE + UPDATE_AFTER) preserving
- // auto-increment
- for (int i = 3; i < recordsWithAutoInc.size(); i++) {
- Tuple2<ChangeType, Long> record = recordsWithAutoInc.get(i);
- assertThat(record.f0).isIn(ChangeType.UPDATE_BEFORE, ChangeType.UPDATE_AFTER);
- assertThat(record.f1).isIn(1L, 2L, 3L);
- }
+ List<Object[]> expected =
+ Arrays.asList(
+ new Object[] {100, null, 1L},
+ new Object[] {200, null, 2L},
+ new Object[] {300, null, 3L});
+ // only 3 INSERT changelogs generated, the rest are empty batches
+ assertLogRecordsEquals(DATA3_ROW_TYPE, records, expected, ChangeType.INSERT, schemaGetter);
}
@Test