Merge remote-tracking branch 'origin/main' into merge_parquet

Signed-off-by: Shailesh-Kumar-Singh <shaileshkumarsingh260@gmail.com>

Author: Shailesh-Kumar-Singh

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java

@@ -44,6 +44,7 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
         SUPPORTED_FIELD_TYPES.addAll(FieldType.keyword());
         SUPPORTED_FIELD_TYPES.addAll(FieldType.date());
         SUPPORTED_FIELD_TYPES.add(FieldType.BOOLEAN);
+        SUPPORTED_FIELD_TYPES.add(FieldType.TEXT);
     }
 
     private static final Set<FilterOperator> STANDARD_FILTER_OPS = Set.of(

sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java

@@ -26,6 +26,7 @@
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.index.engine.dataformat.DataFormat;
 import org.opensearch.index.engine.dataformat.IndexingExecutionEngine;
+import org.opensearch.index.engine.dataformat.MergeResult;
 import org.opensearch.index.engine.dataformat.Merger;
 import org.opensearch.index.engine.dataformat.RefreshInput;
 import org.opensearch.index.engine.dataformat.RefreshResult;
@@ -278,7 +279,8 @@ public RefreshResult refresh(RefreshInput refreshInput) throws IOException {
     /** Returns {@code null} — merge scheduling is not yet implemented for the Lucene format. */
     @Override
     public Merger getMerger() {
-        return null;
+        // TODO: Implement merge support as ParquetMerger
+        return mergeInput -> new MergeResult(Map.of());
     }
 
     /**

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/FieldStorageResolver.java

@@ -92,11 +92,11 @@ public List<FieldStorageInfo> resolve(List<String> fieldNames) {
     }
 
     private static FieldStorageInfo resolveField(String fieldName, String fieldType, Map<String, Object> fieldProps, String primaryFormat) {
-        // Doc values: present for all types except text, unless explicitly disabled
-        boolean hasDocValues = !"text".equals(fieldType) && !Boolean.FALSE.equals(fieldProps.get("doc_values"));
+        // Doc values: present for all types unless explicitly disabled
+        boolean hasDocValues = !Boolean.FALSE.equals(fieldProps.get("doc_values"));
 
-        // Index: only when explicitly set to true in mapping
-        boolean isIndexed = Boolean.TRUE.equals(fieldProps.get("index"));
+        // Index: only when explicitly set to false in mapping - enabled by default.
+        boolean isIndexed = !Boolean.FALSE.equals(fieldProps.get("index"));
 
         // Stored fields: only when explicitly set to true in mapping
         boolean isStored = Boolean.TRUE.equals(fieldProps.get("store"));

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/FieldStorageResolverTests.java

@@ -0,0 +1,69 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.planner;
+
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.cluster.metadata.MappingMetadata;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.core.index.Index;
+import org.opensearch.test.OpenSearchTestCase;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link FieldStorageResolver} field storage resolution.
+ */
+public class FieldStorageResolverTests extends OpenSearchTestCase {
+
+    public void testTextFieldGetsDocValuesInPrimaryFormat() {
+        FieldStorageResolver resolver = newResolver("parquet", Map.of("name", Map.of("type", "text")));
+
+        FieldStorageInfo info = resolver.resolve(List.of("name")).get(0);
+
+        assertEquals("name", info.getFieldName());
+        assertEquals(List.of("parquet"), info.getDocValueFormats());
+        assertEquals(List.of("lucene"), info.getIndexFormats());
+    }
+
+    public void testLongFieldGetsDocValuesInPrimaryFormat() {
+        FieldStorageResolver resolver = newResolver("parquet", Map.of("age", Map.of("type", "long")));
+
+        FieldStorageInfo info = resolver.resolve(List.of("age")).get(0);
+
+        assertEquals("age", info.getFieldName());
+        assertEquals(List.of("parquet"), info.getDocValueFormats());
+        assertEquals(List.of("lucene"), info.getIndexFormats());
+    }
+
+    public void testFieldWithAllStorageDisabledHasNoStorage() {
+        IllegalStateException ex = expectThrows(
+            IllegalStateException.class,
+            () -> newResolver("parquet", Map.of("name", Map.of("type", "text", "doc_values", false, "index", false)))
+        );
+        assertTrue("expected 'no storage' error, got: " + ex.getMessage(), ex.getMessage().contains("has no storage in any format"));
+    }
+
+    private static FieldStorageResolver newResolver(String primaryFormat, Map<String, Map<String, Object>> fieldMappings) {
+        Map<String, Object> mappingSource = Map.of("properties", fieldMappings);
+
+        MappingMetadata mappingMetadata = mock(MappingMetadata.class);
+        when(mappingMetadata.sourceAsMap()).thenReturn(mappingSource);
+
+        IndexMetadata indexMetadata = mock(IndexMetadata.class);
+        when(indexMetadata.getIndex()).thenReturn(new Index("test_index", "uuid"));
+        when(indexMetadata.getSettings()).thenReturn(Settings.builder().put("index.composite.primary_data_format", primaryFormat).build());
+        when(indexMetadata.mapping()).thenReturn(mappingMetadata);
+
+        return new FieldStorageResolver(indexMetadata);
+    }
+}

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/FilterRuleTests.java

@@ -172,7 +172,9 @@ public void testFullTextErrorsWithoutDelegation() {
         RexNode condition = makeFullTextCall(FilterOperator.MATCH_PHRASE.toSqlFunction(), 0, "hello world");
         LogicalFilter filter = LogicalFilter.create(stubScan(table), condition);
 
-        PlannerContext context = buildContext("parquet", Map.of("message", Map.of("type", "keyword")));
+        // index=false strips the inverted index so no backend can satisfy the full-text predicate
+        // natively, forcing the "without delegation" code path under test.
+        PlannerContext context = buildContext("parquet", Map.of("message", Map.of("type", "keyword", "index", false)));
 
         IllegalStateException exception = expectThrows(IllegalStateException.class, () -> runPlanner(filter, context));
         assertTrue(exception.getMessage().contains("No backend can evaluate filter predicate"));

sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeMergeIT.java

@@ -0,0 +1,209 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.composite;
+
+import org.opensearch.action.admin.indices.refresh.RefreshResponse;
+import org.opensearch.action.admin.indices.stats.IndicesStatsResponse;
+import org.opensearch.action.admin.indices.stats.ShardStats;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.be.lucene.LucenePlugin;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.SuppressForbidden;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.core.rest.RestStatus;
+import org.opensearch.index.IndexSettings;
+import org.opensearch.index.engine.CommitStats;
+import org.opensearch.index.engine.dataformat.DataFormatDescriptor;
+import org.opensearch.index.engine.dataformat.DataFormatRegistry;
+import org.opensearch.index.engine.dataformat.ReaderManagerConfig;
+import org.opensearch.index.engine.dataformat.stub.MockDataFormat;
+import org.opensearch.index.engine.dataformat.stub.MockDataFormatPlugin;
+import org.opensearch.index.engine.dataformat.stub.MockReaderManager;
+import org.opensearch.index.engine.exec.EngineReaderManager;
+import org.opensearch.index.engine.exec.coord.DataformatAwareCatalogSnapshot;
+import org.opensearch.index.merge.MergeStats;
+import org.opensearch.index.store.PrecomputedChecksumStrategy;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.SearchBackEndPlugin;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * Integration tests for composite merge operations across single and multiple data format engines.
+ *
+ * Requires JDK 25 and sandbox enabled. Run with:
+ * ./gradlew :sandbox:plugins:composite-engine:internalClusterTest \
+ *   --tests "*.CompositeMergeIT" \
+ *   -Dsandbox.enabled=true
+ */
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 1)
+public class CompositeMergeIT extends OpenSearchIntegTestCase {
+
+    private static final String INDEX_NAME = "test-composite-merge";
+    private static final String MERGE_ENABLED_PROPERTY = "opensearch.pluggable.dataformat.merge.enabled";
+
+    // ── Mock DataFormatPlugin using test framework stubs ──
+
+    public static class MockParquetDataFormatPlugin extends MockDataFormatPlugin implements SearchBackEndPlugin<Object> {
+        private static final MockDataFormat PARQUET_FORMAT = new MockDataFormat("parquet", 0L, Set.of());
+
+        public MockParquetDataFormatPlugin() {
+            super(PARQUET_FORMAT);
+        }
+
+        @Override
+        public Map<String, DataFormatDescriptor> getFormatDescriptors(IndexSettings indexSettings, DataFormatRegistry registry) {
+            return Map.of("parquet", new DataFormatDescriptor("parquet", new PrecomputedChecksumStrategy()));
+        }
+
+        @Override
+        public String name() {
+            return "mock-parquet-backend";
+        }
+
+        @Override
+        public List<String> getSupportedFormats() {
+            return List.of("parquet");
+        }
+
+        @Override
+        public EngineReaderManager<?> createReaderManager(ReaderManagerConfig settings) {
+            return new MockReaderManager("parquet");
+        }
+    }
+
+    // ── Test setup ──
+
+    @Override
+    public void setUp() throws Exception {
+        enableMerge();
+        super.setUp();
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        try {
+            client().admin().indices().prepareDelete(INDEX_NAME).get();
+        } catch (Exception e) {
+            // index may not exist if test failed before creation
+        }
+        super.tearDown();
+        disableMerge();
+    }
+
+    @SuppressForbidden(reason = "enable pluggable dataformat merge for integration testing")
+    private static void enableMerge() {
+        System.setProperty(MERGE_ENABLED_PROPERTY, "true");
+    }
+
+    @SuppressForbidden(reason = "restore pluggable dataformat merge property after test")
+    private static void disableMerge() {
+        System.clearProperty(MERGE_ENABLED_PROPERTY);
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(MockParquetDataFormatPlugin.class, CompositeDataFormatPlugin.class, LucenePlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG, true)
+            .build();
+    }
+
+    // ── Tests ──
+
+    /**
+     * Verifies that background merges are triggered automatically after refresh
+     * when enough segments accumulate to exceed the TieredMergePolicy threshold.
+     * <p>
+     * Flow: index docs across many refresh cycles → each refresh calls
+     * triggerPossibleMerges() → MergeScheduler picks up merge candidates
+     * asynchronously → segment count decreases.
+     */
+    public void testBackgroundMergeSingleEngine() throws Exception {
+        createIndex(INDEX_NAME, singleEngineSettings());
+        ensureGreen(INDEX_NAME);
+
+        // Create enough segments to exceed TieredMergePolicy's default threshold (~10)
+        int totalSegmentsCreated = indexDocsAcrossMultipleRefreshes(15, 5);
+
+        // Wait for async background merges to complete
+        assertBusy(() -> {
+            flush(INDEX_NAME);
+            DataformatAwareCatalogSnapshot snapshot = getCatalogSnapshot();
+            assertTrue(
+                "Expected merges to reduce segment count below " + totalSegmentsCreated + ", but got: " + snapshot.getSegments().size(),
+                snapshot.getSegments().size() < totalSegmentsCreated
+            );
+        });
+
+        MergeStats mergeStats = getMergeStats();
+        assertTrue("Expected at least one merge to have occurred", mergeStats.getTotal() > 0);
+
+        DataformatAwareCatalogSnapshot snapshot = getCatalogSnapshot();
+        assertEquals(Set.of("parquet"), snapshot.getDataFormats());
+    }
+
+    // ── Helpers ──
+
+    private Settings singleEngineSettings() {
+        return Settings.builder()
+            .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+            .put("index.pluggable.dataformat.enabled", true)
+            .put("index.pluggable.dataformat", "composite")
+            .put("index.composite.primary_data_format", "parquet")
+            .putList("index.composite.secondary_data_formats")
+            .build();
+    }
+
+    private int indexDocsAcrossMultipleRefreshes(int refreshCycles, int docsPerCycle) {
+        for (int cycle = 0; cycle < refreshCycles; cycle++) {
+            for (int i = 0; i < docsPerCycle; i++) {
+                IndexResponse response = client().prepareIndex()
+                    .setIndex(INDEX_NAME)
+                    .setSource("field_text", randomAlphaOfLength(10), "field_number", randomIntBetween(1, 1000))
+                    .get();
+                assertEquals(RestStatus.CREATED, response.status());
+            }
+            RefreshResponse refreshResponse = client().admin().indices().prepareRefresh(INDEX_NAME).get();
+            assertEquals(RestStatus.OK, refreshResponse.getStatus());
+        }
+        return refreshCycles;
+    }
+
+    private DataformatAwareCatalogSnapshot getCatalogSnapshot() throws IOException {
+        IndicesStatsResponse statsResponse = client().admin().indices().prepareStats(INDEX_NAME).clear().setStore(true).get();
+        ShardStats shardStats = statsResponse.getIndex(INDEX_NAME).getShards()[0];
+        CommitStats commitStats = shardStats.getCommitStats();
+        assertNotNull(commitStats);
+        assertTrue(commitStats.getUserData().containsKey(DataformatAwareCatalogSnapshot.CATALOG_SNAPSHOT_KEY));
+        return DataformatAwareCatalogSnapshot.deserializeFromString(
+            commitStats.getUserData().get(DataformatAwareCatalogSnapshot.CATALOG_SNAPSHOT_KEY),
+            Function.identity()
+        );
+    }
+
+    private MergeStats getMergeStats() {
+        IndicesStatsResponse statsResponse = client().admin().indices().prepareStats(INDEX_NAME).clear().setMerge(true).get();
+        return statsResponse.getIndex(INDEX_NAME).getShards()[0].getStats().getMerge();
+    }
+}

sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormat.java

@@ -26,21 +26,25 @@
 @ExperimentalApi
 public class CompositeDataFormat extends DataFormat {
 
+    private final DataFormat primaryDataFormat;
     private final List<DataFormat> dataFormats;
 
     /**
-     * Constructs a CompositeDataFormat from the given list of data formats.
+     * Constructs a CompositeDataFormat with a designated primary format and a list of all constituent formats.
      *
-     * @param dataFormats the constituent data formats
+     * @param primaryDataFormat the authoritative data format used for merge operations
+     * @param dataFormats       all constituent data formats (including the primary)
      */
-    public CompositeDataFormat(List<DataFormat> dataFormats) {
+    public CompositeDataFormat(DataFormat primaryDataFormat, List<DataFormat> dataFormats) {
+        this.primaryDataFormat = Objects.requireNonNull(primaryDataFormat, "primaryDataFormat must not be null");
         this.dataFormats = List.copyOf(Objects.requireNonNull(dataFormats, "dataFormats must not be null"));
     }
 
     /**
      * Constructs an empty CompositeDataFormat with no constituent formats.
      */
     public CompositeDataFormat() {
+        this.primaryDataFormat = null;
         this.dataFormats = List.of();
     }
 
@@ -53,6 +57,15 @@ public List<DataFormat> getDataFormats() {
         return dataFormats;
     }
 
+    /**
+     * Returns the primary data format used for merge operations.
+     *
+     * @return the primary data format
+     */
+    public DataFormat getPrimaryDataFormat() {
+        return primaryDataFormat;
+    }
+
     @Override
     public String name() {
         return "composite";