Add Arrow Jni Reader Unittests

cpp/src/jni/orc/jni_wrapper.cpp

@@ -23,6 +23,7 @@
 #include <arrow/util/logging.h>
 #include <cassert>
 #include <string>
+#include <iostream>
 
 #include "org_apache_arrow_adapter_orc_OrcMemoryJniWrapper.h"
 #include "org_apache_arrow_adapter_orc_OrcReaderJniWrapper.h"
@@ -75,22 +76,22 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) {
     return JNI_ERR;
   }
 
-  io_exception_class = CreateGlobalClassReference(env, "java/io/IOException");
-  exception_class = CreateGlobalClassReference(env, "java/lang/Exception");
+  io_exception_class = CreateGlobalClassReference(env, "Ljava/io/IOException;");
+  exception_class = CreateGlobalClassReference(env, "Ljava/lang/Exception;");
 
   orc_field_node_class =
-      CreateGlobalClassReference(env, "/org/apache/arrow/adapter/orc/OrcFieldNode");
+      CreateGlobalClassReference(env, "Lorg/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");
+      env, "Lorg/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");
+      CreateGlobalClassReference(env, "Lorg/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");
+                                "(I[Lorg/apache/arrow/adapter/orc/OrcFieldNode;"
+                                "[Lorg/apache/arrow/adapter/orc/OrcMemoryJniWrapper;)V");
 
   env->ExceptionDescribe();
 
@@ -193,7 +194,10 @@ Java_org_apache_arrow_adapter_orc_OrcReaderJniWrapper_nextStripeReader(JNIEnv* e
 
   if (!status.ok()) {
     return static_cast<jlong>(status.code()) * -1;
-    ;
+  }
+
+  if (!stripe_reader) {
+    return static_cast<jlong>(arrow::StatusCode::Invalid) * -1;
   }
 
   return orc_stripe_reader_holder_.Insert(stripe_reader);
@@ -214,7 +218,8 @@ Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_getSchema(JNIEnv* en
   }
 
   jbyteArray ret = env->NewByteArray(out->size());
-  memcpy(env->GetByteArrayElements(ret, nullptr), out->data(), out->size());
+  auto src = reinterpret_cast<const jbyte*>(out->data());
+  env->SetByteArrayRegion(ret, 0, out->size(), src);
   return ret;
 }
 
@@ -223,10 +228,9 @@ Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_next(JNIEnv* env,
                                                                  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);
-  if (!status.ok()) {
-    delete record_batch;
+  std::shared_ptr<arrow::RecordBatch> record_batch;
+  auto status = stripe_reader->ReadNext(&record_batch);
+  if (!status.ok() || !record_batch) {
     return nullptr;
   }
 
@@ -238,7 +242,7 @@ Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_next(JNIEnv* env,
 
   std::vector<std::shared_ptr<arrow::Buffer>> buffers;
   for (int i = 0; i < schema->num_fields(); ++i) {
-    auto column = (*record_batch)->column(i);
+    auto column = record_batch->column(i);
     auto dataArray = column->data();
     jobject field = env->NewObject(orc_field_node_class, orc_field_node_constructor,
                                    column->length(), column->null_count());
@@ -263,7 +267,7 @@ Java_org_apache_arrow_adapter_orc_OrcStripeReaderJniWrapper_next(JNIEnv* env,
 
   // create OrcRecordBatch
   jobject ret = env->NewObject(record_batch_class, record_batch_constructor,
-                               (*record_batch)->num_rows(), field_array, memory_array);
+                               record_batch->num_rows(), field_array, memory_array);
 
   return ret;
 }

java/README.md

@@ -45,13 +45,13 @@ mvn install -P gandiva -pl gandiva -am -Dgandiva.cpp.build.dir=../../debug
 This library is still in Alpha stages, and subject to API changes without
 deprecation warnings.
 
-## Building and running tests for native orc (optional)
+## Building and running tests for arrow jni (optional)
 Arrow Cpp must be built before this step. The cpp build directory must
 be provided as the value for argument arrow.cpp.build.dir. eg.
 
 ```
 cd java
-mvn install -P native-orc -pl adapter/orc -am -Darrow.cpp.build.dir=../../debug
+mvn install -P native-orc -pl arrow-jni -am -Darrow.cpp.build.dir=../../release
 ```
 
 ## Java Code Style Guide

java/adapter/orc/pom.xml

@@ -33,6 +33,56 @@
             <version>${project.version}</version>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.orc</groupId>
+            <artifactId>orc-core</artifactId>
+            <version>1.5.5</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>commons-logging</groupId>
+                    <artifactId>commons-logging</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>2.2.0</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>commons-logging</groupId>
+                    <artifactId>commons-logging</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.servlet</groupId>
+                    <artifactId>servlet-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hive</groupId>
+            <artifactId>hive-storage-api</artifactId>
+            <version>2.6.0</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <parent>
@@ -47,7 +97,7 @@
     <name>Arrow Orc Adapter</name>
     <packaging>jar</packaging>
     <properties>
-        <arrow.cpp.build.dir>../../../cpp/debug-build/</arrow.cpp.build.dir>
+        <arrow.cpp.build.dir>../../../cpp/release-build/</arrow.cpp.build.dir>
     </properties>
 
     <build>

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

@@ -76,8 +76,7 @@ public class OrcReferenceManager implements ReferenceManager {
   @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);
+    bufRefCnt.addAndGet(increment);
   }
 
   @Override

java/adapter/orc/src/test/java/org/apache/arrow/adapter/orc/OrcReaderTest.java

@@ -0,0 +1,105 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.OrcFile;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+
+
+public class OrcReaderTest {
+
+  @Rule
+  public TemporaryFolder testFolder = new TemporaryFolder();
+
+  private static final int MAX_ALLOCATION = 8 * 1024;
+  private static RootAllocator allocator;
+
+  @BeforeClass
+  public static void beforeClass() {
+    allocator = new RootAllocator(MAX_ALLOCATION);
+  }
+
+  @Test
+  public void testOrcJniReader() throws Exception {
+    TypeDescription schema = TypeDescription.fromString("struct<x:int,y:string>");
+    File testFile = new File(testFolder.getRoot(), "test-orc");
+
+    Writer writer = OrcFile.createWriter(new Path(testFile.getAbsolutePath()),
+                                    OrcFile.writerOptions(new Configuration()).setSchema(schema));
+    VectorizedRowBatch batch = schema.createRowBatch();
+    LongColumnVector longColumnVector = (LongColumnVector) batch.cols[0];
+    BytesColumnVector bytesColumnVector = (BytesColumnVector) batch.cols[1];
+    for (int r = 0; r < 1024; ++r) {
+      int row = batch.size++;
+      longColumnVector.vector[row] = r;
+      byte[] buffer = ("Last-" + (r * 3)).getBytes(StandardCharsets.UTF_8);
+      bytesColumnVector.setRef(row, buffer, 0, buffer.length);
+    }
+    writer.addRowBatch(batch);
+    writer.close();
+
+    OrcReader reader = new OrcReader(testFile.getAbsolutePath(), allocator);
+    assertEquals(1, reader.getNumberOfStripes());
+
+    ArrowReader stripeReader = reader.nextStripeReader(1024);
+    VectorSchemaRoot schemaRoot = stripeReader.getVectorSchemaRoot();
+    stripeReader.loadNextBatch();
+
+    List<FieldVector> fields = schemaRoot.getFieldVectors();
+    assertEquals(2, fields.size());
+
+    IntVector intVector = (IntVector)fields.get(0);
+    VarCharVector varCharVector = (VarCharVector)fields.get(1);
+    for (int i = 0; i < 1024; ++i) {
+      assertEquals(i, intVector.get(i));
+      assertEquals("Last-" + (i * 3), new String(varCharVector.get(i), StandardCharsets.UTF_8));
+    }
+
+    assertFalse(stripeReader.loadNextBatch());
+    assertNull(reader.nextStripeReader(1024));
+
+    stripeReader.close();
+    reader.close();
+  }
+}

java/pom.xml

@@ -685,7 +685,7 @@
 
     <profile>
       <!-- orc java depends on arrow cpp, and arrow cpp isn't enabled by default yet -->
-      <id>native-orc</id>
+      <id>arrow-jni</id>
       <modules>
         <module>format</module>
         <module>memory</module>