Default to parallel reads in serverless mode (50,000 docs/partition)

Instead of requiring explicit opensearch.input.max.docs.per.partition
to enable parallel reads, serverless mode now automatically splits
reads using a default of 50,000 documents per partition. Users can
still override this value. Also removes incorrect documentation about
Classic Serverless not supporting the Slice API.

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

Author: lawofcycles

USER_GUIDE.md

@@ -489,16 +489,14 @@ opensearch.aws.sigv4.region=<region>
 opensearch.aws.sigv4.service.name=aoss
 ```
 
-### Parallel Reads (NextGen Only)
+### Parallel Reads
 
-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.
+Serverless mode automatically splits reads into parallel partitions using PIT + Slice. By default, the connector creates one partition per 50,000 documents in each index.
 
-To enable parallel reads, set `opensearch.input.max.docs.per.partition`:
+To customize the partition size, 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/cfg/ConfigurationOptions.java

@@ -164,6 +164,7 @@ public interface ConfigurationOptions {
 
     /** Input options **/
     String OPENSEARCH_MAX_DOCS_PER_PARTITION = "opensearch.input.max.docs.per.partition";
+    String OPENSEARCH_MAX_DOCS_PER_PARTITION_SERVERLESS_DEFAULT = "50000";
 
     String OPENSEARCH_INPUT_JSON = "opensearch.input.json";
     String OPENSEARCH_INPUT_JSON_DEFAULT = "no";

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

@@ -229,9 +229,8 @@ public static List<PartitionDefinition> findPartitions(Settings settings, Log lo
 
     /**
      * Create partitions for OpenSearch Serverless mode, which doesn't support shard APIs.
-     * 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.
+     * When maxDocsPerPartition is set, creates multiple sliced partitions per index for parallel reads
+     * using PIT + Slice.
      */
     static List<PartitionDefinition> findServerlessPartitions(RestRepository client, Settings settings, MappingSet mappingSet, Log log) {
         Resource readResource = new Resource(settings, true);
@@ -244,13 +243,16 @@ static List<PartitionDefinition> findServerlessPartitions(RestRepository client,
         String[] indices = readResource.index().split(",");
 
         Integer maxDocsPerPartition = settings.getMaxDocsPerPartition();
+        if (maxDocsPerPartition == null) {
+            maxDocsPerPartition = Integer.parseInt(ConfigurationOptions.OPENSEARCH_MAX_DOCS_PER_PARTITION_SERVERLESS_DEFAULT);
+        }
 
         for (String indexName : indices) {
             indexName = indexName.trim();
-            if (maxDocsPerPartition != null) {
-                QueryBuilder query = QueryUtils.parseQueryAndFilters(settings);
-                long numDocs = client.getRestClient().count(indexName, query);
-                int numPartitions = (int) Math.max(1, numDocs / maxDocsPerPartition);
+            QueryBuilder query = QueryUtils.parseQueryAndFilters(settings);
+            long numDocs = client.getRestClient().count(indexName, query);
+            int numPartitions = (int) Math.max(1, numDocs / maxDocsPerPartition);
+            if (numPartitions > 1) {
                 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);

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

@@ -57,8 +57,16 @@ public void testFindServerlessPartitionsSingleIndex() {
         settings.setProperty(ConfigurationOptions.OPENSEARCH_RESOURCE_READ, "test-index");
         settings.setServerlessMode(true);
 
-        List<PartitionDefinition> partitions = RestService.findServerlessPartitions(null, settings, null, LOGGER);
+        RestClient restClient = Mockito.mock(RestClient.class);
+        Mockito.when(restClient.count(Mockito.eq("test-index"), Mockito.any()))
+                .thenReturn(100L);
+
+        RestRepository repository = Mockito.mock(RestRepository.class);
+        Mockito.when(repository.getRestClient()).thenReturn(restClient);
+
+        List<PartitionDefinition> partitions = RestService.findServerlessPartitions(repository, settings, null, LOGGER);
 
+        // 100 docs / 50000 default = 1 partition (no slicing needed)
         assertEquals(1, partitions.size());
         assertEquals("test-index", partitions.get(0).getIndex());
         assertEquals(0, partitions.get(0).getShardId());
@@ -70,7 +78,14 @@ public void testFindServerlessPartitionsMultipleIndices() {
         settings.setProperty(ConfigurationOptions.OPENSEARCH_RESOURCE_READ, "index1,index2,index3");
         settings.setServerlessMode(true);
 
-        List<PartitionDefinition> partitions = RestService.findServerlessPartitions(null, settings, null, LOGGER);
+        RestClient restClient = Mockito.mock(RestClient.class);
+        Mockito.when(restClient.count(Mockito.anyString(), Mockito.any()))
+                .thenReturn(100L);
+
+        RestRepository repository = Mockito.mock(RestRepository.class);
+        Mockito.when(repository.getRestClient()).thenReturn(restClient);
+
+        List<PartitionDefinition> partitions = RestService.findServerlessPartitions(repository, settings, null, LOGGER);
 
         assertEquals(3, partitions.size());
         assertEquals("index1", partitions.get(0).getIndex());