GH-46462: [C++][Parquet] Expose currently thrown EncodedStatistics when checking is_stats_set (#46463)

### Rationale for this change

We are currently throwing away `EncodedStatistics` that are built when calling `is_stats_set`. If we require `EncodedStatistics` we are doing some spurious conversions that could potentially be optimized. For example generating the `TypedStatistics` only when necessary.

### What changes are included in this PR?

- cache the generated `encoded_statistics` on `is_stats_set`
- provide accessor to `encoded_statistics`
- avoid generating the typed statistics if all we care is about `encoded_statistics`

### Are these changes tested?

Existing tests are successful and adapted tests to validate new accessor and `encoded_statistics` values.

### Are there any user-facing changes?

No

* GitHub Issue: #46462

Lead-authored-by: Raúl Cumplido <[email protected]>
Co-authored-by: Antoine Pitrou <[email protected]>
Signed-off-by: Antoine Pitrou <[email protected]>

Author: raulcd

cpp/src/parquet/column_reader.cc

@@ -197,34 +197,8 @@ namespace {
 template <typename H>
 EncodedStatistics ExtractStatsFromHeader(const H& header) {
   EncodedStatistics page_statistics;
-  if (!header.__isset.statistics) {
-    return page_statistics;
-  }
-  const format::Statistics& stats = header.statistics;
-  // Use the new V2 min-max statistics over the former one if it is filled
-  if (stats.__isset.max_value || stats.__isset.min_value) {
-    // TODO: check if the column_order is TYPE_DEFINED_ORDER.
-    if (stats.__isset.max_value) {
-      page_statistics.set_max(stats.max_value);
-    }
-    if (stats.__isset.min_value) {
-      page_statistics.set_min(stats.min_value);
-    }
-  } else if (stats.__isset.max || stats.__isset.min) {
-    // TODO: check created_by to see if it is corrupted for some types.
-    // TODO: check if the sort_order is SIGNED.
-    if (stats.__isset.max) {
-      page_statistics.set_max(stats.max);
-    }
-    if (stats.__isset.min) {
-      page_statistics.set_min(stats.min);
-    }
-  }
-  if (stats.__isset.null_count) {
-    page_statistics.set_null_count(stats.null_count);
-  }
-  if (stats.__isset.distinct_count) {
-    page_statistics.set_distinct_count(stats.distinct_count);
+  if (header.__isset.statistics) {
+    page_statistics = FromThrift(header.statistics);
   }
   return page_statistics;
 }

cpp/src/parquet/metadata.cc

@@ -280,6 +280,7 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
       size_statistics_->Validate(descr_);
     }
     possible_stats_ = nullptr;
+    possible_encoded_stats_ = nullptr;
     possible_geo_stats_ = nullptr;
     InitKeyValueMetadata();
   }
@@ -312,19 +313,17 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
       return false;
     }
     {
-      // Because we are modifying possible_stats_ in a const method
       const std::lock_guard<std::mutex> guard(stats_mutex_);
-      if (possible_stats_ == nullptr) {
-        possible_stats_ = MakeColumnStats(*column_metadata_, descr_);
+      if (possible_encoded_stats_ == nullptr) {
+        possible_encoded_stats_ =
+            std::make_shared<EncodedStatistics>(FromThrift(column_metadata_->statistics));
       }
     }
-    EncodedStatistics encodedStatistics = possible_stats_->Encode();
-    return writer_version_->HasCorrectStatistics(type(), encodedStatistics,
+    return writer_version_->HasCorrectStatistics(type(), *possible_encoded_stats_,
                                                  descr_->sort_order());
   }
 
   inline bool is_geo_stats_set() const {
-    // Because we are modifying possible_geo_stats_ in a const method
     const std::lock_guard<std::mutex> guard(stats_mutex_);
     if (possible_geo_stats_ == nullptr &&
         column_metadata_->__isset.geospatial_statistics) {
@@ -334,8 +333,19 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
     return possible_geo_stats_ != nullptr && possible_geo_stats_->is_valid();
   }
 
+  inline std::shared_ptr<EncodedStatistics> encoded_statistics() const {
+    return is_stats_set() ? possible_encoded_stats_ : nullptr;
+  }
+
   inline std::shared_ptr<Statistics> statistics() const {
-    return is_stats_set() ? possible_stats_ : nullptr;
+    if (is_stats_set()) {
+      const std::lock_guard<std::mutex> guard(stats_mutex_);
+      if (possible_stats_ == nullptr) {
+        possible_stats_ = MakeColumnStats(*column_metadata_, descr_);
+      }
+      return possible_stats_;
+    }
+    return nullptr;
   }
 
   inline std::shared_ptr<SizeStatistics> size_statistics() const {
@@ -425,6 +435,7 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
   }
 
   mutable std::mutex stats_mutex_;
+  mutable std::shared_ptr<EncodedStatistics> possible_encoded_stats_;
   mutable std::shared_ptr<Statistics> possible_stats_;
   mutable std::shared_ptr<geospatial::GeoStatistics> possible_geo_stats_;
   std::vector<Encoding::type> encodings_;
@@ -474,6 +485,10 @@ std::shared_ptr<schema::ColumnPath> ColumnChunkMetaData::path_in_schema() const
   return impl_->path_in_schema();
 }
 
+std::shared_ptr<EncodedStatistics> ColumnChunkMetaData::encoded_statistics() const {
+  return impl_->encoded_statistics();
+}
+
 std::shared_ptr<Statistics> ColumnChunkMetaData::statistics() const {
   return impl_->statistics();
 }

cpp/src/parquet/metadata.h

@@ -143,6 +143,7 @@ class PARQUET_EXPORT ColumnChunkMetaData {
   bool is_stats_set() const;
   bool is_geo_stats_set() const;
   std::shared_ptr<Statistics> statistics() const;
+  std::shared_ptr<EncodedStatistics> encoded_statistics() const;
   std::shared_ptr<SizeStatistics> size_statistics() const;
   std::shared_ptr<geospatial::GeoStatistics> geo_statistics() const;
 

cpp/src/parquet/metadata_test.cc

@@ -158,6 +158,10 @@ TEST(Metadata, TestBuildAccess) {
     ASSERT_EQ(stats_float.max(), rg1_column2->statistics()->EncodeMax());
     ASSERT_EQ(stats_int.min(), rg1_column1->statistics()->EncodeMin());
     ASSERT_EQ(stats_int.max(), rg1_column1->statistics()->EncodeMax());
+    ASSERT_EQ(stats_float.min(), rg1_column2->encoded_statistics()->min());
+    ASSERT_EQ(stats_float.max(), rg1_column2->encoded_statistics()->max());
+    ASSERT_EQ(stats_int.min(), rg1_column1->encoded_statistics()->min());
+    ASSERT_EQ(stats_int.max(), rg1_column1->encoded_statistics()->max());
     ASSERT_EQ(0, rg1_column1->statistics()->null_count());
     ASSERT_EQ(0, rg1_column2->statistics()->null_count());
     ASSERT_EQ(nrows, rg1_column1->statistics()->distinct_count());
@@ -205,6 +209,10 @@ TEST(Metadata, TestBuildAccess) {
     ASSERT_EQ(stats_float.max(), rg2_column2->statistics()->EncodeMax());
     ASSERT_EQ(stats_int.min(), rg1_column1->statistics()->EncodeMin());
     ASSERT_EQ(stats_int.max(), rg1_column1->statistics()->EncodeMax());
+    ASSERT_EQ(stats_float.min(), rg2_column2->encoded_statistics()->min());
+    ASSERT_EQ(stats_float.max(), rg2_column2->encoded_statistics()->max());
+    ASSERT_EQ(stats_int.min(), rg1_column1->encoded_statistics()->min());
+    ASSERT_EQ(stats_int.max(), rg1_column1->encoded_statistics()->max());
     ASSERT_EQ(0, rg2_column1->statistics()->null_count());
     ASSERT_EQ(0, rg2_column2->statistics()->null_count());
     ASSERT_EQ(nrows, rg2_column1->statistics()->distinct_count());

cpp/src/parquet/printer.cc

@@ -156,7 +156,7 @@ void ParquetFilePrinter::DebugPrint(std::ostream& stream, std::list<int> selecte
     // Print column metadata
     for (auto i : selected_columns) {
       auto column_chunk = group_metadata->ColumnChunk(i);
-      std::shared_ptr<Statistics> stats = column_chunk->statistics();
+      std::shared_ptr<EncodedStatistics> stats = column_chunk->encoded_statistics();
 
       const ColumnDescriptor* descr = file_metadata->schema()->Column(i);
       stream << "Column " << i << std::endl;
@@ -165,9 +165,9 @@ void ParquetFilePrinter::DebugPrint(std::ostream& stream, std::list<int> selecte
       }
       stream << "  Values: " << column_chunk->num_values();
       if (column_chunk->is_stats_set()) {
-        std::string min = stats->EncodeMin(), max = stats->EncodeMax();
-        stream << ", Null Values: " << stats->null_count()
-               << ", Distinct Values: " << stats->distinct_count() << std::endl
+        std::string min = stats->min(), max = stats->max();
+        stream << ", Null Values: " << stats->null_count
+               << ", Distinct Values: " << stats->distinct_count << std::endl
                << "  Max: " << FormatStatValue(descr->physical_type(), max)
                << ", Min: " << FormatStatValue(descr->physical_type(), min);
       } else {

cpp/src/parquet/statistics_test.cc

@@ -452,6 +452,13 @@ class TestStatistics : public PrimitiveTypedTest<TestType> {
     EXPECT_TRUE(expected_stats->HasMinMax());
     EXPECT_EQ(expected_stats->EncodeMin(), stats->EncodeMin());
     EXPECT_EQ(expected_stats->EncodeMax(), stats->EncodeMax());
+
+    std::shared_ptr<EncodedStatistics> enc_stats = column_chunk->encoded_statistics();
+    EXPECT_EQ(null_count, enc_stats->null_count);
+    EXPECT_TRUE(enc_stats->has_min);
+    EXPECT_TRUE(enc_stats->has_max);
+    EXPECT_EQ(expected_stats->EncodeMin(), enc_stats->min());
+    EXPECT_EQ(expected_stats->EncodeMax(), enc_stats->max());
   }
 };
 
@@ -800,6 +807,11 @@ void AssertStatsSet(const ApplicationVersion& version,
   stats.set_is_signed(false);
   metadata_builder->SetStatistics(stats);
   ASSERT_EQ(column_chunk->is_stats_set(), expected_is_set);
+  if (expected_is_set) {
+    ASSERT_TRUE(column_chunk->encoded_statistics() != nullptr);
+  } else {
+    ASSERT_TRUE(column_chunk->encoded_statistics() == nullptr);
+  }
 }
 
 // Statistics are restricted for few types in older parquet version

cpp/src/parquet/thrift_internal.h

@@ -262,6 +262,38 @@ static inline AadMetadata FromThrift(format::AesGcmCtrV1 aesGcmCtrV1) {
                      aesGcmCtrV1.supply_aad_prefix};
 }
 
+static inline EncodedStatistics FromThrift(const format::Statistics& stats) {
+  EncodedStatistics out;
+
+  // Use the new V2 min-max statistics over the former one if it is filled
+  if (stats.__isset.max_value || stats.__isset.min_value) {
+    // TODO: check if the column_order is TYPE_DEFINED_ORDER.
+    if (stats.__isset.max_value) {
+      out.set_max(stats.max_value);
+    }
+    if (stats.__isset.min_value) {
+      out.set_min(stats.min_value);
+    }
+  } else if (stats.__isset.max || stats.__isset.min) {
+    // TODO: check created_by to see if it is corrupted for some types.
+    // TODO: check if the sort_order is SIGNED.
+    if (stats.__isset.max) {
+      out.set_max(stats.max);
+    }
+    if (stats.__isset.min) {
+      out.set_min(stats.min);
+    }
+  }
+  if (stats.__isset.null_count) {
+    out.set_null_count(stats.null_count);
+  }
+  if (stats.__isset.distinct_count) {
+    out.set_distinct_count(stats.distinct_count);
+  }
+
+  return out;
+}
+
 static inline geospatial::EncodedGeoStatistics FromThrift(
     const format::GeospatialStatistics& geo_stats) {
   geospatial::EncodedGeoStatistics out;