GH-49360: [C++][ORC] Add stripe statistics API to ORCFileReader

Add public APIs to ORCFileReader for accessing stripe-level and
file-level column statistics as Arrow types, stripe-selective reading,
and Arrow-to-ORC schema mapping. This is the adapter-layer foundation
that the dataset layer will consume for predicate pushdown.

New APIs:
- ReadStripes() for stripe-selective reading
- GetColumnStatistics() / GetStripeColumnStatistics() for stats access
- BuildSchemaManifest() for Arrow-to-ORC column ID mapping
- OrcColumnStatistics, OrcSchemaField, OrcSchemaManifest structs

Author: cbb330

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

@@ -18,6 +18,7 @@
 #include "arrow/adapters/orc/adapter.h"
 
 #include <algorithm>
+#include <cmath>
 #include <list>
 #include <memory>
 #include <sstream>
@@ -33,6 +34,7 @@
 #include "arrow/io/interfaces.h"
 #include "arrow/memory_pool.h"
 #include "arrow/record_batch.h"
+#include "arrow/scalar.h"
 #include "arrow/status.h"
 #include "arrow/table.h"
 #include "arrow/table_builder.h"
@@ -43,6 +45,8 @@
 #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;
@@ -98,6 +102,131 @@ namespace {
 // The following is required by ORC to be uint64_t
 constexpr uint64_t kOrcNaturalWriteSize = 128 * 1024;
 
+// Convert ORC column statistics to Arrow OrcColumnStatistics
+// Returns a Result with the converted statistics, or an error if conversion fails
+Result<OrcColumnStatistics> ConvertColumnStatistics(
+    const liborc::ColumnStatistics* orc_stats) {
+  OrcColumnStatistics stats;
+
+  stats.has_null = orc_stats->hasNull();
+  stats.num_values = static_cast<int64_t>(orc_stats->getNumberOfValues());
+  stats.has_min_max = false;
+  stats.min = nullptr;
+  stats.max = nullptr;
+
+  // Try to extract min/max based on the ORC column statistics type
+  const auto* int_stats =
+      dynamic_cast<const liborc::IntegerColumnStatistics*>(
+          orc_stats);
+  const auto* double_stats =
+      dynamic_cast<const liborc::DoubleColumnStatistics*>(
+          orc_stats);
+  const auto* string_stats =
+      dynamic_cast<const liborc::StringColumnStatistics*>(
+          orc_stats);
+  const auto* date_stats =
+      dynamic_cast<const liborc::DateColumnStatistics*>(
+          orc_stats);
+  const auto* ts_stats =
+      dynamic_cast<const liborc::TimestampColumnStatistics*>(
+          orc_stats);
+  const auto* decimal_stats =
+      dynamic_cast<const liborc::DecimalColumnStatistics*>(
+          orc_stats);
+
+  if (int_stats != nullptr) {
+    if (int_stats->hasMinimum() && int_stats->hasMaximum()) {
+      stats.has_min_max = true;
+      stats.min = std::make_shared<Int64Scalar>(
+          int_stats->getMinimum());
+      stats.max = std::make_shared<Int64Scalar>(
+          int_stats->getMaximum());
+    }
+  } else if (double_stats != nullptr) {
+    if (double_stats->hasMinimum() &&
+        double_stats->hasMaximum()) {
+      double min_val = double_stats->getMinimum();
+      double max_val = double_stats->getMaximum();
+      if (!std::isnan(min_val) && !std::isnan(max_val)) {
+        stats.has_min_max = true;
+        stats.min = std::make_shared<DoubleScalar>(min_val);
+        stats.max = std::make_shared<DoubleScalar>(max_val);
+      }
+    }
+  } else if (string_stats != nullptr) {
+    if (string_stats->hasMinimum() &&
+        string_stats->hasMaximum()) {
+      stats.has_min_max = true;
+      stats.min = std::make_shared<StringScalar>(
+          string_stats->getMinimum());
+      stats.max = std::make_shared<StringScalar>(
+          string_stats->getMaximum());
+    }
+  } else if (date_stats != nullptr) {
+    if (date_stats->hasMinimum() &&
+        date_stats->hasMaximum()) {
+      stats.has_min_max = true;
+      stats.min = std::make_shared<Date32Scalar>(
+          date_stats->getMinimum(), date32());
+      stats.max = std::make_shared<Date32Scalar>(
+          date_stats->getMaximum(), date32());
+    }
+  } else if (ts_stats != nullptr) {
+    if (ts_stats->hasMinimum() && ts_stats->hasMaximum()) {
+      stats.has_min_max = true;
+      // getMinimum/getMaximum return milliseconds.
+      // getMinimumNanos/getMaximumNanos return the
+      // last 6 digits of nanoseconds.
+      int64_t min_millis = ts_stats->getMinimum();
+      int64_t max_millis = ts_stats->getMaximum();
+      int32_t min_nanos = ts_stats->getMinimumNanos();
+      int32_t max_nanos = ts_stats->getMaximumNanos();
+
+      // millis * 1,000,000 + sub-millisecond nanos
+      int64_t min_ns = min_millis * 1000000LL + min_nanos;
+      int64_t max_ns = max_millis * 1000000LL + max_nanos;
+
+      auto ts_type = timestamp(TimeUnit::NANO);
+      stats.min =
+          std::make_shared<TimestampScalar>(min_ns, ts_type);
+      stats.max =
+          std::make_shared<TimestampScalar>(max_ns, ts_type);
+    }
+  } else if (decimal_stats != nullptr) {
+    if (decimal_stats->hasMinimum() &&
+        decimal_stats->hasMaximum()) {
+      liborc::Decimal min_dec = decimal_stats->getMinimum();
+      liborc::Decimal max_dec = decimal_stats->getMaximum();
+
+      if (min_dec.scale != max_dec.scale) {
+        // Corrupted stats: scales should always match within
+        // a column. has_min_max remains false (conservative).
+      } else {
+        stats.has_min_max = true;
+
+        Decimal128 min_d128(min_dec.value.getHighBits(),
+                            min_dec.value.getLowBits());
+        Decimal128 max_d128(max_dec.value.getHighBits(),
+                            max_dec.value.getLowBits());
+
+        // Precision 38 is max for Decimal128; the dataset
+        // layer will Cast() to the actual column type.
+        auto dec_type = decimal128(38, min_dec.scale);
+
+        stats.min =
+            std::make_shared<Decimal128Scalar>(min_d128,
+                                               dec_type);
+        stats.max =
+            std::make_shared<Decimal128Scalar>(max_d128,
+                                               dec_type);
+      }
+    }
+  }
+  // Other types (Boolean, Binary, Collection, etc.) don't have min/max statistics
+
+  return stats;
+}
+
 using internal::checked_cast;
 
 class ArrowInputFile : public liborc::InputStream {
@@ -204,6 +333,52 @@ Status CheckTimeZoneDatabaseAvailability() {
 }
 #endif
 
+// Recursively build OrcSchemaField tree by walking paired ORC and Arrow types
+Status BuildOrcSchemaFieldsRecursive(const liborc::Type* orc_type,
+                                     const std::shared_ptr<Field>& arrow_field,
+                                     OrcSchemaField* out_field) {
+  out_field->field = arrow_field;
+  out_field->orc_column_id = static_cast<int>(orc_type->getColumnId());
+
+  // For struct types, recursively build children
+  if (arrow_field->type()->id() == Type::STRUCT &&
+      orc_type->getKind() == liborc::STRUCT) {
+    const auto& struct_type = checked_cast<const StructType&>(*arrow_field->type());
+    size_t num_children = struct_type.num_fields();
+    out_field->children.resize(num_children);
+
+    for (size_t i = 0; i < num_children; ++i) {
+      const liborc::Type* orc_subtype = orc_type->getSubtype(i);
+      RETURN_NOT_OK(BuildOrcSchemaFieldsRecursive(
+          orc_subtype, struct_type.field(static_cast<int>(i)),
+          &out_field->children[i]));
+    }
+  }
+  // For list types, handle the child
+  else if (arrow_field->type()->id() == Type::LIST &&
+           orc_type->getKind() == liborc::LIST) {
+    const auto& list_type = checked_cast<const ListType&>(*arrow_field->type());
+    out_field->children.resize(1);
+    const liborc::Type* orc_subtype = orc_type->getSubtype(0);
+    RETURN_NOT_OK(BuildOrcSchemaFieldsRecursive(orc_subtype, list_type.value_field(),
+                                                &out_field->children[0]));
+  }
+  // For map types, handle key and value children
+  else if (arrow_field->type()->id() == Type::MAP &&
+           orc_type->getKind() == liborc::MAP) {
+    const auto& map_type = checked_cast<const MapType&>(*arrow_field->type());
+    out_field->children.resize(2);
+    const liborc::Type* key_type = orc_type->getSubtype(0);
+    const liborc::Type* value_type = orc_type->getSubtype(1);
+    RETURN_NOT_OK(BuildOrcSchemaFieldsRecursive(key_type, map_type.key_field(),
+                                                &out_field->children[0]));
+    RETURN_NOT_OK(BuildOrcSchemaFieldsRecursive(value_type, map_type.item_field(),
+                                                &out_field->children[1]));
+  }
+
+  return Status::OK();
+}
+
 }  // namespace
 
 class ORCFileReader::Impl {
@@ -409,6 +584,42 @@ 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()) {
+      return Status::Invalid("stripe_indices cannot be empty");
+    }
+
+    std::vector<std::shared_ptr<Table>> tables;
+    tables.reserve(stripe_indices.size());
+
+    for (int64_t stripe_index : stripe_indices) {
+      ARROW_ASSIGN_OR_RAISE(auto batch, ReadStripe(stripe_index));
+      ARROW_ASSIGN_OR_RAISE(auto table, Table::FromRecordBatches({batch}));
+      tables.push_back(table);
+    }
+
+    return ConcatenateTables(tables);
+  }
+
+  Result<std::shared_ptr<Table>> ReadStripes(const std::vector<int64_t>& stripe_indices,
+                                              const std::vector<int>& include_indices) {
+    if (stripe_indices.empty()) {
+      return Status::Invalid("stripe_indices cannot be empty");
+    }
+
+    std::vector<std::shared_ptr<Table>> tables;
+    tables.reserve(stripe_indices.size());
+
+    for (int64_t stripe_index : stripe_indices) {
+      ARROW_ASSIGN_OR_RAISE(auto batch, ReadStripe(stripe_index, include_indices));
+      ARROW_ASSIGN_OR_RAISE(auto table, Table::FromRecordBatches({batch}));
+      tables.push_back(table);
+    }
+
+    return ConcatenateTables(tables);
+  }
+
   Status SelectStripe(liborc::RowReaderOptions* opts, int64_t stripe) {
     ARROW_RETURN_IF(stripe < 0 || stripe >= NumberOfStripes(),
                     Status::Invalid("Out of bounds stripe: ", stripe));
@@ -548,6 +759,102 @@ class ORCFileReader::Impl {
     return NextStripeReader(batch_size, empty_vec);
   }
 
+  Result<OrcColumnStatistics> GetColumnStatistics(int column_index) {
+    ORC_BEGIN_CATCH_NOT_OK
+    std::unique_ptr<liborc::Statistics> file_stats = reader_->getStatistics();
+    if (column_index < 0 ||
+        static_cast<uint32_t>(column_index) >= file_stats->getNumberOfColumns()) {
+      return Status::Invalid("Column index ", column_index, " out of range [0, ",
+                             file_stats->getNumberOfColumns(), ")");
+    }
+    // NOTE: col_stats is a non-owning pointer into file_stats.
+    // ConvertColumnStatistics copies all values into Arrow scalars synchronously,
+    // so the pointer remains valid for the duration of this call.
+    const liborc::ColumnStatistics* col_stats =
+        file_stats->getColumnStatistics(static_cast<uint32_t>(column_index));
+    return ConvertColumnStatistics(col_stats);
+    ORC_END_CATCH_NOT_OK
+  }
+
+  Result<OrcColumnStatistics> GetStripeColumnStatistics(int64_t stripe_index,
+                                                         int column_index) {
+    ORC_BEGIN_CATCH_NOT_OK
+    if (stripe_index < 0 || stripe_index >= static_cast<int64_t>(stripes_.size())) {
+      return Status::Invalid("Stripe index ", stripe_index, " out of range");
+    }
+    std::unique_ptr<liborc::Statistics> stripe_stats =
+        reader_->getStripeStatistics(static_cast<uint64_t>(stripe_index));
+    if (column_index < 0 ||
+        static_cast<uint32_t>(column_index) >= stripe_stats->getNumberOfColumns()) {
+      return Status::Invalid("Column index ", column_index, " out of range [0, ",
+                             stripe_stats->getNumberOfColumns(), ")");
+    }
+    // NOTE: col_stats is a non-owning pointer into stripe_stats.
+    // ConvertColumnStatistics copies all values into Arrow scalars synchronously,
+    // so the pointer remains valid for the duration of this call.
+    const liborc::ColumnStatistics* col_stats =
+        stripe_stats->getColumnStatistics(static_cast<uint32_t>(column_index));
+    return ConvertColumnStatistics(col_stats);
+    ORC_END_CATCH_NOT_OK
+  }
+
+  Result<std::vector<OrcColumnStatistics>> GetStripeStatistics(
+      int64_t stripe_index, const std::vector<int>& column_indices) {
+    ORC_BEGIN_CATCH_NOT_OK
+    if (stripe_index < 0 || stripe_index >= static_cast<int64_t>(stripes_.size())) {
+      return Status::Invalid("Stripe index ", stripe_index, " out of range");
+    }
+    std::unique_ptr<liborc::Statistics> stripe_stats =
+        reader_->getStripeStatistics(static_cast<uint64_t>(stripe_index));
+    std::vector<OrcColumnStatistics> results;
+    results.reserve(column_indices.size());
+    for (int col_idx : column_indices) {
+      if (col_idx < 0 ||
+          static_cast<uint32_t>(col_idx) >= stripe_stats->getNumberOfColumns()) {
+        return Status::Invalid("Column index ", col_idx, " out of range [0, ",
+                               stripe_stats->getNumberOfColumns(), ")");
+      }
+      // NOTE: col_stats is a non-owning pointer into stripe_stats.
+      // ConvertColumnStatistics copies all values into Arrow scalars synchronously,
+      // so the pointer remains valid for the duration of this call.
+      const liborc::ColumnStatistics* col_stats =
+          stripe_stats->getColumnStatistics(static_cast<uint32_t>(col_idx));
+      ARROW_ASSIGN_OR_RAISE(auto converted, ConvertColumnStatistics(col_stats));
+      results.push_back(std::move(converted));
+    }
+    return results;
+    ORC_END_CATCH_NOT_OK
+  }
+
+  Result<std::shared_ptr<OrcSchemaManifest>> BuildSchemaManifest(
+      const std::shared_ptr<Schema>& arrow_schema) {
+    auto manifest = std::make_shared<OrcSchemaManifest>();
+
+    const liborc::Type& orc_root_type = reader_->getType();
+
+    if (orc_root_type.getKind() != liborc::STRUCT) {
+      return Status::Invalid("ORC root type must be STRUCT");
+    }
+
+    size_t num_fields = arrow_schema->num_fields();
+    if (num_fields != orc_root_type.getSubtypeCount()) {
+      return Status::Invalid("Arrow schema field count (", num_fields,
+                             ") does not match ORC type subtype count (",
+                             orc_root_type.getSubtypeCount(), ")");
+    }
+
+    manifest->schema_fields.resize(num_fields);
+
+    for (size_t i = 0; i < num_fields; ++i) {
+      const liborc::Type* orc_subtype = orc_root_type.getSubtype(i);
+      RETURN_NOT_OK(BuildOrcSchemaFieldsRecursive(
+          orc_subtype, arrow_schema->field(static_cast<int>(i)),
+          &manifest->schema_fields[i]));
+    }
+
+    return manifest;
+  }
+
  private:
   MemoryPool* pool_;
   std::unique_ptr<liborc::Reader> reader_;
@@ -613,6 +920,17 @@ 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(
@@ -678,6 +996,25 @@ std::string ORCFileReader::GetSerializedFileTail() {
   return impl_->GetSerializedFileTail();
 }
 
+Result<OrcColumnStatistics> ORCFileReader::GetColumnStatistics(int column_index) {
+  return impl_->GetColumnStatistics(column_index);
+}
+
+Result<OrcColumnStatistics> ORCFileReader::GetStripeColumnStatistics(
+    int64_t stripe_index, int column_index) {
+  return impl_->GetStripeColumnStatistics(stripe_index, column_index);
+}
+
+Result<std::vector<OrcColumnStatistics>> ORCFileReader::GetStripeStatistics(
+    int64_t stripe_index, const std::vector<int>& column_indices) {
+  return impl_->GetStripeStatistics(stripe_index, column_indices);
+}
+
+Result<std::shared_ptr<OrcSchemaManifest>> ORCFileReader::BuildSchemaManifest(
+    const std::shared_ptr<Schema>& arrow_schema) const {
+  return impl_->BuildSchemaManifest(arrow_schema);
+}
+
 namespace {
 
 class ArrowOutputStream : public liborc::OutputStream {