ARROW-11296: [C++][Python] Add ORC reader options

cpp/src/arrow/CMakeLists.txt

@@ -507,7 +507,11 @@ if(ARROW_JSON)
 endif()
 
 if(ARROW_ORC)
-  list(APPEND ARROW_SRCS adapters/orc/adapter.cc adapters/orc/adapter_util.cc)
+  list(APPEND
+       ARROW_SRCS
+       adapters/orc/adapter.cc
+       adapters/orc/adapter_options.cc
+       adapters/orc/adapter_util.cc)
 endif()
 
 if(CXX_LINKER_SUPPORTS_VERSION_SCRIPT)

cpp/src/arrow/adapters/orc/CMakeLists.txt

@@ -20,7 +20,8 @@
 #
 
 # Headers: top level
-install(FILES adapter.h DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/arrow/adapters/orc")
+install(FILES adapter.h adapter_options.h
+        DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/arrow/adapters/orc")
 
 # pkg-config support
 arrow_add_pkg_config("arrow-orc")

cpp/src/arrow/adapters/orc/adapter.cc

@@ -92,8 +92,7 @@ namespace orc {
 
 namespace {
 
-// The following are required by ORC to be uint64_t
-constexpr uint64_t kOrcWriterBatchSize = 128 * 1024;
+// The following is required by ORC to be uint64_t
 constexpr uint64_t kOrcNaturalWriteSize = 128 * 1024;
 
 using internal::checked_cast;
@@ -632,37 +631,37 @@ class ArrowOutputStream : public liborc::OutputStream {
 
 class ORCFileWriter::Impl {
  public:
-  Status Open(arrow::io::OutputStream* output_stream) {
+  Status Open(arrow::io::OutputStream* output_stream, const WriteOptions& write_options) {
     out_stream_ = std::unique_ptr<liborc::OutputStream>(
         checked_cast<liborc::OutputStream*>(new ArrowOutputStream(*output_stream)));
+    batch_size_ = write_options.batch_size();
+    orc_options_ = write_options.get_orc_writer_options();
     return Status::OK();
   }
 
   Status Write(const Table& table) {
-    std::unique_ptr<liborc::WriterOptions> orc_options =
-        std::unique_ptr<liborc::WriterOptions>(new liborc::WriterOptions());
     ARROW_ASSIGN_OR_RAISE(auto orc_schema, GetOrcType(*(table.schema())));
     ORC_CATCH_NOT_OK(
-        writer_ = liborc::createWriter(*orc_schema, out_stream_.get(), *orc_options))
+        writer_ = liborc::createWriter(*orc_schema, out_stream_.get(), *orc_options_))
 
     int64_t num_rows = table.num_rows();
     const int num_cols_ = table.num_columns();
     std::vector<int64_t> arrow_index_offset(num_cols_, 0);
     std::vector<int> arrow_chunk_offset(num_cols_, 0);
     std::unique_ptr<liborc::ColumnVectorBatch> batch =
-        writer_->createRowBatch(kOrcWriterBatchSize);
+        writer_->createRowBatch(batch_size_);
     liborc::StructVectorBatch* root =
         internal::checked_cast<liborc::StructVectorBatch*>(batch.get());
     while (num_rows > 0) {
       for (int i = 0; i < num_cols_; i++) {
         RETURN_NOT_OK(adapters::orc::WriteBatch(
-            *(table.column(i)), kOrcWriterBatchSize, &(arrow_chunk_offset[i]),
+            *(table.column(i)), batch_size_, &(arrow_chunk_offset[i]),
             &(arrow_index_offset[i]), (root->fields)[i]));
       }
       root->numElements = (root->fields)[0]->numElements;
       writer_->add(*batch);
       batch->clear();
-      num_rows -= kOrcWriterBatchSize;
+      num_rows -= batch_size_;
     }
     return Status::OK();
   }
@@ -675,17 +674,19 @@ class ORCFileWriter::Impl {
  private:
   std::unique_ptr<liborc::Writer> writer_;
   std::unique_ptr<liborc::OutputStream> out_stream_;
+  std::shared_ptr<liborc::WriterOptions> orc_options_;
+  uint64_t batch_size_;
 };
 
 ORCFileWriter::~ORCFileWriter() {}
 
 ORCFileWriter::ORCFileWriter() { impl_.reset(new ORCFileWriter::Impl()); }
 
 Result<std::unique_ptr<ORCFileWriter>> ORCFileWriter::Open(
-    io::OutputStream* output_stream) {
+    io::OutputStream* output_stream, const WriteOptions& writer_options) {
   std::unique_ptr<ORCFileWriter> result =
       std::unique_ptr<ORCFileWriter>(new ORCFileWriter());
-  Status status = result->impl_->Open(output_stream);
+  Status status = result->impl_->Open(output_stream, writer_options);
   RETURN_NOT_OK(status);
   return std::move(result);
 }

cpp/src/arrow/adapters/orc/adapter.h

@@ -21,6 +21,7 @@
 #include <memory>
 #include <vector>
 
+#include "arrow/adapters/orc/adapter_options.h"
 #include "arrow/io/interfaces.h"
 #include "arrow/memory_pool.h"
 #include "arrow/record_batch.h"
@@ -264,8 +265,11 @@ class ARROW_EXPORT ORCFileWriter {
   /// \brief Creates a new ORC writer.
   ///
   /// \param[in] output_stream a pointer to the io::OutputStream to write into
+  /// \param[in] write_options the ORC writer options for Arrow
   /// \return the returned writer object
-  static Result<std::unique_ptr<ORCFileWriter>> Open(io::OutputStream* output_stream);
+  static Result<std::unique_ptr<ORCFileWriter>> Open(
+      io::OutputStream* output_stream,
+      const WriteOptions& write_options = WriteOptions());
 
   /// \brief Write a table
   ///

cpp/src/arrow/adapters/orc/adapter_options.cc

@@ -0,0 +1,216 @@
+// 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 "arrow/adapters/orc/adapter_options.h"
+
+#include "arrow/adapters/orc/adapter.h"
+#include "orc/Common.hh"
+#include "orc/Int128.hh"
+#include "orc/Writer.hh"
+
+namespace liborc = orc;
+
+namespace arrow {
+
+namespace adapters {
+
+namespace orc {
+
+std::string FileVersion::ToString() const {
+  std::stringstream ss;
+  ss << major() << '.' << minor();
+  return ss.str();
+}
+
+const FileVersion& FileVersion::v_0_11() {
+  static FileVersion version(0, 11);
+  return version;
+}
+
+const FileVersion& FileVersion::v_0_12() {
+  static FileVersion version(0, 12);
+  return version;
+}
+
+WriteOptions::WriteOptions()
+    : orc_writer_options_(std::make_shared<liborc::WriterOptions>()), batch_size_(1024) {
+  // PASS
+}
+
+WriteOptions::WriteOptions(const WriteOptions& rhs)
+    : orc_writer_options_(
+          std::make_shared<liborc::WriterOptions>(*(rhs.orc_writer_options_.get()))),
+      batch_size_(rhs.batch_size_) {
+  // PASS
+}
+
+WriteOptions::WriteOptions(WriteOptions& rhs) {
+  // swap orc_writer_options_ with rhs
+  orc_writer_options_.swap(rhs.orc_writer_options_);
+  batch_size_ = rhs.batch_size_;
+}
+
+WriteOptions& WriteOptions::operator=(const WriteOptions& rhs) {
+  if (this != &rhs) {
+    orc_writer_options_.reset(
+        new liborc::WriterOptions(*(rhs.orc_writer_options_.get())));
+    batch_size_ = rhs.batch_size_;
+  }
+  return *this;
+}
+
+WriteOptions::~WriteOptions() {
+  // PASS
+}
+
+WriteOptions& WriteOptions::set_batch_size(uint64_t size) {
+  batch_size_ = size;
+  return *this;
+}
+
+uint64_t WriteOptions::batch_size() const { return batch_size_; }
+
+RleVersion WriteOptions::rle_version() const {
+  return static_cast<RleVersion>(orc_writer_options_->getRleVersion());
+}
+
+WriteOptions& WriteOptions::set_stripe_size(uint64_t size) {
+  orc_writer_options_->setStripeSize(size);
+  return *this;
+}
+
+uint64_t WriteOptions::stripe_size() const {
+  return orc_writer_options_->getStripeSize();
+}
+
+WriteOptions& WriteOptions::set_compression_block_size(uint64_t size) {
+  orc_writer_options_->setCompressionBlockSize(size);
+  return *this;
+}
+
+uint64_t WriteOptions::compression_block_size() const {
+  return orc_writer_options_->getCompressionBlockSize();
+}
+
+WriteOptions& WriteOptions::set_row_index_stride(uint64_t stride) {
+  orc_writer_options_->setRowIndexStride(stride);
+  return *this;
+}
+
+uint64_t WriteOptions::row_index_stride() const {
+  return orc_writer_options_->getRowIndexStride();
+}
+
+WriteOptions& WriteOptions::set_dictionary_key_size_threshold(double val) {
+  orc_writer_options_->setDictionaryKeySizeThreshold(val);
+  return *this;
+}
+
+double WriteOptions::dictionary_key_size_threshold() const {
+  return orc_writer_options_->getDictionaryKeySizeThreshold();
+}
+
+WriteOptions& WriteOptions::set_file_version(const FileVersion& version) {
+  // Only Hive_0_11 and Hive_0_12 version are supported currently
+  uint32_t major = version.major(), minor = version.minor();
+  if (major == 0 && (minor == 11 || minor == 12)) {
+    orc_writer_options_->setFileVersion(liborc::FileVersion(major, minor));
+    return *this;
+  }
+  throw std::logic_error("Unsupported file version specified.");
+}
+
+FileVersion WriteOptions::file_version() const {
+  liborc::FileVersion orc_file_version_ = orc_writer_options_->getFileVersion();
+  return FileVersion(orc_file_version_.getMajor(), orc_file_version_.getMinor());
+}
+
+WriteOptions& WriteOptions::set_compression(CompressionKind comp) {
+  orc_writer_options_->setCompression(static_cast<liborc::CompressionKind>(comp));
+  return *this;
+}
+
+CompressionKind WriteOptions::compression() const {
+  return static_cast<CompressionKind>(orc_writer_options_->getCompression());
+}
+
+WriteOptions& WriteOptions::set_compression_strategy(CompressionStrategy strategy) {
+  orc_writer_options_->setCompressionStrategy(
+      static_cast<liborc::CompressionStrategy>(strategy));
+  return *this;
+}
+
+CompressionStrategy WriteOptions::compression_strategy() const {
+  return static_cast<CompressionStrategy>(orc_writer_options_->getCompressionStrategy());
+}
+
+bool WriteOptions::aligned_bitpacking() const {
+  return orc_writer_options_->getAlignedBitpacking();
+}
+
+WriteOptions& WriteOptions::set_padding_tolerance(double tolerance) {
+  orc_writer_options_->setPaddingTolerance(tolerance);
+  return *this;
+}
+
+double WriteOptions::padding_tolerance() const {
+  return orc_writer_options_->getPaddingTolerance();
+}
+
+WriteOptions& WriteOptions::set_error_stream(std::ostream& err_stream) {
+  orc_writer_options_->setErrorStream(err_stream);
+  return *this;
+}
+
+std::ostream* WriteOptions::error_stream() const {
+  return orc_writer_options_->getErrorStream();
+}
+
+bool WriteOptions::enable_index() const { return orc_writer_options_->getEnableIndex(); }
+
+bool WriteOptions::enable_dictionary() const {
+  return orc_writer_options_->getEnableDictionary();
+}
+
+WriteOptions& WriteOptions::set_columns_use_bloom_filter(
+    const std::set<uint64_t>& columns) {
+  orc_writer_options_->setColumnsUseBloomFilter(columns);
+  return *this;
+}
+
+bool WriteOptions::is_column_use_bloom_filter(uint64_t column) const {
+  return orc_writer_options_->isColumnUseBloomFilter(column);
+}
+
+WriteOptions& WriteOptions::set_bloom_filter_fpp(double fpp) {
+  orc_writer_options_->setBloomFilterFPP(fpp);
+  return *this;
+}
+
+double WriteOptions::bloom_filter_fpp() const {
+  return orc_writer_options_->getBloomFilterFPP();
+}
+
+// delibrately not provide setter to write bloom filter version because
+// we only support UTF8 for now.
+BloomFilterVersion WriteOptions::bloom_filter_version() const {
+  return static_cast<BloomFilterVersion>(orc_writer_options_->getBloomFilterVersion());
+}
+
+}  // namespace orc
+}  // namespace adapters
+}  // namespace arrow