add parquet merge support through a K way merge streaming merge sort

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

Author: Shailesh-Kumar-Singh

sandbox/libs/dataformat-native/rust/Cargo.toml

@@ -51,6 +51,7 @@ once_cell = "1.21.3"
 crc32fast = "1.4"
 parking_lot = "0.12.5"
 lazy_static = "1.4.0"
+rayon = "1.10"
 criterion = { version = "0.5", features = ["async_tokio"] }
 
 # Internal

sandbox/plugins/parquet-data-format/benchmarks/src/main/java/org/opensearch/parquet/benchmark/VSRRotationBenchmark.java

@@ -10,7 +10,10 @@
 
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.Schema;
+import org.opensearch.Version;
+import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.index.IndexSettings;
 import org.opensearch.index.mapper.KeywordFieldMapper;
 import org.opensearch.index.mapper.MappedFieldType;
 import org.opensearch.index.mapper.NumberFieldMapper;
@@ -80,6 +83,7 @@ public class VSRRotationBenchmark {
     private List<MappedFieldType> fieldTypes;
     private VSRManager vsrManager;
     private String filePath;
+    private IndexSettings indexSettings;
 
     @Setup(Level.Trial)
     public void setupTrial() {
@@ -123,7 +127,10 @@ public void setupTrial() {
     public void setup() throws IOException {
         bufferPool = new ArrowBufferPool(Settings.EMPTY);
         filePath = Path.of(System.getProperty("java.io.tmpdir"), "benchmark_vsr_" + System.nanoTime() + ".parquet").toString();
-        vsrManager = new VSRManager(filePath, schema, bufferPool, maxRowsPerVSR, threadPool, runAsync);
+        Settings idxSettings = Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT).build();
+        IndexMetadata indexMetadata = IndexMetadata.builder("benchmark-index").settings(idxSettings).build();
+        indexSettings = new IndexSettings(indexMetadata, Settings.EMPTY);
+        vsrManager = new VSRManager(filePath, indexSettings, schema, bufferPool, maxRowsPerVSR, threadPool, runAsync);
     }
 
     @Benchmark

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java

@@ -9,40 +9,81 @@
 package org.opensearch.parquet;
 
 import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.core.common.unit.ByteSizeUnit;
+import org.opensearch.core.common.unit.ByteSizeValue;
 
 import java.util.List;
 
 /**
- * Node-scoped settings for the Parquet data format plugin.
- *
- * <p>All settings are registered with OpenSearch via
- * {@link ParquetDataFormatPlugin#getSettings()} and can be configured in
- * {@code opensearch.yml} or via cluster settings API.
- *
- * <ul>
- *   <li>{@link #MAX_NATIVE_ALLOCATION} — Maximum native memory allocation for Arrow buffers,
- *       expressed as a percentage of available non-heap system memory (default {@code "10%"}).</li>
- *   <li>{@link #MAX_ROWS_PER_VSR} — Row count threshold that triggers VectorSchemaRoot rotation
- *       during document ingestion (default {@code 50000}).</li>
- * </ul>
+ * Settings for Parquet data format.
  */
 public final class ParquetSettings {
 
     private ParquetSettings() {}
 
-    /** Default maximum native memory allocation as a percentage of available non-heap memory. */
     public static final String DEFAULT_MAX_NATIVE_ALLOCATION = "10%";
-    /** Default maximum number of rows per VectorSchemaRoot before rotation. */
     public static final int DEFAULT_MAX_ROWS_PER_VSR = 50000;
 
-    /** Maximum native memory allocation for Arrow buffers, as a percentage of non-heap memory. */
+    /** Group setting prefix for all Parquet settings. */
+    public static final Setting<Settings> PARQUET_SETTINGS = Setting.groupSetting(
+        "parquet.",
+        Setting.Property.IndexScope
+    );
+
+    /** Maximum row group size in bytes (default 128MB). */
+    public static final Setting<ByteSizeValue> ROW_GROUP_SIZE_BYTES = Setting.byteSizeSetting(
+        "parquet.row_group_size_bytes",
+        new ByteSizeValue(128, ByteSizeUnit.MB),
+        Setting.Property.IndexScope
+    );
+
+    /** Data page size limit in bytes (default 1MB). */
+    public static final Setting<ByteSizeValue> PAGE_SIZE_BYTES = Setting.byteSizeSetting(
+        "parquet.page_size_bytes",
+        new ByteSizeValue(1, ByteSizeUnit.MB),
+        Setting.Property.IndexScope
+    );
+
+    /** Maximum number of rows per data page (default 20000). */
+    public static final Setting<Integer> PAGE_ROW_LIMIT = Setting.intSetting(
+        "parquet.page_row_limit",
+        20000,
+        1,
+        Setting.Property.IndexScope
+    );
+
+    /** Dictionary page size limit in bytes (default 2MB). */
+    public static final Setting<ByteSizeValue> DICT_SIZE_BYTES = Setting.byteSizeSetting(
+        "parquet.dict_size_bytes",
+        new ByteSizeValue(2, ByteSizeUnit.MB),
+        Setting.Property.IndexScope
+    );
+
+    /** Compression codec for Parquet files, e.g. ZSTD, SNAPPY, LZ4_RAW (default LZ4_RAW). */
+    public static final Setting<String> COMPRESSION_TYPE = Setting.simpleString(
+        "parquet.compression_type",
+        "LZ4_RAW",
+        Setting.Property.IndexScope
+    );
+
+    /** Compression level for the chosen codec (default 2, range 1–9). */
+    public static final Setting<Integer> COMPRESSION_LEVEL = Setting.intSetting(
+        "parquet.compression_level",
+        2,
+        1,
+        9,
+        Setting.Property.IndexScope
+    );
+
+    /** Maximum native memory allocation for Arrow buffers, as a percentage of non-heap memory (default 10%). */
     public static final Setting<String> MAX_NATIVE_ALLOCATION = Setting.simpleString(
         "parquet.max_native_allocation",
         DEFAULT_MAX_NATIVE_ALLOCATION,
         Setting.Property.NodeScope
     );
 
-    /** Maximum number of rows per VectorSchemaRoot before rotation is triggered. */
+    /** Maximum rows per VectorSchemaRoot before rotation is triggered (default 50000). */
     public static final Setting<Integer> MAX_ROWS_PER_VSR = Setting.intSetting(
         "parquet.max_rows_per_vsr",
         DEFAULT_MAX_ROWS_PER_VSR,
@@ -52,6 +93,11 @@ private ParquetSettings() {}
 
     /** Returns all settings defined by the Parquet plugin. */
     public static List<Setting<?>> getSettings() {
-        return List.of(MAX_NATIVE_ALLOCATION, MAX_ROWS_PER_VSR);
+        return List.of(
+            PARQUET_SETTINGS,
+            ROW_GROUP_SIZE_BYTES, PAGE_SIZE_BYTES, PAGE_ROW_LIMIT, DICT_SIZE_BYTES,
+            COMPRESSION_TYPE, COMPRESSION_LEVEL,
+            MAX_NATIVE_ALLOCATION, MAX_ROWS_PER_VSR
+        );
     }
 }

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/NativeParquetWriter.java

@@ -11,14 +11,15 @@
 import org.opensearch.common.SetOnce;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Type-safe handle for the native Rust Parquet writer with lifecycle management.
  *
  * <p>Wraps the stateless JNI methods in {@link RustBridge} with a file-scoped lifecycle:
  * <ol>
- *   <li>{@code new NativeParquetWriter(filePath, schemaAddress)} — creates the native writer</li>
+ *   <li>{@code new NativeParquetWriter(filePath, indexName, schemaAddress, sortColumns, reverseSorts, nullsFirst)} — creates the native writer</li>
  *   <li>{@link #write(long, long)} — sends one or more Arrow batches (repeatable)</li>
  *   <li>{@link #flush()} — finalizes the Parquet file and returns metadata</li>
  *   <li>{@link #sync()} — fsyncs the file to durable storage (calls flush if needed)</li>
@@ -37,12 +38,23 @@ public class NativeParquetWriter {
      * Creates a new NativeParquetWriter.
      *
      * @param filePath      the path to the Parquet file to write
+     * @param indexName     the index name for settings lookup
      * @param schemaAddress the native memory address of the Arrow schema
+     * @param sortColumns   the columns to sort by, or empty list for no sorting
+     * @param reverseSorts  whether each sort column is descending, or empty list
+     * @param nullsFirst    whether nulls sort first for each column, or empty list
      * @throws IOException if the native writer creation fails
      */
-    public NativeParquetWriter(String filePath, long schemaAddress) throws IOException {
+    public NativeParquetWriter(
+        String filePath,
+        String indexName,
+        long schemaAddress,
+        List<String> sortColumns,
+        List<Boolean> reverseSorts,
+        List<Boolean> nullsFirst
+    ) throws IOException {
         this.filePath = filePath;
-        RustBridge.createWriter(filePath, schemaAddress);
+        RustBridge.createWriter(filePath, indexName, schemaAddress, sortColumns, reverseSorts, nullsFirst);
     }
 
     /**

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/NativeSettings.java

@@ -0,0 +1,80 @@
+/*
+ * 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.parquet.bridge;
+
+/**
+ * Immutable settings passed to the native Rust writer via JNI.
+ * The Rust side reads values through the getter methods.
+ * All fields are nullable; the native side falls back to defaults when null.
+ */
+public class NativeSettings {
+
+    private final String indexName;
+    private final String compressionType;
+    private final Integer compressionLevel;
+    private final Long pageSizeBytes;
+    private final Integer pageRowLimit;
+    private final Long dictSizeBytes;
+    private final Long rowGroupSizeBytes;
+    private final Boolean bloomFilterEnabled;
+    private final Double bloomFilterFpp;
+    private final Long bloomFilterNdv;
+
+    private NativeSettings(Builder builder) {
+        this.indexName = builder.indexName;
+        this.compressionType = builder.compressionType;
+        this.compressionLevel = builder.compressionLevel;
+        this.pageSizeBytes = builder.pageSizeBytes;
+        this.pageRowLimit = builder.pageRowLimit;
+        this.dictSizeBytes = builder.dictSizeBytes;
+        this.rowGroupSizeBytes = builder.rowGroupSizeBytes;
+        this.bloomFilterEnabled = builder.bloomFilterEnabled;
+        this.bloomFilterFpp = builder.bloomFilterFpp;
+        this.bloomFilterNdv = builder.bloomFilterNdv;
+    }
+
+    public String getIndexName() { return indexName; }
+    public String getCompressionType() { return compressionType; }
+    public Integer getCompressionLevel() { return compressionLevel; }
+    public Long getPageSizeBytes() { return pageSizeBytes; }
+    public Integer getPageRowLimit() { return pageRowLimit; }
+    public Long getDictSizeBytes() { return dictSizeBytes; }
+    public Long getRowGroupSizeBytes() { return rowGroupSizeBytes; }
+    public Boolean getBloomFilterEnabled() { return bloomFilterEnabled; }
+    public Double getBloomFilterFpp() { return bloomFilterFpp; }
+    public Long getBloomFilterNdv() { return bloomFilterNdv; }
+
+    public static Builder builder() { return new Builder(); }
+
+    public static class Builder {
+        private String indexName;
+        private String compressionType;
+        private Integer compressionLevel;
+        private Long pageSizeBytes;
+        private Integer pageRowLimit;
+        private Long dictSizeBytes;
+        private Long rowGroupSizeBytes;
+        private Boolean bloomFilterEnabled;
+        private Double bloomFilterFpp;
+        private Long bloomFilterNdv;
+
+        public Builder indexName(String v) { this.indexName = v; return this; }
+        public Builder compressionType(String v) { this.compressionType = v; return this; }
+        public Builder compressionLevel(Integer v) { this.compressionLevel = v; return this; }
+        public Builder pageSizeBytes(Long v) { this.pageSizeBytes = v; return this; }
+        public Builder pageRowLimit(Integer v) { this.pageRowLimit = v; return this; }
+        public Builder dictSizeBytes(Long v) { this.dictSizeBytes = v; return this; }
+        public Builder rowGroupSizeBytes(Long v) { this.rowGroupSizeBytes = v; return this; }
+        public Builder bloomFilterEnabled(Boolean v) { this.bloomFilterEnabled = v; return this; }
+        public Builder bloomFilterFpp(Double v) { this.bloomFilterFpp = v; return this; }
+        public Builder bloomFilterNdv(Long v) { this.bloomFilterNdv = v; return this; }
+
+        public NativeSettings build() { return new NativeSettings(this); }
+    }
+}