Note: The specified columns must be a contains all columns of primary key, and all columns - * except primary key should be nullable. + *
Note: The specified columns must contain all columns of the primary key, and all columns + * omitted from the specified columns should be nullable, since they are set to null when the + * row doesn't exist. A {@link UpsertWriter#delete(InternalRow)} operation additionally requires + * the specified columns except primary key to be nullable, since it sets them to null. * * @param targetColumns the columns to partial update, */ diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java index 8417855ccb..ef898f9cb5 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java @@ -126,7 +126,6 @@ private static void sanityCheck( targetColumnsSet.set(targetColumnIndex); } - BitSet pkColumnSet = new BitSet(); // check the target columns contains the primary key for (String key : primaryKeys) { int pkIndex = rowType.getFieldIndex(key); @@ -136,7 +135,6 @@ private static void sanityCheck( "The target write columns %s must contain the primary key columns %s.", rowType.project(targetColumns).getFieldNames(), primaryKeys)); } - pkColumnSet.set(pkIndex); } BitSet autoIncrementColumnSet = new BitSet(); @@ -152,17 +150,21 @@ private static void sanityCheck( autoIncrementColumnSet.set(autoIncrementColumnIndex); } - // check the columns not in targetColumns should be nullable + // an omitted column is written as null, so it must be nullable. auto increment columns + // are always omitted and only get their value on the server. for (int i = 0; i < rowType.getFieldCount(); 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()) { + if (!targetColumnsSet.get(i) && !rowType.getTypeAt(i).isNullable()) { + String columnName = rowType.getFieldNames().get(i); + if (autoIncrementColumnSet.get(i)) { throw new IllegalArgumentException( String.format( - "Partial Update requires all columns except primary key to be nullable, but column %s is NOT NULL.", - rowType.getFieldNames().get(i))); + "Partial Update requires the auto increment column %s to be nullable, since it is always omitted from the target columns and assigned by the server.", + columnName)); } + throw new IllegalArgumentException( + String.format( + "Partial Update requires all columns omitted from the target columns to be nullable, but omitted column %s is NOT NULL.", + columnName)); } } } diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java index 03bcb8d967..0b59506297 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java @@ -34,6 +34,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; +import org.apache.fluss.exception.InvalidTargetColumnException; import org.apache.fluss.fs.FsPath; import org.apache.fluss.fs.TestFileSystem; import org.apache.fluss.metadata.DataLakeFormat; @@ -90,6 +91,7 @@ import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** IT case for {@link FlussTable}. */ @@ -955,6 +957,45 @@ void testPartialPutAndDelete() throws Exception { table.close(); } + @Test + void testPartialPutWithNotNullTargetColumn() throws Exception { + Schema schema = + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("c", new BigIntType(false)) + .primaryKey("a") + .build(); + TableDescriptor tableDescriptor = + TableDescriptor.builder().schema(schema).distributedBy(3, "a").build(); + createTable(DATA1_TABLE_PATH_PK, tableDescriptor, true); + + try (Table table = conn.getTable(DATA1_TABLE_PATH_PK)) { + // seed a full row so that column b, outside the target columns, is not null + table.newUpsert().createWriter().upsert(row(1, "bbb", 1L)).get(); + + UpsertWriter upsertWriter = table.newUpsert().partialUpdate("a", "c").createWriter(); + upsertWriter.upsert(row(1, null, 100L)).get(); + + Lookuper lookuper = table.newLookup().createLookuper(); + GenericRow rowKey = row(1); + assertThat(lookupRow(lookuper, rowKey)) + .isEqualTo(compactedRow(schema.getRowType(), new Object[] {1, "bbb", 100L})); + + // updating the same target columns again keeps column b untouched + upsertWriter.upsert(row(1, null, 200L)).get(); + assertThat(lookupRow(lookuper, rowKey)) + .isEqualTo(compactedRow(schema.getRowType(), new Object[] {1, "bbb", 200L})); + + // a partial delete would null out the NOT NULL column c, and only the server can + // tell, since it depends on the stored row + assertThatThrownBy(() -> upsertWriter.delete(row(1, null, 200L)).get()) + .cause() + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessageContaining("but target column c is NOT NULL."); + } + } + @Test void testInvalidPartialUpdate() throws Exception { Schema schema = @@ -975,13 +1016,15 @@ void testInvalidPartialUpdate() throws Exception { .hasMessage( "The target write columns [b] must contain the primary key columns [a]."); - // the column not in the primary key is nullable, should throw exception + // the column omitted from the target columns is NOT NULL, should throw exception assertThatThrownBy(() -> table.newUpsert().partialUpdate("a", "b").createWriter()) .hasMessage( - "Partial Update requires all columns except primary key to be nullable, but column c is NOT NULL."); - assertThatThrownBy(() -> table.newUpsert().partialUpdate("a", "c").createWriter()) - .hasMessage( - "Partial Update requires all columns except primary key to be nullable, but column c is NOT NULL."); + "Partial Update requires all columns omitted from the target columns to be nullable, " + + "but omitted column c is NOT NULL."); + // column c is NOT NULL but listed in the target columns, so it is always provided by + // the writer and should be accepted + assertThatCode(() -> table.newUpsert().partialUpdate("a", "c").createWriter()) + .doesNotThrowAnyException(); assertThatThrownBy(() -> table.newUpsert().partialUpdate("a", "d").createWriter()) .hasMessage( "Can not find target column: d for table test_db_1.test_pk_table_1."); diff --git a/fluss-rust/crates/fluss/src/client/table/upsert.rs b/fluss-rust/crates/fluss/src/client/table/upsert.rs index 28dce4ee79..13dcfff5ea 100644 --- a/fluss-rust/crates/fluss/src/client/table/upsert.rs +++ b/fluss-rust/crates/fluss/src/client/table/upsert.rs @@ -225,8 +225,6 @@ impl UpsertWriterFactory { target_column_set.set(target_index, true); } - let mut pk_column_set = bitvec![0; field_count]; - // check the target columns contains the primary key for primary_key in primary_keys { let pk_index = row_type.get_field_index(primary_key.as_str()); @@ -241,7 +239,6 @@ impl UpsertWriterFactory { ), }); } - pk_column_set.set(pk_index, true); } None => { return Err(IllegalArgument { @@ -267,24 +264,28 @@ impl UpsertWriterFactory { ), }); } - auto_increment_column_set.set(index, true); } } - // check the columns not in targetColumns should be nullable - for i in 0..field_count { - // column not in primary key and not in auto increment column - if !pk_column_set[i] && !auto_increment_column_set[i] { - // the column should be nullable - if !row_type.fields().get(i).unwrap().data_type.is_nullable() { + // an omitted column is written as NULL, so it must be nullable. auto increment columns + // are always omitted and only get their value on the server. + for (i, field) in row_type.fields().iter().enumerate() { + if !target_column_set[i] && !field.data_type.is_nullable() { + if auto_increment_column_set[i] { return Err(IllegalArgument { message: format!( - "Partial Update requires all columns except primary key to be nullable, but column {} is NOT NULL.", - row_type.fields().get(i).unwrap().name() + "Partial Update requires the auto increment column {} to be nullable, since it is always omitted from the target columns and assigned by the server.", + field.name() ), }); } + return Err(IllegalArgument { + message: format!( + "Partial Update requires all columns omitted from the target columns to be nullable, but omitted column {} is NOT NULL.", + field.name() + ), + }); } } @@ -534,8 +535,80 @@ mod tests { ); assert!(result.unwrap_err().to_string().contains( - "Partial Update requires all columns except primary key to be nullable, but column required_field is NOT NULL." + "Partial Update requires all columns omitted from the target columns to be nullable, but omitted column required_field is NOT NULL." + )); + } + + #[test] + fn not_null_column_in_target_columns_is_accepted() { + let row_type = RowType::new(vec![ + DataField::new("id", DataTypes::int().as_non_nullable(), None), + DataField::new( + "required_field", + DataTypes::string().as_non_nullable(), + None, + ), + DataField::new("optional_field", DataTypes::int(), None), + ]); + let primary_keys = vec!["id".to_string()]; + let auto_increment_col_names = vec![]; + + // `required_field` is NOT NULL but listed as a target column, so it is always written + let target_columns = Some(Arc::new(vec![0usize, 1])); + let result = UpsertWriterFactory::sanity_check( + &row_type, + &primary_keys, + &auto_increment_col_names, + &target_columns, + ); + assert!(result.is_ok()); + + // Listing every column is equally valid. + let target_columns = Some(Arc::new(vec![0usize, 1, 2])); + let result = UpsertWriterFactory::sanity_check( + &row_type, + &primary_keys, + &auto_increment_col_names, + &target_columns, + ); + assert!(result.is_ok()); + } + + #[test] + fn omitted_not_null_auto_increment_column_is_rejected() { + let row_type = RowType::new(vec![ + DataField::new("id", DataTypes::int().as_non_nullable(), None), + DataField::new("name", DataTypes::string(), None), + DataField::new("seq", DataTypes::bigint().as_non_nullable(), None), + ]); + let primary_keys = vec!["id".to_string()]; + let auto_increment_col_names = vec!["seq".to_string()]; + let target_columns = Some(Arc::new(vec![0usize, 1])); + + let result = UpsertWriterFactory::sanity_check( + &row_type, + &primary_keys, + &auto_increment_col_names, + &target_columns, + ); + + assert!(result.unwrap_err().to_string().contains( + "Partial Update requires the auto increment column seq to be nullable, since it is always omitted from the target columns and assigned by the server." )); + + // The same table with a nullable auto increment column is fine. + let row_type = RowType::new(vec![ + DataField::new("id", DataTypes::int().as_non_nullable(), None), + DataField::new("name", DataTypes::string(), None), + DataField::new("seq", DataTypes::bigint(), None), + ]); + let result = UpsertWriterFactory::sanity_check( + &row_type, + &primary_keys, + &auto_increment_col_names, + &target_columns, + ); + assert!(result.is_ok()); } } 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 a7ce4bac9c..6d4f9cc197 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 @@ -44,6 +44,14 @@ public class PartialUpdater { private final boolean updatePrimaryKeyOnly; private final DataType[] fieldDataTypes; + /** The target columns that are NOT NULL, checked by {@link #updateRow}. */ + private final BitSet notNullTargetCols = new BitSet(); + + /** The same columns without the primary key, checked by {@link #deleteRow}. */ + private final BitSet notNullNonPkTargetCols = new BitSet(); + + private final String[] fieldNames; + public PartialUpdater(KvFormat kvFormat, short schemaId, Schema schema, int[] targetColumns) { this.targetSchemaId = schemaId; for (int targetColumn : targetColumns) { @@ -54,6 +62,15 @@ public PartialUpdater(KvFormat kvFormat, short schemaId, Schema schema, int[] ta } this.fieldDataTypes = schema.getRowType().getChildren().toArray(new DataType[0]); sanityCheck(schema, targetColumns); + this.fieldNames = schema.getRowType().getFieldNames().toArray(new String[0]); + for (int i = 0; i < fieldDataTypes.length; i++) { + if (partialUpdateCols.get(i) && !fieldDataTypes[i].isNullable()) { + notNullTargetCols.set(i); + if (!primaryKeyCols.get(i)) { + notNullNonPkTargetCols.set(i); + } + } + } // getter for the fields in row flussFieldGetters = new InternalRow.FieldGetter[fieldDataTypes.length]; @@ -65,7 +82,6 @@ public PartialUpdater(KvFormat kvFormat, short schemaId, Schema schema, int[] ta } private void sanityCheck(Schema schema, int[] targetColumns) { - BitSet pkColumnSet = new BitSet(); // check the target columns contains the primary key for (int pkIndex : schema.getPrimaryKeyIndexes()) { if (!partialUpdateCols.get(pkIndex)) { @@ -75,19 +91,28 @@ private void sanityCheck(Schema schema, int[] targetColumns) { schema.getColumnNames(targetColumns), schema.getColumnNames(schema.getPrimaryKeyIndexes()))); } - pkColumnSet.set(pkIndex); } - // check the columns not in targetColumns should be nullable + BitSet autoIncrementCols = new BitSet(); + for (String name : schema.getAutoIncrementColumnNames()) { + autoIncrementCols.set(schema.getRowType().getFieldIndex(name)); + } + + // an omitted column is written as null, so it must be nullable. auto increment columns + // are always omitted and only get their value after the merge. for (int i = 0; i < fieldDataTypes.length; i++) { - // the columns not in primary key should be nullable - if (!pkColumnSet.get(i)) { - if (!fieldDataTypes[i].isNullable()) { + if (!partialUpdateCols.get(i) && !fieldDataTypes[i].isNullable()) { + String columnName = schema.getRowType().getFieldNames().get(i); + if (autoIncrementCols.get(i)) { throw new InvalidTargetColumnException( String.format( - "Partial Update requires all columns except primary key to be nullable, but column %s is NOT NULL.", - schema.getRowType().getFieldNames().get(i))); + "Partial Update requires the auto increment column %s to be nullable, since it is always omitted from the target columns and assigned by the server.", + columnName)); } + throw new InvalidTargetColumnException( + String.format( + "Partial Update requires all columns omitted from the target columns to be nullable, but omitted column %s is NOT NULL.", + columnName)); } } } @@ -107,6 +132,8 @@ public BinaryValue updateRow(@Nullable BinaryValue oldValue, BinaryValue partial return oldValue; } + checkNotNullTargetCols(partialValue); + rowEncoder.startNewRow(); // write each field for (int i = 0; i < fieldDataTypes.length; i++) { @@ -126,6 +153,23 @@ public BinaryValue updateRow(@Nullable BinaryValue oldValue, BinaryValue partial return new BinaryValue(targetSchemaId, rowEncoder.finishRow()); } + /** + * Rejects a null in a non-nullable slot, which would desynchronise the encoded row. Runs before + * any field getter, since a getter deserialises the whole row and would fail first. + */ + private void checkNotNullTargetCols(BinaryValue partialValue) { + for (int i = notNullTargetCols.nextSetBit(0); + i >= 0; + i = notNullTargetCols.nextSetBit(i + 1)) { + if (partialValue.row.isNullAt(i)) { + throw new InvalidTargetColumnException( + String.format( + "Target column %s is NOT NULL but the written row has no value for it.", + fieldNames[i])); + } + } + } + /** * Partial delete the given {@code value}. If all the fields except for {@link * #partialUpdateCols} in {@code value.row} are null, return null. Otherwise, update all the @@ -134,11 +178,20 @@ public BinaryValue updateRow(@Nullable BinaryValue oldValue, BinaryValue partial * * @param value the value to be deleted * @return the value after partial deleted + * @throws InvalidTargetColumnException if a non-primary-key target column is NOT NULL, since it + * would have to be set to null */ public @Nullable BinaryValue deleteRow(BinaryValue value) { if (isFieldsNull(value.row, partialUpdateCols)) { + // the whole row is removed, so no column is set to null return null; } else { + if (!notNullNonPkTargetCols.isEmpty()) { + throw new InvalidTargetColumnException( + String.format( + "Partial Delete sets the target columns to null, so it requires all target columns except primary key to be nullable, but target column %s is NOT NULL.", + fieldNames[notNullNonPkTargetCols.nextSetBit(0)])); + } rowEncoder.startNewRow(); // write each field for (int i = 0; i < fieldDataTypes.length; i++) { 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 38e4be36fd..d82216e6c1 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 @@ -320,14 +320,45 @@ void testInvalidPartialUpdate2() throws Exception { Collections.singletonList( data2kvRecordFactory.ofRecord( "k1".getBytes(), new Object[] {1, null, "str"}))); + // column c is omitted from the target columns, so it would be written as null assertThatThrownBy(() -> kvTablet.putAsLeader(kvRecordBatch, new int[] {0, 1})) .isInstanceOf(InvalidTargetColumnException.class) .hasMessage( - "Partial Update requires all columns except primary key to be nullable, but column c is NOT NULL."); - assertThatThrownBy(() -> kvTablet.putAsLeader(kvRecordBatch, new int[] {0, 2})) + "Partial Update requires all columns omitted from the target columns to be nullable, " + + "but omitted column c is NOT NULL."); + // column c is listed in the target columns, so it is always provided by the writer + assertThatCode(() -> kvTablet.putAsLeader(kvRecordBatch, new int[] {0, 2})) + .doesNotThrowAnyException(); + } + + @Test + void testPartialDeleteWithNotNullTargetColumn() throws Exception { + final Schema schema = + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("c", new StringType(false)) + .primaryKey("a") + .build(); + initLogTabletAndKvTablet(schema, new HashMap<>()); + KvRecordTestUtils.KvRecordFactory recordFactory = + KvRecordTestUtils.KvRecordFactory.of(schema.getRowType()); + // seed a full row so the delete really has to null out column c + kvTablet.putAsLeader( + kvRecordBatchFactory.ofRecords( + Collections.singletonList( + recordFactory.ofRecord( + "k1".getBytes(), new Object[] {1, "bbb", "str"}))), + null); + + KvRecordBatch deleteBatch = + kvRecordBatchFactory.ofRecords( + Collections.singletonList(recordFactory.ofRecord("k1".getBytes(), null))); + assertThatThrownBy(() -> kvTablet.putAsLeader(deleteBatch, new int[] {0, 2})) .isInstanceOf(InvalidTargetColumnException.class) .hasMessage( - "Partial Update requires all columns except primary key to be nullable, but column c is NOT NULL."); + "Partial Delete sets the target columns to null, so it requires all target columns " + + "except primary key to be nullable, but target column c is NOT NULL."); } @Test diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/partialupdate/PartialUpdaterTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/partialupdate/PartialUpdaterTest.java new file mode 100644 index 0000000000..732e872ab2 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/partialupdate/PartialUpdaterTest.java @@ -0,0 +1,226 @@ +/* + * 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.partialupdate; + +import org.apache.fluss.exception.InvalidTargetColumnException; +import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.record.BinaryValue; +import org.apache.fluss.row.compacted.CompactedRow; +import org.apache.fluss.row.compacted.CompactedRowDeserializer; +import org.apache.fluss.types.BigIntType; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.StringType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import static org.apache.fluss.testutils.DataTestUtils.compactedRow; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link PartialUpdater} target column validation. */ +class PartialUpdaterTest { + + private static final short SCHEMA_ID = 1; + + private static final String OMITTED_NOT_NULL_MESSAGE = + "Partial Update requires all columns omitted from the target columns to be nullable, " + + "but omitted column c is NOT NULL."; + + /** {@code a INT NOT NULL} (primary key), {@code b STRING}, {@code c STRING NOT NULL}. */ + private static final Schema SCHEMA = + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("c", new StringType(false)) + .primaryKey("a") + .build(); + + /** {@code (a, b)} primary key, {@code c STRING NOT NULL}, {@code d STRING}. */ + private static final Schema COMPOSITE_PK_SCHEMA = + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("c", new StringType(false)) + .column("d", DataTypes.STRING()) + .primaryKey("a", "b") + .build(); + + /** Same shape as {@link #SCHEMA} but with a nullable {@code c}. */ + private static final Schema NULLABLE_SCHEMA = + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("c", DataTypes.STRING()) + .primaryKey("a") + .build(); + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testNullabilityIsRequiredOnlyForOmittedColumns(KvFormat kvFormat) { + // c is NOT NULL, so it is accepted as a target column and rejected when omitted + assertThatCode(() -> new PartialUpdater(kvFormat, SCHEMA_ID, SCHEMA, new int[] {0, 2})) + .doesNotThrowAnyException(); + assertThatCode(() -> new PartialUpdater(kvFormat, SCHEMA_ID, SCHEMA, new int[] {0, 1, 2})) + .doesNotThrowAnyException(); + assertThatThrownBy(() -> new PartialUpdater(kvFormat, SCHEMA_ID, SCHEMA, new int[] {0, 1})) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage(OMITTED_NOT_NULL_MESSAGE); + + // every column of a composite primary key is NOT NULL and always a target column + assertThatCode( + () -> + new PartialUpdater( + kvFormat, + SCHEMA_ID, + COMPOSITE_PK_SCHEMA, + new int[] {0, 1, 2})) + .doesNotThrowAnyException(); + assertThatThrownBy( + () -> + new PartialUpdater( + kvFormat, + SCHEMA_ID, + COMPOSITE_PK_SCHEMA, + new int[] {0, 1, 3})) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage(OMITTED_NOT_NULL_MESSAGE); + } + + @Test + void testTargetColumnsMustContainPrimaryKey() { + assertThatThrownBy(() -> createPartialUpdater(SCHEMA, new int[] {1, 2})) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage( + "The target write columns [b, c] must contain the primary key columns [a]."); + assertThatThrownBy(() -> createPartialUpdater(COMPOSITE_PK_SCHEMA, new int[] {0, 2, 3})) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage( + "The target write columns [a, c, d] must contain the primary key columns [a, b]."); + } + + @Test + void testAutoIncrementColumnMustBeNullable() { + // an auto increment column is always omitted and only gets its value after the merge, + // so updateRow writes null into it first + assertThatThrownBy( + () -> + createPartialUpdater( + autoIncrementSchema(new BigIntType(false)), + new int[] {0, 1})) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage( + "Partial Update requires the auto increment column c to be nullable, " + + "since it is always omitted from the target columns and assigned by the server."); + assertThatCode( + () -> + createPartialUpdater( + autoIncrementSchema(DataTypes.BIGINT()), new int[] {0, 1})) + .doesNotThrowAnyException(); + } + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testUpdateRowKeepsOmittedColumnsAndRejectsNullInNotNullTargetColumn(KvFormat kvFormat) { + PartialUpdater partialUpdater = + new PartialUpdater(kvFormat, SCHEMA_ID, SCHEMA, new int[] {0, 2}); + + BinaryValue merged = + partialUpdater.updateRow( + binaryValue(SCHEMA, 1, "old", "oldC"), row(1, null, "newC")); + assertThat(merged.row.getString(1).toString()).isEqualTo("old"); + assertThat(merged.row.getString(2).toString()).isEqualTo("newC"); + + // a null in a non-nullable slot has to be caught rather than encoded. the row is typed + // with SCHEMA as the server types it, so the guard has to fire before any deserialization + assertThatThrownBy(() -> partialUpdater.updateRow(null, asServerTypedRow(1, "b", null))) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage("Target column c is NOT NULL but the written row has no value for it."); + } + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testDeleteRowRejectsNotNullTargetColumnUnlessWholeRowIsRemoved(KvFormat kvFormat) { + PartialUpdater partialUpdater = + new PartialUpdater(kvFormat, SCHEMA_ID, SCHEMA, new int[] {0, 2}); + + assertThatThrownBy(() -> partialUpdater.deleteRow(binaryValue(SCHEMA, 1, "b", "c"))) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage( + "Partial Delete sets the target columns to null, so it requires all target columns " + + "except primary key to be nullable, but target column c is NOT NULL."); + + // b is already null, so the row is removed outright and nothing is nulled + assertThat(partialUpdater.deleteRow(binaryValue(SCHEMA, 1, null, "c"))).isNull(); + } + + @ParameterizedTest + @EnumSource(KvFormat.class) + void testDeleteRowSetsNullableTargetColumnsToNull(KvFormat kvFormat) { + PartialUpdater partialUpdater = + new PartialUpdater(kvFormat, SCHEMA_ID, NULLABLE_SCHEMA, new int[] {0, 2}); + + BinaryValue deleted = partialUpdater.deleteRow(binaryValue(NULLABLE_SCHEMA, 1, "b", "c")); + + assertThat(deleted).isNotNull(); + assertThat(deleted.row.getString(1).toString()).isEqualTo("b"); + assertThat(deleted.row.isNullAt(2)).isTrue(); + } + + private static PartialUpdater createPartialUpdater(Schema schema, int[] targetColumns) { + return new PartialUpdater(KvFormat.COMPACTED, SCHEMA_ID, schema, targetColumns); + } + + private static Schema autoIncrementSchema(DataType autoIncrementType) { + return Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("c", autoIncrementType) + .primaryKey("a") + .enableAutoIncrement("c") + .build(); + } + + /** A row of {@link #NULLABLE_SCHEMA}, usable as the partial value for any of the schemas. */ + private static BinaryValue row(int a, String b, String c) { + return binaryValue(NULLABLE_SCHEMA, a, b, c); + } + + /** + * The same bytes as {@link #row}, but typed with {@link #SCHEMA} the way the server types an + * incoming record. Reading any field of it deserialises the whole row and fails on the null + * that {@code c} is not allowed to hold. + */ + private static BinaryValue asServerTypedRow(int a, String b, String c) { + CompactedRow encoded = compactedRow(NULLABLE_SCHEMA.getRowType(), new Object[] {a, b, c}); + byte[] bytes = new byte[encoded.getSizeInBytes()]; + encoded.copyTo(bytes, 0); + DataType[] types = SCHEMA.getRowType().getChildren().toArray(new DataType[0]); + return new BinaryValue( + SCHEMA_ID, CompactedRow.from(types, bytes, new CompactedRowDeserializer(types))); + } + + private static BinaryValue binaryValue(Schema schema, int a, String b, String c) { + return new BinaryValue( + SCHEMA_ID, compactedRow(schema.getRowType(), new Object[] {a, b, c})); + } +} diff --git a/website/docs/table-design/table-types/pk-table.md b/website/docs/table-design/table-types/pk-table.md index 54e22d5020..cae54c5e97 100644 --- a/website/docs/table-design/table-types/pk-table.md +++ b/website/docs/table-design/table-types/pk-table.md @@ -74,6 +74,21 @@ follows: | 1 | 2.0 | t1 | | 2 | 3.0 | t2 | +### Nullability requirements + +A column left out of the written columns is set to `null` when the row does not exist yet, so every column outside the +written columns must be nullable. Columns that are written are always supplied by the writer, so they may be declared +`NOT NULL`. An auto increment column is always left out of the written columns and only receives its value on the +server, so it must be nullable as well. + +Partial delete has a stricter requirement. It sets the written columns other than the primary key to `null`, so it is +rejected when one of them is `NOT NULL`, unless every column outside the written columns is already `null`, in which +case the whole row is removed instead. + +Tables using the `aggregation` merge engine keep the stricter rule for partial update as well: every column other than +the primary key must be nullable, including the written ones. The write is rejected by the server rather than when the +writer is created. + ## Merge Engines The **Merge Engine** in Fluss is a core component designed to efficiently handle and consolidate data updates for Primary Key Tables.