Commiter integration and build fixes (#19612)

* Feature/datafusion 4 (#46)

* Composite document writer pool initial implementation

* Committer interface and lucene based commit engine implementation

* Catalog snapshot changes to create segment view during commit

---------

Co-authored-by: Shashank Gowri <[email protected]>

* fix build for commit integration

Signed-off-by: bharath-techie <[email protected]>

---------

Signed-off-by: bharath-techie <[email protected]>
Co-authored-by: Shashank Gowri <[email protected]>

Author: bharath-techie

.gitignore

@@ -82,5 +82,6 @@ doc-tools/missing-doclet/bin/
 /modules/parquet-data-format/src/main/resources/native/
 /modules/parquet-data-format/jni/target/debug
 
+/modules/parquet-data-format/jni/target/release
 **/Cargo.lock
 /modules/parquet-data-format/jni/

.idea/runConfigurations/Debug_OpenSearch.xml

@@ -1,16 +1,20 @@
 package com.parquet.parquetdataformat.engine;
 
-import org.apache.arrow.vector.types.pojo.Schema;
-import org.opensearch.index.engine.exec.*;
 import com.parquet.parquetdataformat.writer.ParquetDocumentInput;
 import com.parquet.parquetdataformat.writer.ParquetWriter;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.opensearch.index.engine.exec.DataFormat;
+import org.opensearch.index.engine.exec.IndexingExecutionEngine;
+import org.opensearch.index.engine.exec.RefreshInput;
+import org.opensearch.index.engine.exec.RefreshResult;
+import org.opensearch.index.engine.exec.Writer;
+import org.opensearch.index.engine.exec.WriterFileSet;
 import org.opensearch.index.shard.ShardPath;
 
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 
 import static com.parquet.parquetdataformat.engine.ParquetDataFormat.PARQUET_DATA_FORMAT;
@@ -47,13 +51,11 @@
 public class ParquetExecutionEngine implements IndexingExecutionEngine<ParquetDataFormat> {
 
     public static final String FILE_NAME_PREFIX = "parquet_file_generation";
-    AtomicInteger counter;
-    Supplier<Schema> schema;
-    private final List<FileMetadata> filesWrittenAlready = new ArrayList<>();
+    private final Supplier<Schema> schema;
+    private final List<WriterFileSet> filesWrittenAlready = new ArrayList<>();
     private final ShardPath shardPath;
 
     public ParquetExecutionEngine(Supplier<Schema> schema, ShardPath shardPath) {
-        counter = new AtomicInteger(0);
         this.schema = schema;
         this.shardPath = shardPath;
     }
@@ -64,15 +66,15 @@ public List<String> supportedFieldTypes() {
     }
 
     @Override
-    public Writer<ParquetDocumentInput> createWriter() throws IOException {
-        String fileName = Path.of(shardPath.getDataPath().toString(), FILE_NAME_PREFIX + "_" + counter.getAndIncrement() + ".parquet").toString();
-        return new ParquetWriter(fileName, schema.get());
+    public Writer<ParquetDocumentInput> createWriter(long writerGeneration) throws IOException {
+        String fileName = Path.of(shardPath.getDataPath().toString(), FILE_NAME_PREFIX + "_" + writerGeneration + ".parquet").toString();
+        return new ParquetWriter(fileName, schema.get(), writerGeneration);
     }
 
     @Override
     public RefreshResult refresh(RefreshInput refreshInput) throws IOException {
         RefreshResult refreshResult = new RefreshResult();
-        filesWrittenAlready.addAll(refreshInput.getFiles());
+        filesWrittenAlready.addAll(refreshInput.getWriterFiles());
         refreshResult.add(PARQUET_DATA_FORMAT, filesWrittenAlready);
         return refreshResult;
     }

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/engine/ParquetExecutionEngine.java

@@ -8,25 +8,20 @@
 
 package com.parquet.parquetdataformat.vsr;
 
-import com.parquet.parquetdataformat.engine.ParquetDataFormat;
+import com.parquet.parquetdataformat.bridge.ArrowExport;
+import com.parquet.parquetdataformat.bridge.RustBridge;
+import com.parquet.parquetdataformat.memory.MemoryPressureMonitor;
 import com.parquet.parquetdataformat.writer.ParquetDocumentInput;
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.Schema;
-import com.parquet.parquetdataformat.bridge.RustBridge;
-import com.parquet.parquetdataformat.bridge.ArrowExport;
-import com.parquet.parquetdataformat.memory.MemoryPressureMonitor;
+import org.opensearch.index.engine.exec.FlushIn;
+import org.opensearch.index.engine.exec.WriteResult;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
-import org.opensearch.index.engine.exec.FileMetadata;
-import org.opensearch.index.engine.exec.FlushIn;
-import org.opensearch.index.engine.exec.WriteResult;
-
-import static com.parquet.parquetdataformat.engine.ParquetDataFormat.PARQUET_DATA_FORMAT;
-
 /**
  * Manages VectorSchemaRoot lifecycle with integrated memory management and native call wrappers.
  * Provides a high-level interface for Parquet document operations using managed VSR abstractions.
@@ -113,7 +108,7 @@ public WriteResult addToManagedVSR(ParquetDocumentInput document) throws IOExcep
         }
     }
 
-    public FileMetadata flush(FlushIn flushIn) throws IOException {
+    public String flush(FlushIn flushIn) throws IOException {
         System.out.println("[JAVA] flush called, row count: " + managedVSR.getRowCount());
         try {
             // Only flush if we have data
@@ -136,7 +131,7 @@ public FileMetadata flush(FlushIn flushIn) throws IOException {
             }
             System.out.println("[JAVA] Successfully flushed data");
 
-            return new FileMetadata(PARQUET_DATA_FORMAT, fileName);
+            return fileName;
         } catch (Exception e) {
             System.out.println("[JAVA] ERROR in flush: " + e.getMessage());
             throw new IOException("Failed to flush data: " + e.getMessage(), e);

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/vsr/VSRManager.java

@@ -1,14 +1,17 @@
 package com.parquet.parquetdataformat.writer;
 
 import com.parquet.parquetdataformat.vsr.VSRManager;
-import org.opensearch.index.engine.exec.FileMetadata;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.opensearch.index.engine.exec.FileInfos;
 import org.opensearch.index.engine.exec.FlushIn;
 import org.opensearch.index.engine.exec.WriteResult;
 import org.opensearch.index.engine.exec.Writer;
-import org.apache.arrow.vector.types.pojo.Schema;
+import org.opensearch.index.engine.exec.WriterFileSet;
 
 import java.io.IOException;
-import java.util.Optional;
+import java.nio.file.Path;
+
+import static com.parquet.parquetdataformat.engine.ParquetDataFormat.PARQUET_DATA_FORMAT;
 
 /**
  * Parquet file writer implementation that integrates with OpenSearch's Writer interface.
@@ -33,11 +36,13 @@ public class ParquetWriter implements Writer<ParquetDocumentInput> {
     private final String file;
     private final Schema schema;
     private final VSRManager vsrManager;
+    private final long writerGeneration;
 
-    public ParquetWriter(String file, Schema schema) {
+    public ParquetWriter(String file, Schema schema, long writerGeneration) {
         this.file = file;
         this.schema = schema;
         this.vsrManager = new VSRManager(file, schema);
+        this.writerGeneration = writerGeneration;
     }
 
     @Override
@@ -46,8 +51,13 @@ public WriteResult addDoc(ParquetDocumentInput d) throws IOException {
     }
 
     @Override
-    public FileMetadata flush(FlushIn flushIn) throws IOException {
-        return vsrManager.flush(flushIn);
+    public FileInfos flush(FlushIn flushIn) throws IOException {
+        String fileName = vsrManager.flush(flushIn);
+        FileInfos fileInfos = new FileInfos();
+        WriterFileSet writerFileSet = new WriterFileSet(Path.of(fileName).getParent(), writerGeneration);
+        writerFileSet.add(fileName);
+        fileInfos.putWriterFileSet(PARQUET_DATA_FORMAT, writerFileSet);
+        return fileInfos;
     }
 
     @Override
@@ -60,11 +70,6 @@ public void close() {
         vsrManager.close();
     }
 
-    @Override
-    public Optional<FileMetadata> getMetadata() {
-        return Optional.empty();
-    }
-
     @Override
     public ParquetDocumentInput newDocumentInput() {
         // Get a new ManagedVSR from VSRManager for this document input

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/writer/ParquetWriter.java

@@ -160,7 +160,16 @@ pub fn throw_exception(env: &mut JNIEnv, message: &str) {
 pub fn create_object_meta_from_filenames(base_path: &str, filenames: Vec<String>) -> Vec<ObjectMeta> {
     filenames.into_iter().map(|filename| {
         let filename = filename.as_str();
-        let full_path = format!("{}/{}", base_path.trim_end_matches('/'), filename);
+
+        // Handle both full paths and relative filenames
+        let full_path = if filename.starts_with('/') || filename.contains(base_path) {
+            // Already a full path
+            filename.to_string()
+        } else {
+            // Just a filename, needs base_path
+            format!("{}/{}", base_path.trim_end_matches('/'), filename)
+        };
+
         let file_size = fs::metadata(&full_path).map(|m| m.len()).unwrap_or(0);
         let modified = fs::metadata(&full_path)
             .and_then(|m| m.modified())

plugins/engine-datafusion/jni/src/util.rs

@@ -9,14 +9,12 @@
 package org.opensearch.datafusion.search;
 
 import org.opensearch.datafusion.DataFusionQueryJNI;
-import org.opensearch.index.engine.exec.FileMetadata;
+import org.opensearch.index.engine.exec.WriterFileSet;
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.opensearch.datafusion.DataFusionQueryJNI.closeDatafusionReader;
@@ -32,7 +30,7 @@ public class DatafusionReader implements Closeable {
     /**
      * The file metadata collection.
      */
-    public Collection<FileMetadata> files;
+    public Collection<WriterFileSet> files;
     /**
      * The cache pointer.
      */
@@ -44,15 +42,16 @@ public class DatafusionReader implements Closeable {
      * @param directoryPath The directory path
      * @param files The file metadata collection
      */
-    public DatafusionReader(String directoryPath, Collection<FileMetadata> files) {
+    public DatafusionReader(String directoryPath, Collection<WriterFileSet> files) {
         this.directoryPath = directoryPath;
         this.files = files;
         String[] fileNames = new String[0];
         if(files != null) {
             System.out.println("Got the files!!!!!");
-            fileNames = files.stream().map(file -> Path.of(file.fileName()).getFileName().toString()).toArray(String[]::new);
+            fileNames = files.stream()
+                .flatMap(writerFileSet -> writerFileSet.getFiles().stream())
+                .toArray(String[]::new);
         }
-        //String[] fileNames = files.stream().map(file -> Path.of(file.fileName()).getFileName().toString()).toArray(String[]::new);
         System.out.println("File names: " + Arrays.toString(fileNames));
         System.out.println("Directory path: " + directoryPath);
 

plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/search/DatafusionReader.java

@@ -78,6 +78,8 @@ dependencies {
   compileOnly project(':libs:opensearch-plugin-classloader')
   testRuntimeOnly project(':libs:opensearch-plugin-classloader')
 
+  implementation 'org.apache.commons:commons-lang3:3.17.0'
+
   api libs.bundles.lucene
 
   // utilities

server/build.gradle

@@ -0,0 +1,35 @@
+/*
+ * 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.index.engine.exec;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+public final class FileInfos {
+
+    private final Map<DataFormat, WriterFileSet> writerFilesMap;
+
+    public FileInfos() {
+        this.writerFilesMap = new HashMap<>();
+    }
+
+    public Map<DataFormat, WriterFileSet> getWriterFilesMap() {
+        return Collections.unmodifiableMap(writerFilesMap);
+    }
+
+    public void putWriterFileSet(DataFormat format, WriterFileSet writerFileSet) {
+        writerFilesMap.put(format, writerFileSet);
+    }
+
+    public Optional<WriterFileSet> getWriterFileSet(DataFormat format) {
+        return Optional.ofNullable(writerFilesMap.get(format));
+    }
+}

server/src/main/java/org/opensearch/index/engine/exec/FileInfos.java

@@ -8,4 +8,5 @@
 
 package org.opensearch.index.engine.exec;
 
-public record FileMetadata(DataFormat df, String fileName) { }
+public record FileMetadata(String directory, String file) {
+}