Re: [PR] [rust][client][server][gateway] Relax partial update nullability rule [fluss]

["fresh-borzoni (via GitHub)" <gi...@apache.org>]

fresh-borzoni commented on code in PR #3885:
URL: https://github.com/apache/fluss/pull/3885#discussion_r3893664293


##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java:
##########
@@ -959,6 +963,53 @@ 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 null in the NOT NULL target column c is rejected before anything is sent
+            assertThatThrownBy(() -> upsertWriter.upsert(row(1, null, null)))
+                    .isInstanceOf(IllegalArgumentException.class)

Review Comment:
   Can you also assert the row is unchanged and a following upsert still lands?



##########
website/docs/table-design/table-types/pk-table.md:
##########
@@ -74,6 +74,20 @@ follows:
 | 1 | 2.0 | t1 |
 | 2 | 3.0 | t2 |
 
+### Nullability requirements
+
+A column omitted from the target columns (the columns being written) is set to `null` when the row does not exist yet,
+so every omitted column must be nullable. A target column is always supplied by the writer, so it may be declared
+`NOT NULL`. An auto increment column is always omitted from the target 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 target columns other than the primary key to `null`, so it is

Review Comment:
    table.delete.behavior=ignore|disable skips this path entirely, worth mentioning as the workaround. 
   Also a Flink sink on this schema fails on the first delete and keeps restarting, since sink.ignore-delete defaults to false. 
   Should we say deletes aren't available by default here?



##########
fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java:
##########
@@ -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()) {

Review Comment:
   On an auto increment table the isFieldsNull escape above can never fire - the server always fills the auto increment column, so it's never null. I tried to run it and upserts work, every delete throws. 
   On main the schema was rejected outright. Should the check move up to sanityCheck?



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


Review Comment:
   FIRST_ROW and VERSIONED don't support partial update at all, and the old rule was catching some of those early by accident. Now they fail at first write with UnknownServerException. 
   Should we reject them here too, like FlinkTableSink does?
   
   Also Rust has no merge engines yet, so this only applies to Java clients. Shall we file an issue for the Rust side?



##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java:
##########
@@ -152,19 +174,51 @@ private static void sanityCheck(
             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 and not in auto increment column
-            if (!pkColumnSet.get(i) && !autoIncrementColumnSet.get(i)) {
-                // the column should be nullable
-                if (!rowType.getTypeAt(i).isNullable()) {
+        // the first_row and versioned mergers reject partial update outright on the server, at
+        // the first write, so fail at writer creation instead. OVERWRITE is exempt, since the
+        // server merges it with the default merger.
+        if (mergeMode != MergeMode.OVERWRITE) {
+            if (mergeEngineType == MergeEngineType.FIRST_ROW) {
+                throw new IllegalArgumentException(

Review Comment:
   This used to be accepted here and rejected by the server later. Can we add a test?



-- 
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: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org