implement JNI interface on both size

cpp/src/arrow/adapters/orc/jni/jni_wrapper.cpp

@@ -15,43 +15,227 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <string>
+#include <arrow/adapters/orc/adapter.h>
+#include <arrow/array.h>
+#include <arrow/buffer.h>
 #include <arrow/io/api.h>
+#include <arrow/ipc/api.h>
+#include <arrow/util/logging.h>
+#include <cassert>
+#include <string>
 
+#include "org_apache_arrow_adapter_orc_OrcMemoryJniWrapper.h"
 #include "org_apache_arrow_adapter_orc_OrcReaderJniWrapper.h"
-#include "org_apache_arrow_adapter_orc_OrcReaderJniWrapper_StripeReader.h"
+#include "org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper.h"
+
+using ORCFileReader = arrow::adapters::orc::ORCFileReader;
+using RecordBatchReader = arrow::RecordBatchReader;
+
+void ThrowJavaIOException(JNIEnv* env, const std::exception& e) {
+  jclass ioExceptionClass = env->FindClass("java/io/IOException");
+  if (ioExceptionClass != nullptr) {
+    if (env->ThrowNew(ioExceptionClass, e.what())) {
+      // Failed to new IOException. This means another error has occurred in Java
+      // We just propagate this error to caller by doing nothing.
+      ARROW_LOG(ERROR) << "Error occurred when throwing IOException";
+    }
+  } else {
+    ARROW_LOG(ERROR) << "Error occurred when getting IOException class";
+  }
+}
+
+void ThrowJavaException(JNIEnv* env, const std::string& message) {
+  jclass exception = env->FindClass("java/lang/Exception");
+  if (exception != nullptr) {
+    env->ThrowNew(exception, message.c_str());
+  } else {
+    throw std::runtime_error("Can't find java/lang/Exception class");
+  }
+}
+
+jfieldID GetFieldId(JNIEnv* env, jclass this_class, const std::string& sig) {
+  jfieldID ret = env->GetFieldID(this_class, sig.c_str(), "J");
+  if (ret == nullptr) {
+    ThrowJavaException(env, "Unable to get java class field: " + sig);
+  }
+
+  return ret;
+}
+
+std::unique_ptr<ORCFileReader>* GetNativeReader(JNIEnv* env, jobject this_obj) {
+  jlong reader = env->GetLongField(
+      this_obj, GetFieldId(env, env->GetObjectClass(this_obj), "nativeReaderAddress"));
+  return reinterpret_cast<std::unique_ptr<ORCFileReader>*>(reader);
+}
+
+std::shared_ptr<RecordBatchReader>* GetStripeReader(JNIEnv* env, jobject this_obj) {
+  jlong reader = env->GetLongField(
+      this_obj,
+      GetFieldId(env, env->GetObjectClass(this_obj), "nativeStripeReaderAddress"));
+  return reinterpret_cast<std::shared_ptr<RecordBatchReader>*>(reader);
+}
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-JNIEXPORT jboolean JNICALL Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_open
-        (JNIEnv *env, jobject this_obj, jstring file_path) {
-    auto in_file = new std::shared_ptr<arrow::io::ReadableFile>();
-    const char *str= env->GetStringUTFChars(file_path, nullptr);
-    std::string path(str);
-    env->ReleaseStringUTFChars(file_path, str);
+JNIEXPORT jboolean JNICALL Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_open(
+    JNIEnv* env, jobject this_obj, jstring file_path) {
+  std::shared_ptr<arrow::io::ReadableFile> in_file;
+  const char* str = env->GetStringUTFChars(file_path, nullptr);
+  std::string path(str);
+  env->ReleaseStringUTFChars(file_path, str);
 
-    arrow::Status ret;
-    if (path.find("hdfs://") == 0) {
-        return false;
-    } else {
-        ret = arrow::io::ReadableFile::Open(path, in_file);
+  arrow::Status ret;
+  if (path.find("hdfs://") == 0) {
+    return false;
+  } else {
+    ret = arrow::io::ReadableFile::Open(path, &in_file);
+  }
+
+  if (ret.ok()) {
+    auto reader = new std::unique_ptr<ORCFileReader>();
+
+    ret = ORCFileReader::Open(
+        std::static_pointer_cast<arrow::io::RandomAccessFile>(in_file),
+        arrow::default_memory_pool(), reader);
+
+    if (!ret.ok()) {
+      delete reader;
+      ThrowJavaIOException(env, std::invalid_argument("Failed open file" + path));
     }
 
-    if(ret.ok()) {
+    env->SetLongField(
+        this_obj, GetFieldId(env, env->GetObjectClass(this_obj), "nativeReaderAddress"),
+        reinterpret_cast<long>(reader));
+  }
 
-        jclass this_class = env->GetObjectClass(this_obj);
-        jfieldID fidNativeManager = env->GetFieldID(this_class, "nativeReaderAddress", "J");
-        if (fidNativeManager == nullptr)
-        {
-            return false;
-        }
+  return ret.ok();
+}
 
-        env->SetLongField(this_obj, fidNativeManager, (long)in_file);
+JNIEXPORT void JNICALL Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_close(
+    JNIEnv* env, jobject this_obj) {
+  delete GetNativeReader(env, this_obj);
+}
+
+JNIEXPORT jboolean JNICALL Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_seek(
+    JNIEnv* env, jobject this_obj, jint row_number) {
+  auto reader = GetNativeReader(env, this_obj);
+  return (*reader)->Seek(row_number).ok();
+}
+
+JNIEXPORT jint JNICALL
+Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_getNumberOfStripes(
+    JNIEnv* env, jobject this_obj) {
+  auto reader = GetNativeReader(env, this_obj);
+  return (*reader)->NumberOfStripes();
+}
+
+JNIEXPORT jobject JNICALL
+Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_nextStripeReader(JNIEnv* env,
+                                                                       jobject this_obj,
+                                                                       jlong batch_size) {
+  auto reader = GetNativeReader(env, this_obj);
+  auto stripe_reader = new std::shared_ptr<RecordBatchReader>();
+  auto status = (*reader)->NextStripeReader(batch_size, stripe_reader);
+
+  jobject ret = nullptr;
+  if (!status.ok()) {
+    delete stripe_reader;
+    return ret;
+  }
+
+  jclass cls = env->FindClass("/org/apache/arrow/adapter/orc/OrcStripeReaderJniWrapper");
+  ret = env->AllocObject(cls);
+
+  env->SetLongField(ret, GetFieldId(env, cls, "nativeStripeReaderAddress"),
+                    reinterpret_cast<long>(stripe_reader));
+
+  return ret;
+}
+
+JNIEXPORT jbyteArray JNICALL
+Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_getSchema(JNIEnv* env,
+                                                                      jobject this_obj) {
+  auto stripe_reader = GetStripeReader(env, this_obj);
+  auto schema = (*stripe_reader)->schema();
+
+  std::shared_ptr<arrow::Buffer> out;
+  auto status = arrow::ipc::SerializeSchema(*schema, arrow::default_memory_pool(), &out);
+  if (!status.ok()) {
+    return nullptr;
+  }
+
+  jbyteArray ret = env->NewByteArray(out->size());
+  memcpy(env->GetByteArrayElements(ret, nullptr), out->data(), out->size());
+  return ret;
+}
+
+JNIEXPORT jobject JNICALL
+Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_next(JNIEnv* env,
+                                                                 jobject this_obj) {
+  auto stripe_reader = GetStripeReader(env, this_obj);
+  auto record_batch = new std::shared_ptr<arrow::RecordBatch>();
+  auto status = (*stripe_reader)->ReadNext(record_batch);
+  if (!status.ok()) {
+    delete record_batch;
+    return nullptr;
+  }
+
+  auto schema = (*stripe_reader)->schema();
+
+  // create OrcFieldNode[]
+  jclass field_class = env->FindClass("/org/apache/arrow/adapter/orc/OrcFieldNode");
+  jmethodID field_constructor = env->GetMethodID(field_class, "<init>", "(II)V");
+  jobjectArray field_array =
+      env->NewObjectArray(schema->num_fields(), field_class, nullptr);
+
+  std::vector<std::shared_ptr<arrow::Buffer>> buffers;
+  for (int i = 0; i < schema->num_fields(); ++i) {
+    auto column = (*record_batch)->column(i);
+    auto dataArray = column->data();
+    jobject field = env->NewObject(field_class, field_constructor, column->length(),
+                                   column->null_count());
+    env->SetObjectArrayElement(field_array, i, field);
+
+    for (auto& buffer : dataArray->buffers) {
+      buffers.push_back(buffer);
     }
+  }
 
-    return ret.ok();
+  // create OrcMemoryJniWrapper[]
+  jclass memory_class =
+      env->FindClass("/org/apache/arrow/adapter/orc/OrcMemoryJniWrapper");
+  jmethodID memory_constructor = env->GetMethodID(memory_class, "<init>", "(JJJJ)V");
+  jobjectArray memory_array = env->NewObjectArray(buffers.size(), memory_class, nullptr);
+
+  for (int j = 0; j < buffers.size(); ++j) {
+    auto buffer = buffers[j];
+    jobject memory = env->NewObject(memory_class, memory_constructor, &buffer,
+                                    buffer->data(), buffer->size(), buffer->capacity());
+    env->SetObjectArrayElement(memory_array, j, memory);
+  }
+
+  // create OrcRecordBatch
+  jclass ret_cls = env->FindClass("/org/apache/arrow/adapter/orc/OrcRecordBatch");
+  jmethodID ret_constructor =
+      env->GetMethodID(ret_cls, "<init>",
+                       "(I[L/org/apache/arrow/adapter/orc/OrcFieldNode;"
+                       "[L/org/apache/arrow/adapter/orc/OrcMemoryJniWrapper;)V");
+  jobject ret = env->NewObject(ret_cls, ret_constructor, (*record_batch)->num_rows(),
+                               field_array, memory_array);
+
+  return ret;
+}
+
+JNIEXPORT void JNICALL Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_close(
+    JNIEnv* env, jobject this_obj) {
+  delete GetStripeReader(env, this_obj);
+}
+
+JNIEXPORT void JNICALL Java_org_apache_arrow_adapter_orc_OrcMemoryJniWrapper_release(
+    JNIEnv* env, jobject this_obj, jlong address) {
+  delete reinterpret_cast<std::shared_ptr<arrow::Buffer>*>(address);
 }
 
 #ifdef __cplusplus

java/adapter/orc/CMakeLists.txt

@@ -33,6 +33,11 @@ message("generating headers to ${JNI_HEADERS_DIR}")
 add_jar(
         arrow_orc_java
         src/main/java/org/apache/arrow/adapter/orc/OrcReaderJniWrapper.java
+        src/main/java/org/apache/arrow/adapter/orc/OrcStripeReaderJniWrapper.java
+        src/main/java/org/apache/arrow/adapter/orc/OrcMemoryJniWrapper.java
+        src/main/java/org/apache/arrow/adapter/orc/OrcJniUtils.java
+        src/main/java/org/apache/arrow/adapter/orc/OrcRecordBatch.java
+        src/main/java/org/apache/arrow/adapter/orc/OrcFieldNode.java
         GENERATE_NATIVE_HEADERS arrow_orc_java-native
         DESTINATION ${JNI_HEADERS_DIR}
 )

java/adapter/orc/pom.xml

@@ -14,6 +14,20 @@
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.arrow</groupId>
+            <artifactId>arrow-memory</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.arrow</groupId>
+            <artifactId>arrow-vector</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
 
     <parent>
         <groupId>org.apache.arrow</groupId>

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcFieldNode.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+public class OrcFieldNode {
+
+  private final int length;
+  private final int nullCount;
+
+  public OrcFieldNode(int length, int nullCount) {
+    this.length = length;
+    this.nullCount = nullCount;
+  }
+
+  public int getLength() {
+    return length;
+  }
+
+  public int getNullCount() {
+    return nullCount;
+  }
+}

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcJniUtils.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.InvalidPathException;
+import java.nio.file.StandardCopyOption;
+import java.util.UUID;
+
+public class OrcJniUtils {
+  private static final String LIBRARY_NAME = "arrow_orc_jni";
+  private static boolean isLoaded = false;
+
+  public static void loadOrcAdapterLibraryFromJar()
+          throws IOException {
+    synchronized (OrcJniUtils.class) {
+      if (!isLoaded) {
+        final String libraryToLoad = System.mapLibraryName(LIBRARY_NAME);
+        final File libraryFile = moveFileFromJarToTemp(
+                System.getProperty("java.io.tmpdir"), libraryToLoad);
+        System.load(libraryFile.getAbsolutePath());
+        isLoaded = true;
+      }
+    }
+  }
+
+  private static File moveFileFromJarToTemp(final String tmpDir, String libraryToLoad)
+          throws IOException {
+    final File temp = setupFile(tmpDir, libraryToLoad);
+    try (final InputStream is = OrcReaderJniWrapper.class.getClassLoader()
+            .getResourceAsStream(libraryToLoad)) {
+      if (is == null) {
+        throw new InvalidPathException(libraryToLoad,  "file was not found inside JAR.");
+      } else {
+        Files.copy(is, temp.toPath(), StandardCopyOption.REPLACE_EXISTING);
+      }
+    }
+    return temp;
+  }
+
+  private static File setupFile(String tmpDir, String libraryToLoad)
+          throws IOException {
+    final String randomizeFileName = libraryToLoad + UUID.randomUUID();
+    final File temp = new File(tmpDir, randomizeFileName);
+    if (temp.exists() && !temp.delete()) {
+      throw new InvalidPathException(
+              temp.getAbsolutePath(), "File already exists and cannot be removed.");
+    }
+    if (!temp.createNewFile()) {
+      throw new InvalidPathException(temp.getAbsolutePath(), "File could not be created.");
+    }
+    temp.deleteOnExit();
+    return temp;
+  }
+}

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcMemoryJniWrapper.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+import java.io.IOException;
+
+public class OrcMemoryJniWrapper {
+  private final long ownershipAddress;
+
+  private final long memoryAddress;
+
+  private final long size;
+
+  private final long capacity;
+
+  static {
+    try {
+      OrcJniUtils.loadOrcAdapterLibraryFromJar();
+    } catch (IOException e) {
+      throw new ExceptionInInitializerError(e);
+    }
+  }
+
+  OrcMemoryJniWrapper(long ownershipAddress, long memoryAddress, long size, long capacity) {
+    this.ownershipAddress = ownershipAddress;
+    this.memoryAddress = memoryAddress;
+    this.size = size;
+    this.capacity = capacity;
+  }
+
+  long getSize() {
+    return size;
+  }
+
+  long getCapacity() {
+    return capacity;
+  }
+
+  long getMemoryAddress() {
+    return memoryAddress;
+  }
+
+  void release() {
+    release(ownershipAddress);
+  }
+
+  private native void release(long address);
+}

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcReader.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+import java.io.IOException;
+
+public class OrcReader implements AutoCloseable {
+
+  private static final int DEFAULT_BATCH_SIZE = 1024;
+
+  private OrcReaderJniWrapper orcReader;
+
+  private BufferAllocator allocator;
+
+  public OrcReader(String filePath, BufferAllocator allocator) throws IOException {
+    this.allocator = allocator;
+    orcReader = new OrcReaderJniWrapper();
+    orcReader.open(filePath);
+  }
+
+  public ArrowReader nextStripeReader() {
+    OrcStripeReaderJniWrapper stripeReader = orcReader.nextStripeReader(DEFAULT_BATCH_SIZE);
+    if (stripeReader == null) {
+      return null;
+    }
+
+    return new OrcStripeReader(stripeReader, allocator);
+  }
+
+  @Override
+  public void close() throws Exception {
+    orcReader.close();
+  }
+}

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcReaderJniWrapper.java

@@ -17,66 +17,21 @@
 
 package org.apache.arrow.adapter.orc;
 
-import java.io.File;
 import java.io.IOException;
-import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.InvalidPathException;
-import java.nio.file.StandardCopyOption;
-import java.util.UUID;
 
 public class OrcReaderJniWrapper {
-  private static final String LIBRARY_NAME = "arrow_orc_jni";
 
   private long nativeReaderAddress;
 
   static {
     try {
-      loadOrcAdapterLibraryFromJar();
+      OrcJniUtils.loadOrcAdapterLibraryFromJar();
     } catch (IOException e) {
       throw new ExceptionInInitializerError(e);
     }
   }
 
-  private static void loadOrcAdapterLibraryFromJar()
-          throws IOException {
-    final String libraryToLoad = System.mapLibraryName(LIBRARY_NAME);
-    final File libraryFile = moveFileFromJarToTemp(
-            System.getProperty("java.io.tmpdir"), libraryToLoad);
-    System.load(libraryFile.getAbsolutePath());
-  }
-
-
-  private static File moveFileFromJarToTemp(final String tmpDir, String libraryToLoad)
-          throws IOException {
-    final File temp = setupFile(tmpDir, libraryToLoad);
-    try (final InputStream is = OrcReaderJniWrapper.class.getClassLoader()
-            .getResourceAsStream(libraryToLoad)) {
-      if (is == null) {
-        throw new InvalidPathException(libraryToLoad,  "file was not found inside JAR.");
-      } else {
-        Files.copy(is, temp.toPath(), StandardCopyOption.REPLACE_EXISTING);
-      }
-    }
-    return temp;
-  }
-
-  private static File setupFile(String tmpDir, String libraryToLoad)
-          throws IOException {
-    final String randomizeFileName = libraryToLoad + UUID.randomUUID();
-    final File temp = new File(tmpDir, randomizeFileName);
-    if (temp.exists() && !temp.delete()) {
-      throw new InvalidPathException(
-              temp.getAbsolutePath(), "File already exists and cannot be removed.");
-    }
-    if (!temp.createNewFile()) {
-      throw new InvalidPathException(temp.getAbsolutePath(), "File could not be created.");
-    }
-    temp.deleteOnExit();
-    return temp;
-  }
-
-  private native boolean open(String fileName);
+  public native boolean open(String fileName);
 
   public native void close();
 
@@ -84,14 +39,5 @@ public class OrcReaderJniWrapper {
 
   public native int getNumberOfStripes();
 
-  public native StripeReader nextStripeReader(long batchSize);
-
-  public class StripeReader {
-
-    private long nativeStripeReaderAddress;
-
-    public native byte[] getSchema();
-
-    public native boolean next(int numRows, long[] outAddrs, long[] outSizes);
-  }
+  public native OrcStripeReaderJniWrapper nextStripeReader(long batchSize);
 }

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcRecordBatch.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class OrcRecordBatch {
+  public final int length;
+
+  /**
+   * Nodes correspond to the pre-ordered flattened logical schema.
+   */
+  public final List<OrcFieldNode> nodes;
+
+  public final List<OrcMemoryJniWrapper> buffers;
+
+  OrcRecordBatch(int length, OrcFieldNode[] nodes, OrcMemoryJniWrapper[] buffers) {
+    this.length = length;
+    this.nodes = Arrays.asList(nodes);
+    this.buffers = Arrays.asList(buffers);
+  }
+}

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcReferenceManager.java

@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+import io.netty.buffer.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.OwnershipTransferResult;
+import org.apache.arrow.memory.ReferenceManager;
+import org.apache.arrow.util.Preconditions;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class OrcReferenceManager implements ReferenceManager {
+  private final AtomicInteger bufRefCnt = new AtomicInteger(0);
+
+  private OrcMemoryJniWrapper memory;
+
+  OrcReferenceManager(OrcMemoryJniWrapper memory) {
+    this.memory = memory;
+  }
+  /**
+   * Return the reference count.
+   *
+   * @return reference count
+   */
+  @Override
+  public int getRefCount() {
+    return bufRefCnt.get();
+  }
+
+  /**
+   * Decrement this reference manager's reference count by 1 for the associated underlying
+   * memory. If the reference count drops to 0, it implies that ArrowBufs managed by this
+   * reference manager no longer need access to the underlying memory
+   *
+   * @return true if ref count has dropped to 0, false otherwise
+   */
+  @Override
+  public boolean release() {
+    return release(1);
+  }
+
+  /**
+   * Decrement this reference manager's reference count for the associated underlying
+   * memory. If the reference count drops to 0, it implies that ArrowBufs managed by this
+   * reference manager no longer need access to the underlying memory
+   *
+   * @param decrement the count to decrease the reference count by
+   * @return the new reference count
+   */
+  @Override
+  public boolean release(int decrement) {
+    Preconditions.checkState(decrement >= 1,
+            "ref count decrement should be greater than or equal to 1");
+    // decrement the ref count
+    final int refCnt;
+    synchronized (this) {
+      refCnt = bufRefCnt.addAndGet(-decrement);
+      if (refCnt == 0) {
+        // refcount of this reference manager has dropped to 0
+        // release the underlying memory
+        memory.release();
+      }
+    }
+    // the new ref count should be >= 0
+    Preconditions.checkState(refCnt >= 0, "RefCnt has gone negative");
+    return refCnt == 0;
+  }
+
+  /**
+   * Increment this reference manager's reference count by 1 for the associated underlying
+   * memory.
+   */
+  @Override
+  public void retain() {
+    retain(1);
+  }
+
+  /**
+   * Increment this reference manager's reference count by a given amount for the
+   * associated underlying memory.
+   *
+   * @param increment the count to increase the reference count by
+   */
+  @Override
+  public void retain(int increment) {
+    Preconditions.checkArgument(increment > 0, "retain(%d) argument is not positive", increment);
+    final int originalReferenceCount = bufRefCnt.getAndAdd(increment);
+    Preconditions.checkArgument(originalReferenceCount > 0);
+  }
+
+  /**
+   * Create a new ArrowBuf that is associated with an alternative allocator for the purposes of
+   * memory ownership and accounting. This has no impact on the reference counting for the current
+   * ArrowBuf except in the situation where the passed in Allocator is the same as the current buffer.
+   * This operation has no impact on the reference count of this ArrowBuf. The newly created
+   * ArrowBuf with either have a reference count of 1 (in the case that this is the first time this
+   * memory is being associated with the target allocator or in other words allocation manager currently
+   * doesn't hold a mapping for the target allocator) or the current value of the reference count for
+   * the target allocator-reference manager combination + 1 in the case that the provided allocator
+   * already had an association to this underlying memory.
+   *
+   * @param srcBuffer       source ArrowBuf
+   * @param targetAllocator The target allocator to create an association with.
+   * @return A new ArrowBuf which shares the same underlying memory as this ArrowBuf.
+   */
+  @Override
+  public ArrowBuf retain(ArrowBuf srcBuffer, BufferAllocator targetAllocator) {
+    retain();
+    return srcBuffer;
+  }
+
+  /**
+   * Derive a new ArrowBuf from a given source ArrowBuf. The new derived
+   * ArrowBuf will share the same reference count as rest of the ArrowBufs
+   * associated with this reference manager.
+   *
+   * @param sourceBuffer source ArrowBuf
+   * @param index        index (relative to source ArrowBuf) new ArrowBuf should be derived from
+   * @param length       length (bytes) of data in underlying memory that derived buffer will
+   *                     have access to in underlying memory
+   * @return derived buffer
+   */
+  @Override
+  public ArrowBuf deriveBuffer(ArrowBuf sourceBuffer, int index, int length) {
+    final long derivedBufferAddress = sourceBuffer.memoryAddress() + index;
+
+    // create new ArrowBuf
+    final ArrowBuf derivedBuf = new ArrowBuf(
+            this,
+            null,
+            length, // length (in bytes) in the underlying memory chunk for this new ArrowBuf
+            derivedBufferAddress, // starting byte address in the underlying memory for this new ArrowBuf,
+            false);
+
+    return derivedBuf;
+  }
+
+  /**
+   * Transfer the memory accounting ownership of this ArrowBuf to another allocator.
+   * This will generate a new ArrowBuf that carries an association with the underlying memory
+   * for the given ArrowBuf
+   *
+   * @param sourceBuffer    source ArrowBuf
+   * @param targetAllocator The target allocator to create an association with
+   * @return {@link OwnershipTransferResult} with info on transfer result and new buffer
+   */
+  @Override
+  public OwnershipTransferResult transferOwnership(ArrowBuf sourceBuffer, BufferAllocator targetAllocator) {
+    retain();
+    return new OwnershipTransferResult() {
+      @Override
+      public boolean getAllocationFit() {
+        return false;
+      }
+
+      @Override
+      public ArrowBuf getTransferredBuffer() {
+        return sourceBuffer;
+      }
+    };
+  }
+
+  /**
+   * Get the buffer allocator associated with this reference manager
+   *
+   * @return buffer allocator.
+   */
+  @Override
+  public BufferAllocator getAllocator() {
+    return null;
+  }
+
+  /**
+   * Total size (in bytes) of memory underlying this reference manager.
+   *
+   * @return Size (in bytes) of the memory chunk.
+   */
+  @Override
+  public int getSize() {
+    return memory.getSize();
+  }
+
+  /**
+   * Get the total accounted size (in bytes).
+   *
+   * @return accounted size.
+   */
+  @Override
+  public int getAccountedSize() {
+    return 0;
+  }
+}

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcStripeReader.java

@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+import io.netty.buffer.ArrowBuf;
+import org.apache.arrow.flatbuf.MessageHeader;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.arrow.vector.ipc.ReadChannel;
+import org.apache.arrow.vector.ipc.message.*;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.stream.Collectors;
+
+public class OrcStripeReader extends ArrowReader {
+  private OrcStripeReaderJniWrapper stripeReader;
+  private MessageChannelReader schemaReader;
+
+  protected OrcStripeReader(OrcStripeReaderJniWrapper stripeReader, BufferAllocator allocator) {
+    super(allocator);
+    this.stripeReader = stripeReader;
+  }
+
+  /**
+   * Load the next ArrowRecordBatch to the vector schema root if available.
+   *
+   * @return true if a batch was read, false on EOS
+   * @throws IOException on error
+   */
+  @Override
+  public boolean loadNextBatch() throws IOException {
+    OrcRecordBatch recordBatch = stripeReader.next();
+    if (recordBatch == null) {
+      return false;
+    }
+
+    ArrayList<ArrowBuf> buffers = new ArrayList<>();
+    for(OrcMemoryJniWrapper buffer : recordBatch.buffers) {
+      buffers.add(new ArrowBuf(
+              new OrcReferenceManager(buffer),
+              null,
+              (int)buffer.getSize(),
+              buffer.getMemoryAddress(),
+              false));
+    }
+
+    loadRecordBatch(new ArrowRecordBatch(
+            recordBatch.length,
+            recordBatch.nodes.stream()
+              .map(buf -> new ArrowFieldNode(buf.getLength(), buf.getNullCount()))
+              .collect(Collectors.toList()),
+            buffers));
+    return true;
+  }
+
+  /**
+   * Return the number of bytes read from the ReadChannel.
+   *
+   * @return number of bytes read
+   */
+  @Override
+  public long bytesRead() {
+    return 0;
+  }
+
+  /**
+   * Close the underlying read source.
+   *
+   * @throws IOException on error
+   */
+  @Override
+  protected void closeReadSource() throws IOException {
+    stripeReader.close();
+    schemaReader.close();
+  }
+
+  /**
+   * Read the Schema from the source, will be invoked at the beginning the initialization.
+   *
+   * @return the read Schema
+   * @throws IOException on error
+   */
+  @Override
+  protected Schema readSchema() throws IOException {
+    byte[] schemaBytes = stripeReader.getSchema();
+    schemaReader = new MessageChannelReader(
+                      new ReadChannel(
+                        new ByteArrayReadableSeekableByteChannel(schemaBytes)), allocator);
+
+    MessageResult result = schemaReader.readNext();
+
+    if (result == null) {
+      throw new IOException("Unexpected end of input. Missing schema.");
+    }
+
+    if (result.getMessage().headerType() != MessageHeader.Schema) {
+      throw new IOException("Expected schema but header was " + result.getMessage().headerType());
+    }
+
+    return MessageSerializer.deserializeSchema(result.getMessage());
+  }
+
+  /**
+   * Read a dictionary batch from the source, will be invoked after the schema has been read and
+   * called N times, where N is the number of dictionaries indicated by the schema Fields.
+   *
+   * @return the read ArrowDictionaryBatch
+   * @throws IOException on error
+   */
+  @Override
+  protected ArrowDictionaryBatch readDictionary() throws IOException {
+    MessageResult result = schemaReader.readNext();
+
+    if (result == null) {
+      throw new IOException("Unexpected end of input. Expected DictionaryBatch");
+    }
+
+    if (result.getMessage().headerType() != MessageHeader.DictionaryBatch) {
+      throw new IOException("Expected DictionaryBatch but header was " + result.getMessage().headerType());
+    }
+
+    ArrowBuf bodyBuffer = result.getBodyBuffer();
+
+    // For zero-length batches, need an empty buffer to deserialize the batch
+    if (bodyBuffer == null) {
+      bodyBuffer = allocator.getEmpty();
+    }
+
+    return MessageSerializer.deserializeDictionaryBatch(result.getMessage(), bodyBuffer);
+  }
+}

java/adapter/orc/src/main/java/org/apache/arrow/adapter/orc/OrcStripeReaderJniWrapper.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.adapter.orc;
+
+import java.io.IOException;
+
+public class OrcStripeReaderJniWrapper {
+
+  static {
+    try {
+      OrcJniUtils.loadOrcAdapterLibraryFromJar();
+    } catch (IOException e) {
+      throw new ExceptionInInitializerError(e);
+    }
+  }
+
+  private long nativeStripeReaderAddress;
+
+  public native byte[] getSchema();
+
+  public native OrcRecordBatch next();
+
+  public native void close();
+}

java/memory/src/main/java/org/apache/arrow/memory/AllocationManager.java

@@ -70,7 +70,7 @@ public class AllocationManager {
   private volatile BufferLedger owningLedger;
   private volatile long amDestructionTime = 0;
 
-  AllocationManager(BaseAllocator accountingAllocator, int size) {
+  public AllocationManager(BaseAllocator accountingAllocator, int size) {
     Preconditions.checkNotNull(accountingAllocator);
     accountingAllocator.assertOpen();
 
@@ -174,7 +174,7 @@ public class AllocationManager {
         // underlying memory chunk as it is no longer being referenced
         ((BaseAllocator)oldLedger.getAllocator()).releaseBytes(size);
         // free the memory chunk associated with the allocation manager
-        memoryChunk.release();
+        releaseMemory();
         ((BaseAllocator)oldLedger.getAllocator()).getListener().onRelease(size);
         amDestructionTime = System.nanoTime();
         owningLedger = null;
@@ -196,6 +196,10 @@ public class AllocationManager {
     }
   }
 
+  public void releaseMemory() {
+    memoryChunk.release();
+  }
+
   /**
    * Return the size of underlying chunk of memory managed by this Allocation Manager.
    * @return size of memory chunk