Commits of plugin engine-datafusion

.gitattributes

@@ -11,5 +11,6 @@
 *.crt binary
 *.p12 binary
 *.ttf binary
+*.parquet binary
 *.txt text=auto
 CHANGELOG.md merge=union

.github/workflows/datafusion-e2e-test.yml

@@ -0,0 +1,88 @@
+name: DataFusion E2E Integration Test
+
+on:
+  pull_request:
+    branches:
+      - feature/datafusion
+  push:
+    branches:
+      - feature/datafusion
+  workflow_dispatch:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    env:
+      RUSTFLAGS: "-A unused_variables -A unused_mut"
+
+    steps:
+      - name: Checkout OpenSearch
+        uses: actions/checkout@v4
+        with:
+          path: OpenSearch
+
+      - name: Set up JDK
+        uses: actions/setup-java@v4
+        with:
+          distribution: 'temurin'
+          java-version: '21'
+
+      - name: Setup Rust
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          components: rustfmt
+
+      - name: Install Protocol Buffers
+        if: runner.os == 'Linux'
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y protobuf-compiler
+
+      - name: Install Protocol Buffers
+        if: runner.os == 'macOS'
+        run: brew install protobuf
+
+      - name: Install Protocol Buffers
+        if: runner.os == 'Windows'
+        run: choco install protoc
+
+      - name: Setup Gradle
+        uses: gradle/actions/setup-gradle@v3
+
+      - name: Publish OpenSearch to Maven Local
+        working-directory: OpenSearch
+        run: ./gradlew publishToMavenLocal -x test -PrustDebug=true
+
+      - name: Checkout SQL Plugin
+        uses: actions/checkout@v4
+        with:
+          repository: vinaykpud/sql
+          ref: feature/substrait-plan
+          path: opensearch-sql
+
+      - name: Publish SQL Plugin to Maven Local
+        working-directory: opensearch-sql
+        run: ./gradlew publishToMavenLocal -x test
+
+      - name: Run DataFusionReaderManager Tests
+        working-directory: OpenSearch
+        run: ./gradlew :plugins:engine-datafusion:test --tests "org.opensearch.datafusion.DataFusionReaderManagerTests"
+
+      - name: Run IndexFileDeleter Tests
+        working-directory: OpenSearch
+        run: ./gradlew :server:test --tests "org.opensearch.index.engine.exec.coord.IndexFileDeleterTests"
+
+      - name: Run OpenSearch with DataFusion Plugin
+        working-directory: OpenSearch
+        run: |
+          ./gradlew run \
+            --preserve-data \
+            -PremotePlugins="['org.opensearch.plugin:opensearch-job-scheduler:3.3.0.0-SNAPSHOT', 'org.opensearch.plugin:opensearch-sql-plugin:3.3.0.0-SNAPSHOT']" \
+            -PinstalledPlugins="['engine-datafusion']" &
+
+          # Wait for OpenSearch to start
+          timeout 300 bash -c 'until curl -s http://localhost:9200; do sleep 5; done'
+
+      - name: Run SQL CalcitePPLClickBenchIT
+        working-directory: opensearch-sql
+        run: ./gradlew :integ-test:integTest --tests "org.opensearch.sql.calcite.clickbench.CalcitePPLClickBenchIT" -Dtests.method="testDataFusion" -Dtests.cluster=localhost:9200 -Dtests.rest.cluster=localhost:9200 -DignorePrometheus=true -Dtests.clustername=opensearch -Dtests.output=true

.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,15 @@ 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/.rustc_info.json
+
+/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++;

distribution/src/config/jvm.options

@@ -87,6 +87,19 @@ ${error.file}
 21-:-javaagent:agent/opensearch-agent.jar
 21-:--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED
 
+# Heap size settings
+# -Xms32g
+# -Xmx32g
+
+# Enable native memory tracking
+-XX:NativeMemoryTracking=summary
+
+# Allow jcmd to attach to the process (required for 'jcmd VM.native_memory' commands)
+-XX:+UnlockDiagnosticVMOptions
+
+# Enabling debug logs for Allocators in Arrow
+-Darrow.memory.debug.allocator=true
+
 # For cases with high memory-mapped file counts, a lower value can improve stability and
 # prevent issues like "leaked" maps or performance degradation. A value of 1 effectively
 # disables the shared Arena pooling and uses a confined Arena for each MMapDirectory

distribution/src/config/opensearch.yml

@@ -121,3 +121,7 @@ ${path.logs}
 # Once there is no observed impact on performance, this feature flag can be removed.
 #
 #opensearch.experimental.optimization.datetime_formatter_caching.enabled: false
+#
+#
+# Limits the memory pool for datafusion which it uses for query execution.
+#datafusion.search.memory_pool: 1GB

gradle/missing-javadoc.gradle

@@ -160,7 +160,12 @@ 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(":modules:parquet-data-format:benchmarks")
 ]) {
   project.tasks.withType(MissingJavadocTask) {
     isExcluded = true

gradle/run.gradle

@@ -90,6 +90,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/jni/NativeHandle.java

@@ -0,0 +1,94 @@
+/*
+ * 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.jni;
+
+import java.lang.ref.Cleaner;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Base class for type-safe native pointer wrappers.
+ * Provides automatic resource management and prevents use-after-close errors.
+ * Subclasses must implement {@link #doClose()} to release native resources.
+ * Cleaner is used to ensure resources are cleaned up even if the object is not explicitly closed.
+ */
+public abstract class NativeHandle implements AutoCloseable {
+
+    protected final long ptr;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+    protected static final long NULL_POINTER = 0L;
+    private final Cleaner.Cleanable cleanable;
+
+    private static final Cleaner CLEANER = Cleaner.create();
+
+    /**
+     * Creates a new native handle.
+     * @param ptr the native pointer (must not be 0)
+     * @throws IllegalArgumentException if ptr is 0
+     */
+    protected NativeHandle(long ptr) {
+        if (ptr == NULL_POINTER) {
+            throw new IllegalArgumentException("Null native pointer");
+        }
+        this.ptr = ptr;
+        this.cleanable = CLEANER.register(this, new CleanupAction(ptr, this::doClose));
+    }
+
+    /**
+     * Ensures the handle is still open.
+     * @throws IllegalStateException if the handle has been closed
+     */
+    public void ensureOpen() {
+        if (closed.get()) {
+            throw new IllegalStateException("Handle already closed");
+        }
+    }
+
+    /**
+     * Gets the native pointer value.
+     * @return the native pointer
+     * @throws IllegalStateException if the handle has been closed
+     */
+    public long getPointer() {
+        ensureOpen();
+        return ptr;
+    }
+
+    @Override
+    public void close() {
+        if (closed.compareAndSet(false, true)) {
+            cleanable.clean();
+        }
+    }
+
+    /**
+     * Releases the native resource.
+     * Called once when the handle is closed.
+     * Subclasses must implement this to free native memory.
+     */
+    protected abstract void doClose();
+
+    /**
+     * Cleans up the native resource.
+     * Called by the cleaner when the handle is garbage collected.
+     */
+    private static final class CleanupAction implements Runnable {
+        private final long ptr;
+        private final Runnable doClose;
+
+        CleanupAction(long ptr, Runnable doClose) {
+            this.ptr = ptr;
+            this.doClose = doClose;
+        }
+
+        @Override
+        public void run() {
+            doClose.run();
+        }
+    }
+}