Fix NPE that occurs with DELETE w/o clustering key

core/src/main/java/com/scalar/db/transaction/consensuscommit/CommitMutationComposer.java

@@ -9,6 +9,7 @@
 import com.scalar.db.api.ConditionBuilder;
 import com.scalar.db.api.Consistency;
 import com.scalar.db.api.Delete;
+import com.scalar.db.api.DeleteBuilder;
 import com.scalar.db.api.Mutation;
 import com.scalar.db.api.Operation;
 import com.scalar.db.api.Put;
@@ -114,17 +115,20 @@ private Put composePut(Operation base, @Nullable TransactionResult result)
 
   private Delete composeDelete(Operation base, @Nullable TransactionResult result)
       throws ExecutionException {
-    return Delete.newBuilder()
-        .namespace(base.forNamespace().get())
-        .table(base.forTable().get())
-        .partitionKey(getPartitionKey(base, result))
-        .clusteringKey(getClusteringKey(base, result).orElse(null))
-        .consistency(Consistency.LINEARIZABLE)
-        .condition(
-            ConditionBuilder.deleteIf(ConditionBuilder.column(ID).isEqualToText(id))
-                .and(ConditionBuilder.column(STATE).isEqualToInt(TransactionState.DELETED.get()))
-                .build())
-        .build();
+    DeleteBuilder.Buildable deleteBuilder =
+        Delete.newBuilder()
+            .namespace(base.forNamespace().get())
+            .table(base.forTable().get())
+            .partitionKey(getPartitionKey(base, result))
+            .consistency(Consistency.LINEARIZABLE)
+            .condition(
+                ConditionBuilder.deleteIf(ConditionBuilder.column(ID).isEqualToText(id))
+                    .and(
+                        ConditionBuilder.column(STATE).isEqualToInt(TransactionState.DELETED.get()))
+                    .build());
+    getClusteringKey(base, result).ifPresent(deleteBuilder::clusteringKey);
+
+    return deleteBuilder.build();
   }
 
   private Key getPartitionKey(Operation base, @Nullable TransactionResult result)

core/src/test/java/com/scalar/db/transaction/consensuscommit/CommitMutationComposerTest.java

@@ -118,6 +118,25 @@ private Delete prepareDelete() {
         .build();
   }
 
+  private Put preparePutWithoutCk() {
+    Key partitionKey = Key.ofText(ANY_NAME_1, ANY_TEXT_1);
+    return Put.newBuilder()
+        .namespace(ANY_NAMESPACE_NAME)
+        .table(ANY_TABLE_NAME)
+        .partitionKey(partitionKey)
+        .intValue(ANY_NAME_3, ANY_INT_1)
+        .build();
+  }
+
+  private Delete prepareDeleteWithoutCk() {
+    Key partitionKey = Key.ofText(ANY_NAME_1, ANY_TEXT_1);
+    return Delete.newBuilder()
+        .namespace(ANY_NAMESPACE_NAME)
+        .table(ANY_TABLE_NAME)
+        .partitionKey(partitionKey)
+        .build();
+  }
+
   private Get prepareGet() {
     Key partitionKey = Key.ofText(ANY_NAME_1, ANY_TEXT_1);
     Key clusteringKey = Key.ofText(ANY_NAME_2, ANY_TEXT_2);
@@ -220,6 +239,52 @@ public void add_PutAndNullResultGiven_ShouldComposePutWithPutIfCondition()
     assertThat(actual).isEqualTo(expected);
   }
 
+  @Test
+  public void add_PutWithoutCkAndPreparedResultGiven_ShouldComposePutWithPutIfCondition()
+      throws ExecutionException {
+    // Arrange
+    Put put = preparePutWithoutCk();
+    TransactionResult result = prepareResult(TransactionState.PREPARED);
+
+    // Act
+    composer.add(put, result);
+
+    // Assert
+    Put actual = (Put) composer.get().get(0);
+    Put expected =
+        Put.newBuilder()
+            .namespace(put.forNamespace().get())
+            .table(put.forTable().get())
+            .partitionKey(put.getPartitionKey())
+            .consistency(Consistency.LINEARIZABLE)
+            .condition(
+                ConditionBuilder.putIf(ConditionBuilder.column(ID).isEqualToText(ANY_ID))
+                    .and(
+                        ConditionBuilder.column(STATE)
+                            .isEqualToInt(TransactionState.PREPARED.get()))
+                    .build())
+            .bigIntValue(COMMITTED_AT, ANY_TIME_2)
+            .intValue(STATE, TransactionState.COMMITTED.get())
+            .textValue(BEFORE_ID, null)
+            .intValue(BEFORE_STATE, null)
+            .intValue(BEFORE_VERSION, null)
+            .bigIntValue(BEFORE_PREPARED_AT, null)
+            .bigIntValue(BEFORE_COMMITTED_AT, null)
+            .intValue(BEFORE_PREFIX + ANY_NAME_3, null)
+            .booleanValue(BEFORE_PREFIX + ANY_NAME_4, null)
+            .bigIntValue(BEFORE_PREFIX + ANY_NAME_5, null)
+            .floatValue(BEFORE_PREFIX + ANY_NAME_6, null)
+            .doubleValue(BEFORE_PREFIX + ANY_NAME_7, null)
+            .textValue(BEFORE_PREFIX + ANY_NAME_8, null)
+            .blobValue(BEFORE_PREFIX + ANY_NAME_9, (byte[]) null)
+            .dateValue(BEFORE_PREFIX + ANY_NAME_10, null)
+            .timeValue(BEFORE_PREFIX + ANY_NAME_11, null)
+            .timestampValue(BEFORE_PREFIX + ANY_NAME_12, null)
+            .timestampTZValue(BEFORE_PREFIX + ANY_NAME_13, null)
+            .build();
+    assertThat(actual).isEqualTo(expected);
+  }
+
   @Test
   public void add_DeleteAndDeletedResultGiven_ShouldComposeDeleteWithDeleteIfCondition()
       throws ExecutionException {
@@ -267,6 +332,30 @@ public void add_DeleteAndNullResultGiven_ShouldComposeDeleteWithDeleteIfConditio
     assertThat(actual).isEqualTo(delete);
   }
 
+  @Test
+  public void add_DeleteWithoutCkAndDeletedResultGiven_ShouldComposeDeleteWithDeleteIfCondition()
+      throws ExecutionException {
+    // Arrange
+    Delete delete = prepareDeleteWithoutCk();
+    TransactionResult result = prepareResult(TransactionState.DELETED);
+
+    // Act
+    composer.add(delete, result);
+
+    // Assert
+    Delete actual = (Delete) composer.get().get(0);
+    delete =
+        Delete.newBuilder(delete)
+            .consistency(Consistency.LINEARIZABLE)
+            .condition(
+                ConditionBuilder.deleteIf(ConditionBuilder.column(ID).isEqualToText(ANY_ID))
+                    .and(
+                        ConditionBuilder.column(STATE).isEqualToInt(TransactionState.DELETED.get()))
+                    .build())
+            .build();
+    assertThat(actual).isEqualTo(delete);
+  }
+
   @Test
   public void add_SelectionAndPreparedResultGiven_ShouldComposePutForRollforward()
       throws ExecutionException {