Base Flight producer integration with Flight Server

CHANGELOG.md

@@ -38,6 +38,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Added new Setting property UnmodifiableOnRestore to prevent updating settings on restore snapshot ([#16957](https://github.com/opensearch-project/OpenSearch/pull/16957))
 - Introduce Template query ([#16818](https://github.com/opensearch-project/OpenSearch/pull/16818))
 - Propagate the sourceIncludes and excludes fields from fetchSourceContext to FieldsVisitor. ([#17080](https://github.com/opensearch-project/OpenSearch/pull/17080))
+- Arrow Flight server bootstrap logic ([#16962](https://github.com/opensearch-project/OpenSearch/pull/16962))
 
 ### Dependencies
 - Bump `com.google.cloud:google-cloud-core-http` from 2.23.0 to 2.47.0 ([#16504](https://github.com/opensearch-project/OpenSearch/pull/16504))

distribution/src/config/jvm.options

@@ -89,3 +89,4 @@ ${error.file}
 # See please https://bugs.openjdk.org/browse/JDK-8341127 (openjdk/jdk#21283)
 23:-XX:CompileCommand=dontinline,java/lang/invoke/MethodHandle.setAsTypeCache
 23:-XX:CompileCommand=dontinline,java/lang/invoke/MethodHandle.asTypeUncached
+24:-add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED

gradle/libs.versions.toml

@@ -82,7 +82,7 @@ opentelemetry         = "1.46.0"
 opentelemetrysemconv  = "1.29.0-alpha"
 
 # arrow dependencies
-arrow                 = "17.0.0"
+arrow                 = "18.1.0"
 flatbuffers           = "2.0.0"
 
 [libraries]

libs/arrow-memory-shaded/build.gradle

@@ -0,0 +1,82 @@
+/*
+ * 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.
+ */
+
+plugins {
+  id("com.gradleup.shadow")
+}
+
+description = '''Shading dependencies under opensearch namespace for arrow-memory-netty, netty-buffer, netty-common and other common transitive dependencies like
+sl4j, jackson, flatbuffers and common-codec to prevent pollution of server classpath'''
+
+testingConventions.enabled = false
+generatePomFileForShadowPublication.enabled = false
+
+dependencies {
+  runtimeOnly "org.apache.arrow:arrow-memory-netty:${versions.arrow}"
+  runtimeOnly "org.apache.arrow:arrow-memory-core:${versions.arrow}"
+  runtimeOnly "org.apache.arrow:arrow-memory-netty-buffer-patch:${versions.arrow}"
+  runtimeOnly "io.netty:netty-buffer:${versions.netty}"
+  runtimeOnly "io.netty:netty-common:${versions.netty}"
+  runtimeOnly "org.apache.arrow:arrow-vector:${versions.arrow}"
+  runtimeOnly "org.apache.arrow:arrow-format:${versions.arrow}"
+
+  runtimeOnly 'org.checkerframework:checker-qual:3.44.0'
+  runtimeOnly "com.google.flatbuffers:flatbuffers-java:${versions.flatbuffers}"
+  runtimeOnly "org.slf4j:slf4j-api:${versions.slf4j}"
+  runtimeOnly "com.fasterxml.jackson.core:jackson-core:${versions.jackson}"
+  runtimeOnly "com.fasterxml.jackson.core:jackson-databind:${versions.jackson}"
+  runtimeOnly "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}"
+  runtimeOnly "commons-codec:commons-codec:${versions.commonscodec}"
+}
+
+tasks.named("shadowJar").configure {
+  dependencies {
+    include(dependency('org.apache.arrow:'))
+    include(dependency('io.netty:'))
+    include(dependency('org.checkerframework:'))
+    include(dependency('org.slf4j:'))
+    include(dependency('com.google.flatbuffers:'))
+    include(dependency('com.fasterxml.jackson.core:'))
+    include(dependency('commons-codec:'))
+  }
+  manifest {
+    attributes('Multi-Release': 'true')
+  }
+  exclude 'META-INF/maven/**'
+  relocate 'io.netty', 'org.opensearch.shaded.io.netty'
+  relocate 'org.checkerframework', 'org.opensearch.shaded.org.checkerframework'
+  relocate 'org.apache.commons.codec', 'org.opensearch.shaded.commons.codec'
+  relocate 'org.slf4j', 'org.opensearch.shaded.org.slf4j'
+  relocate 'com.google.flatbuffers', 'org.opensearch.shaded.com.google.flatbuffers'
+  relocate 'com.fasterxml.jackson', 'org.opensearch.shaded.com.fasterxml.jackson'
+  mergeServiceFiles()
+  archiveClassifier.set("")
+}
+
+tasks.named('thirdPartyAudit').configure {
+  enabled = false
+}
+
+tasks.register("javadocDummyFiles") {
+  doFirst {
+    def javadocDir = layout.buildDirectory.dir('docs/javadoc').get().asFile
+    javadocDir.mkdirs()
+
+    new File(javadocDir, "element-list").text = ""
+    new File(javadocDir, "package-list").text = ""
+  }
+}
+
+tasks.named('javadoc') { javadocTask ->
+  enabled = false
+  finalizedBy(tasks.named('javadocDummyFiles'))
+}
+
+tasks.named("dependencyLicenses").configure {
+  mapping from: /jackson-.*/, to: 'jackson'
+}

libs/arrow-memory-shaded/licenses/arrow-format-18.1.0.jar.sha1

@@ -0,0 +1 @@
+9d356b6f20620f5619ff85b174f97ae507df4997

libs/arrow-memory-shaded/licenses/arrow-memory-core-18.1.0.jar.sha1

@@ -0,0 +1 @@
+35f4853d512f06759759b40b53bac850867886f8

libs/arrow-memory-shaded/licenses/arrow-memory-netty-18.1.0.jar.sha1

@@ -0,0 +1 @@
+9e9e08d0b548d2c02c632e5daaf176e588810d22

libs/arrow-memory-shaded/licenses/arrow-memory-netty-buffer-patch-18.1.0.jar.sha1

@@ -0,0 +1 @@
+86c8fbdb6ab220603ea3a215f48a7f793ac6a08d

libs/arrow-memory-shaded/licenses/arrow-vector-18.1.0.jar.sha1

@@ -0,0 +1 @@
+b1fb77f4ef36fd52afe480ba12b7da77367eb88c

libs/arrow-memory-shaded/licenses/checker-qual-3.44.0.jar.sha1

@@ -0,0 +1 @@
+e026b198319ea9dd3f221fab367d2099215079e5

libs/arrow-memory-shaded/licenses/checker-qual-LICENSE.txt

@@ -0,0 +1,22 @@
+Checker Framework qualifiers
+Copyright 2004-present by the Checker Framework developers
+
+MIT License:
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

libs/arrow-memory-shaded/licenses/jackson-core-2.18.2.jar.sha1

@@ -0,0 +1 @@
+fb64ccac5c27dca8819418eb4e443a9f496d9ee7

libs/arrow-spi/build.gradle

@@ -10,22 +10,10 @@
  */
 
 testingConventions.enabled = false
+
 dependencies {
   api project(':libs:opensearch-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}"
-  runtimeOnly "org.apache.arrow:arrow-memory-netty-buffer-patch:${versions.arrow}"
-  runtimeOnly "org.apache.arrow:arrow-memory-netty:${versions.arrow}"
-  runtimeOnly "io.netty:netty-buffer:${versions.netty}"
-  runtimeOnly "io.netty:netty-common:${versions.netty}"
-
-  runtimeOnly "com.google.flatbuffers:flatbuffers-java:${versions.flatbuffers}"
-  runtimeOnly "org.slf4j:slf4j-api:${versions.slf4j}"
-  runtimeOnly "com.fasterxml.jackson.core:jackson-databind:${versions.jackson}"
-  api "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}"
-
-  implementation "commons-codec:commons-codec:${versions.commonscodec}"
+  api project(path:':libs:opensearch-arrow-memory-shaded', configuration: 'shadow')
 }
 
 tasks.named('forbiddenApisMain').configure {
@@ -34,55 +22,8 @@ tasks.named('forbiddenApisMain').configure {
 
 tasks.named('thirdPartyAudit').configure {
   ignoreMissingClasses(
-    // Logging frameworks
-    'org.apache.commons.logging.Log',
-    'org.apache.commons.logging.LogFactory',
-    'org.apache.log4j.Level',
-    'org.apache.log4j.Logger',
-    'org.slf4j.impl.StaticLoggerBinder',
-    'org.slf4j.impl.StaticMDCBinder',
-    'org.slf4j.impl.StaticMarkerBinder',
-
-    // Reactor BlockHound
-    'reactor.blockhound.BlockHound$Builder',
-    'reactor.blockhound.integration.BlockHoundIntegration'
-  )
-
-  ignoreViolations(
-    "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\$4",
-    "io.netty.util.internal.PlatformDependent0\$6",
-    "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.util.UnsafeAccess",
-    "io.netty.util.internal.shaded.org.jctools.util.UnsafeLongArrayAccess",
-    "io.netty.util.internal.shaded.org.jctools.util.UnsafeRefArrayAccess",
-    "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",
-    "org.apache.arrow.memory.ArrowBuf",
-    "org.apache.arrow.memory.util.ByteFunctionHelpers",
-    "org.apache.arrow.memory.util.MemoryUtil",
-    "org.apache.arrow.memory.util.MemoryUtil\$1",
-    "org.apache.arrow.memory.util.hash.MurmurHasher",
-    "org.apache.arrow.memory.util.hash.SimpleHasher",
-    "org.apache.arrow.vector.BaseFixedWidthVector",
-    "org.apache.arrow.vector.BitVectorHelper",
-    "org.apache.arrow.vector.Decimal256Vector",
-    "org.apache.arrow.vector.DecimalVector",
-    "org.apache.arrow.vector.util.DecimalUtility",
-    "org.apache.arrow.vector.util.VectorAppender"
+    'org.slf4j.Logger',
+    'org.slf4j.LoggerFactory',
+    'org.apache.commons.codec.binary.Hex'
   )
 }

libs/arrow-spi/src/main/java/org/opensearch/arrow/spi/StreamProducer.java

@@ -11,6 +11,7 @@
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.opensearch.common.annotation.ExperimentalApi;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.tasks.TaskId;
 
 import java.io.Closeable;
@@ -86,7 +87,7 @@ public interface StreamProducer extends Closeable {
      * @param allocator The allocator to use for creating vectors
      * @return A new VectorSchemaRoot instance
      */
-    VectorSchemaRoot createRoot(BufferAllocator allocator);
+    VectorSchemaRoot createRoot(BufferAllocator allocator) throws Exception;
 
     /**
      * Creates a job that will produce the stream data in batches. The job will populate
@@ -97,6 +98,14 @@ public interface StreamProducer extends Closeable {
      */
     BatchedJob createJob(BufferAllocator allocator);
 
+    /**
+     * Returns the deadline for the job execution.
+     * After this deadline, the job should be considered expired.
+     *
+     * @return TimeValue representing the job's deadline
+     */
+    TimeValue getJobDeadline();
+
     /**
      * Provides an estimate of the total number of rows that will be produced.
      *
@@ -122,7 +131,7 @@ interface BatchedJob {
          * @param root The VectorSchemaRoot to populate with data
          * @param flushSignal Signal to coordinate with consumers
          */
-        void run(VectorSchemaRoot root, FlushSignal flushSignal);
+        void run(VectorSchemaRoot root, FlushSignal flushSignal) throws Exception;
 
         /**
          * Called to signal producer when the job is canceled.
@@ -150,8 +159,8 @@ interface FlushSignal {
         /**
          * Blocks until the current batch has been consumed or timeout occurs.
          *
-         * @param timeout Maximum milliseconds to wait
+         * @param timeout Maximum time to wait
          */
-        void awaitConsumption(int timeout);
+        void awaitConsumption(TimeValue timeout);
     }
 }

libs/arrow-spi/src/main/java/org/opensearch/arrow/spi/StreamReader.java

@@ -43,14 +43,16 @@ public interface StreamReader extends Closeable {
      * Blocking request to load next batch into root.
      *
      * @return true if more data was found, false if the stream is exhausted
+     * @throws Exception if an error occurs while loading the next batch
      */
-    boolean next();
+    boolean next() throws Exception;
 
     /**
      * Returns the VectorSchemaRoot associated with this iterator.
      * The content of this root is updated with each successful call to next().
      *
      * @return the VectorSchemaRoot
+     * @throws Exception if an error occurs while retrieving the root
      */
-    VectorSchemaRoot getRoot();
+    VectorSchemaRoot getRoot() throws Exception;
 }

libs/build.gradle

@@ -44,12 +44,12 @@ subprojects {
       dependencies.matching { it instanceof ProjectDependency }.all { ProjectDependency dep ->
         Project depProject = project.project(dep.path)
         if (depProject != null
-          && (false == depProject.path.equals(':libs:opensearch-core') &&
+          && (false == depProject.path.equals(':libs:opensearch-core') && false == depProject.path.equals(':libs:opensearch-arrow-memory-shaded') &&
               false == depProject.path.equals(':libs:opensearch-common'))
           && depProject.path.startsWith(':libs')) {
           throw new InvalidUserDataException("projects in :libs "
             + "may not depend on other projects libs except "
-            + ":libs:opensearch-core or :libs:opensearch-common but "
+            + ":libs:opensearch-core or :libs:opensearch-common  or :libs:opensearch-arrow-memory-shaded but "
             + "${project.path} depends on ${depProject.path}")
         }
       }

libs/flight-core-shaded/build.gradle

@@ -0,0 +1,56 @@
+/*
+ * 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.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+plugins {
+  id("com.gradleup.shadow")
+}
+
+description = '''arrow-flight-core netty shaded dependencies to make it compatible with grpc-netty-shaded'''
+
+testingConventions.enabled = false
+
+// since we are not publishing it
+generatePomFileForShadowPublication.enabled = false
+
+dependencies {
+  runtimeOnly "org.apache.arrow:flight-core:${versions.arrow}"
+}
+
+shadowJar {
+  dependencies {
+    include(dependency('org.apache.arrow:'))
+  }
+  exclude 'META-INF/maven/**'
+  relocate 'io.grpc.netty', 'io.grpc.netty.shaded.io.grpc.netty'
+  relocate 'io.netty', 'io.grpc.netty.shaded.io.netty'
+  mergeServiceFiles()
+  archiveClassifier.set("")
+}
+
+tasks.named('thirdPartyAudit').configure {
+  enabled = false
+}
+
+// creating dummy javadoc to let javadoc check pass when other package use it as a dependency
+tasks.register("javadocDummyFiles") {
+  doFirst {
+    def javadocDir = layout.buildDirectory.dir('docs/javadoc').get().asFile
+    javadocDir.mkdirs()
+
+    new File(javadocDir, "element-list").text = ""
+    new File(javadocDir, "package-list").text = ""
+  }
+}
+
+tasks.named('javadoc') { javadocTask ->
+  enabled = false
+  finalizedBy(tasks.named('javadocDummyFiles'))
+}

libs/flight-core-shaded/licenses/flight-core-18.1.0.jar.sha1

@@ -0,0 +1 @@
+82494895fcb0656967680442f63ce1214e532d52