Enable parallel reads for OpenSearch Serverless via PIT + Slice

Remove the validation that rejected opensearch.input.max.docs.per.partition
in serverless mode. When this setting is provided, the connector now creates
multiple sliced partitions per index using PIT + Slice, enabling parallel
reads across Spark tasks.

This leverages the sliced search support introduced in next-generation
OpenSearch Serverless. On Classic Serverless (which does not support the
Slice API), using this setting will result in a server-side error.

Changes:
- RestService.findServerlessPartitions(): count docs and create sliced
  partitions when maxDocsPerPartition is set
- SearchRequestBuilder.assembleSearchAfterBody(): include slice parameter
  in search_after request body
- ServerlessModeTest: replace rejection test with parallel partition tests
- USER_GUIDE.md: add Serverless documentation section

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

CHANGELOG.md

@@ -21,6 +21,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Add Apache Spark 4.0 support ([#684](https://github.com/opensearch-project/opensearch-hadoop/pull/684))
 - Add dedicated `opensearch.search_after.size` setting for serverless mode page size ([#695](https://github.com/opensearch-project/opensearch-hadoop/pull/695))
 - Add Apache Spark 3.5 support with dedicated opensearch-spark-35 module ([#717](https://github.com/opensearch-project/opensearch-hadoop/pull/717))
+- Add parallel read support for next-generation OpenSearch Serverless via PIT + Slice ([#783](https://github.com/opensearch-project/opensearch-hadoop/pull/783))
 
 ### Changed
 - Switched to more reliable OpenSearch Lucene snapshot location ([#597](https://github.com/opensearch-project/opensearch-hadoop/pull/597))

USER_GUIDE.md

@@ -471,3 +471,34 @@ TBLPROPERTIES(
     'opensearch.aws.sigv4.enabled' = 'true',
     'opensearch.aws.sigv4.region' = 'us-east-1');
 ```
+
+## Amazon OpenSearch Serverless
+
+The connector supports Amazon OpenSearch Serverless (both Classic and NextGen architectures). Serverless collections do not expose shard information, so the connector uses PIT (Point in Time) with search_after for pagination instead of the scroll API.
+
+### Basic Configuration
+
+```
+opensearch.serverless=true
+opensearch.nodes=https://<collection-id>.aoss.<region>.on.aws
+opensearch.port=443
+opensearch.net.ssl=true
+opensearch.nodes.wan.only=true
+opensearch.aws.sigv4.enabled=true
+opensearch.aws.sigv4.region=<region>
+opensearch.aws.sigv4.service.name=aoss
+```
+
+### Parallel Reads (NextGen Only)
+
+By default, serverless mode reads each index serially in a single partition. The next-generation OpenSearch Serverless architecture supports PIT with sliced search, which enables parallel reads across multiple Spark tasks.
+
+To enable parallel reads, set `opensearch.input.max.docs.per.partition`:
+
+```
+opensearch.input.max.docs.per.partition=100000
+```
+
+The connector will count the documents in each index, divide by this value to determine the number of slices, and create one Spark partition per slice. For example, an index with 1,000,000 documents and `max.docs.per.partition=100000` will produce 10 parallel read tasks.
+
+This setting requires the next-generation OpenSearch Serverless architecture. Using it with Classic Serverless collections will result in a server-side error because Classic does not support the Slice API.

mr/src/main/java/org/opensearch/hadoop/rest/RestService.java

@@ -229,15 +229,11 @@ public static List<PartitionDefinition> findPartitions(Settings settings, Log lo
 
     /**
      * Create partitions for OpenSearch Serverless mode, which doesn't support shard APIs.
-     * This function creates a single partition for each index as serverless doesn't expose shard information.
+     * When maxDocsPerPartition is set, creates multiple sliced partitions per index for parallel reads.
+     * This requires PIT + Slice support (available on next-generation OpenSearch Serverless).
+     * On Classic Serverless, using maxDocsPerPartition will result in a server-side error.
      */
     static List<PartitionDefinition> findServerlessPartitions(RestRepository client, Settings settings, MappingSet mappingSet, Log log) {
-        if (settings.getMaxDocsPerPartition() != null) {
-            throw new OpenSearchHadoopIllegalArgumentException(
-                "maxDocsPerPartition setting is not supported in OpenSearch Serverless mode. " +
-                "Serverless does not support Slice API which is required for parallel partition reads.");
-        }
-
         Resource readResource = new Resource(settings, true);
         Mapping resolvedMapping = mappingSet == null ? null : mappingSet.getResolvedView();
         PartitionDefinition.PartitionDefinitionBuilder partitionBuilder = PartitionDefinition.builder(settings, resolvedMapping);
@@ -247,9 +243,22 @@ static List<PartitionDefinition> findServerlessPartitions(RestRepository client,
         List<PartitionDefinition> partitions = new ArrayList<PartitionDefinition>();
         String[] indices = readResource.index().split(",");
 
+        Integer maxDocsPerPartition = settings.getMaxDocsPerPartition();
+
         for (String indexName : indices) {
             indexName = indexName.trim();
-            partitions.add(partitionBuilder.build(indexName, 0, new String[0]));
+            if (maxDocsPerPartition != null) {
+                QueryBuilder query = QueryUtils.parseQueryAndFilters(settings);
+                long numDocs = client.getRestClient().count(indexName, query);
+                int numPartitions = (int) Math.max(1, numDocs / maxDocsPerPartition);
+                log.info(String.format("Serverless parallel read: index [%s] has [%d] docs, creating [%d] sliced partitions", indexName, numDocs, numPartitions));
+                for (int i = 0; i < numPartitions; i++) {
+                    PartitionDefinition.Slice slice = new PartitionDefinition.Slice(i, numPartitions);
+                    partitions.add(partitionBuilder.build(indexName, 0, slice, new String[0]));
+                }
+            } else {
+                partitions.add(partitionBuilder.build(indexName, 0, new String[0]));
+            }
         }
 
         return partitions;

mr/src/main/java/org/opensearch/hadoop/rest/SearchRequestBuilder.java

@@ -377,6 +377,15 @@ private BytesArray assembleSearchAfterBody() {
         JacksonJsonGenerator generator = new JacksonJsonGenerator(out);
         try {
             generator.writeBeginObject();
+            if (slice != null && slice.max > 1) {
+                generator.writeFieldName("slice");
+                generator.writeBeginObject();
+                generator.writeFieldName("id");
+                generator.writeNumber(slice.id);
+                generator.writeFieldName("max");
+                generator.writeNumber(slice.max);
+                generator.writeEndObject();
+            }
             generator.writeFieldName("query");
             generator.writeBeginObject();
             root.toJson(generator);

mr/src/test/java/org/opensearch/hadoop/rest/ServerlessModeTest.java

@@ -14,7 +14,6 @@
 import org.apache.commons.logging.impl.NoOpLog;
 import org.junit.Test;
 import org.mockito.Mockito;
-import org.opensearch.hadoop.OpenSearchHadoopIllegalArgumentException;
 import org.opensearch.hadoop.cfg.ConfigurationOptions;
 import org.opensearch.hadoop.cfg.PropertiesSettings;
 import org.opensearch.hadoop.cfg.Settings;
@@ -79,14 +78,54 @@ public void testFindServerlessPartitionsMultipleIndices() {
         assertEquals("index3", partitions.get(2).getIndex());
     }
 
-    @Test(expected = OpenSearchHadoopIllegalArgumentException.class)
-    public void testFindServerlessPartitionsRejectsMaxDocsPerPartition() {
+    @Test
+    public void testFindServerlessPartitionsWithMaxDocsPerPartition() {
         Settings settings = new PropertiesSettings();
         settings.setProperty(ConfigurationOptions.OPENSEARCH_RESOURCE_READ, "test-index");
         settings.setServerlessMode(true);
+        settings.setMaxDocsPerPartition(500);
+
+        RestClient restClient = Mockito.mock(RestClient.class);
+        Mockito.when(restClient.count(Mockito.eq("test-index"), Mockito.any()))
+                .thenReturn(2000L);
+
+        RestRepository repository = Mockito.mock(RestRepository.class);
+        Mockito.when(repository.getRestClient()).thenReturn(restClient);
+
+        List<PartitionDefinition> partitions = RestService.findServerlessPartitions(repository, settings, null, LOGGER);
+
+        assertEquals(4, partitions.size());
+        for (int i = 0; i < 4; i++) {
+            assertEquals("test-index", partitions.get(i).getIndex());
+            assertNotNull(partitions.get(i).getSlice());
+            assertEquals(i, partitions.get(i).getSlice().id);
+            assertEquals(4, partitions.get(i).getSlice().max);
+        }
+    }
+
+    @Test
+    public void testFindServerlessPartitionsWithMaxDocsPerPartitionMultipleIndices() {
+        Settings settings = new PropertiesSettings();
+        settings.setProperty(ConfigurationOptions.OPENSEARCH_RESOURCE_READ, "index1,index2");
+        settings.setServerlessMode(true);
         settings.setMaxDocsPerPartition(1000);
 
-        RestService.findServerlessPartitions(null, settings, null, LOGGER);
+        RestClient restClient = Mockito.mock(RestClient.class);
+        Mockito.when(restClient.count(Mockito.eq("index1"), Mockito.any()))
+                .thenReturn(3000L);
+        Mockito.when(restClient.count(Mockito.eq("index2"), Mockito.any()))
+                .thenReturn(1500L);
+
+        RestRepository repository = Mockito.mock(RestRepository.class);
+        Mockito.when(repository.getRestClient()).thenReturn(restClient);
+
+        List<PartitionDefinition> partitions = RestService.findServerlessPartitions(repository, settings, null, LOGGER);
+
+        // index1: 3000/1000 = 3 partitions, index2: 1500/1000 = 1 partition
+        assertEquals(4, partitions.size());
+        assertEquals("index1", partitions.get(0).getIndex());
+        assertEquals("index1", partitions.get(2).getIndex());
+        assertEquals("index2", partitions.get(3).getIndex());
     }
 
     @Test