Extract arrow-base plugin

plugins/arrow-base/build.gradle

@@ -0,0 +1,102 @@
+/*
+ * 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 from: "$rootDir/gradle/fips.gradle"
+
+opensearchplugin {
+  description = 'Bundles Apache Arrow and exposes the transport integration contracts ' +
+    'Plugins that carry native Arrow data declare extendedPlugins = [\'arrow-base\'] so they share ' +
+    'one copy of Arrow classes across the cluster.'
+  classname = 'org.opensearch.arrow.plugin.ArrowBasePlugin'
+}
+
+dependencies {
+  // Arrow core
+  api "org.apache.arrow:arrow-vector:${versions.arrow}"
+  api "org.apache.arrow:arrow-format:${versions.arrow}"
+  api "org.apache.arrow:arrow-memory-core:${versions.arrow}"
+  api "org.apache.arrow:arrow-memory-netty:${versions.arrow}"
+  api "org.apache.arrow:arrow-memory-netty-buffer-patch:${versions.arrow}"
+
+  // Transitive deps required by Arrow at runtime
+  api "io.netty:netty-buffer:${versions.netty}"
+  api "io.netty:netty-common:${versions.netty}"
+  api "com.google.flatbuffers:flatbuffers-java:${versions.flatbuffers}"
+  api "org.slf4j:slf4j-api:${versions.slf4j}"
+  api "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson_annotations}"
+  api "com.fasterxml.jackson.core:jackson-databind:${versions.jackson}"
+  api "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:${versions.jackson}"
+  api "commons-codec:commons-codec:${versions.commonscodec}"
+
+  compileOnly 'org.checkerframework:checker-qual:3.44.0'
+}
+
+tasks.named("dependencyLicenses").configure {
+  mapping from: /netty-.*/, to: 'netty'
+  mapping from: /jackson-.*/, to: 'jackson'
+}
+
+test {
+  systemProperty 'io.netty.allocator.numDirectArenas', '1'
+  systemProperty 'io.netty.noUnsafe', 'false'
+  systemProperty 'io.netty.tryUnsafe', 'true'
+  systemProperty 'io.netty.tryReflectionSetAccessible', 'true'
+  jvmArgs += ["--add-opens", "java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED"]
+}
+
+tasks.named('thirdPartyAudit').configure {
+  ignoreMissingClasses(
+    'org.apache.commons.logging.Log',
+    'org.apache.commons.logging.LogFactory',
+
+    // from Log4j (deliberate, Netty will fallback to Log4j 2)
+    'org.apache.log4j.Level',
+    'org.apache.log4j.Logger',
+
+    'reactor.blockhound.BlockHound$Builder',
+    'reactor.blockhound.integration.BlockHoundIntegration'
+  )
+  ignoreViolations(
+    'io.netty.buffer.AbstractAllocatorEvent',
+    'io.netty.buffer.AbstractBufferEvent',
+    'io.netty.buffer.AbstractChunkEvent',
+    'io.netty.buffer.AdaptivePoolingAllocator$AdaptiveByteBuf',
+    'io.netty.buffer.AdaptivePoolingAllocator$Chunk',
+    'io.netty.buffer.AllocateBufferEvent',
+    'io.netty.buffer.AllocateChunkEvent',
+    'io.netty.buffer.FreeBufferEvent',
+    'io.netty.buffer.FreeChunkEvent',
+    'io.netty.buffer.PooledByteBufAllocator',
+    'io.netty.buffer.ReallocateBufferEvent',
+
+    'io.netty.util.internal.PlatformDependent0',
+    'io.netty.util.internal.PlatformDependent0$1',
+    'io.netty.util.internal.PlatformDependent0$2',
+    'io.netty.util.internal.PlatformDependent0$3',
+    'io.netty.util.internal.PlatformDependent0$5',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueConsumerNodeRef',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueProducerNodeRef',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields',
+    'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueProducerFields',
+    'io.netty.util.internal.shaded.org.jctools.queues.LinkedQueueNode',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueConsumerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueProducerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueConsumerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerLimitField',
+    'io.netty.util.internal.shaded.org.jctools.queues.unpadded.MpscUnpaddedArrayQueueConsumerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.unpadded.MpscUnpaddedArrayQueueProducerIndexField',
+    'io.netty.util.internal.shaded.org.jctools.queues.unpadded.MpscUnpaddedArrayQueueProducerLimitField',
+    'io.netty.util.internal.shaded.org.jctools.util.UnsafeAccess',
+    'io.netty.util.internal.shaded.org.jctools.util.UnsafeLongArrayAccess',
+    'io.netty.util.internal.shaded.org.jctools.util.UnsafeRefArrayAccess',
+    'org.apache.arrow.memory.util.MemoryUtil',
+    'org.apache.arrow.memory.util.MemoryUtil$1'
+  )
+}

plugins/arrow-base/src/main/java/org/opensearch/arrow/memory/ArrowAllocatorService.java

@@ -0,0 +1,47 @@
+/*
+ * 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.arrow.memory;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.opensearch.common.annotation.ExperimentalApi;
+
+/**
+ * Node-level Arrow allocator service. Plugins that need an Arrow {@link BufferAllocator}
+ * obtain a child from this service instead of creating their own {@link RootAllocator},
+ * so every Arrow buffer on a node shares the same root. Arrow's
+ * {@link org.apache.arrow.memory.AllocationManager} associate check on cross-allocator
+ * operations requires {@code source.getRoot() == target.getRoot()}, and this is what
+ * makes producer-to-consumer zero-copy handoff work across plugin boundaries.
+ *
+ * <p>One instance per node, bound into Guice by {@code ArrowBasePlugin}. Consumers inject
+ * the interface and do not depend on a concrete implementation.
+ *
+ * @opensearch.experimental
+ */
+@ExperimentalApi
+public interface ArrowAllocatorService {
+
+    /**
+     * Creates a named child allocator with its own memory limit. Callers own the returned
+     * allocator and must close it.
+     *
+     * @param name descriptive name for debugging (e.g., "flight", "analytics-search")
+     * @param limit maximum bytes this child can allocate; Arrow enforces this on every
+     *              {@code buffer()} call and walks the parent chain, throwing
+     *              {@link org.apache.arrow.memory.OutOfMemoryException} on breach
+     */
+    BufferAllocator newChildAllocator(String name, long limit);
+
+    /** Current bytes allocated across all descendants of the root. */
+    long getAllocatedMemory();
+
+    /** Peak bytes allocated since this service started. */
+    long getPeakMemoryAllocation();
+}

plugins/arrow-base/src/main/java/org/opensearch/arrow/memory/DefaultArrowAllocatorService.java

@@ -0,0 +1,63 @@
+/*
+ * 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.arrow.memory;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.Closeable;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+/**
+ * Default {@link ArrowAllocatorService} backed by a single {@link RootAllocator}. The
+ * underlying {@link org.apache.arrow.memory.AllocationManager} implementation is selected
+ * by Arrow based on classpath and system properties; {@code arrow-base} ships
+ * {@code arrow-memory-netty}, so by default Netty-backed allocation is used. Switching to
+ * {@code arrow-memory-unsafe} is a build-time choice and does not require changes here.
+ */
+@SuppressWarnings("removal")
+public final class DefaultArrowAllocatorService implements ArrowAllocatorService, Closeable {
+
+    private static final Logger logger = LogManager.getLogger(DefaultArrowAllocatorService.class);
+    private final RootAllocator root;
+
+    /** Creates a new service with an unbounded root; child allocators carry their own limits. */
+    public DefaultArrowAllocatorService() {
+        this.root = AccessController.doPrivileged((PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE));
+    }
+
+    @Override
+    public BufferAllocator newChildAllocator(String name, long limit) {
+        return root.newChildAllocator(name, 0, limit);
+    }
+
+    @Override
+    public long getAllocatedMemory() {
+        return root.getAllocatedMemory();
+    }
+
+    @Override
+    public long getPeakMemoryAllocation() {
+        return root.getPeakMemoryAllocation();
+    }
+
+    @Override
+    public void close() {
+        try {
+            root.close();
+        } catch (IllegalStateException e) {
+            // Outstanding child allocators remain open — likely a consumer plugin that didn't
+            // clean up. Log at warn so the leak is visible without crashing shutdown.
+            logger.warn("Arrow root allocator closed with outstanding children: {}", e.getMessage());
+        }
+    }
+}

plugins/arrow-base/src/main/java/org/opensearch/arrow/memory/package-info.java

@@ -0,0 +1,15 @@
+/*
+ * 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.
+ */
+
+/**
+ * Node-level Arrow memory management. Plugins that produce or consume Arrow data
+ * obtain child allocators from {@link ArrowAllocatorService},
+ * which roots them under a single per-node {@link org.apache.arrow.memory.RootAllocator}
+ * so cross-plugin buffer handoffs pass Arrow's {@code AllocationManager.associate} check.
+ */
+package org.opensearch.arrow.memory;

plugins/arrow-base/src/main/java/org/opensearch/arrow/plugin/ArrowBasePlugin.java

@@ -0,0 +1,88 @@
+/*
+ * 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.arrow.plugin;
+
+import org.opensearch.arrow.memory.ArrowAllocatorService;
+import org.opensearch.arrow.memory.DefaultArrowAllocatorService;
+import org.opensearch.arrow.transport.ArrowBatchResponse;
+import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.inject.AbstractModule;
+import org.opensearch.common.inject.Module;
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.env.Environment;
+import org.opensearch.env.NodeEnvironment;
+import org.opensearch.plugins.ExtensiblePlugin;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.script.ScriptService;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.client.Client;
+import org.opensearch.watcher.ResourceWatcherService;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Supplier;
+
+/**
+ * Node-level home for Apache Arrow. Bundles the Arrow libraries and exposes the transport
+ * integration contracts ({@link ArrowBatchResponse} and friends) plus the node-level
+ * {@link ArrowAllocatorService}. Plugins that produce or consume native Arrow data declare
+ * {@code extendedPlugins = ['arrow-base']} so they share this plugin's classloader — a
+ * single copy of the Arrow classes lets Arrow-carrying transport types cross plugin
+ * boundaries.
+ *
+ * <p>Implements {@link ExtensiblePlugin} purely for the classloader side-effect of being
+ * a valid parent in {@code extendedPlugins} chains. This plugin does not define or consume
+ * any SPI — do not add {@code loadExtensions()} hooks here.
+ */
+public class ArrowBasePlugin extends Plugin implements ExtensiblePlugin {
+
+    private DefaultArrowAllocatorService allocatorService;
+
+    /** Default constructor. */
+    public ArrowBasePlugin() {}
+
+    @Override
+    public Collection<Object> createComponents(
+        Client client,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ResourceWatcherService resourceWatcherService,
+        ScriptService scriptService,
+        NamedXContentRegistry xContentRegistry,
+        Environment environment,
+        NodeEnvironment nodeEnvironment,
+        NamedWriteableRegistry namedWriteableRegistry,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<RepositoriesService> repositoriesServiceSupplier
+    ) {
+        this.allocatorService = new DefaultArrowAllocatorService();
+        return List.of(allocatorService);
+    }
+
+    @Override
+    public Collection<Module> createGuiceModules() {
+        return List.of(new AbstractModule() {
+            @Override
+            protected void configure() {
+                bind(ArrowAllocatorService.class).toInstance(allocatorService);
+            }
+        });
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (allocatorService != null) {
+            allocatorService.close();
+        }
+    }
+}

plugins/arrow-base/src/main/java/org/opensearch/arrow/plugin/package-info.java

@@ -0,0 +1,10 @@
+/*
+ * 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.
+ */
+
+/** Plugin entry point for the arrow-base plugin. */
+package org.opensearch.arrow.plugin;