GH-50689: [C++][Parquet] Add IEEE-754 total order and nan count for floating types

cpp/apidoc/Doxyfile

@@ -2487,6 +2487,7 @@ PREDEFINED             = __attribute__(x)= \
                          ARROW_SUPPRESS_DEPRECATION_WARNING= \
                          ARROW_UNSUPPRESS_DEPRECATION_WARNING= \
                          GANDIVA_EXPORT= \
+                         PARQUET_DEPRECATED(x)= \
                          PARQUET_EXPORT=
 
 # If the MACRO_EXPANSION and EXPAND_ONLY_PREDEF tags are set to YES then this

cpp/src/arrow/dataset/file_parquet.cc

@@ -31,6 +31,7 @@
 #include "arrow/dataset/scanner.h"
 #include "arrow/filesystem/path_util.h"
 #include "arrow/table.h"
+#include "arrow/type_traits.h"
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/future.h"
 #include "arrow/util/iterator.h"
@@ -370,7 +371,7 @@ std::optional<compute::Expression> ParquetFileFragment::EvaluateStatisticsAsExpr
     const parquet::Statistics& statistics) {
   auto field_expr = compute::field_ref(field_ref);
 
-  bool may_have_null = !statistics.HasNullCount() || statistics.null_count() > 0;
+  bool may_have_null = !statistics.HasNullCount() || statistics.null_count() != 0;
   // Optimize for corner case where all values are nulls
   if (statistics.num_values() == 0) {
     // If there are no non-null values, column `field_ref` in the fragment
@@ -379,6 +380,30 @@ std::optional<compute::Expression> ParquetFileFragment::EvaluateStatisticsAsExpr
     return is_null(std::move(field_expr));
   }
 
+  auto with_null = [&](compute::Expression expression) {
+    if (may_have_null) {
+      return compute::or_(std::move(expression), is_null(field_expr));
+    }
+    return expression;
+  };
+  auto is_nan_expression = [&] { return compute::call("is_nan", {field_expr}); };
+
+  const bool is_floating_point = is_floating(field.type()->id());
+  const bool all_nan = is_floating_point && statistics.HasNanCount() &&
+                       statistics.nan_count() == statistics.num_values();
+  if (all_nan) {
+    return with_null(is_nan_expression());
+  }
+
+  if (field.type()->id() == Type::HALF_FLOAT) {
+    // TODO: Arrow compute has no HALF_FLOAT scalar comparison kernels, so numeric
+    // statistics expressions cannot be bound. GH-46858 tracks the scalar
+    // representation, while GH-50512 explains why HALF_FLOAT cannot simply be
+    // added to NumericTypes(). Statistics pruning is optional, so skip it instead
+    // of returning NotImplemented and failing the scan.
+    return std::nullopt;
+  }
+
   std::shared_ptr<Scalar> min, max;
   if (!StatisticsAsScalars(statistics, &min, &max).ok()) {
     return std::nullopt;
@@ -393,10 +418,11 @@ std::optional<compute::Expression> ParquetFileFragment::EvaluateStatisticsAsExpr
     if (min->Equals(*max)) {
       auto single_value = compute::equal(field_expr, compute::literal(std::move(min)));
 
-      if (!may_have_null) {
-        return single_value;
+      if (is_floating_point &&
+          (!statistics.HasNanCount() || statistics.nan_count() != 0)) {
+        single_value = compute::or_(std::move(single_value), is_nan_expression());
       }
-      return compute::or_(std::move(single_value), is_null(std::move(field_expr)));
+      return with_null(std::move(single_value));
     }
 
     auto lower_bound = compute::greater_equal(field_expr, compute::literal(min));
@@ -419,10 +445,10 @@ std::optional<compute::Expression> ParquetFileFragment::EvaluateStatisticsAsExpr
     } else {
       in_range = compute::and_(std::move(lower_bound), std::move(upper_bound));
     }
-    if (may_have_null) {
-      return compute::or_(std::move(in_range), compute::is_null(std::move(field_expr)));
+    if (is_floating_point && (!statistics.HasNanCount() || statistics.nan_count() != 0)) {
+      in_range = compute::or_(std::move(in_range), is_nan_expression());
     }
-    return in_range;
+    return with_null(std::move(in_range));
   }
   return std::nullopt;
 }

cpp/src/arrow/dataset/file_parquet_test.cc

@@ -38,6 +38,7 @@
 #include "arrow/testing/util.h"
 #include "arrow/type.h"
 #include "arrow/type_fwd.h"
+#include "arrow/util/float16.h"
 #include "arrow/util/io_util.h"
 #include "arrow/util/logging_internal.h"
 #include "arrow/util/range.h"
@@ -921,7 +922,7 @@ TEST(TestParquetStatistics, NullMax) {
   auto statistics = reader->RowGroup(0)->metadata()->ColumnChunk(0)->statistics();
   auto stat_expression =
       ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *statistics);
-  EXPECT_EQ(stat_expression->ToString(), "(x >= 1)");
+  EXPECT_EQ(stat_expression->ToString(), "((x >= 1) or is_nan(x))");
 }
 
 TEST(TestParquetStatistics, NoNullCount) {
@@ -974,6 +975,95 @@ TEST(TestParquetStatistics, NoNullCount) {
   }
 }
 
+template <typename T>
+void TestNaNCount(const std::shared_ptr<DataType>& type,
+                  const ::parquet::schema::NodePtr& parquet_node) {
+  auto field = ::arrow::field("x", type);
+  auto dataset_schema = ::arrow::schema({field});
+  ::parquet::ColumnDescriptor descr(parquet_node, 0, 0);
+  auto encode = [](T value) {
+    return std::string(reinterpret_cast<const char*>(&value), sizeof(value));
+  };
+  auto check_expression = [&](const std::optional<compute::Expression>& expression,
+                              const char* expected) {
+    ASSERT_TRUE(expression.has_value());
+    EXPECT_EQ(expected, expression->ToString());
+    ASSERT_OK(expression->Bind(*dataset_schema));
+  };
+
+  ::parquet::EncodedStatistics encoded_stats;
+  encoded_stats.set_min(encode(T{1})).set_max(encode(T{100})).set_null_count(0);
+  auto stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 10);
+  auto expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  check_expression(expression, "(((x >= 1) and (x <= 100)) or is_nan(x))");
+
+  encoded_stats.set_nan_count(0);
+  stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 10);
+  expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  check_expression(expression, "((x >= 1) and (x <= 100))");
+
+  encoded_stats.set_nan_count(2);
+  stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 10);
+  expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  check_expression(expression, "(((x >= 1) and (x <= 100)) or is_nan(x))");
+
+  encoded_stats.set_null_count(1);
+  stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 10);
+  expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  check_expression(expression,
+                   "((((x >= 1) and (x <= 100)) or is_nan(x)) or "
+                   "is_null(x, {nan_is_null=false}))");
+
+  encoded_stats.set_null_count(0);
+  encoded_stats.ClearMinMax();
+  stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 2);
+  expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  check_expression(expression, "is_nan(x)");
+}
+
+TEST(TestParquetStatistics, NaNCount) {
+  TestNaNCount<float>(float32(),
+                      ::parquet::schema::Float("x", ::parquet::Repetition::REQUIRED));
+  TestNaNCount<double>(float64(),
+                       ::parquet::schema::Double("x", ::parquet::Repetition::REQUIRED));
+}
+
+TEST(TestParquetStatistics, HalfFloatNaNCount) {
+  auto field = ::arrow::field("x", float16());
+  auto parquet_node = ::parquet::schema::PrimitiveNode::Make(
+      "x", ::parquet::Repetition::REQUIRED, ::parquet::LogicalType::Float16(),
+      ::parquet::Type::FIXED_LEN_BYTE_ARRAY, 2);
+  ::parquet::ColumnDescriptor descr(parquet_node, 0, 0);
+  auto encode = [](util::Float16 value) {
+    const auto bytes = value.ToLittleEndian();
+    return std::string(reinterpret_cast<const char*>(bytes.data()), bytes.size());
+  };
+
+  ::parquet::EncodedStatistics encoded_stats;
+  encoded_stats.set_min(encode(util::Float16(-1.0f)))
+      .set_max(encode(util::Float16(1.0f)))
+      .set_null_count(0)
+      .set_nan_count(1);
+  auto stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 3);
+  auto expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  ASSERT_FALSE(expression.has_value());
+
+  encoded_stats.ClearMinMax();
+  encoded_stats.set_nan_count(3);
+  stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 3);
+  expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  ASSERT_TRUE(expression.has_value());
+  EXPECT_EQ("is_nan(x)", expression->ToString());
+  ASSERT_OK(expression->Bind(*::arrow::schema({field})));
+
+  encoded_stats.set_null_count(1);
+  stats = ::parquet::Statistics::Make(&descr, &encoded_stats, 3);
+  expression = ParquetFileFragment::EvaluateStatisticsAsExpression(*field, *stats);
+  ASSERT_TRUE(expression.has_value());
+  EXPECT_EQ("(is_nan(x) or is_null(x, {nan_is_null=false}))", expression->ToString());
+  ASSERT_OK(expression->Bind(*::arrow::schema({field})));
+}
+
 TEST_F(TestParquetFileFormat, MultithreadedScanRegression) {
   // GH-38438: This test is similar to MultithreadedScan, but it try to use self
   // designed Executor and DelayedBufferReader to mock async execution to make

cpp/src/parquet/arrow/arrow_reader_writer_test.cc

@@ -24,6 +24,8 @@
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
+#include <algorithm>
+#include <array>
 #include <cstdint>
 #include <functional>
 #include <set>
@@ -59,6 +61,7 @@
 #include "arrow/util/key_value_metadata.h"
 #include "arrow/util/logging_internal.h"
 #include "arrow/util/range.h"
+#include "arrow/util/ubsan.h"
 
 #ifdef ARROW_CSV
 #  include "arrow/csv/api.h"
@@ -75,6 +78,7 @@
 #include "parquet/arrow/writer.h"
 #include "parquet/column_writer.h"
 #include "parquet/file_writer.h"
+#include "parquet/page_index.h"
 #include "parquet/properties.h"
 #include "parquet/test_util.h"
 #include "parquet/types.h"
@@ -3672,6 +3676,54 @@ TEST(TestArrowReadWrite, NonUniqueDictionaryValues) {
   }
 }
 
+TEST(TestArrowReadWrite, FloatingDictionaryBits) {
+  // Float32: -sNaN(payload=1), +qNaN(payload=2), +0, -0, 1.0f.
+  const std::array<uint32_t, 5> dictionary_bits{0xff800001, 0x7fc00002, 0x00000000,
+                                                0x80000000, 0x3f800000};
+  std::vector<float> dictionary_values(dictionary_bits.size());
+  std::transform(dictionary_bits.begin(), dictionary_bits.end(),
+                 dictionary_values.begin(),
+                 [](uint32_t bits) { return ::arrow::util::SafeCopy<float>(bits); });
+  std::shared_ptr<Array> dictionary;
+  ::arrow::ArrayFromVector<::arrow::FloatType>(dictionary_values, &dictionary);
+  auto indices = ArrayFromJSON(::arrow::int32(), "[0, 0, 1, 2, 3, 4]");
+  ASSERT_OK_AND_ASSIGN(auto values, DictionaryArray::FromArrays(indices, dictionary));
+  auto table =
+      Table::Make(::arrow::schema({::arrow::field("values", values->type())}), {values});
+
+  auto properties = WriterProperties::Builder().enable_write_page_index()->build();
+  ASSERT_OK_AND_ASSIGN(auto buffer,
+                       WriteTableToBuffer(table, table->num_rows(), properties));
+  auto parquet_reader = ParquetFileReader::Open(std::make_shared<BufferReader>(buffer));
+  auto metadata = parquet_reader->metadata();
+  ASSERT_EQ(ColumnOrder::IEEE_754_TOTAL_ORDER,
+            metadata->schema()->Column(0)->column_order().get_order());
+  auto statistics = metadata->RowGroup(0)->ColumnChunk(0)->statistics();
+  ASSERT_TRUE(statistics->HasNanCount());
+  ASSERT_EQ(3, statistics->nan_count());
+
+  auto column_index =
+      parquet_reader->GetPageIndexReader()->RowGroup(0)->GetColumnIndex(0);
+  ASSERT_NE(nullptr, column_index);
+  ASSERT_TRUE(column_index->has_nan_counts());
+  EXPECT_THAT(column_index->nan_counts(), ::testing::ElementsAre(3));
+
+  std::unique_ptr<FileReader> arrow_reader;
+  FileReaderBuilder builder;
+  ASSERT_OK(builder.Open(std::make_shared<BufferReader>(buffer)));
+  ASSERT_OK(builder.Build(&arrow_reader));
+  ASSERT_OK_AND_ASSIGN(auto read_table, arrow_reader->ReadTable());
+  auto actual =
+      checked_pointer_cast<::arrow::FloatArray>(read_table->column(0)->chunk(0));
+  const std::array<uint32_t, 6> expected_bits{dictionary_bits[0], dictionary_bits[0],
+                                              dictionary_bits[1], dictionary_bits[2],
+                                              dictionary_bits[3], dictionary_bits[4]};
+  for (int64_t value_index = 0; value_index < actual->length(); ++value_index) {
+    ASSERT_EQ(expected_bits[value_index],
+              ::arrow::util::SafeCopy<uint32_t>(actual->Value(value_index)));
+  }
+}
+
 TEST(TestArrowReadWrite, DictionaryIndexBitwidthRoundtrip) {
   // GH-30302: the bitwidth of Arrow dictionary indices should be preserved
   for (const auto& index_type :