PHOENIX-7738 Tests with random filter expressions and segment scan

.asf.yaml

@@ -40,6 +40,8 @@ github:
     squash: true
     merge: false
     rebase: false
+  autolink_jira:
+    - PHOENIX
 notifications:
   commits: commits@phoenix.apache.org
   issues: issues@phoenix.apache.org

phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/ConcurrentDifferentSegmentCountIT.java

@@ -69,8 +69,10 @@ private void runConcurrentDifferentSegmentCountTest(String hashKeyName, ScalarAt
                                 TOTAL_ITEMS, SPLIT_FREQUENCY, 75);
 
         // Perform full scan for comparison baseline
-        List<Map<String, AttributeValue>> fullScanItemsPhoenix = performFullScanWithPagination(phoenixDBClientV2, tableName, false);
-        List<Map<String, AttributeValue>> fullScanItemsDDB = performFullScanWithPagination(dynamoDbClient, tableName, false);
+        List<Map<String, AttributeValue>> fullScanItemsPhoenix =
+                performFullScanWithPagination(phoenixDBClientV2, tableName, false, 0);
+        List<Map<String, AttributeValue>> fullScanItemsDDB =
+                performFullScanWithPagination(dynamoDbClient, tableName, false, 0);
 
         // Execute concurrent segment scans with different totalSegments
         List<Map<String, AttributeValue>> segmentScan3Items = performConcurrentSegmentScansWithDifferentCounts(tableName);
@@ -141,8 +143,9 @@ private List<Map<String, AttributeValue>> performAllSegmentScans(String tableNam
 
         // Scan each segment sequentially within this thread
         for (int segment = 0; segment < totalSegments; segment++) {
-            List<Map<String, AttributeValue>> segmentItems = 
-                scanSingleSegmentWithPagination(tableName, segment, totalSegments, SCAN_LIMIT, false, false);
+            List<Map<String, AttributeValue>> segmentItems =
+                    scanSingleSegmentWithPagination(tableName, segment, totalSegments, SCAN_LIMIT,
+                            false, false, 0);
             allItems.addAll(segmentItems);
             LOGGER.info("Segment {}/{} scan found {} items", segment, totalSegments, segmentItems.size());
         }

phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/ConcurrentSegmentScanIT.java

@@ -70,8 +70,10 @@ private void runConcurrentSegmentScanTest(String hashKeyName, ScalarAttributeTyp
                                 TOTAL_ITEMS, SPLIT_FREQUENCY, 50);
 
         // Perform full scan for comparison
-        List<Map<String, AttributeValue>> fullScanItemsPhoenix = performFullScanWithPagination(phoenixDBClientV2, tableName, false);
-        List<Map<String, AttributeValue>> fullScanItemsDDB = performFullScanWithPagination(dynamoDbClient, tableName, false);
+        List<Map<String, AttributeValue>> fullScanItemsPhoenix =
+                performFullScanWithPagination(phoenixDBClientV2, tableName, false, 0);
+        List<Map<String, AttributeValue>> fullScanItemsDDB =
+                performFullScanWithPagination(dynamoDbClient, tableName, false, 0);
 
         // Execute concurrent segment scans with concurrent splits
         List<Map<String, AttributeValue>> concurrentSegmentItems = 
@@ -138,8 +140,9 @@ private List<Map<String, AttributeValue>> performConcurrentSegmentScansWithSplit
     }
 
     private List<Map<String, AttributeValue>> scanSegmentWithPagination(String tableName, int segment) {
-        List<Map<String, AttributeValue>> segmentItems = 
-            scanSingleSegmentWithPagination(tableName, segment, TOTAL_SEGMENTS, SCAN_LIMIT, true, false);
+        List<Map<String, AttributeValue>> segmentItems =
+                scanSingleSegmentWithPagination(tableName, segment, TOTAL_SEGMENTS, SCAN_LIMIT,
+                        true, false, 0);
         LOGGER.info("Segment {} scan completed with {} items", segment, segmentItems.size());
         return segmentItems;
     }

phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/ScanIndex3IT.java

@@ -40,13 +40,15 @@
 import org.junit.rules.TestName;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import software.amazon.awssdk.services.dynamodb.model.BatchWriteItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.CreateTableRequest;
 import software.amazon.awssdk.services.dynamodb.model.DeleteItemRequest;
-import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.PutRequest;
 import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
 import software.amazon.awssdk.services.dynamodb.model.ScanResponse;
 import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
 import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.WriteRequest;
 
 import static org.apache.phoenix.query.BaseTest.setUpConfigForMiniCluster;
 
@@ -87,6 +89,18 @@ public static void initialize() throws Exception {
         createTableAndInsertData();
     }
 
+    private static void executeBatchWrite(String tableName, List<WriteRequest> batch) {
+        if (batch.isEmpty()) {
+            return;
+        }
+        Map<String, List<WriteRequest>> requestItems = new HashMap<>();
+        requestItems.put(tableName, new ArrayList<>(batch));
+        BatchWriteItemRequest batchRequest =
+                BatchWriteItemRequest.builder().requestItems(requestItems).build();
+        phoenixDBClientV2.batchWriteItem(batchRequest);
+        dynamoDbClient.batchWriteItem(batchRequest);
+    }
+
     private static void createTableAndInsertData() {
         CreateTableRequest createTableRequest = DDLTestUtils.getCreateTableRequest(
                 TABLE_NAME, "pk", ScalarAttributeType.S, "sk", ScalarAttributeType.N);
@@ -96,6 +110,7 @@ private static void createTableAndInsertData() {
         phoenixDBClientV2.createTable(createTableRequest);
         dynamoDbClient.createTable(createTableRequest);
 
+        List<WriteRequest> batch = new ArrayList<>();
         for (int i = 0; i < NUM_RECORDS; i++) {
             Map<String, AttributeValue> item = new HashMap<>();
             item.put("pk", AttributeValue.builder().s("pk_" + (i % 100)).build());
@@ -126,10 +141,15 @@ private static void createTableAndInsertData() {
             nestedList.add(AttributeValue.builder().n(String.valueOf(i % 40)).build());
             nestedList.add(AttributeValue.builder().ss("set_in_list_" + (i % 3), "list_common").build());
             item.put("itms", AttributeValue.builder().l(nestedList).build());
-
-            PutItemRequest putRequest = PutItemRequest.builder().tableName(TABLE_NAME).item(item).build();
-            phoenixDBClientV2.putItem(putRequest);
-            dynamoDbClient.putItem(putRequest);
+
+            WriteRequest writeRequest =
+                    WriteRequest.builder().putRequest(PutRequest.builder().item(item).build())
+                            .build();
+            batch.add(writeRequest);
+            if (batch.size() >= 25 || i == NUM_RECORDS - 1) {
+                executeBatchWrite(TABLE_NAME, batch);
+                batch.clear();
+            }
         }
 
         for (int i = 0; i < NUM_RECORDS; i++) {

phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/SegmentScan1IT.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.ddb;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
+
+/**
+ * Segment scan tests - Part 1.
+ */
+@RunWith(Parameterized.class)
+public class SegmentScan1IT extends SegmentScanIT {
+
+    public SegmentScan1IT(String hashKeyName, ScalarAttributeType hashKeyType, String sortKeyName,
+            ScalarAttributeType sortKeyType, boolean useFilter) {
+        super(hashKeyName, hashKeyType, sortKeyName, sortKeyType, useFilter);
+    }
+
+    @Parameterized.Parameters(name = "Hash_{1}_Sort_{3}_Filter_{4}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+                // Hash key only
+                {"pk", ScalarAttributeType.S, null, null, true},
+                {"pk", ScalarAttributeType.N, null, null, false},
+                {"pk", ScalarAttributeType.B, null, null, true},
+
+                // String Hash Key combinations
+                {"pk", ScalarAttributeType.S, "sk", ScalarAttributeType.S, false},
+                {"pk", ScalarAttributeType.S, "sk", ScalarAttributeType.N, true},
+                {"pk", ScalarAttributeType.S, "sk", ScalarAttributeType.B, false},
+
+                // Number Hash Key combinations
+                {"pk", ScalarAttributeType.N, "sk", ScalarAttributeType.S, true},
+                {"pk", ScalarAttributeType.N, "sk", ScalarAttributeType.N, false},
+                {"pk", ScalarAttributeType.N, "sk", ScalarAttributeType.B, true}
+        });
+    }
+}

phoenix-ddb-rest/src/test/java/org/apache/phoenix/ddb/SegmentScan2IT.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.ddb;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
+
+/**
+ * Segment scan tests - Part 2.
+ */
+@RunWith(Parameterized.class)
+public class SegmentScan2IT extends SegmentScanIT {
+
+    public SegmentScan2IT(String hashKeyName, ScalarAttributeType hashKeyType, String sortKeyName,
+            ScalarAttributeType sortKeyType, boolean useFilter) {
+        super(hashKeyName, hashKeyType, sortKeyName, sortKeyType, useFilter);
+    }
+
+    @Parameterized.Parameters(name = "Hash_{1}_Sort_{3}_Filter_{4}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+                // Binary Hash Key combinations
+                {"pk", ScalarAttributeType.B, "sk", ScalarAttributeType.S, false},
+                {"pk", ScalarAttributeType.B, "sk", ScalarAttributeType.N, true},
+                {"pk", ScalarAttributeType.B, "sk", ScalarAttributeType.B, false},
+
+                // Additional combinations
+                {"pk", ScalarAttributeType.S, null, null, true},
+                {"pk", ScalarAttributeType.N, null, null, false},
+                {"pk", ScalarAttributeType.B, null, null, true},
+                {"pk", ScalarAttributeType.B, "sk", ScalarAttributeType.S, false},
+                {"pk", ScalarAttributeType.S, "sk", ScalarAttributeType.N, true},
+                {"pk", ScalarAttributeType.N, "sk", ScalarAttributeType.B, false}
+        });
+    }
+}