removing CSV codec and integrating with parquet module

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

Author: bharath-techie

gradle/missing-javadoc.gradle

@@ -165,7 +165,6 @@ configure([
   project(":plugins:engine-datafusion"), //TODO
   project(":server"),
   project(":modules:parquet-data-format"),
-  project(":plugins:dataformat-csv"), //TODO
 ]) {
   project.tasks.withType(MissingJavadocTask) {
     isExcluded = true

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

@@ -7,7 +7,9 @@
  */
 package com.parquet.parquetdataformat;
 
+import com.parquet.parquetdataformat.engine.ParquetDataFormat;
 import com.parquet.parquetdataformat.fields.ParquetFieldUtil;
+import com.parquet.parquetdataformat.engine.read.ParquetDataSourceCodec;
 import com.parquet.parquetdataformat.writer.ParquetWriter;
 import org.opensearch.index.engine.DataFormatPlugin;
 import org.opensearch.index.engine.exec.DataFormat;
@@ -18,8 +20,12 @@
 import org.opensearch.plugins.DataSourcePlugin;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.vectorized.execution.search.spi.DataSourceCodec;
 
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
 
 /**
  * OpenSearch plugin that provides Parquet data format support for indexing operations.
@@ -58,9 +64,23 @@ public <T extends DataFormat> IndexingExecutionEngine<T> indexingEngine(MapperSe
         return (IndexingExecutionEngine<T>) new ParquetExecutionEngine(() -> ParquetFieldUtil.getSchema(mapperService), shardPath);
     }
 
+    private Class<? extends DataFormat> getDataFormatType() {
+        return ParquetDataFormat.class;
+    }
+
     @Override
     public DataFormat getDataFormat() {
-        return null;
+        return new ParquetDataFormat();
+    }
+
+    @Override
+    public Optional<Map<org.opensearch.vectorized.execution.search.DataFormat, DataSourceCodec>> getDataSourceCodecs() {
+        Map<org.opensearch.vectorized.execution.search.DataFormat, DataSourceCodec> codecs = new HashMap<>();
+        ParquetDataSourceCodec parquetDataSourceCodec = new ParquetDataSourceCodec();
+        // TODO : version it correctly - similar to lucene codecs?
+        codecs.put(parquetDataSourceCodec.getDataFormat(), new ParquetDataSourceCodec());
+        return Optional.of(codecs);
+        // return Optional.empty();
     }
 
     // for testing locally only

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/bridge/RustBridge.java

@@ -11,21 +11,21 @@
 
 /**
  * JNI bridge to the native Rust Parquet writer implementation.
- * 
+ *
  * <p>This class provides the interface between Java and the native Rust library
  * that handles low-level Parquet file operations. It automatically loads the
  * appropriate native library for the current platform and architecture.
- * 
+ *
  * <p>Supported platforms:
  * <ul>
  *   <li>Windows (x86, x86_64, aarch64)</li>
  *   <li>macOS (x86_64, aarch64/arm64)</li>
  *   <li>Linux (x86, x86_64, aarch64)</li>
  * </ul>
- * 
+ *
  * <p>The native library is extracted from resources and loaded as a temporary file,
  * which is automatically cleaned up on JVM shutdown.
- * 
+ *
  * <p>All native methods operate on Arrow C Data Interface pointers and return
  * integer status codes for error handling.
  */
@@ -83,19 +83,37 @@ private static void loadNativeLibrary() {
     public static native void write(String file, long arrayAddress, long schemaAddress) throws IOException;
     public static native void closeWriter(String file) throws IOException;
     public static native void flushToDisk(String file) throws IOException;
-    
+
     // State and metrics methods handled on Rust side
     public static native boolean writerExists(String file);
     public static native long getWriteCount(String file);
     public static native long getTotalRows(String file);
     public static native String[] getActiveWriters();
-    
+
     // Validation helpers that could be implemented natively for better performance
     public static boolean isValidFileName(String fileName) {
         return fileName != null && !fileName.trim().isEmpty();
     }
-    
+
     public static boolean isValidMemoryAddress(long address) {
         return address != 0;
     }
+
+
+    // DATAFUSION specific native methods starts here
+
+    // Record batch and streaming related methods
+    public static native String nativeNextBatch(long streamPtr);
+
+    public static native void nativeCloseStream(long streamPtr);
+
+
+    // Native method declarations - these will be implemented in the JNI library
+    public static native void nativeRegisterDirectory(String tableName, String directoryPath, String[] files, long runtimeId);
+
+    public static native long nativeCreateSessionContext(String[] configKeys, String[] configValues);
+
+    public static native long nativeExecuteSubstraitQuery(long sessionContextPtr, byte[] substraitPlan);
+
+    public static native void nativeCloseSessionContext(long sessionContextPtr);
 }

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

@@ -6,7 +6,7 @@
  * compatible open source license.
  */
 
-package com.parquet.parquetdataformat.read;
+package com.parquet.parquetdataformat.engine.read;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -20,6 +20,11 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
+import static com.parquet.parquetdataformat.bridge.RustBridge.nativeCloseSessionContext;
+import static com.parquet.parquetdataformat.bridge.RustBridge.nativeCreateSessionContext;
+import static com.parquet.parquetdataformat.bridge.RustBridge.nativeExecuteSubstraitQuery;
+import static com.parquet.parquetdataformat.bridge.RustBridge.nativeRegisterDirectory;
+
 /**
  * Datasource codec implementation for parquet files
  */
@@ -33,7 +38,7 @@ public class ParquetDataSourceCodec implements DataSourceCodec {
     // JNI library loading
     static {
         try {
-            JniLibraryLoader.loadLibrary();
+            //JniLibraryLoader.loadLibrary();
             logger.info("DataFusion JNI library loaded successfully");
         } catch (Exception e) {
             logger.error("Failed to load DataFusion JNI library", e);
@@ -135,13 +140,4 @@ public CompletableFuture<Void> closeSessionContext(long sessionContextId) {
     public DataFormat getDataFormat() {
         return DataFormat.CSV;
     }
-
-    // Native method declarations - these will be implemented in the JNI library
-    private static native void nativeRegisterDirectory(String tableName, String directoryPath, String[] files, long runtimeId);
-
-    private static native long nativeCreateSessionContext(String[] configKeys, String[] configValues);
-
-    private static native long nativeExecuteSubstraitQuery(long sessionContextPtr, byte[] substraitPlan);
-
-    private static native void nativeCloseSessionContext(long sessionContextPtr);
 }

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

@@ -6,14 +6,17 @@
  * compatible open source license.
  */
 
-package com.parquet.parquetdataformat.read;
+package com.parquet.parquetdataformat.engine.read;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.vectorized.execution.search.spi.RecordBatchStream;
 
 import java.util.concurrent.CompletableFuture;
 
+import static com.parquet.parquetdataformat.bridge.RustBridge.nativeCloseStream;
+import static com.parquet.parquetdataformat.bridge.RustBridge.nativeNextBatch;
+
 /**
  * TODO : this need not be here - nothing specific to parquet - move to LIB ?
  * Native implementation of RecordBatchStream that wraps a JNI stream pointer.
@@ -111,9 +114,4 @@ public void close() {
             }
         }
     }
-
-    // Native method declarations
-    private static native String nativeNextBatch(long streamPtr);
-
-    private static native void nativeCloseStream(long streamPtr);
 }

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/engine/read/package-info.java

@@ -10,4 +10,4 @@
  * CSV data format implementation for DataFusion integration.
  * Provides CSV file reading capabilities through DataFusion query engine.
  */
-package com.parquet.parquetdataformat.read;
+package com.parquet.parquetdataformat.engine.read;

modules/parquet-data-format/src/main/resources/META-INF/services/org.opensearch.vectorized.execution.search.spi.DataSourceCodec

@@ -1 +1 @@
-org.opensearch.datafusion.csv.CsvDataSourceCodec
+com.parquet.parquetdataformat.engine.read.ParquetDataSourceCodec

modules/parquet-data-format/src/main/rust/Cargo.toml

@@ -8,9 +8,56 @@ name = "parquet_dataformat_jni"
 crate-type = ["cdylib"]
 
 [dependencies]
-jni = "0.21.1"
-arrow = { version = "53.0.0", features = ["ffi"] }
-parquet = "53.0.0"
+
+# DataFusion dependencies
+datafusion = "49.0.0"
+datafusion-substrait = "49.0.0"
+arrow = { version = "54.0.0", features = ["ffi"] }
+
+arrow-array = "54.0.0"
+arrow-schema = "54.0.0"
+arrow-buffer = "54.0.0"
+
+# JNI dependencies
+jni = "0.21"
+
+# Async runtime
+tokio = { version = "1.0", features = ["full"] }
+futures = "0.3"
+futures-util = "0.3"
+
+# Serialization
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+
+# Error handling
+anyhow = "1.0"
+thiserror = "1.0"
+
+# Logging
+log = "0.4"
+
+# Parquet support
+parquet = "54.0.0"
+
+# Object store for file access
+object_store = "0.11"
+url = "2.0"
+
+# Substrait support
+substrait = "0.47"
+prost = "0.13"
+
+# Temporary directory support
+tempfile = "3.0"
+
+#jni = "0.21.1"
+#arrow = { version = "53.0.0", features = ["ffi"] }
+#parquet = "53.0.0"
 lazy_static = "1.4.0"
 dashmap = "7.0.0-rc2"
 chrono = "0.4"
+
+
+[build-dependencies]
+cbindgen = "0.27"

modules/parquet-data-format/src/main/rust/src/context.rs

@@ -28,7 +28,7 @@ impl SessionContextManager {
         directory_path: &str,
         options: HashMap<String, String>,
     ) -> Result<u64> {
-        // Placeholder implementation - would register csv directory as table
+        // Placeholder implementation - would register parquet directory as table
         log::info!("Registering directory: {} at path: {} with options: {:?}",
                    table_name, directory_path, options);