Merge branch 'feature/3.x-lucene' into agg-perf-lucene

Author: jainankitk

CHANGELOG.md

@@ -17,6 +17,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add metrics for the merged segment warmer feature ([#18929](https://github.com/opensearch-project/OpenSearch/pull/18929))
 - Handle deleted documents for filter rewrite sub-aggregation optimization ([#19643](https://github.com/opensearch-project/OpenSearch/pull/19643))
 - Add bulk collect API for filter rewrite sub-aggregation optimization ([#19933](https://github.com/opensearch-project/OpenSearch/pull/19933))
+- Allow collectors take advantage of preaggregated data using collectRange API ([#20009](https://github.com/opensearch-project/OpenSearch/pull/20009))
 - Add pointer based lag metric in pull-based ingestion ([#19635](https://github.com/opensearch-project/OpenSearch/pull/19635))
 - Introduced internal API for retrieving metadata about requested indices from transport actions  ([#18523](https://github.com/opensearch-project/OpenSearch/pull/18523))
 - Add cluster defaults for merge autoThrottle, maxMergeThreads, and maxMergeCount; Add segment size filter to the merged segment warmer ([#19629](https://github.com/opensearch-project/OpenSearch/pull/19629))
@@ -27,6 +28,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add async periodic flush task support for pull-based ingestion ([#19878](https://github.com/opensearch-project/OpenSearch/pull/19878))
 - Add support for context aware segments ([#19098](https://github.com/opensearch-project/OpenSearch/pull/19098))
 - Implement GRPC FunctionScoreQuery ([#19888](https://github.com/opensearch-project/OpenSearch/pull/19888))
+- Support pull-based ingestion message mappers and raw payload support ([#19765](https://github.com/opensearch-project/OpenSearch/pull/19765)]
 
 ### Changed
 - Combining filter rewrite and skip list to optimize sub aggregation([#19573](https://github.com/opensearch-project/OpenSearch/pull/19573))
@@ -55,6 +57,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Thread Context Preservation by gRPC Interceptor ([#19776](https://github.com/opensearch-project/OpenSearch/pull/19776))
 - Update NoOpResult constructors in the Engine to be public ([#19950](https://github.com/opensearch-project/OpenSearch/pull/19950))
 - Refactor the TranslogStats and RequestCacheStats class to use the Builder pattern instead of constructors ([#19961](https://github.com/opensearch-project/OpenSearch/pull/19961))
+- Refactor the IndexPressutreStats, DeviceStats and TransportStats class to use the Builder pattern instead of constructors ([#19991](https://github.com/opensearch-project/OpenSearch/pull/19991))
 
 ### Fixed
 - Fix Allocation and Rebalance Constraints of WeightFunction are incorrectly reset ([#19012](https://github.com/opensearch-project/OpenSearch/pull/19012))
@@ -72,6 +75,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add S3Repository.LEGACY_MD5_CHECKSUM_CALCULATION to list of repository-s3 settings ([#19788](https://github.com/opensearch-project/OpenSearch/pull/19788))
 - Fix NullPointerException when restoring remote snapshot with missing shard size information ([#19684](https://github.com/opensearch-project/OpenSearch/pull/19684))
 - Fix NPE of ScriptScoreQuery ([#19650](https://github.com/opensearch-project/OpenSearch/pull/19650))
+- Fix ClassCastException in FlightClientChannel for requests larger than 16KB ([#20010](https://github.com/opensearch-project/OpenSearch/pull/20010))
 
 ### Dependencies
 - Update to Gradle 9.2 ([#19575](https://github.com/opensearch-project/OpenSearch/pull/19575)) ([#19856](https://github.com/opensearch-project/OpenSearch/pull/19856))
@@ -112,6 +116,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Deprecated existing constructors in GetStats, FlushStats and QueryCacheStats in favor of the new Builder ([#19935](https://github.com/opensearch-project/OpenSearch/pull/19935))
 - Deprecated existing constructors in FieldDataStats and CompletionStats in favor of the new Builder ([#19936](https://github.com/opensearch-project/OpenSearch/pull/19936))
 - Deprecated existing constructors in TranslogStats and RequestCacheStats in favor of the new Builder ([#19961](https://github.com/opensearch-project/OpenSearch/pull/19961))
+- Deprecated existing constructors in IndexPressutreStats, DeviceStats and TransportStats in favor of the new Builder ([#19991](https://github.com/opensearch-project/OpenSearch/pull/19991))
 
 ### Removed
 

README.md

@@ -3,8 +3,8 @@
 </a>
 
 [![License](https://img.shields.io/badge/license-Apache%20v2-blue.svg)](https://github.com/opensearch-project/OpenSearch/blob/main/LICENSE.txt)
-[![LFX Health Score](https://insights.production.lfx.dev/api/badge/health-score?project=opensearch-foundation)](https://insights.linuxfoundation.org/project/opensearch-foundation)
-[![LFX Active Contributors](https://insights.production.lfx.dev/api/badge/active-contributors?project=opensearch-foundation&repos=https://github.com/opensearch-project/OpenSearch)](https://insights.linuxfoundation.org/project/opensearch-foundation/repository/opensearch-project-opensearch)
+[![LFX Health Score](https://insights.linuxfoundation.org/api/badge/health-score?project=opensearch-foundation)](https://insights.linuxfoundation.org/project/opensearch-foundation)
+[![LFX Active Contributors](https://insights.linuxfoundation.org/api/badge/active-contributors?project=opensearch-foundation)](https://insights.linuxfoundation.org/project/opensearch-foundation)
 [![Code Coverage](https://codecov.io/gh/opensearch-project/OpenSearch/branch/main/graph/badge.svg)](https://codecov.io/gh/opensearch-project/OpenSearch)
 ![GitHub release (latest SemVer)](https://img.shields.io/github/v/release/opensearch-project/OpenSearch?sort=semver)
 [![Linkedin](https://img.shields.io/badge/Follow-Linkedin-blue)](https://www.linkedin.com/company/opensearch-project)

buildSrc/src/main/resources/forbidden/opensearch-test-signatures.txt

@@ -26,3 +26,8 @@ com.carrotsearch.randomizedtesting.annotations.Nightly @ We don't run nightly te
 org.junit.Test @defaultMessage Just name your test method testFooBar
 
 java.lang.Math#random() @ Use one of the various randomization methods from LuceneTestCase or OpenSearchTestCase for reproducibility
+
+@defaultMessage Disabling assertions is a JVM-wide setting and will break other tests
+java.lang.ClassLoader#setDefaultAssertionStatus(boolean)
+java.lang.ClassLoader#setPackageAssertionStatus(java.lang.String, boolean)
+java.lang.ClassLoader#setClassAssertionStatus(java.lang.String, boolean)

modules/aggs-matrix-stats/src/main/java/org/opensearch/search/aggregations/matrix/stats/MatrixStatsAggregator.java

@@ -32,6 +32,7 @@
 package org.opensearch.search.aggregations.matrix.stats;
 
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.DocIdStream;
 import org.apache.lucene.search.ScoreMode;
 import org.opensearch.common.lease.Releasables;
 import org.opensearch.common.util.BigArrays;
@@ -113,6 +114,16 @@ public void collect(int doc, long bucket) throws IOException {
                 }
             }
 
+            @Override
+            public void collect(DocIdStream stream, long owningBucketOrd) throws IOException {
+                super.collect(stream, owningBucketOrd);
+            }
+
+            @Override
+            public void collectRange(int min, int max) throws IOException {
+                super.collectRange(min, max);
+            }
+
             /**
              * return a map of field names and data
              */

plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightClientChannel.java

@@ -18,7 +18,6 @@
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.core.action.ActionListener;
-import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.common.transport.BoundTransportAddress;
@@ -35,7 +34,6 @@
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -351,8 +349,7 @@ private void notifyListener(ActionListener<Void> listener, CompletableFuture<Voi
     }
 
     private Ticket serializeToTicket(BytesReference reference) {
-        byte[] data = Arrays.copyOfRange(((BytesArray) reference).array(), 0, reference.length());
-        return new Ticket(data);
+        return new Ticket(BytesReference.toBytes(reference));
     }
 
     @Override

plugins/arrow-flight-rpc/src/test/java/org/opensearch/arrow/flight/transport/FlightClientChannelTests.java

@@ -13,6 +13,7 @@
 import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.transport.TransportResponse;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.StreamTransportResponseHandler;
@@ -588,4 +589,82 @@ public void testSetMessageListenerTwice() {
         IllegalStateException exception = assertThrows(IllegalStateException.class, () -> flightTransport.setMessageListener(listener2));
         assertEquals("Cannot set message listener twice", exception.getMessage());
     }
+
+    static class LargeTestRequest extends TestRequest {
+        private final String largeData;
+
+        LargeTestRequest(String data) {
+            this.largeData = data;
+        }
+
+        LargeTestRequest(StreamInput in) throws IOException {
+            super(in);
+            this.largeData = in.readString();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(largeData);
+        }
+    }
+
+    public void testLargeRequest() throws Exception {
+        String action = "internal:test/large";
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Exception> error = new AtomicReference<>();
+
+        streamTransportService.registerRequestHandler(action, ThreadPool.Names.SAME, LargeTestRequest::new, (request, channel, task) -> {
+            try {
+                channel.sendResponseBatch(new TestResponse("OK"));
+                channel.completeStream();
+            } catch (Exception e) {
+                try {
+                    channel.sendResponse(e);
+                } catch (IOException ex) {}
+            }
+        });
+
+        LargeTestRequest testRequest = new LargeTestRequest("X".repeat(20 * 1024));
+
+        streamTransportService.sendRequest(
+            remoteNode,
+            action,
+            testRequest,
+            TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STREAM).build(),
+            new StreamTransportResponseHandler<TestResponse>() {
+                @Override
+                public void handleStreamResponse(StreamTransportResponse<TestResponse> streamResponse) {
+                    try {
+                        while (streamResponse.nextResponse() != null) {
+                        }
+                        streamResponse.close();
+                    } catch (Exception e) {
+                        error.set(e);
+                    } finally {
+                        latch.countDown();
+                    }
+                }
+
+                @Override
+                public void handleException(TransportException exp) {
+                    error.set(exp);
+                    latch.countDown();
+                }
+
+                @Override
+                public String executor() {
+                    return ThreadPool.Names.SAME;
+                }
+
+                @Override
+                public TestResponse read(StreamInput in) throws IOException {
+                    return new TestResponse(in);
+                }
+            }
+        );
+
+        assertTrue(latch.await(TIMEOUT_SEC, TimeUnit.SECONDS));
+        assertNull(error.get());
+    }
 }

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/IngestFromKafkaIT.java

@@ -778,6 +778,123 @@ public void testAllActiveIngestionPeriodicFlush() throws Exception {
 
         waitForSearchableDocs(10, Arrays.asList(nodeA));
         waitForState(() -> getPeriodicFlushCount(nodeA, indexName) >= 1);
+    }
+
+    public void testRawPayloadMapperIngestion() throws Exception {
+        // Start cluster
+        internalCluster().startClusterManagerOnlyNode();
+        final String nodeA = internalCluster().startDataOnlyNode();
+
+        // Publish 2 valid messages
+        String validMessage1 = "{\"name\":\"alice\",\"age\":30}";
+        String validMessage2 = "{\"name\":\"bob\",\"age\":25}";
+        produceData(validMessage1);
+        produceData(validMessage2);
+
+        // Create index with raw_payload mapper
+        createIndex(
+            indexName,
+            Settings.builder()
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .put("ingestion_source.type", "kafka")
+                .put("ingestion_source.param.topic", topicName)
+                .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
+                .put("ingestion_source.pointer.init.reset", "earliest")
+                .put("ingestion_source.mapper_type", "raw_payload")
+                .put("ingestion_source.error_strategy", "drop")
+                .put("ingestion_source.all_active", true)
+                .build(),
+            "{\"properties\":{\"name\":{\"type\": \"text\"},\"age\":{\"type\": \"integer\"}}}}"
+        );
+
+        ensureGreen(indexName);
+
+        // Wait for both messages to be indexed
+        waitForSearchableDocs(2, List.of(nodeA));
+
+        // Verify stats show 2 processed messages
+        waitForState(() -> {
+            PollingIngestStats stats = client().admin().indices().prepareStats(indexName).get().getIndex(indexName).getShards()[0]
+                .getPollingIngestStats();
+            return stats != null
+                && stats.getMessageProcessorStats().totalProcessedCount() == 2L
+                && stats.getConsumerStats().totalPolledCount() == 2L
+                && stats.getConsumerStats().totalPollerMessageFailureCount() == 0L
+                && stats.getConsumerStats().totalPollerMessageDroppedCount() == 0L
+                && stats.getMessageProcessorStats().totalInvalidMessageCount() == 0L;
+        });
+
+        // Validate document content
+        SearchResponse searchResponse = client().prepareSearch(indexName).get();
+        assertEquals(2, searchResponse.getHits().getHits().length);
+        for (int i = 0; i < searchResponse.getHits().getHits().length; i++) {
+            Map<String, Object> source = searchResponse.getHits().getHits()[i].getSourceAsMap();
+            assertTrue(source.containsKey("name"));
+            assertTrue(source.containsKey("age"));
+        }
+
+        // Publish invalid JSON message
+        String invalidJsonMessage = "{ invalid json";
+        produceData(invalidJsonMessage);
+
+        // Wait for consumer to encounter the error and drop it
+        waitForState(() -> {
+            PollingIngestStats stats = client().admin().indices().prepareStats(indexName).get().getIndex(indexName).getShards()[0]
+                .getPollingIngestStats();
+            return stats != null
+                && stats.getConsumerStats().totalPolledCount() == 3L
+                && stats.getConsumerStats().totalPollerMessageFailureCount() == 1L
+                && stats.getConsumerStats().totalPollerMessageDroppedCount() == 1L
+                && stats.getMessageProcessorStats().totalProcessedCount() == 2L;
+        });
+
+        // Publish message with invalid content that will fail at processor level
+        String invalidFieldTypeMessage = "{\"name\":123,\"age\":\"not a number\"}";
+        produceData(invalidFieldTypeMessage);
+
+        // Wait for processor to encounter the error
+        waitForState(() -> {
+            PollingIngestStats stats = client().admin().indices().prepareStats(indexName).get().getIndex(indexName).getShards()[0]
+                .getPollingIngestStats();
+            return stats != null
+                && stats.getConsumerStats().totalPolledCount() == 4L
+                && stats.getConsumerStats().totalPollerMessageFailureCount() == 1L
+                && stats.getMessageProcessorStats().totalProcessedCount() == 3L
+                && stats.getMessageProcessorStats().totalFailedCount() == 1L
+                && stats.getMessageProcessorStats().totalFailuresDroppedCount() == 1L;
+        });
+
+        // Pause ingestion, reset to offset 0, and resume
+        pauseIngestion(indexName);
+        waitForState(() -> {
+            GetIngestionStateResponse ingestionState = getIngestionState(indexName);
+            return ingestionState.getShardStates().length == 1
+                && ingestionState.getFailedShards() == 0
+                && ingestionState.getShardStates()[0].isPollerPaused()
+                && ingestionState.getShardStates()[0].getPollerState().equalsIgnoreCase("paused");
+        });
+
+        // Resume with reset to offset 0 (will re-process the 2 valid messages)
+        resumeIngestion(indexName, 0, ResumeIngestionRequest.ResetSettings.ResetMode.OFFSET, "0");
+        waitForState(() -> {
+            GetIngestionStateResponse ingestionState = getIngestionState(indexName);
+            return ingestionState.getShardStates().length == 1
+                && ingestionState.getShardStates()[0].isPollerPaused() == false
+                && (ingestionState.getShardStates()[0].getPollerState().equalsIgnoreCase("polling")
+                    || ingestionState.getShardStates()[0].getPollerState().equalsIgnoreCase("processing"));
+        });
 
+        // Wait for the 3 messages to be processed by the processor after reset (1 will be dropped by the poller)
+        waitForState(() -> {
+            PollingIngestStats stats = client().admin().indices().prepareStats(indexName).get().getIndex(indexName).getShards()[0]
+                .getPollingIngestStats();
+            return stats != null && stats.getMessageProcessorStats().totalProcessedCount() == 3L;
+        });
+
+        // Verify still only 2 documents (no duplicates must be indexed)
+        RangeQueryBuilder query = new RangeQueryBuilder("age").gte(0);
+        SearchResponse response = client().prepareSearch(indexName).setQuery(query).get();
+        assertThat(response.getHits().getTotalHits().value(), is(2L));
     }
 }