GH-47449: [C++][Parquet] Do not drop all Statistics if SortOrder is UNKNOWN (#47466)

### Rationale for this change

Currently we drop all statistics if `SortOrder` is `UNKNOWN`. This seems too broad and there are some statistics, like `null_count` that could be maintained.

https://github.com/apache/arrow/blob/6f6138b7eedece0841b04f4e235e3bedf5a3ee29/cpp/src/parquet/metadata.cc#L330-L335

Clearing `min/max` but allowing to keep `null_count` when `SortOrder` is `UNKNOWN` would allow users to use them.

### What changes are included in this PR?

Maintain Statistics when reading them if `SortOrder::UNKNOWK` but clear min/max

### Are these changes tested?

Yes, there is a file on parquet-testing which allows us to validate this exact scenario.

### Are there any user-facing changes?

No changes to APIs, users will be able to read statistics on this case.

* GitHub Issue: #47449

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

Author: raulcd

cpp/src/parquet/metadata.cc

@@ -327,17 +327,18 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
   // 2) Statistics must not be corrupted
   inline bool is_stats_set() const {
     DCHECK(writer_version_ != nullptr);
-    // If the column statistics don't exist or column sort order is unknown
-    // we cannot use the column stats
-    if (!column_metadata_->__isset.statistics ||
-        descr_->sort_order() == SortOrder::UNKNOWN) {
+    if (!column_metadata_->__isset.statistics) {
       return false;
     }
     {
       const std::lock_guard<std::mutex> guard(stats_mutex_);
       if (possible_encoded_stats_ == nullptr) {
         possible_encoded_stats_ =
             std::make_shared<EncodedStatistics>(FromThrift(column_metadata_->statistics));
+        if (descr_->sort_order() == SortOrder::UNKNOWN) {
+          // If the column SortOrder is Unknown we can't trust max/min.
+          possible_encoded_stats_->ClearMinMax();
+        }
       }
     }
     return writer_version_->HasCorrectStatistics(type(), *possible_encoded_stats_,
@@ -1588,11 +1589,6 @@ bool ApplicationVersion::HasCorrectStatistics(Type::type col_type,
     return true;
   }
 
-  // Unknown sort order has incorrect stats
-  if (SortOrder::UNKNOWN == sort_order) {
-    return false;
-  }
-
   // PARQUET-251
   if (VersionLt(PARQUET_251_FIXED_VERSION())) {
     return false;

cpp/src/parquet/statistics.cc

@@ -336,6 +336,11 @@ CleanStatistic(std::pair<T, T> min_max, LogicalType::Type::type) {
   return min_max;
 }
 
+std::optional<std::pair<Int96, Int96>> CleanStatistic(std::pair<Int96, Int96> min_max,
+                                                      LogicalType::Type::type) {
+  return min_max;
+}
+
 // In case of floating point types, the following rules are applied (as per
 // upstream parquet-mr):
 // - If any of min/max is NaN, return nothing.
@@ -573,7 +578,9 @@ class TypedStatisticsImpl : public TypedStatistics<DType> {
         min_buffer_(AllocateBuffer(pool_, 0)),
         max_buffer_(AllocateBuffer(pool_, 0)),
         logical_type_(LogicalTypeId(descr_)) {
-    comparator_ = MakeComparator<DType>(descr);
+    if (descr->sort_order() != SortOrder::UNKNOWN) {
+      comparator_ = MakeComparator<DType>(descr);
+    }
     TypedStatisticsImpl::Reset();
   }
 
@@ -732,6 +739,7 @@ class TypedStatisticsImpl : public TypedStatistics<DType> {
       return;
     }
 
+    if (comparator_ == nullptr) return;
     SetMinMaxPair(comparator_->GetMinMax(values));
   }
 
@@ -832,6 +840,7 @@ class TypedStatisticsImpl : public TypedStatistics<DType> {
   }
 
   void SetMinMaxPair(std::pair<T, T> min_max) {
+    if (comparator_ == nullptr) return;
     // CleanStatistic can return a nullopt in case of erroneous values, e.g. NaN
     auto maybe_min_max = CleanStatistic(min_max, logical_type_);
     if (!maybe_min_max) return;
@@ -894,7 +903,7 @@ void TypedStatisticsImpl<DType>::Update(const T* values, int64_t num_values,
   IncrementNullCount(null_count);
   IncrementNumValues(num_values);
 
-  if (num_values == 0) return;
+  if (num_values == 0 || comparator_ == nullptr) return;
   SetMinMaxPair(comparator_->GetMinMax(values, num_values));
 }
 
@@ -909,7 +918,7 @@ void TypedStatisticsImpl<DType>::UpdateSpaced(const T* values, const uint8_t* va
   IncrementNullCount(null_count);
   IncrementNumValues(num_values);
 
-  if (num_values == 0) return;
+  if (num_values == 0 || comparator_ == nullptr) return;
   SetMinMaxPair(comparator_->GetMinMaxSpaced(values, num_spaced_values, valid_bits,
                                              valid_bits_offset));
 }
@@ -1104,6 +1113,7 @@ std::shared_ptr<Statistics> Statistics::Make(
     MAKE_STATS(BOOLEAN, BooleanType);
     MAKE_STATS(INT32, Int32Type);
     MAKE_STATS(INT64, Int64Type);
+    MAKE_STATS(INT96, Int96Type);
     MAKE_STATS(FLOAT, FloatType);
     MAKE_STATS(DOUBLE, DoubleType);
     MAKE_STATS(BYTE_ARRAY, ByteArrayType);

cpp/src/parquet/statistics.h

@@ -163,6 +163,14 @@ class PARQUET_EXPORT EncodedStatistics {
     }
   }
 
+  // Clear Min Max.
+  void ClearMinMax() {
+    has_max = false;
+    max_.clear();
+    has_min = false;
+    min_.clear();
+  }
+
   bool is_set() const {
     return has_min || has_max || has_null_count || has_distinct_count;
   }

cpp/src/parquet/statistics_test.cc

@@ -916,7 +916,7 @@ TEST(CorrectStatistics, Basics) {
   AssertStatsSet(version, props, schema.Column(1), true);
   AssertStatsSet(version, props, schema.Column(2), true);
   AssertStatsSet(version, props, schema.Column(3), true);
-  AssertStatsSet(version, props, schema.Column(4), false);
+  AssertStatsSet(version, props, schema.Column(4), true);
   AssertStatsSet(version, props, schema.Column(5), true);
 }
 
@@ -1632,6 +1632,36 @@ TEST(TestStatisticsSortOrderFloatNaN, NaNAndNullsInfiniteLoop) {
   AssertUnsetMinMax(stats, nans_but_last, &all_but_last_valid);
 }
 
+// Test read statistics for column with UNKNOWN sort order
+TEST(TestStatisticsSortOrder, UNKNOWN) {
+  std::string dir_string(test::get_data_dir());
+  std::stringstream ss;
+  ss << dir_string << "/int96_from_spark.parquet";
+  auto path = ss.str();
+
+  // The file contains a single column of INT96 type (deprecated)
+  // with SortOrder UNKNOWN.
+  // It contains 6 values with a single null value.
+  // The null_count statistics value is preserved.
+  auto file_reader = ParquetFileReader::OpenFile(path);
+  auto rg_reader = file_reader->RowGroup(0);
+  auto metadata = rg_reader->metadata();
+  auto column_schema = metadata->schema()->Column(0);
+  ASSERT_EQ(SortOrder::UNKNOWN, column_schema->sort_order());
+
+  auto column_chunk = metadata->ColumnChunk(0);
+  ASSERT_TRUE(column_chunk->is_stats_set());
+
+  std::shared_ptr<EncodedStatistics> enc_stats = column_chunk->encoded_statistics();
+  ASSERT_TRUE(enc_stats->has_null_count);
+  ASSERT_FALSE(enc_stats->has_distinct_count);
+  ASSERT_FALSE(enc_stats->has_min);
+  ASSERT_FALSE(enc_stats->has_max);
+  ASSERT_EQ(1, enc_stats->null_count);
+  ASSERT_FALSE(enc_stats->is_max_value_exact.has_value());
+  ASSERT_FALSE(enc_stats->is_min_value_exact.has_value());
+}
+
 // Test statistics for binary column with UNSIGNED sort order
 TEST(TestStatisticsSortOrderMinMax, Unsigned) {
   std::string dir_string(test::get_data_dir());