Interface refactor and performance optimization

- refactor JNI interface to make the underlying resource id explicit
- add concurrent map for resource management
- hold reference to class and method field to avoid unnecessary overhead

cpp/src/arrow/adapters/orc/jni/concurrent_map.h

@@ -0,0 +1,80 @@
+/*
+ * 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
+ */
+
+#ifndef JNI_ID_TO_MODULE_MAP_H
+#define JNI_ID_TO_MODULE_MAP_H
+
+#include <memory>
+#include <utility>
+
+namespace arrow {
+
+template <typename HOLDER>
+class concurrentMap {
+public:
+    concurrentMap() : module_id_(kInitModuleId) {}
+
+    jlong Insert(HOLDER holder) {
+        mtx_.lock();
+        jlong result = module_id_++;
+        map_.insert(std::pair<jlong, HOLDER>(result, holder));
+        mtx_.unlock();
+        return result;
+    }
+
+    void Erase(jlong module_id) {
+        mtx_.lock();
+        map_.erase(module_id);
+        mtx_.unlock();
+    }
+
+    HOLDER Lookup(jlong module_id) {
+        HOLDER result = nullptr;
+        try {
+            result = map_.at(module_id);
+        } catch (const std::out_of_range &e) {
+        }
+        if (result != nullptr) {
+            return result;
+        }
+        mtx_.lock();
+        try {
+            result = map_.at(module_id);
+        } catch (const std::out_of_range &e) {
+        }
+        mtx_.unlock();
+        return result;
+    }
+
+    void Clear() {
+      mtx_.lock();
+      map_.clear();
+      mtx_.unlock();
+    }
+
+private:
+    static const int kInitModuleId = 4;
+
+    long module_id_;
+    std::mutex mtx_;
+    // map from module ids returned to Java and module pointers
+    std::unordered_map<jlong, HOLDER> map_;
+};
+
+}  // namespace gandiva
+
+#endif  // JNI_ID_TO_MODULE_MAP_H

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

@@ -28,140 +28,198 @@
 #include "org_apache_arrow_adapter_orc_OrcReaderJniWrapper.h"
 #include "org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper.h"
 
+#include "concurrent_map.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";
-  }
+static jclass io_exception_class;
+static jclass exception_class;
+
+static jclass orc_field_node_class;
+static jmethodID orc_field_node_constructor;
+
+static jclass orc_memory_class;
+static jmethodID orc_memory_constructor;
+
+static jclass record_batch_class;
+static jmethodID record_batch_constructor;
+
+static jint JNI_VERSION = JNI_VERSION_1_6;
+
+static arrow::concurrentMap<std::shared_ptr<arrow::Buffer>> buffer_holder_;
+static arrow::concurrentMap<std::shared_ptr<RecordBatchReader>> orc_stripe_reader_holder_;
+static arrow::concurrentMap<std::shared_ptr<ORCFileReader>> orc_file_reader_holder_;
+
+jclass CreateGlobalClassReference(JNIEnv* env, const char* class_name) {
+  jclass local_class = env->FindClass(class_name);
+  jclass global_class = (jclass)env->NewGlobalRef(local_class);
+  env->DeleteLocalRef(local_class);
+  return global_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");
+jmethodID GetMethodID(JNIEnv* env, jclass this_class, const char* name, const char* sig) {
+  jmethodID ret = env->GetMethodID(this_class, name, sig);
   if (ret == nullptr) {
-    ThrowJavaException(env, "Unable to get java class field: " + sig);
+    std::string error_message = "Unable to find method " + std::string(name) +
+                                " within signature" + std::string(sig);
+    env->ThrowNew(exception_class, error_message.c_str());
   }
 
   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);
+jint JNI_OnLoad(JavaVM* vm, void* reserved) {
+  JNIEnv* env;
+  if (vm->GetEnv(reinterpret_cast<void**>(&env), JNI_VERSION) != JNI_OK) {
+    return JNI_ERR;
+  }
+
+  io_exception_class = CreateGlobalClassReference(env, "java/io/IOException");
+  exception_class = CreateGlobalClassReference(env, "java/lang/Exception");
+
+  orc_field_node_class =
+      CreateGlobalClassReference(env, "/org/apache/arrow/adapter/orc/OrcFieldNode");
+  orc_field_node_constructor = GetMethodID(env, orc_field_node_class, "<init>", "(II)V");
+
+  orc_memory_class = CreateGlobalClassReference(
+      env, "/org/apache/arrow/adapter/orc/OrcMemoryJniWrapper");
+  orc_memory_constructor = GetMethodID(env, orc_memory_class, "<init>", "(JJJJ)V");
+
+  record_batch_class =
+      CreateGlobalClassReference(env, "/org/apache/arrow/adapter/orc/OrcRecordBatch");
+  record_batch_constructor = GetMethodID(env, record_batch_class, "<init>",
+                                "(I[L/org/apache/arrow/adapter/orc/OrcFieldNode;"
+                                "[L/org/apache/arrow/adapter/orc/OrcMemoryJniWrapper;)V");
+
+  env->ExceptionDescribe();
+
+  return JNI_VERSION;
 }
 
-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);
+void JNI_OnUnload(JavaVM* vm, void* reserved) {
+  JNIEnv* env;
+  vm->GetEnv(reinterpret_cast<void**>(&env), JNI_VERSION);
+  env->DeleteGlobalRef(io_exception_class);
+  env->DeleteGlobalRef(exception_class);
+  env->DeleteGlobalRef(orc_field_node_class);
+  env->DeleteGlobalRef(orc_memory_class);
+  env->DeleteGlobalRef(record_batch_class);
+
+  buffer_holder_.Clear();
+  orc_stripe_reader_holder_.Clear();
+  orc_file_reader_holder_.Clear();
+}
+
+std::shared_ptr<ORCFileReader> GetNativeReader(jlong id) {
+  return orc_file_reader_holder_.Lookup(id);
+}
+
+std::shared_ptr<RecordBatchReader> GetStripeReader(jlong id) {
+  return orc_stripe_reader_holder_.Lookup(id);
+}
+
+int jstr_to_cstr(JNIEnv* env, jstring jstr, char* cstr, size_t cstr_len) {
+  int32_t jlen, clen;
+
+  clen = env->GetStringUTFLength(jstr);
+  if (clen > (int32_t)cstr_len) return -ENAMETOOLONG;
+  jlen = env->GetStringLength(jstr);
+  env->GetStringUTFRegion(jstr, 0, jlen, cstr);
+  if (env->ExceptionCheck()) return -EIO;
+  return 0;
+}
+
+std::string JStringToCString(JNIEnv* env, jstring string) {
+  int32_t jlen, clen;
+  clen = env->GetStringUTFLength(string);
+  jlen = env->GetStringLength(string);
+  std::vector<char> buffer(clen);
+  env->GetStringUTFRegion(string, 0, jlen, buffer.data());
+  return std::string(buffer.data(), clen);
 }
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-JNIEXPORT jboolean JNICALL Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_open(
-    JNIEnv* env, jobject this_obj, jstring file_path) {
+JNIEXPORT jlong JNICALL Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_open(
+    JNIEnv* env, jclass this_cls, 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);
+
+  std::string path = JStringToCString(env, file_path);
 
   arrow::Status ret;
   if (path.find("hdfs://") == 0) {
-    return false;
+    env->ThrowNew(io_exception_class, "hdfs path not support yet.");
   } else {
     ret = arrow::io::ReadableFile::Open(path, &in_file);
   }
 
   if (ret.ok()) {
-    auto reader = new std::unique_ptr<ORCFileReader>();
+    std::unique_ptr<ORCFileReader> reader;
 
     ret = ORCFileReader::Open(
         std::static_pointer_cast<arrow::io::RandomAccessFile>(in_file),
-        arrow::default_memory_pool(), reader);
+        arrow::default_memory_pool(), &reader);
 
     if (!ret.ok()) {
-      delete reader;
-      ThrowJavaIOException(env, std::invalid_argument("Failed open file" + path));
+      env->ThrowNew(io_exception_class, std::string("Failed open file" + path).c_str());
     }
 
-    env->SetLongField(
-        this_obj, GetFieldId(env, env->GetObjectClass(this_obj), "nativeReaderAddress"),
-        reinterpret_cast<long>(reader));
+    return orc_file_reader_holder_.Insert(
+        std::shared_ptr<ORCFileReader>(reader.release()));
   }
 
-  return ret.ok();
+  return static_cast<jlong>(ret.code()) * -1;
 }
 
 JNIEXPORT void JNICALL Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_close(
-    JNIEnv* env, jobject this_obj) {
-  delete GetNativeReader(env, this_obj);
+    JNIEnv* env, jclass this_cls, jlong id) {
+  orc_file_reader_holder_.Erase(id);
 }
 
 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();
+    JNIEnv* env, jclass this_cls, jlong id, jint row_number) {
+  auto reader = orc_file_reader_holder_.Lookup(id);
+  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();
+Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_getNumberOfStripes(JNIEnv* env,
+                                                                         jclass this_cls,
+                                                                         jlong id) {
+  auto reader = orc_file_reader_holder_.Lookup(id);
+  return reader->NumberOfStripes();
 }
 
-JNIEXPORT jobject JNICALL
+JNIEXPORT jlong JNICALL
 Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_nextStripeReader(JNIEnv* env,
-                                                                       jobject this_obj,
+                                                                       jclass this_cls,
+                                                                       jlong id,
                                                                        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);
+  auto reader = GetNativeReader(id);
+  std::shared_ptr<RecordBatchReader> stripe_reader;
+  auto status = reader->NextStripeReader(batch_size, &stripe_reader);
 
-  jobject ret = nullptr;
   if (!status.ok()) {
-    delete stripe_reader;
-    return ret;
+    return static_cast<jlong>(status.code()) * -1;
+    ;
   }
 
-  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;
+  return orc_stripe_reader_holder_.Insert(stripe_reader);
 }
 
 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();
+                                                                      jclass this_cls,
+                                                                      jlong id) {
+  auto stripe_reader = GetStripeReader(id);
+  auto schema = stripe_reader->schema();
 
   std::shared_ptr<arrow::Buffer> out;
-  auto status = arrow::ipc::SerializeSchema(*schema, arrow::default_memory_pool(), &out);
+  auto status =
+      arrow::ipc::SerializeSchema(*schema, nullptr, arrow::default_memory_pool(), &out);
   if (!status.ok()) {
     return nullptr;
   }
@@ -173,29 +231,28 @@ Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_getSchema(JNIEnv* en
 
 JNIEXPORT jobject JNICALL
 Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_next(JNIEnv* env,
-                                                                 jobject this_obj) {
-  auto stripe_reader = GetStripeReader(env, this_obj);
+                                                                 jclass this_cls,
+                                                                 jlong id) {
+  auto stripe_reader = GetStripeReader(id);
   auto record_batch = new std::shared_ptr<arrow::RecordBatch>();
-  auto status = (*stripe_reader)->ReadNext(record_batch);
+  auto status = stripe_reader->ReadNext(record_batch);
   if (!status.ok()) {
     delete record_batch;
     return nullptr;
   }
 
-  auto schema = (*stripe_reader)->schema();
+  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);
+      env->NewObjectArray(schema->num_fields(), orc_field_node_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());
+    jobject field = env->NewObject(orc_field_node_class, orc_field_node_constructor,
+                                   column->length(), column->null_count());
     env->SetObjectArrayElement(field_array, i, field);
 
     for (auto& buffer : dataArray->buffers) {
@@ -204,38 +261,32 @@ Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_next(JNIEnv* env,
   }
 
   // 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);
+  jobjectArray memory_array =
+      env->NewObjectArray(buffers.size(), orc_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());
+    jobject memory = env->NewObject(orc_memory_class, orc_memory_constructor,
+                                    buffer_holder_.Insert(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);
+  jobject ret = env->NewObject(record_batch_class, record_batch_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);
+    JNIEnv* env, jclass this_cls, jlong id) {
+  buffer_holder_.Erase(id);
 }
 
 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);
+    JNIEnv* env, jobject this_obj, jlong id) {
+  buffer_holder_.Erase(id);
 }
 
 #ifdef __cplusplus

java/README.md

@@ -51,7 +51,7 @@ be provided as the value for argument arrow.cpp.build.dir. eg.
 
 ```
 cd java
-mvn install -P native-orc -pl adapter/orc -am -Dgandiva.cpp.build.dir=../../debug
+mvn install -P native-orc -pl adapter/orc -am -Darrow.cpp.build.dir=../../debug
 ```
 
 ## Java Code Style Guide

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

@@ -17,21 +17,29 @@
 
 package org.apache.arrow.adapter.orc;
 
-public class OrcFieldNode {
+/**
+ * Metadata about Vectors/Arrays that is passed via JNI interface
+ */
+class OrcFieldNode {
 
   private final int length;
   private final int nullCount;
 
+  /**
+   * Construct a new instance.
+   * @param length the number of values written.
+   * @param nullCount the number of null values.
+   */
   public OrcFieldNode(int length, int nullCount) {
     this.length = length;
     this.nullCount = nullCount;
   }
 
-  public int getLength() {
+  int getLength() {
     return length;
   }
 
-  public int getNullCount() {
+  int getNullCount() {
     return nullCount;
   }
 }

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

@@ -25,11 +25,18 @@ import java.nio.file.InvalidPathException;
 import java.nio.file.StandardCopyOption;
 import java.util.UUID;
 
-public class OrcJniUtils {
+/**
+ * Helper class for JNI related operations.
+ */
+class OrcJniUtils {
   private static final String LIBRARY_NAME = "arrow_orc_jni";
   private static boolean isLoaded = false;
 
-  public static void loadOrcAdapterLibraryFromJar()
+  /**
+   * Load arrow orc jni library from jar.
+   * @throws IOException throws IOException in case target library can not be found.
+   */
+  static void loadOrcAdapterLibraryFromJar()
           throws IOException {
     synchronized (OrcJniUtils.class) {
       if (!isLoaded) {

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

@@ -19,8 +19,11 @@ package org.apache.arrow.adapter.orc;
 
 import java.io.IOException;
 
-public class OrcMemoryJniWrapper {
-  private final long ownershipAddress;
+/**
+ * Wrapper for orc memory allocated by native code.
+ */
+class OrcMemoryJniWrapper implements AutoCloseable {
+  private final long id;
 
   private final long memoryAddress;
 
@@ -36,8 +39,15 @@ public class OrcMemoryJniWrapper {
     }
   }
 
-  OrcMemoryJniWrapper(long ownershipAddress, long memoryAddress, long size, long capacity) {
-    this.ownershipAddress = ownershipAddress;
+  /**
+   * Construct a new instance.
+   * @param id unique id of the underlying memory.
+   * @param memoryAddress starting memory address of the the underlying memory.
+   * @param size size of the valid data.
+   * @param capacity allocated memory size.
+   */
+  OrcMemoryJniWrapper(long id, long memoryAddress, long size, long capacity) {
+    this.id = id;
     this.memoryAddress = memoryAddress;
     this.size = size;
     this.capacity = capacity;
@@ -55,9 +65,10 @@ public class OrcMemoryJniWrapper {
     return memoryAddress;
   }
 
-  void release() {
-    release(ownershipAddress);
+  @Override
+  public void close() {
+    release(id);
   }
 
-  private native void release(long address);
+  private native void release(long id);
 }

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

@@ -22,32 +22,64 @@ import java.io.IOException;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.vector.ipc.ArrowReader;
 
-
+/**
+ *  Orc Reader that allow accessing orc stripes in Orc file.
+ *  This orc reader basically acts like an ArrowReader iterator that
+ *  iterate over orc stripes. Each stripe will be accessed via an
+ *  ArrowReader.
+ */
 public class OrcReader implements AutoCloseable {
-
-  private static final int DEFAULT_BATCH_SIZE = 1024;
-
-  private OrcReaderJniWrapper orcReader;
-
   private BufferAllocator allocator;
 
+  private final long id;
+
+  /**
+   * Create an OrcReader that iterate over orc stripes.
+   * @param filePath file path to target file, currently only support local file.
+   * @param allocator allocator provided to ArrowReader.
+   * @throws IOException throws exception in case of file not found
+   */
   public OrcReader(String filePath, BufferAllocator allocator) throws IOException {
     this.allocator = allocator;
-    orcReader = new OrcReaderJniWrapper();
-    orcReader.open(filePath);
+    id = OrcReaderJniWrapper.open(filePath);
   }
 
-  public ArrowReader nextStripeReader() {
-    OrcStripeReaderJniWrapper stripeReader = orcReader.nextStripeReader(DEFAULT_BATCH_SIZE);
-    if (stripeReader == null) {
+  /**
+   * Seek to designated row. Invoke NextStripeReader() after seek
+   * will return stripe reader starting from designated row.
+   * @param rowNumber the rows number to seek
+   * @return true if seek operation is succeeded
+   */
+  public boolean seek(int rowNumber) {
+    return OrcReaderJniWrapper.seek(id, rowNumber);
+  }
+
+  /**
+   * Get a stripe level ArrowReader with specified batchSize in each record batch.
+   *
+   * @param batchSize the number of rows loaded on each iteration
+   * @return ArrowReader that iterate over current stripes
+   */
+  public ArrowReader nextStripeReader(long batchSize) {
+    long stripeReaderId = OrcReaderJniWrapper.nextStripeReader(id, batchSize);
+    if (stripeReaderId < 0) {
       return null;
     }
 
-    return new OrcStripeReader(stripeReader, allocator);
+    return new OrcStripeReader(stripeReaderId, allocator);
+  }
+
+  /**
+   * The number of stripes in the file.
+   *
+   * @return number of stripes
+   */
+  public int getNumberOfStripes() {
+    return OrcReaderJniWrapper.getNumberOfStripes(id);
   }
 
   @Override
   public void close() throws Exception {
-    orcReader.close();
+    OrcReaderJniWrapper.close(id);
   }
 }

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

@@ -19,10 +19,10 @@ package org.apache.arrow.adapter.orc;
 
 import java.io.IOException;
 
-public class OrcReaderJniWrapper {
-
-  private long nativeReaderAddress;
-
+/**
+ * JNI wrapper for Orc reader
+ */
+class OrcReaderJniWrapper {
   static {
     try {
       OrcJniUtils.loadOrcAdapterLibraryFromJar();
@@ -31,13 +31,41 @@ public class OrcReaderJniWrapper {
     }
   }
 
-  public native boolean open(String fileName);
+  /**
+   * Construct a orc file reader over the target file
+   * @param fileName absolute file path of target file
+   * @return id of the orc reader instance if file opened successfully,
+   * otherwise return error code * -1.
+   */
+  static native long open(String fileName);
 
-  public native void close();
+  /**
+   * Close the underlying reader and release related resources.
+   * @param readerId id of the reader instance.
+   */
+  static native void close(long readerId);
 
-  public native boolean seek(int rowNumber);
+  /**
+   *  Seek to designated row. Invoke nextStripeReader() after seek
+   *  will return id of stripe reader starting from designated row.
+   * @param readerId id of the reader instance
+   * @param rowNumber the rows number to seek
+   * @return true if seek operation is succeeded
+   */
+  static native boolean seek(long readerId, int rowNumber);
 
-  public native int getNumberOfStripes();
+  /**
+   * The number of stripes in the file.
+   * @param readerId id of the reader instance
+   * @return number of stripes
+   */
+  static native int getNumberOfStripes(long readerId);
 
-  public native OrcStripeReaderJniWrapper nextStripeReader(long batchSize);
+  /**
+   * Get a stripe level ArrowReader with specified batchSize in each record batch.
+   * @param readerId id of the reader instance
+   * @param batchSize the number of rows loaded on each iteration
+   * @return id of the stripe reader instance.
+   */
+  static native long nextStripeReader(long readerId, long batchSize);
 }

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

@@ -20,16 +20,22 @@ package org.apache.arrow.adapter.orc;
 import java.util.Arrays;
 import java.util.List;
 
-public class OrcRecordBatch {
-  public final int length;
+class OrcRecordBatch {
+  final int length;
 
   /**
    * Nodes correspond to the pre-ordered flattened logical schema.
    */
-  public final List<OrcFieldNode> nodes;
+  final List<OrcFieldNode> nodes;
 
-  public final List<OrcMemoryJniWrapper> buffers;
+  final List<OrcMemoryJniWrapper> buffers;
 
+  /**
+   * Construct a new instance
+   * @param length number of records included in current batch
+   * @param nodes meta data for each fields
+   * @param buffers buffers for underlying data
+   */
   OrcRecordBatch(int length, OrcFieldNode[] nodes, OrcMemoryJniWrapper[] buffers) {
     this.length = length;
     this.nodes = Arrays.asList(nodes);

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

@@ -17,6 +17,8 @@
 
 package org.apache.arrow.adapter.orc;
 
+import io.netty.buffer.ArrowBuf;
+
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.arrow.memory.BufferAllocator;
@@ -24,8 +26,10 @@ import org.apache.arrow.memory.OwnershipTransferResult;
 import org.apache.arrow.memory.ReferenceManager;
 import org.apache.arrow.util.Preconditions;
 
-import io.netty.buffer.ArrowBuf;
-
+/**
+ * A simple reference manager implementation for memory allocated by native code.
+ * The underlying memory will be released when reference count reach zero.
+ */
 public class OrcReferenceManager implements ReferenceManager {
   private final AtomicInteger bufRefCnt = new AtomicInteger(0);
 
@@ -35,36 +39,16 @@ public class OrcReferenceManager implements ReferenceManager {
     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,
@@ -76,7 +60,7 @@ public class OrcReferenceManager implements ReferenceManager {
       if (refCnt == 0) {
         // refcount of this reference manager has dropped to 0
         // release the underlying memory
-        memory.release();
+        memory.close();
       }
     }
     // the new ref count should be >= 0
@@ -84,21 +68,11 @@ public class OrcReferenceManager implements ReferenceManager {
     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);
@@ -106,38 +80,12 @@ public class OrcReferenceManager implements ReferenceManager {
     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;
@@ -153,15 +101,6 @@ public class OrcReferenceManager implements ReferenceManager {
     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();
@@ -178,31 +117,16 @@ public class OrcReferenceManager implements ReferenceManager {
     };
   }
 
-  /**
-   * 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 (int)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

@@ -17,6 +17,8 @@
 
 package org.apache.arrow.adapter.orc;
 
+import io.netty.buffer.ArrowBuf;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.stream.Collectors;
@@ -33,28 +35,28 @@ import org.apache.arrow.vector.ipc.message.MessageResult;
 import org.apache.arrow.vector.ipc.message.MessageSerializer;
 import org.apache.arrow.vector.types.pojo.Schema;
 import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
-import io.netty.buffer.ArrowBuf;
-
-
+/**
+ * Orc stripe that load data into ArrowRecordBatch.
+ */
 public class OrcStripeReader extends ArrowReader {
-  private OrcStripeReaderJniWrapper stripeReader;
+  private final long id;
   private MessageChannelReader schemaReader;
 
-  protected OrcStripeReader(OrcStripeReaderJniWrapper stripeReader, BufferAllocator allocator) {
+  /**
+   * Construct a new instance.
+   * @param id id of the stripe reader instance
+   * @param allocator memory allocator for accounting.
+   */
+  OrcStripeReader(long id, BufferAllocator allocator) {
     super(allocator);
-    this.stripeReader = stripeReader;
+    this.id = id;
   }
 
-  /**
-   * 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();
+    OrcRecordBatch recordBatch = OrcStripeReaderJniWrapper.next(id);
     if (recordBatch == null) {
       return false;
     }
@@ -78,36 +80,21 @@ public class OrcStripeReader extends ArrowReader {
     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();
+    OrcStripeReaderJniWrapper.close(id);
     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();
+    byte[] schemaBytes = OrcStripeReaderJniWrapper.getSchema(id);
     schemaReader = new MessageChannelReader(
                       new ReadChannel(
                         new ByteArrayReadableSeekableByteChannel(schemaBytes)), allocator);
@@ -125,33 +112,8 @@ public class OrcStripeReader extends ArrowReader {
     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);
+    throw new NotImplementedException();
   }
 }

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

@@ -19,7 +19,10 @@ package org.apache.arrow.adapter.orc;
 
 import java.io.IOException;
 
-public class OrcStripeReaderJniWrapper {
+/**
+ * JNI wrapper for orc stripe reader.
+ */
+class OrcStripeReaderJniWrapper {
 
   static {
     try {
@@ -29,11 +32,24 @@ public class OrcStripeReaderJniWrapper {
     }
   }
 
-  private long nativeStripeReaderAddress;
+  /**
+   * Get the schema of current stripe.
+   * @param readerId id of the stripe reader instance.
+   * @return serialized schema.
+   */
+  static native byte[] getSchema(long readerId);
 
-  public native byte[] getSchema();
+  /**
+   * Load next record batch.
+   * @param readerId id of the stripe reader instance.
+   * @return loaded record batch, return null when reached
+   * the end of current stripe.
+   */
+  static native OrcRecordBatch next(long readerId);
 
-  public native OrcRecordBatch next();
-
-  public native void close();
+  /**
+   * Release resources of underlying reader.
+   * @param readerId id of the stripe reader instance.
+   */
+  static native void close(long readerId);
 }

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;
 
-  public AllocationManager(BaseAllocator accountingAllocator, int size) {
+  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
-        releaseMemory();
+        memoryChunk.release();
         ((BaseAllocator)oldLedger.getAllocator()).getListener().onRelease(size);
         amDestructionTime = System.nanoTime();
         owningLedger = null;
@@ -196,10 +196,6 @@ 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