[client] Reject NOT NULL violations before Java table writes

fluss-client/src/main/java/org/apache/fluss/client/table/writer/AbstractTableWriter.java

@@ -26,6 +26,7 @@
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
 
 import javax.annotation.Nullable;
 
@@ -127,4 +128,28 @@ protected void checkFieldCount(InternalRow row) {
                             + row.getFieldCount());
         }
     }
+
+    static void checkNotNullConstraints(
+            InternalRow row, TableInfo tableInfo, @Nullable int[] targetColumns) {
+        RowType rowType = tableInfo.getRowType();
+        if (targetColumns == null) {
+            for (int i = 0; i < rowType.getFieldCount(); i++) {
+                checkColumnNotNull(row, rowType, tableInfo, i);
+            }
+        } else {
+            for (int targetColumn : targetColumns) {
+                checkColumnNotNull(row, rowType, tableInfo, targetColumn);
+            }
+        }
+    }
+
+    private static void checkColumnNotNull(
+            InternalRow row, RowType rowType, TableInfo tableInfo, int index) {
+        if (!rowType.getTypeAt(index).isNullable() && row.isNullAt(index)) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Column '%s' at position %d of table %s is NOT NULL, but the row has a null value for it.",
+                            rowType.getFieldNames().get(index), index, tableInfo.getTablePath()));
+        }
+    }
 }

fluss-client/src/main/java/org/apache/fluss/client/table/writer/AppendWriterImpl.java

@@ -80,6 +80,7 @@ class AppendWriterImpl extends AbstractTableWriter implements AppendWriter {
      */
     public CompletableFuture<AppendResult> append(InternalRow row) {
         checkFieldCount(row);
+        checkNotNullConstraints(row, tableInfo, null /* targetColumns */);
 
         PhysicalTablePath physicalPath = getPhysicalPath(row);
         byte[] bucketKey = bucketKeyEncoder != null ? bucketKeyEncoder.encodeKey(row) : null;

fluss-client/src/main/java/org/apache/fluss/client/table/writer/MultiTableWriterImpl.java

@@ -387,6 +387,7 @@ WriteRecord build(MultiTableWriteRecord.Operation operation, InternalRow row) {
         }
 
         WriteRecord buildAppendRecord(InternalRow row) {
+            AbstractTableWriter.checkNotNullConstraints(row, tableInfo, null /* targetColumns */);
             PhysicalTablePath physicalPath = getPhysicalPath(row);
             byte[] bucketKey =
                     logBucketKeyEncoder != null ? logBucketKeyEncoder.encodeKey(row) : null;
@@ -424,6 +425,7 @@ private static final class PrimaryTableWriteState extends TableWriteState {
         final KvFormat kvFormat;
         final WriteFormat kvWriteFormat;
         final RowEncoder kvRowEncoder;
+        final int[] primaryKeyIndexes;
 
         PrimaryTableWriteState(TablePath tablePath, TableInfo tableInfo) {
             super(tablePath, tableInfo);
@@ -444,6 +446,7 @@ private static final class PrimaryTableWriteState extends TableWriteState {
             this.kvFormat = tableInfo.getTableConfig().getKvFormat();
             this.kvWriteFormat = WriteFormat.fromKvFormat(this.kvFormat);
             this.kvRowEncoder = RowEncoder.create(this.kvFormat, rowType);
+            this.primaryKeyIndexes = tableInfo.getSchema().getPrimaryKeyIndexes();
         }
 
         @Override
@@ -464,6 +467,7 @@ WriteRecord build(MultiTableWriteRecord.Operation operation, InternalRow row) {
         }
 
         WriteRecord buildUpsertRecord(InternalRow row) {
+            AbstractTableWriter.checkNotNullConstraints(row, tableInfo, null /* targetColumns */);
             byte[] key = primaryKeyEncoder.encodeKey(row);
             return WriteRecord.forUpsert(
                     tableInfo,
@@ -477,6 +481,7 @@ WriteRecord buildUpsertRecord(InternalRow row) {
         }
 
         WriteRecord buildDeleteRecord(InternalRow row) {
+            AbstractTableWriter.checkNotNullConstraints(row, tableInfo, primaryKeyIndexes);
             byte[] key = primaryKeyEncoder.encodeKey(row);
             return WriteRecord.forDelete(
                     tableInfo,

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

@@ -45,6 +45,7 @@ class UpsertWriterImpl extends AbstractTableWriter implements UpsertWriter {
     private final TableInfo tableInfo;
     private final KeyEncoder primaryKeyEncoder;
     private final @Nullable int[] targetColumns;
+    private final int[] primaryKeyIndexes;
 
     // same to primaryKeyEncoder if the bucket key is the same to the primary key
     private final KeyEncoder bucketKeyEncoder;
@@ -80,6 +81,7 @@ class UpsertWriterImpl extends AbstractTableWriter implements UpsertWriter {
                 partialUpdateColumns);
 
         this.targetColumns = partialUpdateColumns;
+        this.primaryKeyIndexes = tableInfo.getSchema().getPrimaryKeyIndexes();
         // encode primary key using physical primary key
         this.primaryKeyEncoder =
                 KeyEncoder.ofPrimaryKeyEncoder(
@@ -176,6 +178,7 @@ private static void sanityCheck(
     @Override
     public CompletableFuture<UpsertResult> upsert(InternalRow row) {
         checkFieldCount(row);
+        checkNotNullConstraints(row, tableInfo, targetColumns);
         byte[] key = primaryKeyEncoder.encodeKey(row);
         byte[] bucketKey =
                 bucketKeyEncoder == primaryKeyEncoder ? key : bucketKeyEncoder.encodeKey(row);
@@ -202,6 +205,7 @@ public CompletableFuture<UpsertResult> upsert(InternalRow row) {
     @Override
     public CompletableFuture<DeleteResult> delete(InternalRow row) {
         checkFieldCount(row);
+        checkNotNullConstraints(row, tableInfo, primaryKeyIndexes);
         byte[] key = primaryKeyEncoder.encodeKey(row);
         byte[] bucketKey =
                 bucketKeyEncoder == primaryKeyEncoder ? key : bucketKeyEncoder.encodeKey(row);