Fix based on feedback

Author: brfrn169

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

@@ -159,7 +159,10 @@ public void check(Scan scan, TransactionContext context) throws ExecutionExcepti
         for (Selection.Conjunction conjunction : scan.getConjunctions()) {
           for (ConditionalExpression condition : conjunction.getConditions()) {
             String column = condition.getColumn().getName();
-            if (metadata.getSecondaryIndexNames().contains(column)) {
+            // If the column is an indexed column but is part of the primary key, it's allowed
+            if (metadata.getSecondaryIndexNames().contains(column)
+                && !tableMetadata.getPartitionKeyNames().contains(column)
+                && !tableMetadata.getClusteringKeyNames().contains(column)) {
               throw new IllegalArgumentException(
                   CoreError
                       .CONSENSUS_COMMIT_CONDITION_ON_INDEXED_COLUMNS_NOT_ALLOWED_IN_CROSS_PARTITION_SCAN_IN_SERIALIZABLE

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

@@ -495,6 +495,12 @@ public void checkForScan_WithIndexKeyUsingClusteringKeyInSerializable_ShouldNotT
     Set<String> secondaryIndexNames = new LinkedHashSet<>(Collections.singletonList("idx_col"));
     when(tableMetadata.getSecondaryIndexNames()).thenReturn(secondaryIndexNames);
 
+    // Mock the underlying TableMetadata
+    TableMetadata mockTableMetadata = mock(TableMetadata.class);
+    when(tableMetadata.getTableMetadata()).thenReturn(mockTableMetadata);
+    when(mockTableMetadata.getPartitionKeyNames()).thenReturn(new LinkedHashSet<>());
+    when(mockTableMetadata.getClusteringKeyNames()).thenReturn(new LinkedHashSet<>());
+
     Scan scan =
         ScanAll.newBuilder()
             .namespace("ns")
@@ -517,6 +523,12 @@ public void checkForScan_WithIndexKeyUsingClusteringKeyInSerializable_ShouldNotT
     Set<String> secondaryIndexNames = new LinkedHashSet<>(Collections.singletonList("idx_col"));
     when(tableMetadata.getSecondaryIndexNames()).thenReturn(secondaryIndexNames);
 
+    // Mock the underlying TableMetadata
+    TableMetadata mockTableMetadata = mock(TableMetadata.class);
+    when(tableMetadata.getTableMetadata()).thenReturn(mockTableMetadata);
+    when(mockTableMetadata.getPartitionKeyNames()).thenReturn(new LinkedHashSet<>());
+    when(mockTableMetadata.getClusteringKeyNames()).thenReturn(new LinkedHashSet<>());
+
     Scan scan =
         ScanAll.newBuilder()
             .namespace("ns")
@@ -531,6 +543,64 @@ public void checkForScan_WithIndexKeyUsingClusteringKeyInSerializable_ShouldNotT
     assertThatCode(() -> checker.check(scan, context)).doesNotThrowAnyException();
   }
 
+  @Test
+  public void
+      checkForScan_ScanAllWithConditionOnIndexedPartitionKeyColumnInSerializable_ShouldNotThrowException() {
+    // Arrange
+    Set<String> secondaryIndexNames = new LinkedHashSet<>(Collections.singletonList("idx_col"));
+    when(tableMetadata.getSecondaryIndexNames()).thenReturn(secondaryIndexNames);
+
+    // Mock the underlying TableMetadata - indexed column is also a partition key
+    TableMetadata mockTableMetadata = mock(TableMetadata.class);
+    when(tableMetadata.getTableMetadata()).thenReturn(mockTableMetadata);
+    LinkedHashSet<String> partitionKeys = new LinkedHashSet<>();
+    partitionKeys.add("idx_col");
+    when(mockTableMetadata.getPartitionKeyNames()).thenReturn(partitionKeys);
+    when(mockTableMetadata.getClusteringKeyNames()).thenReturn(new LinkedHashSet<>());
+
+    Scan scan =
+        ScanAll.newBuilder()
+            .namespace("ns")
+            .table("tbl")
+            .all()
+            .where(ConditionBuilder.column("idx_col").isEqualToInt(100))
+            .build();
+    TransactionContext context =
+        new TransactionContext("txId", null, Isolation.SERIALIZABLE, false, false);
+
+    // Act Assert
+    assertThatCode(() -> checker.check(scan, context)).doesNotThrowAnyException();
+  }
+
+  @Test
+  public void
+      checkForScan_ScanAllWithConditionOnIndexedClusteringKeyColumnInSerializable_ShouldNotThrowException() {
+    // Arrange
+    Set<String> secondaryIndexNames = new LinkedHashSet<>(Collections.singletonList("idx_col"));
+    when(tableMetadata.getSecondaryIndexNames()).thenReturn(secondaryIndexNames);
+
+    // Mock the underlying TableMetadata - indexed column is also a clustering key
+    TableMetadata mockTableMetadata = mock(TableMetadata.class);
+    when(tableMetadata.getTableMetadata()).thenReturn(mockTableMetadata);
+    when(mockTableMetadata.getPartitionKeyNames()).thenReturn(new LinkedHashSet<>());
+    LinkedHashSet<String> clusteringKeys = new LinkedHashSet<>();
+    clusteringKeys.add("idx_col");
+    when(mockTableMetadata.getClusteringKeyNames()).thenReturn(clusteringKeys);
+
+    Scan scan =
+        ScanAll.newBuilder()
+            .namespace("ns")
+            .table("tbl")
+            .all()
+            .where(ConditionBuilder.column("idx_col").isEqualToInt(100))
+            .build();
+    TransactionContext context =
+        new TransactionContext("txId", null, Isolation.SERIALIZABLE, false, false);
+
+    // Act Assert
+    assertThatCode(() -> checker.check(scan, context)).doesNotThrowAnyException();
+  }
+
   @Test
   public void
       checkForScan_RegularScanWithConditionOnIndexedColumnInSerializable_ShouldNotThrowException() {