Feature/orc adapter 2 review

cpp/src/arrow/CMakeLists.txt

@@ -1016,7 +1016,7 @@ endif()
 
 if(ARROW_ORC)
   arrow_add_object_library(ARROW_ORC adapters/orc/adapter.cc adapters/orc/options.cc
-                           adapters/orc/util.cc)
+                           adapters/orc/statistics.cc adapters/orc/util.cc)
   foreach(ARROW_ORC_TARGET ${ARROW_ORC_TARGETS})
     target_link_libraries(${ARROW_ORC_TARGET} PRIVATE orc::orc)
     if(ARROW_ORC_VERSION VERSION_LESS "2.0.0")

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

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

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

@@ -39,10 +39,11 @@
 #include "arrow/type.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/checked_cast.h"
-#include "arrow/util/decimal.h"
 #include "arrow/util/key_value_metadata.h"
 #include "arrow/util/macros.h"
 #include "orc/Exceptions.hh"
+#include "orc/Statistics.hh"
+#include "orc/Type.hh"
 
 // alias to not interfere with nested orc namespace
 namespace liborc = orc;
@@ -217,7 +218,7 @@ class ORCFileReader::Impl {
     std::unique_ptr<liborc::Reader> liborc_reader;
     ORC_CATCH_NOT_OK(liborc_reader = createReader(std::move(io_wrapper), options));
     pool_ = pool;
-    reader_ = std::move(liborc_reader);
+    reader_ = std::shared_ptr<liborc::Reader>(liborc_reader.release());
     current_row_ = 0;
 
     return Init();
@@ -409,6 +410,44 @@ class ORCFileReader::Impl {
     return ReadBatch(opts, schema, stripes_[static_cast<size_t>(stripe)].num_rows);
   }
 
+  Result<std::shared_ptr<Table>> ReadStripes(
+      const std::vector<int64_t>& stripe_indices) {
+    if (stripe_indices.empty()) {
+      ARROW_ASSIGN_OR_RAISE(auto schema, ReadSchema());
+      return Table::MakeEmpty(schema);
+    }
+
+    std::vector<std::shared_ptr<RecordBatch>> batches;
+    batches.reserve(stripe_indices.size());
+    for (int64_t stripe_index : stripe_indices) {
+      ARROW_ASSIGN_OR_RAISE(auto batch, ReadStripe(stripe_index));
+      batches.push_back(std::move(batch));
+    }
+    ARROW_ASSIGN_OR_RAISE(auto schema, ReadSchema());
+    return Table::FromRecordBatches(schema, std::move(batches));
+  }
+
+  Result<std::shared_ptr<Table>> ReadStripes(const std::vector<int64_t>& stripe_indices,
+                                              const std::vector<int>& include_indices) {
+    if (stripe_indices.empty()) {
+      liborc::RowReaderOptions opts = DefaultRowReaderOptions();
+      RETURN_NOT_OK(SelectIndices(&opts, include_indices));
+      ARROW_ASSIGN_OR_RAISE(auto schema, ReadSchema(opts));
+      return Table::MakeEmpty(schema);
+    }
+
+    std::vector<std::shared_ptr<RecordBatch>> batches;
+    batches.reserve(stripe_indices.size());
+    for (int64_t stripe_index : stripe_indices) {
+      ARROW_ASSIGN_OR_RAISE(auto batch, ReadStripe(stripe_index, include_indices));
+      batches.push_back(std::move(batch));
+    }
+    liborc::RowReaderOptions opts = DefaultRowReaderOptions();
+    RETURN_NOT_OK(SelectIndices(&opts, include_indices));
+    ARROW_ASSIGN_OR_RAISE(auto schema, ReadSchema(opts));
+    return Table::FromRecordBatches(schema, std::move(batches));
+  }
+
   Status SelectStripe(liborc::RowReaderOptions* opts, int64_t stripe) {
     ARROW_RETURN_IF(stripe < 0 || stripe >= NumberOfStripes(),
                     Status::Invalid("Out of bounds stripe: ", stripe));
@@ -526,7 +565,7 @@ class ORCFileReader::Impl {
                                              pool_);
   }
 
-  Result<std::shared_ptr<RecordBatchReader>> GetRecordBatchReader(
+  Result<std::unique_ptr<RecordBatchReader>> GetRecordBatchReader(
       int64_t batch_size, const std::vector<std::string>& include_names) {
     liborc::RowReaderOptions opts = DefaultRowReaderOptions();
     if (!include_names.empty()) {
@@ -539,7 +578,7 @@ class ORCFileReader::Impl {
     row_reader = reader_->createRowReader(opts);
     ORC_END_CATCH_NOT_OK
 
-    return std::make_shared<OrcStripeReader>(std::move(row_reader), schema, batch_size,
+    return std::make_unique<OrcStripeReader>(std::move(row_reader), schema, batch_size,
                                              pool_);
   }
 
@@ -548,9 +587,21 @@ class ORCFileReader::Impl {
     return NextStripeReader(batch_size, empty_vec);
   }
 
+  std::shared_ptr<FileMetaData> GetFileMetaData() {
+    try {
+      auto file_stats =
+          std::shared_ptr<const liborc::Statistics>(reader_->getStatistics().release());
+      return std::make_shared<FileMetaData>(reader_, std::move(file_stats));
+    } catch (...) {
+      return nullptr;
+    }
+  }
+
+  const liborc::Type& GetORCType() { return reader_->getType(); }
+
  private:
   MemoryPool* pool_;
-  std::unique_ptr<liborc::Reader> reader_;
+  std::shared_ptr<liborc::Reader> reader_;
   std::vector<StripeInformation> stripes_;
   int64_t current_row_;
 };
@@ -613,14 +664,25 @@ Result<std::shared_ptr<RecordBatch>> ORCFileReader::ReadStripe(
   return impl_->ReadStripe(stripe, include_names);
 }
 
+Result<std::shared_ptr<Table>> ORCFileReader::ReadStripes(
+    const std::vector<int64_t>& stripe_indices) {
+  return impl_->ReadStripes(stripe_indices);
+}
+
+Result<std::shared_ptr<Table>> ORCFileReader::ReadStripes(
+    const std::vector<int64_t>& stripe_indices,
+    const std::vector<int>& include_indices) {
+  return impl_->ReadStripes(stripe_indices, include_indices);
+}
+
 Status ORCFileReader::Seek(int64_t row_number) { return impl_->Seek(row_number); }
 
 Result<std::shared_ptr<RecordBatchReader>> ORCFileReader::NextStripeReader(
     int64_t batch_size) {
   return impl_->NextStripeReader(batch_size);
 }
 
-Result<std::shared_ptr<RecordBatchReader>> ORCFileReader::GetRecordBatchReader(
+Result<std::unique_ptr<RecordBatchReader>> ORCFileReader::GetRecordBatchReader(
     int64_t batch_size, const std::vector<std::string>& include_names) {
   return impl_->GetRecordBatchReader(batch_size, include_names);
 }
@@ -678,6 +740,12 @@ std::string ORCFileReader::GetSerializedFileTail() {
   return impl_->GetSerializedFileTail();
 }
 
+std::shared_ptr<FileMetaData> ORCFileReader::GetFileMetaData() {
+  return impl_->GetFileMetaData();
+}
+
+const ::orc::Type& ORCFileReader::GetORCType() { return impl_->GetORCType(); }
+
 namespace {
 
 class ArrowOutputStream : public liborc::OutputStream {

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

@@ -22,15 +22,19 @@
 #include <vector>
 
 #include "arrow/adapters/orc/options.h"
+#include "arrow/adapters/orc/statistics.h"
 #include "arrow/io/interfaces.h"
 #include "arrow/memory_pool.h"
 #include "arrow/record_batch.h"
 #include "arrow/status.h"
-#include "arrow/type.h"
 #include "arrow/type_fwd.h"
 #include "arrow/util/macros.h"
 #include "arrow/util/visibility.h"
 
+namespace orc {
+class Type;
+}  // namespace orc
+
 namespace arrow {
 namespace adapters {
 namespace orc {
@@ -129,6 +133,29 @@ class ARROW_EXPORT ORCFileReader {
   Result<std::shared_ptr<RecordBatch>> ReadStripe(
       int64_t stripe, const std::vector<std::string>& include_names);
 
+  /// \brief Read multiple selected stripes as a Table
+  ///
+  /// Reads only the specified stripes and concatenates them into a single table.
+  /// This is useful for stripe-selective reading based on predicate pushdown.
+  /// If stripe_indices is empty, returns an empty table with the file's schema.
+  ///
+  /// \param[in] stripe_indices the indices of stripes to read
+  /// \return the returned Table containing data from the selected stripes
+  Result<std::shared_ptr<Table>> ReadStripes(
+      const std::vector<int64_t>& stripe_indices);
+
+  /// \brief Read multiple selected stripes with column selection as a Table
+  ///
+  /// Reads only the specified stripes and selected columns, concatenating them
+  /// into a single table.
+  /// If stripe_indices is empty, returns an empty table with the selected schema.
+  ///
+  /// \param[in] stripe_indices the indices of stripes to read
+  /// \param[in] include_indices the selected field indices to read
+  /// \return the returned Table containing data from the selected stripes and columns
+  Result<std::shared_ptr<Table>> ReadStripes(const std::vector<int64_t>& stripe_indices,
+                                              const std::vector<int>& include_indices);
+
   /// \brief Seek to designated row. Invoke NextStripeReader() after seek
   ///        will return stripe reader starting from designated row.
   ///
@@ -171,7 +198,7 @@ class ARROW_EXPORT ORCFileReader {
   /// \param[in] include_names the selected field names to read, if not empty
   /// (otherwise all fields are read)
   /// \return the record batch iterator
-  Result<std::shared_ptr<RecordBatchReader>> GetRecordBatchReader(
+  Result<std::unique_ptr<RecordBatchReader>> GetRecordBatchReader(
       int64_t batch_size, const std::vector<std::string>& include_names);
 
   /// \brief The number of stripes in the file
@@ -267,6 +294,14 @@ class ARROW_EXPORT ORCFileReader {
   /// \return A KeyValueMetadata object containing the ORC metadata
   Result<std::shared_ptr<const KeyValueMetadata>> ReadMetadata();
 
+  /// \brief Get file-level metadata view.
+  std::shared_ptr<FileMetaData> GetFileMetaData();
+
+  /// \brief Get the ORC type tree for column ID mapping.
+  ///
+  /// \return reference to the root ORC Type (STRUCT), owned by the reader.
+  const ::orc::Type& GetORCType();
+
  private:
   class Impl;
   std::unique_ptr<Impl> impl_;