[server] Add logic to protect delete a not exists record for kv table (#31)

Author: swuferhong

fluss-connectors/fluss-connector-flink/src/test/java/com/alibaba/fluss/connector/flink/sink/FlinkTableSinkITCase.java

@@ -458,6 +458,9 @@ void testDeleteAndUpdateStmtOnPkTable() throws Exception {
                                 + " primary key (a) not enforced"
                                 + ")",
                         tableName));
+        // test delete without data.
+        tBatchEnv.executeSql("DELETE FROM " + tableName + " WHERE a = 5").await();
+
         List<String> insertValues =
                 Arrays.asList(
                         "(1, 3501, 'Beijing')",

fluss-server/src/main/java/com/alibaba/fluss/server/kv/KvTablet.java

@@ -260,6 +260,7 @@ public LogAppendInfo putAsLeader(
                         ValueDecoder valueDecoder =
                                 new ValueDecoder(readContext.getRowDecoder(schemaId));
 
+                        int appendedRecordCount = 0;
                         for (KvRecord kvRecord : kvRecords.records(readContext)) {
                             byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
                             KvPreWriteBuffer.Key key = KvPreWriteBuffer.Key.of(keyBytes);
@@ -277,11 +278,13 @@ public LogAppendInfo putAsLeader(
                                     // if newRow is null, it means the row should be deleted
                                     if (newRow == null) {
                                         walBuilder.append(RowKind.DELETE, oldRow);
+                                        appendedRecordCount += 1;
                                         kvPreWriteBuffer.delete(key, logOffset++);
                                     } else {
                                         // otherwise, it's a partial update, should produce -U,+U
                                         walBuilder.append(RowKind.UPDATE_BEFORE, oldRow);
                                         walBuilder.append(RowKind.UPDATE_AFTER, newRow);
+                                        appendedRecordCount += 2;
                                         kvPreWriteBuffer.put(
                                                 key,
                                                 ValueEncoder.encodeValue(schemaId, newRow),
@@ -299,6 +302,7 @@ public LogAppendInfo putAsLeader(
                                             updateRow(oldRow, kvRecord.getRow(), partialUpdater);
                                     walBuilder.append(RowKind.UPDATE_BEFORE, oldRow);
                                     walBuilder.append(RowKind.UPDATE_AFTER, newRow);
+                                    appendedRecordCount += 2;
                                     // logOffset is for -U, logOffset + 1 is for +U, we need to use
                                     // the log offset for +U
                                     kvPreWriteBuffer.put(
@@ -312,6 +316,7 @@ public LogAppendInfo putAsLeader(
                                     //  of the input row are set to null.
                                     BinaryRow newRow = kvRecord.getRow();
                                     walBuilder.append(RowKind.INSERT, newRow);
+                                    appendedRecordCount += 1;
                                     kvPreWriteBuffer.put(
                                             key,
                                             ValueEncoder.encodeValue(schemaId, newRow),
@@ -320,8 +325,21 @@ public LogAppendInfo putAsLeader(
                             }
                         }
 
-                        // now, we can build the full log
-                        return logTablet.appendAsLeader(walBuilder.build());
+                        // if appendedRecordCount is 0, it means there is no record to append, we
+                        // should not append.
+                        if (appendedRecordCount > 0) {
+                            // now, we can build the full log.
+                            return logTablet.appendAsLeader(walBuilder.build());
+                        } else {
+                            return new LogAppendInfo(
+                                    logEndOffsetOfPrevBatch - 1,
+                                    logEndOffsetOfPrevBatch - 1,
+                                    0L,
+                                    0L,
+                                    0,
+                                    0,
+                                    false);
+                        }
                     } catch (Throwable t) {
                         // While encounter error here, the CDC logs may fail writing to disk,
                         // and the client probably will resend the batch. If we do not remove the