native interface abstractions and indexing integration with sandbox plugins (#20926)

Signed-off-by: bharath-techie <bharath78910@gmail.com>

Author: bharath-techie

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;
     }

plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DataFusionAnalyticsBackend.java

@@ -14,12 +14,8 @@
 import org.opensearch.index.engine.exec.coord.CatalogSnapshot;
 
 /**
- * SPI adapter that delegates to the real {@link DataFusionPlugin} instance.
- *
- * <p>{@code PluginsService} extension discovery requires either a no-arg constructor
- * or a constructor taking the enclosing plugin. Since {@code DataFusionPlugin}
- * itself needs a {@code Settings} constructor for plugin loading, this thin
- * adapter bridges the two requirements.
+ * SPI adapter that delegates to the real {@link DataFusionPlugin} instance,
+ * or to a child backend plugin if one was discovered via ExtensiblePlugin.
  */
 public class DataFusionAnalyticsBackend implements AnalyticsBackEndPlugin {
 
@@ -29,18 +25,32 @@ public DataFusionAnalyticsBackend(DataFusionPlugin plugin) {
         this.plugin = plugin;
     }
 
+    private AnalyticsBackEndPlugin delegate() {
+        // If a child plugin (e.g. sandbox analytics-backend-datafusion) registered,
+        // delegate to it so it can override the bridge implementation.
+        if (plugin.getChildBackends().isEmpty() == false) {
+            return plugin.getChildBackends().get(0);
+        }
+        return plugin;
+    }
+
     @Override
     public String name() {
         return plugin.name();
     }
 
     @Override
     public EngineBridge<?, ?, ?> bridge(CatalogSnapshot snapshot) {
-        return plugin.bridge(snapshot);
+        return delegate().bridge(snapshot);
     }
 
     @Override
     public SqlOperatorTable operatorTable() {
-        return plugin.operatorTable();
+        return delegate().operatorTable();
+    }
+
+    @Override
+    public boolean supportsSearchExecEngine() {
+        return delegate().supportsSearchExecEngine();
     }
 }

plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DataFusionPlugin.java

@@ -35,7 +35,9 @@
 import org.opensearch.index.engine.exec.FileMetadata;
 import org.opensearch.index.shard.ShardPath;
 import org.opensearch.plugins.ActionPlugin;
+import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.SearchAnalyticsBackEndPlugin;
 import org.opensearch.plugins.SearchEnginePlugin;
 import org.opensearch.plugins.spi.vectorized.DataFormat;
 import org.opensearch.plugins.spi.vectorized.DataSourceCodec;
@@ -72,9 +74,13 @@
  * Main plugin class for OpenSearch DataFusion integration.
  *
  */
-public class DataFusionPlugin extends Plugin implements ActionPlugin, SearchEnginePlugin, AnalyticsBackEndPlugin {
+public class DataFusionPlugin extends Plugin implements ActionPlugin, SearchEnginePlugin, AnalyticsBackEndPlugin, ExtensiblePlugin, SearchAnalyticsBackEndPlugin {
 
     private DataFusionService dataFusionService;
+
+    public DataFusionService getDataFusionService() {
+        return dataFusionService;
+    }
     private final boolean isDataFusionEnabled;
 
     /**
@@ -224,6 +230,56 @@ public String name() {
     public SqlOperatorTable operatorTable() {
         return null;
     }
+
+    // Forward AnalyticsBackEndPlugin extensions from child plugins (e.g. analytics-backend-datafusion)
+    private final List<AnalyticsBackEndPlugin> childBackends = new ArrayList<>();
+
+    @Override
+    public void loadExtensions(ExtensionLoader loader) {
+        for (AnalyticsBackEndPlugin ext : loader.loadExtensions(AnalyticsBackEndPlugin.class)) {
+            // Inject ourselves so child backends can access the DataFusionService
+            if (ext instanceof ParentAware) {
+                ((ParentAware) ext).setParentPlugin(this);
+            }
+            childBackends.add(ext);
+        }
+    }
+
+    public List<AnalyticsBackEndPlugin> getChildBackends() {
+        return childBackends;
+    }
+
+    /** Marker interface for child backends that need the parent plugin. */
+    // ---- SearchAnalyticsBackEndPlugin (delegates to child backend if available) ----
+
+    private SearchAnalyticsBackEndPlugin getChildSearchBackend() {
+        for (AnalyticsBackEndPlugin child : childBackends) {
+            if (child instanceof SearchAnalyticsBackEndPlugin) {
+                return (SearchAnalyticsBackEndPlugin) child;
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public org.opensearch.index.engine.exec.CatalogSnapshotAwareReaderManager<?> createReaderManager(
+            org.opensearch.plugins.spi.vectorized.DataFormat format, ShardPath shardPath) throws IOException {
+        SearchAnalyticsBackEndPlugin child = getChildSearchBackend();
+        if (child != null) return child.createReaderManager(format, shardPath);
+        return null;
+    }
+
+    @Override
+    public org.opensearch.index.engine.exec.SearchExecEngine<?, ?> createSearchExecEngine(
+            org.opensearch.plugins.spi.vectorized.DataFormat format, ShardPath shardPath) throws IOException {
+        SearchAnalyticsBackEndPlugin child = getChildSearchBackend();
+        if (child != null) return child.createSearchExecEngine(format, shardPath);
+        return null;
+    }
+
+    public interface ParentAware {
+        void setParentPlugin(DataFusionPlugin parent);
+    }
 //
 //    @Override
 //    public List<Setting<?>> getSettings() {

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/backend/jni/NativeHandle.java

@@ -0,0 +1,70 @@
+/*
+ * 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.analytics.backend.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.
+ */
+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();
+
+    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));
+    }
+
+    public void ensureOpen() {
+        if (closed.get()) {
+            throw new IllegalStateException("Handle already closed");
+        }
+    }
+
+    public long getPointer() {
+        ensureOpen();
+        return ptr;
+    }
+
+    @Override
+    public void close() {
+        if (closed.compareAndSet(false, true)) {
+            cleanable.clean();
+        }
+    }
+
+    protected abstract void doClose();
+
+    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();
+        }
+    }
+}

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsBackEndPlugin.java

@@ -12,8 +12,9 @@
 import org.opensearch.analytics.backend.EngineBridge;
 import org.opensearch.index.engine.exec.coord.CatalogSnapshot;
 
+
 /**
- * SPI extension point for back-end query engines (DataFusion, Lucene, etc.).
+ * SPI extension point for back-end query engines (DataFusion, Lucene, etc.)
  * @opensearch.internal
  */
 public interface AnalyticsBackEndPlugin {
@@ -25,4 +26,9 @@ public interface AnalyticsBackEndPlugin {
 
     /** Supported functions as a Calcite operator table, or null if the back-end adds no functions. */
     SqlOperatorTable operatorTable();
+
+    /** Whether this backend supports the SearchExecEngine path via CompositeEngine. */
+    default boolean supportsSearchExecEngine() {
+        return false;
+    }
 }

sandbox/plugins/analytics-backend-datafusion/build.gradle

@@ -9,14 +9,28 @@
 opensearchplugin {
   description = 'DataFusion native execution engine plugin for the query engine.'
   classname = 'org.opensearch.be.datafusion.DataFusionPlugin'
-  extendedPlugins = ['analytics-engine']
+  extendedPlugins = ['engine-datafusion']
 }
 
 dependencies {
-  // Shared types and SPI interfaces (EngineBridge, AnalyticsBackEndPlugin, etc.)
-  // Also provides calcite-core transitively via api.
-  api project(':sandbox:libs:analytics-framework')
+  // All compileOnly — provided by parent classloaders at runtime
+  compileOnly project(':sandbox:libs:analytics-framework')
+  compileOnly project(':plugins:engine-datafusion')
+  compileOnly project(':libs:opensearch-vectorized-exec-spi')
+  compileOnly "org.apache.arrow:arrow-memory-core:18.3.0"
+  compileOnly "org.apache.arrow:arrow-vector:18.3.0"
+  compileOnly "org.apache.arrow:arrow-c-data:18.3.0"
+  compileOnly "org.apache.arrow:arrow-format:18.3.0"
+  compileOnly "org.apache.calcite:calcite-core:1.41.0"
+  compileOnly "com.google.protobuf:protobuf-java:${versions.protobuf}"
+
+  // Substrait — only new jars we bundle
+  implementation('io.substrait:core:0.67.0') { transitive = false }
+  implementation('io.substrait:isthmus:0.67.0') { transitive = false }
 }
 
-// TODO: Remove once back-end is built out with test suite
 testingConventions.enabled = false
+
+tasks.withType(JavaCompile).configureEach {
+  options.compilerArgs -= '-Werror'
+}

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackend.java

@@ -0,0 +1,84 @@
+/*
+ * 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.be.datafusion;
+
+import org.apache.calcite.sql.SqlOperatorTable;
+import org.opensearch.analytics.backend.EngineBridge;
+import org.opensearch.analytics.spi.AnalyticsBackEndPlugin;
+import org.opensearch.index.engine.exec.CatalogSnapshotAwareReaderManager;
+import org.opensearch.index.engine.exec.SearchExecEngine;
+import org.opensearch.index.engine.exec.WriterFileSet;
+import org.opensearch.index.engine.exec.coord.CatalogSnapshot;
+import org.opensearch.index.shard.ShardPath;
+import org.opensearch.plugins.SearchAnalyticsBackEndPlugin;
+import org.opensearch.plugins.spi.vectorized.DataFormat;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+
+public class DataFusionAnalyticsBackend
+    implements AnalyticsBackEndPlugin, SearchAnalyticsBackEndPlugin, org.opensearch.datafusion.DataFusionPlugin.ParentAware {
+
+    private org.opensearch.datafusion.DataFusionPlugin parentPlugin;
+
+    public DataFusionAnalyticsBackend() {}
+
+    @Override
+    public void setParentPlugin(org.opensearch.datafusion.DataFusionPlugin parent) {
+        this.parentPlugin = parent;
+    }
+
+    private long getRuntimePtr() {
+        return parentPlugin.getDataFusionService().getRuntimePointer();
+    }
+
+    @Override
+    public String name() {
+        return "DataFusion";
+    }
+
+    @Override
+    public EngineBridge<?, ?, ?> bridge(CatalogSnapshot snapshot) {
+        Collection<WriterFileSet> files = snapshot.getSearchableFiles("parquet");
+        if (files.isEmpty() || files.stream().allMatch(wfs -> wfs.getFiles().isEmpty())) {
+            throw new IllegalStateException("No parquet files available in catalog snapshot");
+        }
+        String dir = files.stream().findFirst().map(WriterFileSet::getDirectory).orElse("");
+        DatafusionReader reader = new DatafusionReader(dir, files);
+        return new SandboxDataFusionBridge(getRuntimePtr(), reader);
+    }
+
+    @Override
+    public SqlOperatorTable operatorTable() {
+        return null;
+    }
+
+    @Override
+    public boolean supportsSearchExecEngine() {
+        return true;
+    }
+
+    // ---- SearchAnalyticsBackEndPlugin ----
+
+    @Override
+    public List<DataFormat> getSupportedFormats() {
+        return List.of(DataFormat.PARQUET);
+    }
+
+    @Override
+    public CatalogSnapshotAwareReaderManager<?> createReaderManager(DataFormat format, ShardPath shardPath) throws IOException {
+        return new DatafusionReaderManager(format, shardPath);
+    }
+
+    @Override
+    public SearchExecEngine<?, ?> createSearchExecEngine(DataFormat format, ShardPath shardPath) throws IOException {
+        return new DatafusionSearchExecEngine(getRuntimePtr());
+    }
+}