[Draft]Datafusion and Parquet plugin config interaction

.gitignore

@@ -10,6 +10,12 @@ CLAUDE.md
 build-idea/
 out/
 
+modules/parquet-data-format/src/main/rust/target/*
+libs/dataformat-csv/jni/target/*
+libs/dataformat-csv/src/main/resources/*
+plugins/dataformat-csv/src/main/resources/*
+libs/dataformat-csv/jni/Cargo.lock
+
 # include shared intellij config
 !.idea/inspectionProfiles/Project_Default.xml
 !.idea/runConfigurations/Debug_OpenSearch.xml
@@ -68,3 +74,14 @@ testfixtures_shared/
 
 # build files generated
 doc-tools/missing-doclet/bin/
+/plugins/dataformat-csv/jni/target
+/plugins/dataformat-csv/jni/Cargo.lock
+
+/modules/parquet-data-format/src/main/rust/target
+/modules/parquet-data-format/src/main/rust/debug
+/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/

buildSrc/src/main/java/org/opensearch/gradle/testclusters/RunTask.java

@@ -168,6 +168,8 @@ public void beforeStart() {
             firstNode.setting("discovery.seed_hosts", LOCALHOST_ADDRESS_PREFIX + DEFAULT_TRANSPORT_PORT);
             cluster.setPreserveDataDir(preserveData);
             for (OpenSearchNode node : cluster.getNodes()) {
+                // TODO : remove this - this disables assertions
+                node.jvmArgs(" -da ");
                 if (node != firstNode) {
                     node.setHttpPort(String.valueOf(httpPort));
                     httpPort++;

gradle/missing-javadoc.gradle

@@ -160,7 +160,11 @@ configure([
   project(":test:fixtures:hdfs-fixture"),
   project(":test:fixtures:s3-fixture"),
   project(":test:framework"),
-  project(":test:logger-usage")
+  project(":test:logger-usage"),
+  project(":libs:opensearch-vectorized-exec-spi"), // TODO
+  project(":plugins:engine-datafusion"), //TODO
+  project(":server"),
+  project(":modules:parquet-data-format"),
 ]) {
   project.tasks.withType(MissingJavadocTask) {
     isExcluded = true

gradle/run.gradle

@@ -52,6 +52,30 @@ testClusters {
         }
       }
     }
+
+    if (findProperty("remotePlugins")) {
+      remotePlugins = Eval.me(remotePlugins)
+      for (String coords : remotePlugins) {
+        if (coords.startsWith('/') || coords.startsWith('file:')) {
+          // Direct file path
+          plugin(project.layout.file(project.provider { new File(coords) }))
+        } else {
+          // Maven coordinates
+          def config = project.configurations.detachedConfiguration(
+            project.dependencies.create(coords + '@zip')
+          )
+          config.resolutionStrategy.cacheChangingModulesFor 0, 'seconds'
+          project.repositories.mavenLocal()
+          project.repositories {
+            maven {
+              name = 'OpenSearch Snapshots'
+              url = 'https://central.sonatype.com/repository/maven-snapshots/'
+            }
+          }
+          plugin(project.layout.file(project.provider { config.singleFile }))
+        }
+      }
+    }
   }
 
 }

libs/common/src/main/java/org/opensearch/common/annotation/processor/ApiAnnotationProcessor.java

@@ -85,20 +85,20 @@ public boolean process(Set<? extends TypeElement> annotations, RoundEnvironment
             Set.of(PublicApi.class, ExperimentalApi.class, DeprecatedApi.class)
         );
 
-        for (var element : elements) {
-            validate(element);
-
-            if (!checkPackage(element)) {
-                continue;
-            }
-
-            // Skip all not-public elements
-            checkPublicVisibility(null, element);
-
-            if (element instanceof TypeElement) {
-                process((TypeElement) element);
-            }
-        }
+//        for (var element : elements) {
+//            validate(element);
+//
+//            if (!checkPackage(element)) {
+//                continue;
+//            }
+//
+//            // Skip all not-public elements
+//            checkPublicVisibility(null, element);
+//
+//            if (element instanceof TypeElement) {
+//                process((TypeElement) element);
+//            }
+//        }
 
         return false;
     }

libs/vectorized-exec-spi/build.gradle

@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+apply plugin: 'opensearch.build'
+
+description = 'Vectorized engine common interfaces for OpenSearch'
+
+dependencies {
+  api project(':libs:opensearch-core')
+  api project(':libs:opensearch-common')
+
+  testImplementation(project(":test:framework")) {
+    exclude group: 'org.opensearch', module: 'vectorized-exec-spi'
+  }
+}
+
+tasks.named('forbiddenApisMain').configure {
+  replaceSignatureFiles 'jdk-signatures'
+}
+
+jarHell.enabled = false
+
+test {
+  systemProperty 'tests.security.manager', 'false'
+}

libs/vectorized-exec-spi/src/main/java/org/opensearch/vectorized/execution/package-info.java

@@ -0,0 +1,13 @@
+/*
+ * 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.
+ */
+
+/**
+ * DataFusion integration for OpenSearch.
+ * Provides JNI bindings and core functionality for DataFusion query engine.
+ */
+package org.opensearch.vectorized.execution;

libs/vectorized-exec-spi/src/main/java/org/opensearch/vectorized/execution/search/CatalogSearcher.java

@@ -0,0 +1,12 @@
+/*
+ * 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.vectorized.execution.search;
+
+public class CatalogSearcher {
+}

libs/vectorized-exec-spi/src/main/java/org/opensearch/vectorized/execution/search/DataFormat.java

@@ -0,0 +1,34 @@
+/*
+ * 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.vectorized.execution.search;
+
+import org.opensearch.common.annotation.ExperimentalApi;
+
+/**
+ DataFormat supported by OpenSearch
+ */
+@ExperimentalApi
+public enum DataFormat {
+    /** CSV Format*/
+    CSV("parquet"),
+    PARQUET("parquet"),
+
+    /** Text Format */
+    Text("text");
+
+    private final String name;
+
+    DataFormat(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+}

libs/vectorized-exec-spi/src/main/java/org/opensearch/vectorized/execution/search/IndexReader.java

@@ -0,0 +1,12 @@
+/*
+ * 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.vectorized.execution.search;
+
+public class IndexReader {
+}

libs/vectorized-exec-spi/src/main/java/org/opensearch/vectorized/execution/search/spi/ConfigUpdateListener.java

@@ -0,0 +1,19 @@
+/*
+ * 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.vectorized.execution.search.spi;
+
+import org.opensearch.vectorized.execution.search.DataFormat;
+
+/**
+ * Listener for configuration updates.
+ * DataFusionPlugin implements this to receive updates from format-specific plugins (e.g., Parquet).
+ */
+public interface ConfigUpdateListener {
+    void onSessionConfigUpdate(SessionConfig sessionConfig);
+}

libs/vectorized-exec-spi/src/main/java/org/opensearch/vectorized/execution/search/spi/DataSourceCodec.java

@@ -0,0 +1,72 @@
+/*
+ * 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.vectorized.execution.search.spi;
+
+import org.opensearch.vectorized.execution.search.DataFormat;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Service Provider Interface for DataFusion data source codecs.
+ * Implementations provide access to different data formats (CSV, Parquet, etc.)
+ * through the DataFusion query engine.
+ */
+public interface DataSourceCodec {
+
+    /**
+     * Register a directory containing data files with the runtime environment to prewarm cache
+     * This ideally should be used as part of each refresh - equivalent of acquire searcher
+     * where we register the files associated with this particular refresh point
+     * @param directoryPath the path to the directory containing data files
+     * @param fileNames the list of file names to register
+     * @param runtimeId the runtime environment ID
+     * @return a CompletableFuture that completes when registration is done
+     */
+    CompletableFuture<Void> registerDirectory(String directoryPath, List<String> fileNames, long runtimeId);
+
+    /**
+     * Create a new session context for query execution.
+     *
+     * @param globalRuntimeEnvId the global runtime environment ID
+     * @return a CompletableFuture containing the session context ID
+     */
+    CompletableFuture<Long> createSessionContext(long globalRuntimeEnvId);
+
+    /**
+     * Execute a Substrait query plan.
+     *
+     * @param sessionContextId the session context ID
+     * @param substraitPlanBytes the serialized Substrait query plan
+     * @return a CompletableFuture containing the result stream
+     */
+    CompletableFuture<RecordBatchStream> executeSubstraitQuery(long sessionContextId, byte[] substraitPlanBytes);
+
+    /**
+     * Close a session context and free associated resources.
+     *
+     * @param sessionContextId the session context ID to close
+     * @return a CompletableFuture that completes when the context is closed
+     */
+    CompletableFuture<Void> closeSessionContext(long sessionContextId);
+
+    /**
+     * Returns the data format name
+     */
+    DataFormat getDataFormat();
+
+    /**
+     * Override the engine config with the session config
+     * @return a CompletableFuture containing the overridden engine config
+     */
+    EngineConfig updateEngineConfig(EngineConfig config);
+
+    void attachListener(ConfigUpdateListener listener);
+
+}

libs/vectorized-exec-spi/src/main/java/org/opensearch/vectorized/execution/search/spi/EngineConfig.java

@@ -0,0 +1,40 @@
+/*
+ * 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.vectorized.execution.search.spi;
+
+/**
+ * Configuration interface for data formats.
+ * Format-specific configs (e.g., Parquet) publish updates when settings change.
+ * DataFusionPlugin subscribes to receive updates and holds the final config.
+ */
+public interface EngineConfig {
+
+    /**
+     * Gets the session config
+     * @return The session config
+     */
+    SessionConfig getSessionConfig();
+
+    NativeConfiguration getNativeConfiguration();
+
+    /**
+     * Updates the session config by merging values
+     *
+     * @param sessionConfig The session config to merge from
+     * @return
+     */
+    EngineConfig updateSessionConfig(SessionConfig sessionConfig);
+
+    /**
+     * Updates the listing table options
+     * @param nativeConfiguration The new listing table options
+     */
+    EngineConfig updateNativeConfiguration(NativeConfiguration nativeConfiguration);
+
+}