ARROW-6720:[JAVA][C++]Support Parquet Read and Write in Java

cpp/src/jni/CMakeLists.txt

@@ -22,3 +22,7 @@
 if(ARROW_ORC)
   add_subdirectory(orc)
 endif()
+
+if(ARROW_PARQUET)
+  add_subdirectory(parquet)
+endif()

cpp/src/jni/parquet/CMakeLists.txt

@@ -0,0 +1,56 @@
+# 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.
+
+#
+# arrow_parquet_jni
+#
+
+project(arrow_parquet_jni)
+
+cmake_minimum_required(VERSION 3.11)
+
+find_package(JNI REQUIRED)
+
+add_custom_target(arrow_parquet_jni)
+
+set(JNI_HEADERS_DIR "${CMAKE_CURRENT_BINARY_DIR}/generated")
+
+add_subdirectory(../../../../java/adapter/parquet ./java)
+
+set(ARROW_BUILD_STATIC OFF)
+
+add_arrow_lib(arrow_parquet_jni
+              BUILD_SHARED
+              SOURCES
+              jni_wrapper.cpp
+              hdfs_connector.cc
+              file_connector.cc
+              parquet_reader.cc
+              parquet_writer.cc
+              OUTPUTS
+              ARROW_PARQUET_JNI_LIBRARIES
+              SHARED_LINK_LIBS
+              parquet_shared
+              arrow_shared
+              EXTRA_INCLUDES
+              ${JNI_HEADERS_DIR}
+              PRIVATE_INCLUDES
+              ${JNI_INCLUDE_DIRS}
+              DEPENDENCIES
+              arrow_parquet_java)
+
+add_dependencies(arrow_parquet_jni ${ARROW_PARQUET_JNI_LIBRARIES})

cpp/src/jni/parquet/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 <mutex>
+#include <unordered_map>
+#include <utility>
+
+#include "arrow/util/macros.h"
+
+namespace arrow {
+namespace jni {
+
+/**
+ * An utility class that map module id to module pointers.
+ * @tparam Holder class of the object to hold.
+ */
+template <typename Holder>
+class ConcurrentMap {
+ public:
+  ConcurrentMap() : module_id_(init_module_id_) {}
+
+  jlong Insert(Holder holder) {
+    std::lock_guard<std::mutex> lock(mtx_);
+    jlong result = module_id_++;
+    map_.insert(std::pair<jlong, Holder>(result, holder));
+    return result;
+  }
+
+  void Erase(jlong module_id) {
+    std::lock_guard<std::mutex> lock(mtx_);
+    map_.erase(module_id);
+  }
+
+  Holder Lookup(jlong module_id) {
+    std::lock_guard<std::mutex> lock(mtx_);
+    auto it = map_.find(module_id);
+    if (it != map_.end()) {
+      return it->second;
+    }
+    return NULLPTR;
+  }
+
+  void Clear() {
+    std::lock_guard<std::mutex> lock(mtx_);
+    map_.clear();
+  }
+
+ private:
+  // Initialize the module id starting value to a number greater than zero
+  // to allow for easier debugging of uninitialized java variables.
+  static constexpr int init_module_id_ = 4;
+
+  int64_t module_id_;
+  std::mutex mtx_;
+  // map from module ids returned to Java and module pointers
+  std::unordered_map<jlong, Holder> map_;
+};
+
+}  // namespace jni
+}  // namespace arrow
+
+#endif  // JNI_ID_TO_MODULE_MAP_H

cpp/src/jni/parquet/connector.h

@@ -0,0 +1,57 @@
+// 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.
+
+#ifndef CONNECTOR_H
+#define CONNECTOR_H
+
+#include <arrow/array.h>
+#include <arrow/buffer.h>
+#include <arrow/io/interfaces.h>
+#include <arrow/status.h>
+#include <cstring>
+#include <iostream>
+#include <memory>
+#include <string>
+
+namespace jni {
+namespace parquet {
+
+/// \brief A base class for Connectors
+///
+/// This class is used by ParquetReader and ParquetWriter to handle
+/// parquet files from different resources, local and hdfs for now using
+/// unified API.
+class Connector {
+ public:
+  Connector() = default;
+  virtual ~Connector() {}
+  std::string GetFileName() { return file_path_; }
+  virtual ::arrow::Status OpenReadable(bool option) = 0;
+  virtual ::arrow::Status OpenWritable(bool option, int replication) = 0;
+  virtual std::shared_ptr<::arrow::io::RandomAccessFile> GetReader() = 0;
+  virtual std::shared_ptr<::arrow::io::OutputStream> GetWriter() = 0;
+  virtual void TearDown() = 0;
+
+ protected:
+  std::string file_path_;
+  std::string dir_path_;
+  virtual ::arrow::Status Mkdir(std::string path) = 0;
+};
+}  // namespace parquet
+}  // namespace jni
+
+#endif

cpp/src/jni/parquet/file_connector.cc

@@ -0,0 +1,92 @@
+// 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.
+
+#include "jni/parquet/file_connector.h"
+
+#include <atomic>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+#include <numeric>
+#include <sstream>  // IWYU pragma: keep
+#include <string>
+#include <vector>
+#include "arrow/filesystem/localfs.h"
+#include "arrow/filesystem/path_util.h"
+
+namespace jni {
+namespace parquet {
+
+FileConnector::FileConnector(std::string path) {
+  file_path_ = path;
+  auto path_pair = arrow::fs::internal::GetAbstractPathParent(file_path_);
+  dir_path_ = path_pair.first;
+}
+
+FileConnector::~FileConnector() {}
+
+void FileConnector::TearDown() {
+  if (file_writer_) {
+    file_writer_->Close();
+  }
+}
+
+::arrow::Status FileConnector::OpenReadable(bool option) {
+  ::arrow::Status msg = ::arrow::io::ReadableFile::Open(file_path_, &file_reader_);
+  if (!msg.ok()) {
+    std::cerr << "Open file failed, file name is " << file_path_ << ", error is : " << msg
+              << std::endl;
+    return msg;
+  }
+  return msg;
+}
+
+::arrow::Status FileConnector::OpenWritable(bool option, int replication) {
+  ::arrow::Status msg;
+  if (!dir_path_.empty()) {
+    msg = Mkdir(dir_path_);
+    if (!msg.ok()) {
+      std::cerr << "Mkdir for path failed " << dir_path_ << ", error is : " << msg
+                << std::endl;
+      return msg;
+    }
+  }
+
+  msg = ::arrow::io::FileOutputStream::Open(file_path_, false, &file_writer_);
+  if (!msg.ok()) {
+    std::cerr << "Open file failed, file name is " << file_path_ << ", error is : " << msg
+              << std::endl;
+    return msg;
+  }
+  return msg;
+}
+
+::arrow::Status FileConnector::Mkdir(std::string path) {
+  std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs =
+      std::make_shared<::arrow::fs::LocalFileSystem>();
+  ::arrow::Status msg = local_fs->CreateDir(path);
+  if (!msg.ok()) {
+    std::cerr << "Make Directory for " << path << " failed, error is: " << msg
+              << std::endl;
+    return msg;
+  }
+  return msg;
+}
+
+}  // namespace parquet
+}  // namespace jni

cpp/src/jni/parquet/file_connector.h

@@ -0,0 +1,68 @@
+// 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.
+
+#ifndef FILE_CONNECTOR_H
+#define FILE_CONNECTOR_H
+
+#include <arrow/io/file.h>
+#include <memory>
+#include <string>
+#include "jni/parquet/connector.h"
+
+namespace jni {
+namespace parquet {
+
+/// \brief A connector to handle local parquet file
+///
+/// This class is derived from base class "Connector", used to provide
+/// an unified interface for ParquetReader and ParquetWriter to handle
+/// files from different sources.
+/// Member methods are wrappers of some methods under arrow/io/file.h
+class FileConnector : public Connector {
+ public:
+  /// \brief Construction of this class
+  /// \param[in] path local file path
+  explicit FileConnector(std::string path);
+  ~FileConnector();
+
+  /// \brief Open local parquet file as readable handler
+  /// \param[in] option a param holder, not used
+  ::arrow::Status OpenReadable(bool option);
+
+  /// \brief Open local parquet file as writable handler
+  /// \param[in] option a param holder, not used
+  /// \param[in] replication a param holder, not used
+  ::arrow::Status OpenWritable(bool option, int replication);
+
+  /// \brief Get reader handler
+  std::shared_ptr<::arrow::io::RandomAccessFile> GetReader() { return file_reader_; }
+
+  /// \brief Get writer handler
+  std::shared_ptr<::arrow::io::OutputStream> GetWriter() { return file_writer_; }
+
+  /// \brief Tear down connection and handlers
+  void TearDown();
+
+ protected:
+  ::arrow::Status Mkdir(std::string path);
+  std::shared_ptr<::arrow::io::ReadableFile> file_reader_;
+  std::shared_ptr<::arrow::io::OutputStream> file_writer_;
+};
+}  // namespace parquet
+}  // namespace jni
+
+#endif