From 9b2995402a11f74ed1cacc0c9df6bbeeaf22a08b Mon Sep 17 00:00:00 2001 From: Zehua Zou Date: Tue, 4 Aug 2026 22:53:05 +0800 Subject: [PATCH 1/2] Add IEEE-754 total order and nan count for floating types --- cpp/apidoc/Doxyfile | 1 + cpp/src/arrow/dataset/file_parquet.cc | 40 +- cpp/src/arrow/dataset/file_parquet_test.cc | 92 +++- .../parquet/arrow/arrow_reader_writer_test.cc | 52 ++ cpp/src/parquet/arrow/index_test.cc | 125 ++++- cpp/src/parquet/arrow/reader_internal.cc | 9 + cpp/src/parquet/encoder.cc | 27 +- cpp/src/parquet/encoding_test.cc | 67 +++ cpp/src/parquet/file_serialize_test.cc | 63 +++ cpp/src/parquet/file_writer.cc | 38 +- cpp/src/parquet/metadata.cc | 115 ++-- cpp/src/parquet/page_index.cc | 22 +- cpp/src/parquet/page_index.h | 9 + cpp/src/parquet/page_index_test.cc | 38 ++ cpp/src/parquet/properties.h | 28 +- cpp/src/parquet/schema.cc | 21 + cpp/src/parquet/schema.h | 8 +- cpp/src/parquet/schema_internal.h | 5 +- cpp/src/parquet/schema_test.cc | 12 + cpp/src/parquet/statistics.cc | 520 +++++++++++++++--- cpp/src/parquet/statistics.h | 55 +- cpp/src/parquet/statistics_test.cc | 114 +++- cpp/src/parquet/thrift_internal.h | 7 + cpp/src/parquet/types.cc | 2 + cpp/src/parquet/types.h | 5 +- cpp/submodules/parquet-testing | 2 +- docs/source/python/parquet.rst | 2 +- 27 files changed, 1256 insertions(+), 223 deletions(-) diff --git a/cpp/apidoc/Doxyfile b/cpp/apidoc/Doxyfile index 82688718d92e..56acf971b2bf 100644 --- a/cpp/apidoc/Doxyfile +++ b/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 diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc index ba0e93f09d40..aae8945463b2 100644 --- a/cpp/src/arrow/dataset/file_parquet.cc +++ b/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 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 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 min, max; if (!StatisticsAsScalars(statistics, &min, &max).ok()) { return std::nullopt; @@ -393,10 +418,11 @@ std::optional 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 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; } diff --git a/cpp/src/arrow/dataset/file_parquet_test.cc b/cpp/src/arrow/dataset/file_parquet_test.cc index 0d86f5eaa0d3..96d73ed5d774 100644 --- a/cpp/src/arrow/dataset/file_parquet_test.cc +++ b/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 +void TestNaNCount(const std::shared_ptr& 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(&value), sizeof(value)); + }; + auto check_expression = [&](const std::optional& 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(float32(), + ::parquet::schema::Float("x", ::parquet::Repetition::REQUIRED)); + TestNaNCount(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(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 diff --git a/cpp/src/parquet/arrow/arrow_reader_writer_test.cc b/cpp/src/parquet/arrow/arrow_reader_writer_test.cc index 2bdbc38b3647..cf6b38ac8476 100644 --- a/cpp/src/parquet/arrow/arrow_reader_writer_test.cc +++ b/cpp/src/parquet/arrow/arrow_reader_writer_test.cc @@ -24,6 +24,8 @@ #include "gmock/gmock.h" #include "gtest/gtest.h" +#include +#include #include #include #include @@ -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 dictionary_bits{0xff800001, 0x7fc00002, 0x00000000, + 0x80000000, 0x3f800000}; + std::vector dictionary_values(dictionary_bits.size()); + std::transform(dictionary_bits.begin(), dictionary_bits.end(), + dictionary_values.begin(), + [](uint32_t bits) { return ::arrow::util::SafeCopy(bits); }); + std::shared_ptr 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(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 arrow_reader; + FileReaderBuilder builder; + ASSERT_OK(builder.Open(std::make_shared(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 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(actual->Value(value_index))); + } +} + TEST(TestArrowReadWrite, DictionaryIndexBitwidthRoundtrip) { // GH-30302: the bitwidth of Arrow dictionary indices should be preserved for (const auto& index_type : diff --git a/cpp/src/parquet/arrow/index_test.cc b/cpp/src/parquet/arrow/index_test.cc index b12265cf4ee7..96ded352ed1f 100644 --- a/cpp/src/parquet/arrow/index_test.cc +++ b/cpp/src/parquet/arrow/index_test.cc @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -36,6 +37,7 @@ #include "arrow/testing/gtest_util.h" #include "arrow/testing/random.h" #include "arrow/util/checked_cast.h" +#include "arrow/util/ubsan.h" #include "parquet/arrow/reader.h" #include "parquet/arrow/reader_internal.h" @@ -47,6 +49,7 @@ #include "parquet/file_writer.h" #include "parquet/page_index.h" #include "parquet/properties.h" +#include "parquet/test_util.h" using arrow::Array; using arrow::Buffer; @@ -75,6 +78,7 @@ struct ColumnIndexObject { std::vector max_values; BoundaryOrder::type boundary_order = BoundaryOrder::Unordered; std::vector null_counts; + std::vector nan_counts; ColumnIndexObject() = default; @@ -82,12 +86,14 @@ struct ColumnIndexObject { const std::vector& min_values, const std::vector& max_values, BoundaryOrder::type boundary_order, - const std::vector& null_counts) + const std::vector& null_counts, + const std::vector& nan_counts) : null_pages(null_pages), min_values(min_values), max_values(max_values), boundary_order(boundary_order), - null_counts(null_counts) {} + null_counts(null_counts), + nan_counts(nan_counts) {} explicit ColumnIndexObject(const ColumnIndex* column_index) { if (column_index == nullptr) { @@ -100,12 +106,15 @@ struct ColumnIndexObject { if (column_index->has_null_counts()) { null_counts = column_index->null_counts(); } + if (column_index->has_nan_counts()) { + nan_counts = column_index->nan_counts(); + } } bool operator==(const ColumnIndexObject& b) const { return null_pages == b.null_pages && min_values == b.min_values && max_values == b.max_values && boundary_order == b.boundary_order && - null_counts == b.null_counts; + null_counts == b.null_counts && nan_counts == b.nan_counts; } }; @@ -245,22 +254,22 @@ TEST_F(ParquetPageIndexRoundTripTest, SimpleRoundTrip) { ::testing::ElementsAre( ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(1)}, /*max_values=*/{encode_int64(3)}, BoundaryOrder::Ascending, - /*null_counts=*/{1}}, + /*null_counts=*/{1}, /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{"a"}, /*max_values=*/{"d"}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(1)}, /*max_values=*/{encode_int64(2)}, BoundaryOrder::Ascending, - /*null_counts=*/{2}}, + /*null_counts=*/{2}, /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(5)}, /*max_values=*/{encode_int64(6)}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{"f"}, /*max_values=*/{"f"}, BoundaryOrder::Ascending, - /*null_counts=*/{1}}, + /*null_counts=*/{1}, /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(3)}, /*max_values=*/{encode_int64(3)}, BoundaryOrder::Ascending, - /*null_counts=*/{1}})); + /*null_counts=*/{1}, /*nan_counts=*/{}})); } TEST_F(ParquetPageIndexRoundTripTest, SimpleRoundTripWithStatsDisabled) { @@ -314,17 +323,17 @@ TEST_F(ParquetPageIndexRoundTripTest, SimpleRoundTripWithColumnStatsDisabled) { empty_column_index, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{"a"}, /*max_values=*/{"d"}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(1)}, /*max_values=*/{encode_int64(2)}, BoundaryOrder::Ascending, - /*null_counts=*/{2}}, + /*null_counts=*/{2}, /*nan_counts=*/{}}, empty_column_index, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{"f"}, /*max_values=*/{"f"}, BoundaryOrder::Ascending, - /*null_counts=*/{1}}, + /*null_counts=*/{1}, /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(3)}, /*max_values=*/{encode_int64(3)}, BoundaryOrder::Ascending, - /*null_counts=*/{1}})); + /*null_counts=*/{1}, /*nan_counts=*/{}})); } TEST_F(ParquetPageIndexRoundTripTest, DropLargeStats) { @@ -346,7 +355,7 @@ TEST_F(ParquetPageIndexRoundTripTest, DropLargeStats) { ::testing::ElementsAre( ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{"short_string"}, /*max_values=*/{"short_string"}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{}}, ColumnIndexObject{})); } @@ -373,18 +382,21 @@ TEST_F(ParquetPageIndexRoundTripTest, MultiplePages) { /*min_values=*/{encode_int64(1), encode_int64(3), encode_int64(6), ""}, /*max_values=*/{encode_int64(2), encode_int64(4), encode_int64(6), ""}, BoundaryOrder::Ascending, - /*null_counts=*/{0, 0, 1, 2}}, + /*null_counts=*/{0, 0, 1, 2}, + /*nan_counts=*/{}}, ColumnIndexObject{/*null_pages=*/{false, false, false, true}, /*min_values=*/{"a", "c", "f", ""}, /*max_values=*/{"b", "d", "f", ""}, BoundaryOrder::Ascending, - /*null_counts=*/{0, 0, 1, 2}})); + /*null_counts=*/{0, 0, 1, 2}, /*nan_counts=*/{}})); } TEST_F(ParquetPageIndexRoundTripTest, DoubleWithNaNs) { - auto writer_properties = WriterProperties::Builder() - .enable_write_page_index() - ->max_row_group_length(3) /* 3 rows per row group */ - ->build(); + auto writer_properties = + WriterProperties::Builder() + .enable_write_page_index() + ->max_row_group_length(3) /* 3 rows per row group */ + ->floating_point_column_order(ColumnOrder::TYPE_DEFINED_ORDER) + ->build(); // Create table to write with NaNs. auto vectors = std::vector>(4); @@ -411,17 +423,17 @@ TEST_F(ParquetPageIndexRoundTripTest, DoubleWithNaNs) { ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_double(0.1)}, /*max_values=*/{encode_double(1.0)}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{1}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_double(-0.0)}, /*max_values=*/{encode_double(+0.0)}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{1}}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_double(-0.0)}, /*max_values=*/{encode_double(+0.0)}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{1}}, ColumnIndexObject{ /* Page with only NaN values does not have column index built */})); } @@ -446,11 +458,11 @@ TEST_F(ParquetPageIndexRoundTripTest, EnablePerColumn) { ::testing::ElementsAre( ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(0)}, /*max_values=*/{encode_int64(0)}, BoundaryOrder::Ascending, - /*null_counts=*/{0}}, + /*null_counts=*/{0}, /*nan_counts=*/{}}, ColumnIndexObject{/* page index of c1 is disabled */}, ColumnIndexObject{/*null_pages=*/{false}, /*min_values=*/{encode_int64(2)}, /*max_values=*/{encode_int64(2)}, BoundaryOrder::Ascending, - /*null_counts=*/{0}})); + /*null_counts=*/{0}, /*nan_counts=*/{}})); } class ParquetBloomFilterRoundTripTest : public ::testing::Test, @@ -663,4 +675,67 @@ TEST_F(ParquetBloomFilterRoundTripTest, ThrowForBoolean) { ::testing::HasSubstr("BloomFilterBuilder does not support boolean type")); } +TEST(ParquetPageIndex, FloatingPointOrders) { + auto reader = ParquetFileReader::OpenFile( + test::get_data_file("floating_orders_nan_count.parquet")); + auto metadata = reader->metadata(); + ASSERT_EQ(6, metadata->num_columns()); + ASSERT_EQ(5, metadata->num_row_groups()); + + for (int column = 0; column < metadata->num_columns(); ++column) { + const auto expected_order = column % 2 == 0 ? ColumnOrder::IEEE_754_TOTAL_ORDER + : ColumnOrder::TYPE_DEFINED_ORDER; + ASSERT_EQ(expected_order, + metadata->schema()->Column(column)->column_order().get_order()); + } + + constexpr int kAllNaNRowGroup = 2; + constexpr int64_t kNaNCount = 10; + // FLOAT, DOUBLE, and FLOAT16 -qNaN and +qNaN bounds. + const std::array expected_min_bits{0xffffffff, 0xffffffffffffffff, 0xffff}; + const std::array expected_max_bits{0x7fffffff, 0x7fffffffffffffff, 0x7fff}; + auto scalar_bits = [](const ::arrow::Scalar& scalar) -> uint64_t { + switch (scalar.type->id()) { + case ::arrow::Type::FLOAT: + return ::arrow::util::SafeCopy( + static_cast(scalar).value); + case ::arrow::Type::DOUBLE: + return ::arrow::util::SafeCopy( + static_cast(scalar).value); + case ::arrow::Type::HALF_FLOAT: + return static_cast(scalar).value; + default: + throw ParquetException("Unexpected floating-point scalar type"); + } + }; + auto page_index_reader = reader->GetPageIndexReader(); + ASSERT_NE(nullptr, page_index_reader); + auto row_group_index = page_index_reader->RowGroup(kAllNaNRowGroup); + ASSERT_NE(nullptr, row_group_index); + auto row_group = metadata->RowGroup(kAllNaNRowGroup); + for (int column = 0; column < metadata->num_columns(); ++column) { + SCOPED_TRACE(::testing::Message() << "column=" << column); + auto statistics = row_group->ColumnChunk(column)->statistics(); + ASSERT_NE(nullptr, statistics); + ASSERT_TRUE(statistics->HasNanCount()); + ASSERT_EQ(kNaNCount, statistics->nan_count()); + if (column % 2 != 0) { + ASSERT_FALSE(statistics->HasMinMax()); + ASSERT_EQ(nullptr, row_group_index->GetColumnIndex(column)); + continue; + } + ASSERT_TRUE(statistics->HasMinMax()); + auto column_index = row_group_index->GetColumnIndex(column); + ASSERT_NE(nullptr, column_index); + ASSERT_TRUE(column_index->has_nan_counts()); + EXPECT_THAT(column_index->nan_counts(), ::testing::ElementsAre(kNaNCount)); + + std::shared_ptr<::arrow::Scalar> min; + std::shared_ptr<::arrow::Scalar> max; + ASSERT_OK(StatisticsAsScalars(*statistics, &min, &max)); + ASSERT_EQ(expected_min_bits[column / 2], scalar_bits(*min)); + ASSERT_EQ(expected_max_bits[column / 2], scalar_bits(*max)); + } +} + } // namespace parquet::arrow diff --git a/cpp/src/parquet/arrow/reader_internal.cc b/cpp/src/parquet/arrow/reader_internal.cc index 12f36fe39cf8..b11b1775e487 100644 --- a/cpp/src/parquet/arrow/reader_internal.cc +++ b/cpp/src/parquet/arrow/reader_internal.cc @@ -270,6 +270,15 @@ Status ByteArrayStatisticsAsScalars(const Statistics& statistics, return ExtractDecimalMinMaxFromBytes(statistics.EncodeMin(), statistics.EncodeMax(), *logical_type, min, max); } + if (logical_type->type() == LogicalType::Type::FLOAT16) { + *min = std::make_shared<::arrow::HalfFloatScalar>( + ::arrow::util::Float16::FromLittleEndian( + reinterpret_cast(statistics.EncodeMin().data()))); + *max = std::make_shared<::arrow::HalfFloatScalar>( + ::arrow::util::Float16::FromLittleEndian( + reinterpret_cast(statistics.EncodeMax().data()))); + return Status::OK(); + } std::shared_ptr<::arrow::DataType> type; if (statistics.descr()->physical_type() == Type::FIXED_LEN_BYTE_ARRAY) { type = ::arrow::fixed_size_binary(statistics.descr()->type_length()); diff --git a/cpp/src/parquet/encoder.cc b/cpp/src/parquet/encoder.cc index 3e469df277b8..9d21d50e739f 100644 --- a/cpp/src/parquet/encoder.cc +++ b/cpp/src/parquet/encoder.cc @@ -18,7 +18,6 @@ #include "parquet/encoding.h" #include -#include #include #include #include @@ -424,6 +423,26 @@ template struct DictEncoderTraits { using c_type = typename DType::c_type; using MemoTableType = ::arrow::internal::ScalarMemoTable; + + static c_type DictKey(c_type value) { return value; } +}; + +template <> +struct DictEncoderTraits { + using MemoTableType = ::arrow::internal::ScalarMemoTable; + + static uint32_t DictKey(float value) { + return ::arrow::util::SafeCopy(value); + } +}; + +template <> +struct DictEncoderTraits { + using MemoTableType = ::arrow::internal::ScalarMemoTable; + + static uint64_t DictKey(double value) { + return ::arrow::util::SafeCopy(value); + } }; template <> @@ -684,7 +703,8 @@ inline void DictEncoderImpl::Put(const T& v) { }; int32_t memo_index; - PARQUET_THROW_NOT_OK(memo_table_.GetOrInsert(v, on_found, on_not_found, &memo_index)); + PARQUET_THROW_NOT_OK(memo_table_.GetOrInsert(DictEncoderTraits::DictKey(v), + on_found, on_not_found, &memo_index)); buffered_indices_.push_back(memo_index); } @@ -813,7 +833,8 @@ void DictEncoderImpl::PutDictionary(const ::arrow::Array& values) { dict_encoded_size_ += static_cast(sizeof(typename DType::c_type) * data.length()); for (int64_t i = 0; i < data.length(); i++) { int32_t unused_memo_index; - PARQUET_THROW_NOT_OK(memo_table_.GetOrInsert(data.Value(i), &unused_memo_index)); + PARQUET_THROW_NOT_OK(memo_table_.GetOrInsert( + DictEncoderTraits::DictKey(data.Value(i)), &unused_memo_index)); } } diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc index 831829e4a210..df25e25914a3 100644 --- a/cpp/src/parquet/encoding_test.cc +++ b/cpp/src/parquet/encoding_test.cc @@ -18,6 +18,7 @@ #include #include +#include #include #include #include @@ -29,6 +30,7 @@ #include "arrow/array.h" #include "arrow/array/builder_binary.h" #include "arrow/array/builder_dict.h" +#include "arrow/array/builder_primitive.h" #include "arrow/array/concatenate.h" #include "arrow/compute/cast.h" #include "arrow/testing/gtest_util.h" @@ -41,6 +43,7 @@ #include "arrow/util/bitmap_writer.h" #include "arrow/util/checked_cast.h" #include "arrow/util/endian.h" +#include "arrow/util/hashing.h" #include "arrow/util/string.h" #include "parquet/encoding.h" #include "parquet/platform.h" @@ -466,6 +469,70 @@ TEST(TestDictionaryEncoding, CannotDictDecodeBoolean) { ASSERT_THROW(MakeDictDecoder(nullptr), ParquetException); } +template +void TestFloatingDictionaryBits(const std::array& bits, + int num_entries) { + using T = typename DType::c_type; + std::array values; + std::transform(bits.begin(), bits.end(), values.begin(), + [](UInt value) { return ::arrow::util::SafeCopy(value); }); + + auto encoder = MakeTypedEncoder(Encoding::PLAIN, true); + auto dictionary = dynamic_cast*>(encoder.get()); + ASSERT_NE(nullptr, dictionary); + encoder->Put(values.data(), values.size()); + ASSERT_EQ(num_entries, dictionary->num_entries()); + + auto buffer = AllocateBuffer(default_memory_pool(), dictionary->dict_encoded_size()); + dictionary->WriteDict(buffer->mutable_data()); + const UInt* encoded = reinterpret_cast(buffer->data()); + for (int value_index = 0; value_index < num_entries; ++value_index) { + EXPECT_EQ(bits[value_index], encoded[value_index]); + } + + using ArrowType = std::conditional_t, + ::arrow::FloatType, ::arrow::DoubleType>; + typename ::arrow::TypeTraits::BuilderType builder; + ASSERT_OK( + builder.AppendValues(std::vector(values.begin(), values.begin() + num_entries))); + std::shared_ptr<::arrow::Array> values_array; + ASSERT_OK(builder.Finish(&values_array)); + auto direct_encoder = MakeTypedEncoder(Encoding::PLAIN, true); + auto direct_dictionary = dynamic_cast*>(direct_encoder.get()); + ASSERT_NE(nullptr, direct_dictionary); + direct_dictionary->PutDictionary(*values_array); + ASSERT_EQ(num_entries, direct_dictionary->num_entries()); + auto direct_buffer = + AllocateBuffer(default_memory_pool(), direct_dictionary->dict_encoded_size()); + direct_dictionary->WriteDict(direct_buffer->mutable_data()); + const UInt* direct_encoded = reinterpret_cast(direct_buffer->data()); + for (int value_index = 0; value_index < num_entries; ++value_index) { + EXPECT_EQ(bits[value_index], direct_encoded[value_index]); + } +} + +TEST(TestDictionaryEncoding, FloatingPointBits) { + // Float32: +sNaN(payload=1), +qNaN(payload=2), +0, -0, +sNaN(payload=1). + TestFloatingDictionaryBits( + std::array{0x7f800001, 0x7fc00002, 0x00000000, 0x80000000, 0x7f800001}, + 4); + // Float64: +sNaN(payload=1), +qNaN(payload=2), +0, -0, +sNaN(payload=1). + TestFloatingDictionaryBits( + std::array{0x7ff0000000000001, 0x7ff8000000000002, 0x0000000000000000, + 0x8000000000000000, 0x7ff0000000000001}, + 4); +} + +TEST(TestDictionaryEncoding, NaNHashCollision) { + // Distinct NaN values that collided under the previous dictionary hash. + const std::array bits{0x7ff3b2b800075724, 0xfff0445d001b3a31}; + ASSERT_EQ(::arrow::internal::ScalarHelper::ComputeHash( + ::arrow::util::SafeCopy(bits[0])), + ::arrow::internal::ScalarHelper::ComputeHash( + ::arrow::util::SafeCopy(bits[1]))); + TestFloatingDictionaryBits(bits, 2); +} + // ---------------------------------------------------------------------- // Shared arrow builder decode tests diff --git a/cpp/src/parquet/file_serialize_test.cc b/cpp/src/parquet/file_serialize_test.cc index f287e493a908..7d4700dc9a57 100644 --- a/cpp/src/parquet/file_serialize_test.cc +++ b/cpp/src/parquet/file_serialize_test.cc @@ -18,6 +18,8 @@ #include #include +#include + #include "arrow/testing/gtest_compat.h" #include "arrow/util/config.h" @@ -480,6 +482,67 @@ TEST(ParquetRoundtrip, AllNulls) { EXPECT_THAT(def_levels, ElementsAre(0, 0, 0)); } +TEST(TestFileWriter, FloatingPointColumnOrder) { + schema::NodeVector fields{ + schema::Float("float", Repetition::REQUIRED), + schema::Double("double", Repetition::REQUIRED), + schema::PrimitiveNode::Make("float16", Repetition::REQUIRED, LogicalType::Float16(), + Type::FIXED_LEN_BYTE_ARRAY, 2), + schema::Int32("int", Repetition::REQUIRED)}; + + auto schema = std::static_pointer_cast( + GroupNode::Make("schema", Repetition::REQUIRED, fields)); + + auto assert_type_lengths = [](const SchemaDescriptor* schema) { + ASSERT_EQ(-1, schema->Column(0)->type_length()); + ASSERT_EQ(-1, schema->Column(1)->type_length()); + ASSERT_EQ(2, schema->Column(2)->type_length()); + ASSERT_EQ(-1, schema->Column(3)->type_length()); + }; + + auto write_orders = [&](ColumnOrder::type order) { + auto properties = + WriterProperties::Builder().floating_point_column_order(order)->build(); + auto sink = CreateOutputStream(); + auto writer = ParquetFileWriter::Open(sink, schema, properties); + assert_type_lengths(writer->schema()); + writer->Close(); + auto writer_metadata = writer->metadata(); + PARQUET_ASSIGN_OR_THROW(auto buffer, sink->Finish()); + auto file_metadata = + ParquetFileReader::Open( + std::make_shared<::arrow::io::BufferReader>(std::move(buffer))) + ->metadata(); + return std::pair{std::move(writer_metadata), std::move(file_metadata)}; + }; + + auto assert_orders = [](const SchemaDescriptor* schema, + ColumnOrder::type floating_point_order) { + for (int column_index = 0; column_index < 3; ++column_index) { + ASSERT_EQ(floating_point_order, + schema->Column(column_index)->column_order().get_order()); + } + ASSERT_EQ(ColumnOrder::TYPE_DEFINED_ORDER, + schema->Column(3)->column_order().get_order()); + }; + + SchemaDescriptor input_schema; + input_schema.Init(schema); + assert_type_lengths(&input_schema); + assert_orders(&input_schema, ColumnOrder::TYPE_DEFINED_ORDER); + + auto [ieee_writer, ieee_file] = write_orders(ColumnOrder::IEEE_754_TOTAL_ORDER); + assert_orders(&input_schema, ColumnOrder::TYPE_DEFINED_ORDER); + assert_orders(ieee_writer->schema(), ColumnOrder::IEEE_754_TOTAL_ORDER); + assert_orders(ieee_file->schema(), ColumnOrder::IEEE_754_TOTAL_ORDER); + + auto [type_writer, type_file] = write_orders(ColumnOrder::TYPE_DEFINED_ORDER); + assert_orders(&input_schema, ColumnOrder::TYPE_DEFINED_ORDER); + assert_orders(type_writer->schema(), ColumnOrder::TYPE_DEFINED_ORDER); + assert_orders(type_file->schema(), ColumnOrder::TYPE_DEFINED_ORDER); + EXPECT_THROW(ieee_file->AppendRowGroups(*type_file), ParquetException); +} + } // namespace test } // namespace parquet diff --git a/cpp/src/parquet/file_writer.cc b/cpp/src/parquet/file_writer.cc index ec303408f363..ab2f0324fab0 100644 --- a/cpp/src/parquet/file_writer.cc +++ b/cpp/src/parquet/file_writer.cc @@ -23,9 +23,11 @@ #include #include +#include "arrow/util/checked_cast.h" #include "arrow/util/endian.h" #include "arrow/util/key_value_metadata.h" #include "arrow/util/logging_internal.h" +#include "generated/parquet_types.h" #include "parquet/bloom_filter_writer.h" #include "parquet/column_writer.h" #include "parquet/encryption/encryption_internal.h" @@ -34,6 +36,7 @@ #include "parquet/page_index.h" #include "parquet/platform.h" #include "parquet/schema.h" +#include "parquet/schema_internal.h" using arrow::MemoryPool; @@ -335,15 +338,46 @@ class RowGroupSerializer : public RowGroupWriter::Contents { // An implementation of ParquetFileWriter::Contents that deals with the Parquet // file structure, Thrift serialization, and other internal matters +namespace { + +std::shared_ptr MakeWriterSchema(const GroupNode& input_schema, + const WriterProperties& properties) { + std::vector elements; + schema::ToParquet(&input_schema, &elements); + for (auto& element : elements) { + if (element.__isset.type && !element.__isset.type_length) { + // Unflatten reads the in-memory value even though non-FLBA schemas omit it. + element.type_length = -1; + } + } + auto root = schema::Unflatten(elements.data(), static_cast(elements.size())); + auto writer_schema = std::shared_ptr( + ::arrow::internal::checked_pointer_cast(std::move(root))); + + SchemaDescriptor descr; + descr.Init(writer_schema); + std::vector column_orders(descr.num_columns(), ColumnOrder::type_defined_); + for (int column_index = 0; column_index < descr.num_columns(); ++column_index) { + if (schema::IsFloatingPoint(*descr.Column(column_index))) { + column_orders[column_index] = ColumnOrder(properties.floating_point_column_order()); + } + } + descr.updateColumnOrders(column_orders); + return writer_schema; +} + +} // namespace + class FileSerializer : public ParquetFileWriter::Contents { public: static std::unique_ptr Open( std::shared_ptr sink, std::shared_ptr schema, std::shared_ptr properties, std::shared_ptr key_value_metadata) { + auto writer_schema = MakeWriterSchema(*schema, *properties); std::unique_ptr result( - new FileSerializer(std::move(sink), std::move(schema), std::move(properties), - std::move(key_value_metadata))); + new FileSerializer(std::move(sink), std::move(writer_schema), + std::move(properties), std::move(key_value_metadata))); return result; } diff --git a/cpp/src/parquet/metadata.cc b/cpp/src/parquet/metadata.cc index 98f60df63dd4..656c2d70abb8 100644 --- a/cpp/src/parquet/metadata.cc +++ b/cpp/src/parquet/metadata.cc @@ -92,46 +92,6 @@ std::string ParquetVersionToString(ParquetVersion::type ver) { namespace { -template -std::shared_ptr MakeTypedColumnStats(const format::ColumnMetaData& metadata, - const ColumnDescriptor* descr, - ::arrow::MemoryPool* pool) { - const auto& statistics = metadata.statistics; - const std::string kEmpty = ""; - const std::string* encoded_min = &kEmpty; - const std::string* encoded_max = &kEmpty; - bool has_min_max = false; - std::optional min_exact = std::nullopt; - std::optional max_exact = std::nullopt; - - switch (GetStatisticsMinMaxField(*descr)) { - case StatisticsMinMaxField::kMinValueMaxValue: - encoded_min = &statistics.min_value; - encoded_max = &statistics.max_value; - has_min_max = statistics.__isset.max_value && statistics.__isset.min_value; - min_exact = statistics.__isset.is_min_value_exact - ? std::optional(statistics.is_min_value_exact) - : std::nullopt; - max_exact = statistics.__isset.is_max_value_exact - ? std::optional(statistics.is_max_value_exact) - : std::nullopt; - break; - case StatisticsMinMaxField::kLegacyMinMax: - encoded_min = &statistics.min; - encoded_max = &statistics.max; - has_min_max = statistics.__isset.max && statistics.__isset.min; - break; - case StatisticsMinMaxField::kInvalid: - break; - } - - return MakeStatistics( - descr, *encoded_min, *encoded_max, metadata.num_values - statistics.null_count, - statistics.null_count, statistics.distinct_count, has_min_max, - statistics.__isset.null_count, statistics.__isset.distinct_count, min_exact, - max_exact, pool); -} - std::shared_ptr MakeColumnGeometryStats( const format::ColumnMetaData& metadata, const ColumnDescriptor* descr) { if (metadata.__isset.geospatial_statistics) { @@ -145,6 +105,7 @@ std::shared_ptr MakeColumnGeometryStats( std::shared_ptr MakeColumnStats(const format::ColumnMetaData& meta_data, const ColumnDescriptor* descr, + const EncodedStatistics& encoded_statistics, ::arrow::MemoryPool* pool) { auto metadata_type = LoadEnumSafe(&meta_data.type); if (descr->physical_type() != metadata_type) { @@ -152,27 +113,8 @@ std::shared_ptr MakeColumnStats(const format::ColumnMetaData& meta_d "ColumnMetaData type does not match ColumnDescriptor physical type: " + TypeToString(metadata_type) + " vs. " + TypeToString(descr->physical_type())); } - switch (metadata_type) { - case Type::BOOLEAN: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::INT32: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::INT64: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::INT96: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::DOUBLE: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::FLOAT: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::BYTE_ARRAY: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::FIXED_LEN_BYTE_ARRAY: - return MakeTypedColumnStats(meta_data, descr, pool); - case Type::UNDEFINED: - break; - } - throw ParquetException("Can't decode page statistics for selected column type"); + return Statistics::Make(descr, &encoded_statistics, + meta_data.num_values - encoded_statistics.null_count, pool); } // Get KeyValueMetadata from parquet Thrift RowGroup or ColumnChunk metadata. @@ -371,7 +313,8 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl { const std::lock_guard guard(stats_mutex_); if (possible_stats_ == nullptr) { possible_stats_ = - MakeColumnStats(*column_metadata_, descr_, properties_.memory_pool()); + MakeColumnStats(*column_metadata_, descr_, *possible_encoded_stats_, + properties_.memory_pool()); } return possible_stats_; } @@ -936,6 +879,12 @@ class FileMetaData::FileMetaDataImpl { auto msg = "AppendRowGroups requires equal schemas.\n" + diff_output.str(); throw ParquetException(msg); } + for (int column_index = 0; column_index < schema()->num_columns(); ++column_index) { + if (schema()->Column(column_index)->column_order().get_order() != + other->schema()->Column(column_index)->column_order().get_order()) { + throw ParquetException("AppendRowGroups requires equal column orders."); + } + } // ARROW-13654: `other` may point to self, be careful not to enter an infinite loop const int n = other->num_row_groups(); @@ -1036,10 +985,20 @@ class FileMetaData::FileMetaDataImpl { // update ColumnOrder std::vector column_orders; if (metadata_->__isset.column_orders) { + if (metadata_->column_orders.size() != static_cast(schema_.num_columns())) { + throw ParquetException( + "Malformed schema: ColumnOrder count does not match number of columns"); + } column_orders.reserve(metadata_->column_orders.size()); - for (auto& column_order : metadata_->column_orders) { + for (size_t i = 0; i < metadata_->column_orders.size(); ++i) { + const auto& column_order = metadata_->column_orders[i]; if (column_order.__isset.TYPE_ORDER) { column_orders.push_back(ColumnOrder::type_defined_); + } else if (column_order.__isset.IEEE_754_TOTAL_ORDER) { + const auto* column = schema_.Column(static_cast(i)); + column_orders.push_back(schema::IsFloatingPoint(*column) + ? ColumnOrder::ieee_754_total_order_ + : ColumnOrder::unknown_); } else { column_orders.push_back(ColumnOrder::unknown_); } @@ -1607,6 +1566,9 @@ bool ApplicationVersion::VersionEq(const ApplicationVersion& other_version) cons bool ApplicationVersion::HasCorrectStatistics(Type::type col_type, const EncodedStatistics& statistics, SortOrder::type sort_order) const { + if (sort_order == SortOrder::TOTAL_ORDER) { + return true; + } // parquet-cpp version 1.3.0 and parquet-mr 1.10.0 onwards stats are computed // correctly for all types if ((application_ == "parquet-cpp" && VersionLt(PARQUET_CPP_FIXED_STATS_VERSION())) || @@ -2101,16 +2063,22 @@ class FileMetaDataBuilder::FileMetaDataBuilderImpl { metadata_->__set_version(file_version); metadata_->__set_created_by(properties_->created_by()); - // Users cannot set the `ColumnOrder` since we do not have user defined sort order - // in the spec yet. - // We always default to `TYPE_DEFINED_ORDER`. We can expose it in - // the API once we have user defined sort orders in the Parquet format. - // TypeDefinedOrder implies choose SortOrder based on ConvertedType/PhysicalType - format::TypeDefinedOrder type_defined_order; - format::ColumnOrder column_order; - column_order.__set_TYPE_ORDER(type_defined_order); - column_order.__isset.TYPE_ORDER = true; - metadata_->column_orders.resize(schema_->num_columns(), column_order); + metadata_->column_orders.reserve(schema_->num_columns()); + for (int column_index = 0; column_index < schema_->num_columns(); ++column_index) { + format::ColumnOrder column_order; + switch (schema_->Column(column_index)->column_order().get_order()) { + case ColumnOrder::TYPE_DEFINED_ORDER: + column_order.__set_TYPE_ORDER(format::TypeDefinedOrder{}); + break; + case ColumnOrder::IEEE_754_TOTAL_ORDER: + column_order.__set_IEEE_754_TOTAL_ORDER(format::IEEE754TotalOrder{}); + break; + case ColumnOrder::UNDEFINED: + case ColumnOrder::UNKNOWN: + throw ParquetException("Invalid writer column order"); + } + metadata_->column_orders.push_back(std::move(column_order)); + } metadata_->__isset.column_orders = true; // if plaintext footer, set footer signing algorithm @@ -2138,6 +2106,7 @@ class FileMetaDataBuilder::FileMetaDataBuilderImpl { auto file_meta_data = std::unique_ptr(new FileMetaData()); file_meta_data->impl_->metadata_ = std::move(metadata_); file_meta_data->impl_->InitSchema(); + file_meta_data->impl_->InitColumnOrders(); file_meta_data->impl_->InitKeyValueMetadata(); return file_meta_data; } diff --git a/cpp/src/parquet/page_index.cc b/cpp/src/parquet/page_index.cc index 1d2faebd251c..f04ed836b608 100644 --- a/cpp/src/parquet/page_index.cc +++ b/cpp/src/parquet/page_index.cc @@ -31,6 +31,7 @@ #include "parquet/metadata.h" #include "parquet/page_index.h" #include "parquet/schema.h" +#include "parquet/schema_internal.h" #include "parquet/statistics.h" #include "parquet/thrift_internal.h" @@ -101,7 +102,9 @@ class TypedColumnIndexImpl : public TypedColumnIndex { column_index_.min_values.size() != num_pages || column_index_.max_values.size() != num_pages || (column_index_.__isset.null_counts && - column_index_.null_counts.size() != num_pages)) { + column_index_.null_counts.size() != num_pages) || + (column_index_.__isset.nan_counts && + column_index_.nan_counts.size() != num_pages)) { throw ParquetException("Invalid column index"); } @@ -153,6 +156,12 @@ class TypedColumnIndexImpl : public TypedColumnIndex { return column_index_.null_counts; } + bool has_nan_counts() const override { return column_index_.__isset.nan_counts; } + + const std::vector& nan_counts() const override { + return column_index_.nan_counts; + } + const std::vector& non_null_page_indices() const override { return non_null_page_indices_; } @@ -496,6 +505,7 @@ class ColumnIndexBuilderImpl final : public ColumnIndexBuilder { /// Initialize the null_counts vector as set. Invalid null_counts vector from /// any page will invalidate the null_counts vector of the column index. column_index_.__isset.null_counts = true; + column_index_.__isset.nan_counts = schema::IsFloatingPoint(*descr_); column_index_.boundary_order = format::BoundaryOrder::UNORDERED; } @@ -534,6 +544,13 @@ class ColumnIndexBuilderImpl final : public ColumnIndexBuilder { column_index_.null_counts.clear(); } + if (column_index_.__isset.nan_counts && stats.has_nan_count) { + column_index_.nan_counts.emplace_back(stats.nan_count); + } else { + column_index_.__isset.nan_counts = false; + column_index_.nan_counts.clear(); + } + if (size_stats.is_set()) { const auto& page_def_level_hist = size_stats.definition_level_histogram; const auto& page_ref_level_hist = size_stats.repetition_level_histogram; @@ -568,6 +585,9 @@ class ColumnIndexBuilderImpl final : public ColumnIndexBuilder { if (!column_index_.__isset.null_counts) { column_index_.null_counts.clear(); } + if (!column_index_.__isset.nan_counts) { + column_index_.nan_counts.clear(); + } /// Decode min/max values according to the data type. const size_t non_null_page_count = non_null_page_indices_.size(); diff --git a/cpp/src/parquet/page_index.h b/cpp/src/parquet/page_index.h index 67e682885328..f368b9be9b71 100644 --- a/cpp/src/parquet/page_index.h +++ b/cpp/src/parquet/page_index.h @@ -73,6 +73,15 @@ class PARQUET_EXPORT ColumnIndex { /// available. virtual const std::vector& null_counts() const = 0; + /// \brief Whether per-page NaN count information is available. + virtual bool has_nan_counts() const = 0; + + /// \brief An optional vector with the number of NaN values in each data page. + /// + /// `has_nan_counts` should be called first to determine if this information is + /// available. + virtual const std::vector& nan_counts() const = 0; + /// \brief A vector of page indices for non-null pages. virtual const std::vector& non_null_page_indices() const = 0; diff --git a/cpp/src/parquet/page_index_test.cc b/cpp/src/parquet/page_index_test.cc index 3a7308c1c6bc..21a486e14401 100644 --- a/cpp/src/parquet/page_index_test.cc +++ b/cpp/src/parquet/page_index_test.cc @@ -18,10 +18,13 @@ #include "parquet/page_index.h" #include +#include +#include #include #include "arrow/io/file.h" #include "arrow/util/float16.h" +#include "arrow/util/ubsan.h" #include "parquet/file_reader.h" #include "parquet/metadata.h" #include "parquet/schema.h" @@ -507,6 +510,9 @@ void TestWriteTypedColumnIndex(schema::NodePtr node, int16_t max_repetition_level = 0, const std::vector& page_levels = {}) { const bool build_size_stats = !page_levels.empty(); + const bool has_nan_counts = + std::all_of(page_stats.begin(), page_stats.end(), + [](const EncodedStatistics& stats) { return stats.has_nan_count; }); if (build_size_stats) { ASSERT_EQ(page_levels.size(), page_stats.size()); } @@ -536,6 +542,7 @@ void TestWriteTypedColumnIndex(schema::NodePtr node, for (const auto& column_index : column_indexes) { ASSERT_EQ(boundary_order, column_index->boundary_order()); ASSERT_EQ(has_null_counts, column_index->has_null_counts()); + ASSERT_EQ(has_nan_counts, column_index->has_nan_counts()); const size_t num_pages = column_index->null_pages().size(); if (build_size_stats) { ASSERT_EQ(num_pages * (max_repetition_level + 1), @@ -551,6 +558,9 @@ void TestWriteTypedColumnIndex(schema::NodePtr node, if (has_null_counts) { ASSERT_EQ(page_stats[i].null_count, column_index->null_counts()[i]); } + if (has_nan_counts) { + ASSERT_EQ(page_stats[i].nan_count, column_index->nan_counts()[i]); + } if (build_size_stats) { ASSERT_NO_FATAL_FAILURE(VerifyPageLevelHistogram( i, page_levels[i].def_levels, column_index->definition_level_histograms())); @@ -669,6 +679,34 @@ TEST(PageIndex, WriteFloat16ColumnIndex) { /*has_null_counts=*/false); } +TEST(PageIndex, WriteFloatTotalOrder) { + auto encode = [](uint32_t bits) { + const auto value = ::arrow::util::SafeCopy(bits); + return std::string(reinterpret_cast(&value), sizeof(value)); + }; + + std::vector page_stats(2); + // IEEE 754 encodings: + // page 0: [-1.0f, +1.0f], with no NaNs. + // page 1: [-qNaN(payload=1), +qNaN(payload=1)], all NaNs. + page_stats[0].set_min(encode(0xbf800000)).set_max(encode(0x3f800000)); + page_stats[0].set_nan_count(0); + page_stats[1].set_min(encode(0xffc00001)).set_max(encode(0x7fc00001)); + page_stats[1].set_nan_count(2); + + auto node = schema::Float("c1"); + std::static_pointer_cast(node)->SetColumnOrder( + ColumnOrder::ieee_754_total_order_); + TestWriteTypedColumnIndex(node, page_stats, BoundaryOrder::Unordered, + /*has_null_counts=*/false); + + page_stats[1].has_nan_count = false; + std::static_pointer_cast(node)->SetColumnOrder( + ColumnOrder::ieee_754_total_order_); + TestWriteTypedColumnIndex(std::move(node), page_stats, BoundaryOrder::Unordered, + /*has_null_counts=*/false); +} + TEST(PageIndex, WriteColumnIndexWithAllNullPages) { // All values are null. std::vector page_stats(3); diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h index e2244a1176e3..70af593a9e54 100644 --- a/cpp/src/parquet/properties.h +++ b/cpp/src/parquet/properties.h @@ -375,6 +375,7 @@ class PARQUET_EXPORT WriterProperties { store_decimal_as_integer_(false), page_checksum_enabled_(false), size_statistics_level_(DEFAULT_SIZE_STATISTICS_LEVEL), + floating_point_column_order_(ColumnOrder::IEEE_754_TOTAL_ORDER), content_defined_chunking_enabled_(false), content_defined_chunking_options_({}) {} @@ -391,6 +392,7 @@ class PARQUET_EXPORT WriterProperties { store_decimal_as_integer_(properties.store_decimal_as_integer()), page_checksum_enabled_(properties.page_checksum_enabled()), size_statistics_level_(properties.size_statistics_level()), + floating_point_column_order_(properties.floating_point_column_order()), sorting_columns_(properties.sorting_columns()), default_column_properties_(properties.default_column_properties()), content_defined_chunking_enabled_( @@ -864,6 +866,20 @@ class PARQUET_EXPORT WriterProperties { return this; } + /// \brief Set the column order for all floating-point columns. + /// + /// The supported values are IEEE_754_TOTAL_ORDER and TYPE_DEFINED_ORDER. + /// The default is IEEE_754_TOTAL_ORDER. + Builder* floating_point_column_order(ColumnOrder::type order) { + if (order != ColumnOrder::IEEE_754_TOTAL_ORDER && + order != ColumnOrder::TYPE_DEFINED_ORDER) { + throw ParquetException("Unsupported floating-point column order: ", + static_cast(order)); + } + floating_point_column_order_ = order; + return this; + } + /// \brief Build the WriterProperties with the builder parameters. /// \return The WriterProperties defined by the builder. std::shared_ptr build() { @@ -901,8 +917,8 @@ class PARQUET_EXPORT WriterProperties { pool_, dictionary_pagesize_limit_, write_batch_size_, max_row_group_length_, pagesize_, max_rows_per_page_, version_, created_by_, page_checksum_enabled_, size_statistics_level_, std::move(file_encryption_properties_), - default_column_properties_, column_properties, data_page_version_, - store_decimal_as_integer_, std::move(sorting_columns_), + default_column_properties_, column_properties, floating_point_column_order_, + data_page_version_, store_decimal_as_integer_, std::move(sorting_columns_), content_defined_chunking_enabled_, content_defined_chunking_options_)); } @@ -921,6 +937,7 @@ class PARQUET_EXPORT WriterProperties { bool store_decimal_as_integer_; bool page_checksum_enabled_; SizeStatisticsLevel size_statistics_level_; + ColumnOrder::type floating_point_column_order_; std::shared_ptr file_encryption_properties_; @@ -976,6 +993,10 @@ class PARQUET_EXPORT WriterProperties { return size_statistics_level_; } + inline ColumnOrder::type floating_point_column_order() const { + return floating_point_column_order_; + } + inline Encoding::type dictionary_index_encoding() const { if (parquet_version_ == ParquetVersion::PARQUET_1_0) { return Encoding::PLAIN_DICTIONARY; @@ -1084,6 +1105,7 @@ class PARQUET_EXPORT WriterProperties { std::shared_ptr file_encryption_properties, const ColumnProperties& default_column_properties, const std::unordered_map& column_properties, + ColumnOrder::type floating_point_column_order, ParquetDataPageVersion data_page_version, bool store_short_decimal_as_integer, std::vector sorting_columns, bool content_defined_chunking_enabled, CdcOptions content_defined_chunking_options) @@ -1099,6 +1121,7 @@ class PARQUET_EXPORT WriterProperties { store_decimal_as_integer_(store_short_decimal_as_integer), page_checksum_enabled_(page_write_checksum_enabled), size_statistics_level_(size_statistics_level), + floating_point_column_order_(floating_point_column_order), file_encryption_properties_(file_encryption_properties), sorting_columns_(std::move(sorting_columns)), default_column_properties_(default_column_properties), @@ -1118,6 +1141,7 @@ class PARQUET_EXPORT WriterProperties { bool store_decimal_as_integer_; bool page_checksum_enabled_; SizeStatisticsLevel size_statistics_level_; + ColumnOrder::type floating_point_column_order_; std::shared_ptr file_encryption_properties_; diff --git a/cpp/src/parquet/schema.cc b/cpp/src/parquet/schema.cc index 0cfa49c21c16..e3ec6fd083f8 100644 --- a/cpp/src/parquet/schema.cc +++ b/cpp/src/parquet/schema.cc @@ -620,6 +620,12 @@ void ToParquet(const GroupNode* schema, std::vector* out) schema->VisitConst(&visitor); } +bool IsFloatingPoint(const ColumnDescriptor& descr) { + return descr.physical_type() == Type::FLOAT || descr.physical_type() == Type::DOUBLE || + (descr.physical_type() == Type::FIXED_LEN_BYTE_ARRAY && descr.logical_type() && + descr.logical_type()->type() == LogicalType::Type::FLOAT16); +} + // ---------------------------------------------------------------------- // Schema printing @@ -947,6 +953,21 @@ std::string ColumnDescriptor::ToString() const { return ss.str(); } +SortOrder::type ColumnDescriptor::sort_order() const { + const auto& la = logical_type(); + const auto pt = physical_type(); + switch (column_order().get_order()) { + case ColumnOrder::IEEE_754_TOTAL_ORDER: + return schema::IsFloatingPoint(*this) ? SortOrder::TOTAL_ORDER : SortOrder::UNKNOWN; + case ColumnOrder::UNKNOWN: + return SortOrder::UNKNOWN; + case ColumnOrder::TYPE_DEFINED_ORDER: + case ColumnOrder::UNDEFINED: + return la ? GetSortOrder(la, pt) : GetSortOrder(converted_type(), pt); + } + return SortOrder::UNKNOWN; +} + int ColumnDescriptor::type_scale() const { return primitive_node_->decimal_metadata().scale; } diff --git a/cpp/src/parquet/schema.h b/cpp/src/parquet/schema.h index 65732603ea1d..e9208e206f19 100644 --- a/cpp/src/parquet/schema.h +++ b/cpp/src/parquet/schema.h @@ -376,17 +376,15 @@ class PARQUET_EXPORT ColumnDescriptor { ColumnOrder column_order() const { return primitive_node_->column_order(); } - SortOrder::type sort_order() const { - const auto& la = logical_type(); - auto pt = physical_type(); - return la ? GetSortOrder(la, pt) : GetSortOrder(converted_type(), pt); - } + SortOrder::type sort_order() const; // Whether ColumnOrder-governed min/max values have a supported ordering. bool can_use_min_max() const { switch (column_order().get_order()) { case ColumnOrder::TYPE_DEFINED_ORDER: return sort_order() != SortOrder::UNKNOWN; + case ColumnOrder::IEEE_754_TOTAL_ORDER: + return sort_order() == SortOrder::TOTAL_ORDER; case ColumnOrder::UNDEFINED: // If there is no defined column order, the obsolete min and max fields // in the Statistics object are to be used, and they are always sorted diff --git a/cpp/src/parquet/schema_internal.h b/cpp/src/parquet/schema_internal.h index c0cfffc87e2b..4d4ec30b2d1a 100644 --- a/cpp/src/parquet/schema_internal.h +++ b/cpp/src/parquet/schema_internal.h @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -// Non-public Thrift schema serialization utilities +// Non-public schema utilities #pragma once @@ -50,5 +50,8 @@ std::unique_ptr Unflatten(const format::SchemaElement* elements, int lengt PARQUET_EXPORT void ToParquet(const GroupNode* schema, std::vector* out); +PARQUET_EXPORT +bool IsFloatingPoint(const ColumnDescriptor& descr); + } // namespace schema } // namespace parquet diff --git a/cpp/src/parquet/schema_test.cc b/cpp/src/parquet/schema_test.cc index 859f14a34d91..55085b657bc6 100644 --- a/cpp/src/parquet/schema_test.cc +++ b/cpp/src/parquet/schema_test.cc @@ -684,6 +684,18 @@ TEST(TestColumnDescriptor, CanUseStats) { node = PrimitiveNode::Make("name", Repetition::REQUIRED, Type::INT96); // INT96 has no defined sort order in the Parquet type-defined ordering. EXPECT_FALSE(ColumnDescriptor(node, 0, 0).can_use_min_max()); + + node = Float("name"); + primitive_node = std::static_pointer_cast(node); + primitive_node->SetColumnOrder(ColumnOrder::ieee_754_total_order_); + EXPECT_EQ(SortOrder::TOTAL_ORDER, ColumnDescriptor(node, 0, 0).sort_order()); + EXPECT_TRUE(ColumnDescriptor(node, 0, 0).can_use_min_max()); + + node = Int32("name"); + primitive_node = std::static_pointer_cast(node); + primitive_node->SetColumnOrder(ColumnOrder::ieee_754_total_order_); + EXPECT_EQ(SortOrder::UNKNOWN, ColumnDescriptor(node, 0, 0).sort_order()); + EXPECT_FALSE(ColumnDescriptor(node, 0, 0).can_use_min_max()); } class TestSchemaDescriptor : public ::testing::Test { diff --git a/cpp/src/parquet/statistics.cc b/cpp/src/parquet/statistics.cc index d43998ef78f2..288193363116 100644 --- a/cpp/src/parquet/statistics.cc +++ b/cpp/src/parquet/statistics.cc @@ -19,6 +19,8 @@ #include #include +#include +#include #include #include #include @@ -349,6 +351,45 @@ struct CompareHelper { } }; +float ToArrowFloat(float value) { return value; } + +double ToArrowFloat(double value) { return value; } + +Float16 ToArrowFloat(const FLBA& value) { + DCHECK_NE(value.ptr, nullptr); + return Float16::FromLittleEndian(value.ptr); +} + +template +std::strong_ordering TotalOrderCompareBits(T lhs, T rhs) { + // https://parquet.apache.org/blog/2026/05/29/taming-floating-point-statistics-in-apache-parquet-ieee-754-total-order-and-nan-counts/ + auto lhs_bits = SafeCopy(lhs); + auto rhs_bits = SafeCopy(rhs); + using UInt = std::make_unsigned_t; + constexpr int sign_shift = sizeof(Int) * 8 - 1; + lhs_bits ^= static_cast(static_cast(lhs_bits >> sign_shift) >> 1); + rhs_bits ^= static_cast(static_cast(rhs_bits >> sign_shift) >> 1); + return lhs_bits <=> rhs_bits; +} + +std::strong_ordering TotalOrderCompare(float lhs, float rhs) { + static_assert(std::numeric_limits::is_iec559); + // TODO: Use std::strong_order once all supported standard libraries implement its + // floating-point overloads. + return TotalOrderCompareBits(lhs, rhs); +} + +std::strong_ordering TotalOrderCompare(double lhs, double rhs) { + static_assert(std::numeric_limits::is_iec559); + // TODO: Use std::strong_order once all supported standard libraries implement its + // floating-point overloads. + return TotalOrderCompareBits(lhs, rhs); +} + +std::strong_ordering TotalOrderCompare(Float16 lhs, Float16 rhs) { + return TotalOrderCompareBits(lhs.bits(), rhs.bits()); +} + using ::std::optional; // A usable min/max pair always satisfies min <= max. The reverse ordering @@ -523,6 +564,66 @@ class TypedComparatorImpl int type_length_; }; +template +class TotalOrderComparatorImpl + : public TypedComparator::DType> { + public: + using T = typename RebindLogical::c_type; + + bool Compare(const T& lhs, const T& rhs) const override { + return std::is_lt(TotalOrderCompare(ToArrowFloat(lhs), ToArrowFloat(rhs))); + } + + std::pair GetMinMax(const T* values, int64_t length) const override { + DCHECK_GT(length, 0); + T min = SafeLoad(values); + T max = min; + for (int64_t value_index = 1; value_index < length; ++value_index) { + const T value = SafeLoad(values + value_index); + min = std::is_lt(TotalOrderCompare(ToArrowFloat(value), ToArrowFloat(min))) ? value + : min; + max = std::is_lt(TotalOrderCompare(ToArrowFloat(max), ToArrowFloat(value))) ? value + : max; + } + return {min, max}; + } + + std::pair GetMinMaxSpaced(const T* values, int64_t length, + const uint8_t* valid_bits, + int64_t valid_bits_offset) const override { + DCHECK_GT(length, 0); + T min{}; + T max{}; + bool has_value = false; + ::arrow::internal::VisitSetBitRunsVoid( + valid_bits, valid_bits_offset, length, [&](int64_t position, int64_t run_length) { + int64_t value_index = 0; + if (!has_value) { + const T value = SafeLoad(values + position); + min = value; + max = value; + has_value = true; + value_index = 1; + } + for (; value_index < run_length; ++value_index) { + const T value = SafeLoad(values + position + value_index); + min = std::is_lt(TotalOrderCompare(ToArrowFloat(value), ToArrowFloat(min))) + ? value + : min; + max = std::is_lt(TotalOrderCompare(ToArrowFloat(max), ToArrowFloat(value))) + ? value + : max; + } + }); + DCHECK(has_value); + return {min, max}; + } + + std::pair GetMinMax(const ::arrow::Array& values) const override { + ParquetException::NYI(values.type()->ToString()); + } +}; + // ARROW-11675: A hand-written version of GetMinMax(), to work around // what looks like a MSVC code generation bug. // This does not seem to be required for GetMinMaxSpaced(). @@ -599,6 +700,75 @@ LogicalType::Type::type LogicalTypeId(const Statistics& stats) { return LogicalTypeId(stats.descr()); } +template +concept ArrowFloatValue = + std::same_as || std::same_as || std::same_as; + +template +bool IsNaNValue(T value) { + return std::isnan(value); +} + +template <> +bool IsNaNValue(Float16 value) { + return value.is_nan(); +} + +template + requires(sort_order == SortOrder::SIGNED || sort_order == SortOrder::TOTAL_ORDER) +class FloatingValueSummary { + public: + void Add(const T& value) { + if (IsNaNValue(value)) { + ++nan_count_; + if constexpr (sort_order == SortOrder::TOTAL_ORDER) { + if (is_all_nan_) { + if (bounds_.has_value()) { + UpdateBounds(value); + } else { + bounds_.emplace(value, value); + } + } + } + } else { + if (is_all_nan_) { + bounds_.emplace(value, value); + is_all_nan_ = false; + } else { + UpdateBounds(value); + } + } + } + + int64_t nan_count() const { return nan_count_; } + + const std::optional>& bounds() const { return bounds_; } + + private: + static bool Less(const T& lhs, const T& rhs) { + if constexpr (sort_order == SortOrder::TOTAL_ORDER) { + return std::is_lt(TotalOrderCompare(lhs, rhs)); + } else { + return lhs < rhs; + } + } + + void UpdateBounds(const T& value) { + DCHECK(bounds_.has_value()); + auto& min = bounds_->first; + auto& max = bounds_->second; + if (Less(value, min)) { + min = value; + } else if (Less(max, value)) { + max = value; + } + } + + int64_t nan_count_ = 0; + bool is_all_nan_ = true; + std::optional> bounds_; +}; + template class TypedStatisticsImpl : public TypedStatistics { public: @@ -610,14 +780,16 @@ class TypedStatisticsImpl : public TypedStatistics { pool_(pool), min_buffer_(AllocateBuffer(pool_, 0)), max_buffer_(AllocateBuffer(pool_, 0)), - logical_type_(LogicalTypeId(descr_)) { + logical_type_(LogicalTypeId(descr_)), + is_half_float_(logical_type_ == LogicalType::Type::FLOAT16) { if (descr->sort_order() != SortOrder::UNKNOWN) { comparator_ = MakeComparator(descr); } TypedStatisticsImpl::Reset(); } - // Create stats from provided values. + // Only used by the deprecated MakeStatistics overload. Remove it after that + // overload has been removed. TypedStatisticsImpl(const T& min, const T& max, int64_t num_values, int64_t null_count, int64_t distinct_count) : pool_(default_memory_pool()), @@ -637,19 +809,20 @@ class TypedStatisticsImpl : public TypedStatistics { // Create stats from a thrift Statistics object. TypedStatisticsImpl(const ColumnDescriptor* descr, const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, - int64_t null_count, int64_t distinct_count, bool has_min_max, - bool has_null_count, bool has_distinct_count, MemoryPool* pool) + int64_t null_count, int64_t distinct_count, int64_t nan_count, + bool has_min_max, bool has_null_count, bool has_distinct_count, + bool has_nan_count, MemoryPool* pool) : TypedStatisticsImpl(descr, encoded_min, encoded_max, num_values, null_count, - distinct_count, has_min_max, has_null_count, - has_distinct_count, + distinct_count, nan_count, has_min_max, has_null_count, + has_distinct_count, has_nan_count, /*is_min_value_exact=*/std::nullopt, /*is_max_value_exact=*/std::nullopt, pool) {} TypedStatisticsImpl(const ColumnDescriptor* descr, const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, - int64_t null_count, int64_t distinct_count, bool has_min_max, - bool has_null_count, bool has_distinct_count, - std::optional is_min_value_exact, + int64_t null_count, int64_t distinct_count, int64_t nan_count, + bool has_min_max, bool has_null_count, bool has_distinct_count, + bool has_nan_count, std::optional is_min_value_exact, std::optional is_max_value_exact, MemoryPool* pool) : TypedStatisticsImpl(descr, pool) { TypedStatisticsImpl::IncrementNumValues(num_values); @@ -658,6 +831,12 @@ class TypedStatisticsImpl : public TypedStatistics { } else { has_null_count_ = false; } + if (has_nan_count) { + statistics_.nan_count = nan_count; + has_nan_count_ = true; + } else { + has_nan_count_ = false; + } if (has_distinct_count) { SetDistinctCount(distinct_count); } else { @@ -665,8 +844,17 @@ class TypedStatisticsImpl : public TypedStatistics { } if (has_min_max) { - PlainDecode(encoded_min, &min_); - PlainDecode(encoded_max, &max_); + if constexpr (std::same_as || std::same_as) { + T decoded_min; + T decoded_max; + PlainDecode(encoded_min, &decoded_min); + PlainDecode(encoded_max, &decoded_max); + Copy(decoded_min, &min_, min_buffer_.get()); + Copy(decoded_max, &max_, max_buffer_.get()); + } else { + PlainDecode(encoded_min, &min_); + PlainDecode(encoded_max, &max_); + } statistics_.is_min_value_exact = is_min_value_exact; statistics_.is_max_value_exact = is_max_value_exact; } @@ -677,6 +865,7 @@ class TypedStatisticsImpl : public TypedStatistics { bool HasDistinctCount() const override { return has_distinct_count_; }; bool HasMinMax() const override { return has_min_max_; } bool HasNullCount() const override { return has_null_count_; }; + bool HasNanCount() const override { return has_nan_count_; } void IncrementNullCount(int64_t n) override { statistics_.null_count += n; @@ -713,8 +902,12 @@ class TypedStatisticsImpl : public TypedStatistics { if (!MinMaxEqual(other)) return false; } - return null_count() == other.null_count() && - distinct_count() == other.distinct_count() && + return HasNullCount() == other.HasNullCount() && + (!HasNullCount() || null_count() == other.null_count()) && + HasDistinctCount() == other.HasDistinctCount() && + (!HasDistinctCount() || distinct_count() == other.distinct_count()) && + HasNanCount() == other.HasNanCount() && + (!HasNanCount() || nan_count() == other.nan_count()) && num_values() == other.num_values() && is_min_value_exact() == other.is_min_value_exact() && is_max_value_exact() == other.is_max_value_exact(); @@ -749,6 +942,11 @@ class TypedStatisticsImpl : public TypedStatistics { // Otherwise clear has_distinct_count_ as distinct count cannot be merged. this->has_distinct_count_ = false; } + if (has_nan_count_ && other.HasNanCount()) { + statistics_.nan_count += other.nan_count(); + } else { + has_nan_count_ = false; + } // Do not clear min/max here if the other side does not provide // min/max which may happen when other is an empty stats or all // its values are null and/or NaN. @@ -773,6 +971,42 @@ class TypedStatisticsImpl : public TypedStatistics { } if (comparator_ == nullptr) return; + + if constexpr (std::same_as || std::same_as || + std::same_as) { + auto visit_valid_indices = [&](auto&& visit) { + ::arrow::internal::VisitSetBitRunsVoid( + values.null_bitmap_data(), values.offset(), values.length(), + [&](int64_t position, int64_t run_length) { + for (int64_t value_index = 0; value_index < run_length; ++value_index) { + visit(position + value_index); + } + }); + }; + if constexpr (std::same_as || std::same_as) { + using ArrayType = typename ::arrow::CTypeTraits::ArrayType; + const auto& array = checked_cast(values); + UpdateFloatingBounds( + [&](auto&& visit) { + visit_valid_indices( + [&](int64_t value_index) { visit(array.Value(value_index)); }); + }, + update_counts); + return; + } else if (is_half_float_) { + DCHECK_EQ(values.type_id(), ::arrow::Type::HALF_FLOAT); + const auto& array = checked_cast(values); + UpdateFloatingBounds( + [&](auto&& visit) { + visit_valid_indices([&](int64_t value_index) { + visit(Float16::FromBits(array.Value(value_index))); + }); + }, + update_counts); + return; + } + } + SetMinMaxPair(comparator_->GetMinMax(values)); } @@ -812,11 +1046,15 @@ class TypedStatisticsImpl : public TypedStatistics { if (HasDistinctCount()) { s.set_distinct_count(this->distinct_count()); } + if (HasNanCount()) { + s.set_nan_count(this->nan_count()); + } return s; } int64_t null_count() const override { return statistics_.null_count; } int64_t distinct_count() const override { return statistics_.distinct_count; } + int64_t nan_count() const override { return statistics_.nan_count; } int64_t num_values() const override { return num_values_; } std::optional is_min_value_exact() const override { return statistics_.is_min_value_exact; @@ -830,6 +1068,7 @@ class TypedStatisticsImpl : public TypedStatistics { bool has_min_max_ = false; bool has_null_count_ = false; bool has_distinct_count_ = false; + bool has_nan_count_ = false; T min_; T max_; ::arrow::MemoryPool* pool_; @@ -843,6 +1082,7 @@ class TypedStatisticsImpl : public TypedStatistics { std::shared_ptr> comparator_; std::shared_ptr min_buffer_, max_buffer_; LogicalType::Type::type logical_type_ = LogicalType::Type::NONE; + bool is_half_float_ = false; void PlainEncode(const T& src, std::string* dst) const; void PlainDecode(const std::string& src, T* dst) const; @@ -858,6 +1098,7 @@ class TypedStatisticsImpl : public TypedStatistics { void ResetCounts() { this->statistics_.null_count = 0; this->statistics_.distinct_count = 0; + this->statistics_.nan_count = 0; this->num_values_ = 0; } @@ -870,18 +1111,84 @@ class TypedStatisticsImpl : public TypedStatistics { this->has_distinct_count_ = false; // Null count calculation is cheap and enabled by default. this->has_null_count_ = true; + // NaN counts are collected alongside floating-point bounds and enabled by + // default. + if constexpr (std::same_as || std::same_as) { + this->has_nan_count_ = true; + } else if constexpr (std::same_as) { + this->has_nan_count_ = is_half_float_; + } else { + this->has_nan_count_ = false; + } + } + + template + void UpdateFloatingBoundsWithOrder(VisitValues&& visit_values, bool update_nan_count) { + using ArrowFloat = decltype(ToArrowFloat(std::declval())); + + FloatingValueSummary summary; + std::invoke(std::forward(visit_values), + [&](const auto& value) { summary.Add(value); }); + if (has_nan_count_ && update_nan_count) { + statistics_.nan_count += summary.nan_count(); + } + const auto& bounds = summary.bounds(); + if (bounds.has_value()) { + if constexpr (std::same_as) { + DCHECK(is_half_float_); + const auto min = bounds->first.ToLittleEndian(); + const auto max = bounds->second.ToLittleEndian(); + SetMinMaxPair({FLBA{min.data()}, FLBA{max.data()}}); + } else { + SetMinMaxPair(bounds.value()); + } + } + } + + template + void UpdateFloatingBounds(VisitValues&& visit_values, bool update_nan_count) { + if (descr_->sort_order() == SortOrder::TOTAL_ORDER) { + UpdateFloatingBoundsWithOrder( + std::forward(visit_values), update_nan_count); + } else { + DCHECK_EQ(descr_->sort_order(), SortOrder::SIGNED); + UpdateFloatingBoundsWithOrder( + std::forward(visit_values), update_nan_count); + } } void SetMinMaxPair(std::pair 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_); + auto maybe_min_max = descr_->sort_order() == SortOrder::TOTAL_ORDER + ? std::optional>(min_max) + : CleanStatistic(min_max, logical_type_); if (!maybe_min_max) return; auto min = maybe_min_max.value().first; auto max = maybe_min_max.value().second; - if (!has_min_max_) { + bool replace_all_nan_bounds = false; + if constexpr (std::same_as || std::same_as || + std::same_as) { + if (descr_->sort_order() == SortOrder::TOTAL_ORDER) { + DCHECK((!std::same_as) || is_half_float_); + + const bool min_is_nan = IsNaNValue(ToArrowFloat(min)); + DCHECK_EQ(min_is_nan, IsNaNValue(ToArrowFloat(max))); + bool current_bounds_are_nan = false; + if (has_min_max_) { + const bool current_min_is_nan = IsNaNValue(ToArrowFloat(min_)); + DCHECK_EQ(current_min_is_nan, IsNaNValue(ToArrowFloat(max_))); + current_bounds_are_nan = current_min_is_nan; + } + if (min_is_nan && has_min_max_ && !current_bounds_are_nan) { + return; + } + replace_all_nan_bounds = !min_is_nan && has_min_max_ && current_bounds_are_nan; + } + } + + if (!has_min_max_ || replace_all_nan_bounds) { has_min_max_ = true; Copy(min, &min_, min_buffer_.get()); Copy(max, &max_, max_buffer_.get()); @@ -905,6 +1212,13 @@ inline bool TypedStatisticsImpl::MinMaxEqual( template bool TypedStatisticsImpl::MinMaxEqual( const TypedStatisticsImpl& other) const { + if constexpr (std::same_as || std::same_as) { + if (descr_->sort_order() == SortOrder::TOTAL_ORDER && + other.descr_->sort_order() == SortOrder::TOTAL_ORDER) { + return std::is_eq(TotalOrderCompare(min_, other.min_)) && + std::is_eq(TotalOrderCompare(max_, other.max_)); + } + } return min_ == other.min_ && max_ == other.max_; } @@ -937,6 +1251,20 @@ void TypedStatisticsImpl::Update(const T* values, int64_t num_values, IncrementNumValues(num_values); if (num_values == 0 || comparator_ == nullptr) return; + if constexpr (std::same_as || std::same_as || + std::same_as) { + const bool use_floating_bounds = !std::same_as || is_half_float_; + if (use_floating_bounds) { + UpdateFloatingBounds( + [&](auto&& visit) { + for (int64_t value_index = 0; value_index < num_values; ++value_index) { + visit(ToArrowFloat(SafeLoad(values + value_index))); + } + }, + true); + return; + } + } SetMinMaxPair(comparator_->GetMinMax(values, num_values)); } @@ -952,6 +1280,24 @@ void TypedStatisticsImpl::UpdateSpaced(const T* values, const uint8_t* va IncrementNumValues(num_values); if (num_values == 0 || comparator_ == nullptr) return; + if constexpr (std::same_as || std::same_as || + std::same_as) { + const bool use_floating_bounds = !std::same_as || is_half_float_; + if (use_floating_bounds) { + UpdateFloatingBounds( + [&](auto&& visit) { + ::arrow::internal::VisitSetBitRunsVoid( + valid_bits, valid_bits_offset, num_spaced_values, + [&](int64_t position, int64_t run_length) { + for (int64_t value_index = 0; value_index < run_length; ++value_index) { + visit(ToArrowFloat(SafeLoad(values + position + value_index))); + } + }); + }, + true); + return; + } + } SetMinMaxPair(comparator_->GetMinMaxSpaced(values, num_spaced_values, valid_bits, valid_bits_offset)); } @@ -993,50 +1339,66 @@ std::shared_ptr DoMakeComparator(Type::type physical_type, LogicalType::Type::type logical_type, SortOrder::type sort_order, int type_length) { - if (SortOrder::SIGNED == sort_order) { - switch (physical_type) { - case Type::BOOLEAN: - return std::make_shared>(); - case Type::INT32: - return std::make_shared>(); - case Type::INT64: - return std::make_shared>(); - case Type::INT96: - return std::make_shared>(); - case Type::FLOAT: - return std::make_shared>(); - case Type::DOUBLE: - return std::make_shared>(); - case Type::BYTE_ARRAY: - return std::make_shared>(); - case Type::FIXED_LEN_BYTE_ARRAY: - if (logical_type == LogicalType::Type::FLOAT16) { - return std::make_shared>( - type_length); - } - return std::make_shared>(type_length); - default: - ParquetException::NYI("Signed Compare not implemented"); - } - } else if (SortOrder::UNSIGNED == sort_order) { - switch (physical_type) { - case Type::INT32: - return std::make_shared>(); - case Type::INT64: - return std::make_shared>(); - case Type::INT96: - return std::make_shared>(); - case Type::BYTE_ARRAY: - return std::make_shared>(); - case Type::FIXED_LEN_BYTE_ARRAY: - return std::make_shared>(type_length); - default: - ParquetException::NYI("Unsigned Compare not implemented"); - } - } else { - throw ParquetException("UNKNOWN Sort Order"); + switch (sort_order) { + case SortOrder::SIGNED: + switch (physical_type) { + case Type::BOOLEAN: + return std::make_shared>(); + case Type::INT32: + return std::make_shared>(); + case Type::INT64: + return std::make_shared>(); + case Type::INT96: + return std::make_shared>(); + case Type::FLOAT: + return std::make_shared>(); + case Type::DOUBLE: + return std::make_shared>(); + case Type::BYTE_ARRAY: + return std::make_shared>(); + case Type::FIXED_LEN_BYTE_ARRAY: + if (logical_type == LogicalType::Type::FLOAT16) { + return std::make_shared>( + type_length); + } + return std::make_shared>(type_length); + default: + ParquetException::NYI("Signed Compare not implemented"); + } + case SortOrder::UNSIGNED: + switch (physical_type) { + case Type::INT32: + return std::make_shared>(); + case Type::INT64: + return std::make_shared>(); + case Type::INT96: + return std::make_shared>(); + case Type::BYTE_ARRAY: + return std::make_shared>(); + case Type::FIXED_LEN_BYTE_ARRAY: + return std::make_shared>(type_length); + default: + ParquetException::NYI("Unsigned Compare not implemented"); + } + case SortOrder::TOTAL_ORDER: + switch (physical_type) { + case Type::FLOAT: + return std::make_shared>(); + case Type::DOUBLE: + return std::make_shared>(); + case Type::FIXED_LEN_BYTE_ARRAY: + if (logical_type == LogicalType::Type::FLOAT16) { + return std::make_shared>(); + } + break; + default: + break; + } + throw ParquetException( + "Total order comparison is only supported for floating-point types"); + default: + throw ParquetException("UNKNOWN Sort Order"); } - return nullptr; } } // namespace @@ -1111,20 +1473,20 @@ std::shared_ptr Statistics::Make(const ColumnDescriptor* descr, DCHECK(encoded_stats != nullptr); return Make(descr, encoded_stats->min(), encoded_stats->max(), num_values, encoded_stats->null_count, encoded_stats->distinct_count, - encoded_stats->has_min && encoded_stats->has_max, + encoded_stats->nan_count, encoded_stats->has_min && encoded_stats->has_max, encoded_stats->has_null_count, encoded_stats->has_distinct_count, - encoded_stats->is_min_value_exact, encoded_stats->is_max_value_exact, pool); + encoded_stats->has_nan_count, encoded_stats->is_min_value_exact, + encoded_stats->is_max_value_exact, pool); } -std::shared_ptr Statistics::Make(const ColumnDescriptor* descr, - const std::string& encoded_min, - const std::string& encoded_max, - int64_t num_values, int64_t null_count, - int64_t distinct_count, bool has_min_max, - bool has_null_count, bool has_distinct_count, - ::arrow::MemoryPool* pool) { +std::shared_ptr Statistics::Make( + const ColumnDescriptor* descr, const std::string& encoded_min, + const std::string& encoded_max, int64_t num_values, int64_t null_count, + int64_t distinct_count, int64_t nan_count, bool has_min_max, bool has_null_count, + bool has_distinct_count, bool has_nan_count, ::arrow::MemoryPool* pool) { return Statistics::Make(descr, encoded_min, encoded_max, num_values, null_count, - distinct_count, has_min_max, has_null_count, has_distinct_count, + distinct_count, nan_count, has_min_max, has_null_count, + has_distinct_count, has_nan_count, /*is_min_value_exact=*/std::nullopt, /*is_max_value_exact=*/std::nullopt, pool); } @@ -1132,17 +1494,17 @@ std::shared_ptr Statistics::Make(const ColumnDescriptor* descr, std::shared_ptr Statistics::Make( const ColumnDescriptor* descr, const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, int64_t null_count, - int64_t distinct_count, bool has_min_max, bool has_null_count, - bool has_distinct_count, std::optional is_min_value_exact, + int64_t distinct_count, int64_t nan_count, bool has_min_max, bool has_null_count, + bool has_distinct_count, bool has_nan_count, std::optional is_min_value_exact, std::optional is_max_value_exact, ::arrow::MemoryPool* pool) { - return VisitType(descr->physical_type(), - [&](auto* type) -> std::shared_ptr { - using DType = std::decay_t; - return std::make_shared>( - descr, encoded_min, encoded_max, num_values, null_count, - distinct_count, has_min_max, has_null_count, has_distinct_count, - is_min_value_exact, is_max_value_exact, pool); - }); + return VisitType( + descr->physical_type(), [&](auto* type) -> std::shared_ptr { + using DType = std::decay_t; + return std::make_shared>( + descr, encoded_min, encoded_max, num_values, null_count, distinct_count, + nan_count, has_min_max, has_null_count, has_distinct_count, has_nan_count, + is_min_value_exact, is_max_value_exact, pool); + }); } } // namespace parquet diff --git a/cpp/src/parquet/statistics.h b/cpp/src/parquet/statistics.h index 0d55409f746e..d0d7cb5f8f84 100644 --- a/cpp/src/parquet/statistics.h +++ b/cpp/src/parquet/statistics.h @@ -134,11 +134,13 @@ class PARQUET_EXPORT EncodedStatistics { int64_t null_count = 0; int64_t distinct_count = 0; + int64_t nan_count = 0; bool has_min = false; bool has_max = false; bool has_null_count = false; bool has_distinct_count = false; + bool has_nan_count = false; // When all values in the statistics are null, it is set to true. // Otherwise, at least one value is not null, or we are not sure at all. @@ -173,7 +175,7 @@ class PARQUET_EXPORT EncodedStatistics { } bool is_set() const { - return has_min || has_max || has_null_count || has_distinct_count; + return has_min || has_max || has_null_count || has_distinct_count || has_nan_count; } bool is_signed() const { return is_signed_; } @@ -203,6 +205,12 @@ class PARQUET_EXPORT EncodedStatistics { has_distinct_count = true; return *this; } + + EncodedStatistics& set_nan_count(int64_t value) { + nan_count = value; + has_nan_count = true; + return *this; + } }; /// \brief Base type for computing column statistics while writing a file @@ -226,15 +234,17 @@ class PARQUET_EXPORT Statistics { /// \param[in] num_values total number of values /// \param[in] null_count number of null values /// \param[in] distinct_count number of distinct values + /// \param[in] nan_count number of NaN values /// \param[in] has_min_max whether the min/max statistics are set /// \param[in] has_null_count whether the null_count statistics are set /// \param[in] has_distinct_count whether the distinct_count statistics are set + /// \param[in] has_nan_count whether the nan_count statistics are set /// \param[in] pool a memory pool to use for any memory allocations, optional static std::shared_ptr Make( const ColumnDescriptor* descr, const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, int64_t null_count, - int64_t distinct_count, bool has_min_max, bool has_null_count, - bool has_distinct_count, + int64_t distinct_count, int64_t nan_count, bool has_min_max, bool has_null_count, + bool has_distinct_count, bool has_nan_count, ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()); /// \brief Create a new statistics instance given a column schema @@ -245,17 +255,19 @@ class PARQUET_EXPORT Statistics { /// \param[in] num_values total number of values /// \param[in] null_count number of null values /// \param[in] distinct_count number of distinct values + /// \param[in] nan_count number of NaN values /// \param[in] has_min_max whether the min/max statistics are set /// \param[in] has_null_count whether the null_count statistics are set /// \param[in] has_distinct_count whether the distinct_count statistics are set + /// \param[in] has_nan_count whether the nan_count statistics are set /// \param[in] is_min_value_exact whether the min value is exact /// \param[in] is_max_value_exact whether the max value is exact /// \param[in] pool a memory pool to use for any memory allocations, optional static std::shared_ptr Make( const ColumnDescriptor* descr, const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, int64_t null_count, - int64_t distinct_count, bool has_min_max, bool has_null_count, - bool has_distinct_count, std::optional is_min_value_exact, + int64_t distinct_count, int64_t nan_count, bool has_min_max, bool has_null_count, + bool has_distinct_count, bool has_nan_count, std::optional is_min_value_exact, std::optional is_max_value_exact, ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()); @@ -279,6 +291,12 @@ class PARQUET_EXPORT Statistics { /// \brief The number of distinct values, may not be set virtual int64_t distinct_count() const = 0; + /// \brief Return true if the count of NaN values is set + virtual bool HasNanCount() const = 0; + + /// \brief The number of NaN values, may not be set + virtual int64_t nan_count() const = 0; + /// \brief The number of non-null values in the column virtual int64_t num_values() const = 0; @@ -316,6 +334,10 @@ class PARQUET_EXPORT Statistics { virtual bool Equals(const Statistics& other) const = 0; protected: + // Only used by the deprecated MakeStatistics overload. Remove it after that + // overload has been removed. + PARQUET_DEPRECATED( + "Deprecated in 26.0.0. Use a ColumnDescriptor-based overload instead.") static std::shared_ptr Make(Type::type physical_type, const void* min, const void* max, int64_t num_values, int64_t null_count, int64_t distinct_count); @@ -409,8 +431,10 @@ std::shared_ptr> MakeStatistics(const typename DType::c_t int64_t num_values, int64_t null_count, int64_t distinct_count) { + ARROW_SUPPRESS_DEPRECATION_WARNING return std::static_pointer_cast>(Statistics::Make( DType::type_num, &min, &max, num_values, null_count, distinct_count)); + ARROW_UNSUPPRESS_DEPRECATION_WARNING } /// \brief Typed version of Statistics::Make @@ -418,11 +442,12 @@ template std::shared_ptr> MakeStatistics( const ColumnDescriptor* descr, const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, int64_t null_count, - int64_t distinct_count, bool has_min_max, bool has_null_count, - bool has_distinct_count, ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()) { + int64_t distinct_count, int64_t nan_count, bool has_min_max, bool has_null_count, + bool has_distinct_count, bool has_nan_count, + ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()) { return std::static_pointer_cast>(Statistics::Make( - descr, encoded_min, encoded_max, num_values, null_count, distinct_count, - has_min_max, has_null_count, has_distinct_count, + descr, encoded_min, encoded_max, num_values, null_count, distinct_count, nan_count, + has_min_max, has_null_count, has_distinct_count, has_nan_count, /*is_min_value_exact=*/std::nullopt, /*is_max_value_exact=*/std::nullopt, pool)); } @@ -431,14 +456,14 @@ template std::shared_ptr> MakeStatistics( const ColumnDescriptor* descr, const std::string& encoded_min, const std::string& encoded_max, int64_t num_values, int64_t null_count, - int64_t distinct_count, bool has_min_max, bool has_null_count, - bool has_distinct_count, std::optional is_min_value_exact, + int64_t distinct_count, int64_t nan_count, bool has_min_max, bool has_null_count, + bool has_distinct_count, bool has_nan_count, std::optional is_min_value_exact, std::optional is_max_value_exact, ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()) { - return std::static_pointer_cast>( - Statistics::Make(descr, encoded_min, encoded_max, num_values, null_count, - distinct_count, has_min_max, has_null_count, has_distinct_count, - is_min_value_exact, is_max_value_exact, pool)); + return std::static_pointer_cast>(Statistics::Make( + descr, encoded_min, encoded_max, num_values, null_count, distinct_count, nan_count, + has_min_max, has_null_count, has_distinct_count, has_nan_count, is_min_value_exact, + is_max_value_exact, pool)); } } // namespace parquet diff --git a/cpp/src/parquet/statistics_test.cc b/cpp/src/parquet/statistics_test.cc index bf0961c4fc6a..7017d96cef95 100644 --- a/cpp/src/parquet/statistics_test.cc +++ b/cpp/src/parquet/statistics_test.cc @@ -27,6 +27,7 @@ #include #include "arrow/array.h" +#include "arrow/array/builder_primitive.h" #include "arrow/buffer.h" #include "arrow/memory_pool.h" #include "arrow/testing/builder.h" @@ -322,9 +323,10 @@ class TestStatistics : public PrimitiveTypedTest { auto statistics2 = MakeStatistics( this->schema_.Column(0), encoded_min, encoded_max, this->values_.size(), - /*null_count=*/0, /*distinct_count=*/0, + /*null_count=*/0, /*distinct_count=*/0, /*nan_count=*/0, /*has_min_max=*/true, /*has_null_count=*/true, /*has_distinct_count=*/true, - /*is_min_value_exact=*/true, /*is_max_value_exact=*/true); + /*has_nan_count=*/false, /*is_min_value_exact=*/true, + /*is_max_value_exact=*/true); auto statistics3 = MakeStatistics(this->schema_.Column(0)); std::vector valid_bits( @@ -337,8 +339,9 @@ class TestStatistics : public PrimitiveTypedTest { // Use old API without is_{min/max}_value_exact auto statistics4 = MakeStatistics( this->schema_.Column(0), encoded_min, encoded_max, this->values_.size(), - /*null_count=*/0, /*distinct_count=*/0, - /*has_min_max=*/true, /*has_null_count=*/true, /*has_distinct_count=*/true); + /*null_count=*/0, /*distinct_count=*/0, /*nan_count=*/0, + /*has_min_max=*/true, /*has_null_count=*/true, /*has_distinct_count=*/true, + /*has_nan_count=*/false); ASSERT_EQ(encoded_min, statistics2->EncodeMin()); ASSERT_EQ(encoded_max, statistics2->EncodeMax()); ASSERT_EQ(statistics1->min(), statistics2->min()); @@ -572,8 +575,9 @@ void TestStatistics::TestMinMaxEncode() { auto statistics2 = MakeStatistics( this->schema_.Column(0), encoded_min, encoded_max, this->values_.size(), /*null_count=*/0, - /*distinct_count=*/0, /*has_min_max=*/true, /*has_null_count=*/true, - /*has_distinct_count=*/true, /*is_min_value_exact=*/true, + /*distinct_count=*/0, /*nan_count=*/0, /*has_min_max=*/true, + /*has_null_count=*/true, /*has_distinct_count=*/true, /*has_nan_count=*/false, + /*is_min_value_exact=*/true, /*is_max_value_exact=*/true); ASSERT_EQ(encoded_min, statistics2->EncodeMin()); @@ -1467,21 +1471,43 @@ class TestFloatStatistics : public ::testing::Test { auto some_nan_stats = MakeStatistics(descr); // Ingesting only nans should not yield valid min max AssertUnsetMinMax(some_nan_stats, all_nans); + ASSERT_TRUE(some_nan_stats->HasNanCount()); + ASSERT_EQ(static_cast(all_nans.size()), some_nan_stats->nan_count()); // Ingesting a mix of NaNs and non-NaNs should yield a valid min max. AssertMinMaxAre(some_nan_stats, some_nans, min, max); + ASSERT_EQ(static_cast(all_nans.size() + 3), some_nan_stats->nan_count()); // Ingesting only nans after a valid min/max, should have no effect AssertMinMaxAre(some_nan_stats, all_nans, min, max); + ASSERT_EQ(static_cast(all_nans.size() * 2 + 3), some_nan_stats->nan_count()); some_nan_stats = MakeStatistics(descr); AssertUnsetMinMax(some_nan_stats, all_nans, &valid_bitmap); + ASSERT_EQ(7, some_nan_stats->nan_count()); // NaNs should not pollute min max when excluded via null bitmap. AssertMinMaxAre(some_nan_stats, some_nans, &valid_bitmap_no_nans, min, max); // Ingesting NaNs with a null bitmap should not change the result. AssertMinMaxAre(some_nan_stats, some_nans, &valid_bitmap, min, max); + ASSERT_EQ(9, some_nan_stats->nan_count()); // An array that doesn't start with NaN auto other_stats = MakeStatistics(descr); AssertMinMaxAre(other_stats, other_nans, min, max); + + auto missing_nan_count = MakeStatistics( + descr, /*encoded_min=*/{}, /*encoded_max=*/{}, /*num_values=*/0, + /*null_count=*/0, /*distinct_count=*/0, /*nan_count=*/0, + /*has_min_max=*/false, /*has_null_count=*/true, + /*has_distinct_count=*/false, /*has_nan_count=*/false); + AssertMinMaxAre(missing_nan_count, some_nans, min, max); + ASSERT_FALSE(missing_nan_count->HasNanCount()); + + auto missing_spaced_nan_count = MakeStatistics( + descr, /*encoded_min=*/{}, /*encoded_max=*/{}, /*num_values=*/0, + /*null_count=*/0, /*distinct_count=*/0, /*nan_count=*/0, + /*has_min_max=*/false, /*has_null_count=*/true, + /*has_distinct_count=*/false, /*has_nan_count=*/false); + AssertMinMaxAre(missing_spaced_nan_count, some_nans, &valid_bitmap_no_nans, min, max); + ASSERT_FALSE(missing_spaced_nan_count->HasNanCount()); } void TestNaNs(); @@ -1670,6 +1696,82 @@ TYPED_TEST(TestFloatStatistics, NegativeZeros) { this->TestNegativeZeroes(); } TYPED_TEST(TestFloatStatistics, NaNs) { this->TestNaNs(); } TYPED_TEST(TestFloatStatistics, Infinities) { this->TestInfinities(); } +template +void TestNativeTotalOrder(UInt negative_nan_bits, UInt positive_nan_bits) { + using T = typename DType::c_type; + auto node = schema::PrimitiveNode::Make("f", Repetition::REQUIRED, DType::type_num); + std::static_pointer_cast(node)->SetColumnOrder( + ColumnOrder::ieee_754_total_order_); + ColumnDescriptor descr(node, 0, 0); + + const T negative_nan = SafeCopy(negative_nan_bits); + const T positive_nan = SafeCopy(positive_nan_bits); + const T negative_zero = -T{0}; + const T positive_zero = T{0}; + std::array mixed{negative_nan, positive_zero, negative_zero, positive_nan}; + auto stats = MakeStatistics(&descr); + stats->Update(mixed.data(), mixed.size(), 0); + ASSERT_TRUE(stats->HasNanCount()); + ASSERT_EQ(2, stats->nan_count()); + ASSERT_TRUE(stats->HasMinMax()); + ASSERT_TRUE(std::signbit(stats->min())); + ASSERT_FALSE(std::signbit(stats->max())); + + std::array all_nan{positive_nan, negative_nan}; + stats->Reset(); + stats->Update(all_nan.data(), all_nan.size(), 0); + ASSERT_EQ(2, stats->nan_count()); + ASSERT_EQ(negative_nan_bits, SafeCopy(stats->min())); + ASSERT_EQ(positive_nan_bits, SafeCopy(stats->max())); + + auto same = MakeStatistics(&descr); + same->Update(all_nan.data(), all_nan.size(), 0); + ASSERT_TRUE(stats->Equals(*same)); + + auto numeric = MakeStatistics(&descr); + std::array values{T{1}}; + numeric->Update(values.data(), values.size(), 0); + stats->Merge(*numeric); + ASSERT_EQ(2, stats->nan_count()); + ASSERT_EQ(T{1}, stats->min()); + ASSERT_EQ(T{1}, stats->max()); +} + +TEST(TestFloatStatistics, TotalOrder) { + // -qNaN(payload=1), +qNaN(payload=1). + TestNativeTotalOrder(uint32_t{0xffc00001}, uint32_t{0x7fc00001}); + // -qNaN(payload=1), +qNaN(payload=1). + TestNativeTotalOrder(uint64_t{0xfff8000000000001}, + uint64_t{0x7ff8000000000001}); +} + +TEST(TestFloatStatistics, TotalOrderFloat16) { + // -qNaN(payload=1), +qNaN(payload=1). + BufferedFloat16 negative_nan(Float16::FromBits(0xfe01)); + BufferedFloat16 positive_nan(Float16::FromBits(0x7e01)); + BufferedFloat16 negative_zero(-Float16::zero()); + BufferedFloat16 positive_zero(Float16::zero()); + std::array values{FLBA{negative_nan.bytes()}, FLBA{positive_zero.bytes()}, + FLBA{negative_zero.bytes()}, FLBA{positive_nan.bytes()}}; + auto node = schema::PrimitiveNode::Make( + "f", Repetition::REQUIRED, LogicalType::Float16(), Type::FIXED_LEN_BYTE_ARRAY, 2); + std::static_pointer_cast(node)->SetColumnOrder( + ColumnOrder::ieee_754_total_order_); + ColumnDescriptor descr(node, 0, 0); + auto stats = MakeStatistics(&descr); + stats->Update(values.data(), values.size(), 0); + ASSERT_EQ(2, stats->nan_count()); + // 0x8000 and 0x0000 are Float16 -0 and +0. + ASSERT_EQ(Float16::FromBits(0x8000), Float16::FromLittleEndian(stats->min().ptr)); + ASSERT_EQ(Float16::FromBits(0x0000), Float16::FromLittleEndian(stats->max().ptr)); + + std::array all_nan{FLBA{positive_nan.bytes()}, FLBA{negative_nan.bytes()}}; + stats->Reset(); + stats->Update(all_nan.data(), all_nan.size(), 0); + ASSERT_EQ(0xfe01, Float16::FromLittleEndian(stats->min().ptr).bits()); + ASSERT_EQ(0x7e01, Float16::FromLittleEndian(stats->max().ptr).bits()); +} + // ARROW-7376 TEST(TestStatisticsSortOrderFloatNaN, NaNAndNullsInfiniteLoop) { constexpr int kNumValues = 8; diff --git a/cpp/src/parquet/thrift_internal.h b/cpp/src/parquet/thrift_internal.h index 971e6ccebc9d..d3f9978d0904 100644 --- a/cpp/src/parquet/thrift_internal.h +++ b/cpp/src/parquet/thrift_internal.h @@ -268,6 +268,7 @@ static inline StatisticsMinMaxField GetStatisticsMinMaxField( const ColumnDescriptor& descr) { switch (descr.column_order().get_order()) { case ColumnOrder::TYPE_DEFINED_ORDER: + case ColumnOrder::IEEE_754_TOTAL_ORDER: return descr.sort_order() != SortOrder::UNKNOWN ? StatisticsMinMaxField::kMinValueMaxValue : StatisticsMinMaxField::kInvalid; @@ -312,6 +313,9 @@ static inline EncodedStatistics FromThrift(const format::Statistics& stats, if (stats.__isset.distinct_count) { out.set_distinct_count(stats.distinct_count); } + if (stats.__isset.nan_count) { + out.set_nan_count(stats.nan_count); + } return out; } @@ -523,6 +527,9 @@ static inline format::Statistics ToThrift(const EncodedStatistics& stats) { if (stats.has_distinct_count) { statistics.__set_distinct_count(stats.distinct_count); } + if (stats.has_nan_count) { + statistics.__set_nan_count(stats.nan_count); + } return statistics; } diff --git a/cpp/src/parquet/types.cc b/cpp/src/parquet/types.cc index 9d7604faec30..97555f19eb1e 100644 --- a/cpp/src/parquet/types.cc +++ b/cpp/src/parquet/types.cc @@ -455,6 +455,8 @@ SortOrder::type GetSortOrder(const std::shared_ptr& logical_t ColumnOrder ColumnOrder::undefined_ = ColumnOrder(ColumnOrder::UNDEFINED); ColumnOrder ColumnOrder::type_defined_ = ColumnOrder(ColumnOrder::TYPE_DEFINED_ORDER); +ColumnOrder ColumnOrder::ieee_754_total_order_ = + ColumnOrder(ColumnOrder::IEEE_754_TOTAL_ORDER); ColumnOrder ColumnOrder::unknown_ = ColumnOrder(ColumnOrder::UNKNOWN); // Static methods for LogicalType class diff --git a/cpp/src/parquet/types.h b/cpp/src/parquet/types.h index 687353aa9bcb..790273623f8e 100644 --- a/cpp/src/parquet/types.h +++ b/cpp/src/parquet/types.h @@ -125,7 +125,7 @@ struct Repetition { // Parquet file. These stats are discarded for types that need unsigned. // See PARQUET-686. struct SortOrder { - enum type { SIGNED, UNSIGNED, UNKNOWN }; + enum type { SIGNED, UNSIGNED, TOTAL_ORDER, UNKNOWN }; }; namespace schema { @@ -604,6 +604,8 @@ class PARQUET_EXPORT ColumnOrder { UNDEFINED, // File metadata uses TypeDefinedOrder from the Parquet format. TYPE_DEFINED_ORDER, + // File metadata uses IEEE754TotalOrder from the Parquet format. + IEEE_754_TOTAL_ORDER, // Column order value unsupported by this reader. UNKNOWN }; @@ -614,6 +616,7 @@ class PARQUET_EXPORT ColumnOrder { static ColumnOrder undefined_; static ColumnOrder type_defined_; + static ColumnOrder ieee_754_total_order_; static ColumnOrder unknown_; private: diff --git a/cpp/submodules/parquet-testing b/cpp/submodules/parquet-testing index e74785d85a4e..cf1eed4dfb45 160000 --- a/cpp/submodules/parquet-testing +++ b/cpp/submodules/parquet-testing @@ -1 +1 @@ -Subproject commit e74785d85a4ecee829e1e405444d6a1b24b8bc9c +Subproject commit cf1eed4dfb45794be0f35c42d1d323fd617d763f diff --git a/docs/source/python/parquet.rst b/docs/source/python/parquet.rst index 4fc1368fb270..5c2c1bfb6c8b 100644 --- a/docs/source/python/parquet.rst +++ b/docs/source/python/parquet.rst @@ -302,7 +302,7 @@ such as the row groups and column chunk metadata and statistics: num_columns: 4 num_rows: 3 - total_byte_size: 290 + total_byte_size: 272 sorting_columns: () >>> metadata.row_group(0).column(0) From c9aac54b4e73eb85c886777d3a58f62c74f46ccc Mon Sep 17 00:00:00 2001 From: Zehua Zou Date: Wed, 5 Aug 2026 17:45:43 +0800 Subject: [PATCH 2/2] address review --- cpp/src/parquet/column_writer.cc | 13 ++- cpp/src/parquet/metadata.cc | 6 +- cpp/src/parquet/schema.cc | 18 +--- cpp/src/parquet/schema.h | 8 +- cpp/src/parquet/schema_test.cc | 7 -- cpp/src/parquet/statistics.cc | 159 ++++++++++++++++-------------- cpp/src/parquet/statistics.h | 3 +- cpp/src/parquet/thrift_internal.h | 3 +- cpp/src/parquet/types.h | 2 +- docs/source/python/parquet.rst | 4 +- 10 files changed, 110 insertions(+), 113 deletions(-) diff --git a/cpp/src/parquet/column_writer.cc b/cpp/src/parquet/column_writer.cc index 653f28f64bde..a34487878c4f 100644 --- a/cpp/src/parquet/column_writer.cc +++ b/cpp/src/parquet/column_writer.cc @@ -83,6 +83,11 @@ namespace parquet { namespace { +bool CanWriteLegacyStatistics(const ColumnDescriptor& descr) { + return descr.column_order().get_order() == ColumnOrder::TYPE_DEFINED_ORDER && + descr.sort_order() == SortOrder::SIGNED; +} + // Visitor that extracts the value buffer from a FlatArray at a given offset. struct ValueBufferSlicer { template @@ -1013,7 +1018,7 @@ void ColumnWriterImpl::BuildDataPageV1(int64_t definition_levels_rle_size, uncompressed_data_->mutable_data()); auto [page_stats, page_size_stats] = GetPageStatistics(); page_stats.ApplyStatSizeLimits(properties_->max_statistics_size(descr_->path())); - page_stats.set_is_signed(SortOrder::SIGNED == descr_->sort_order()); + page_stats.set_is_signed(CanWriteLegacyStatistics(*descr_)); ResetPageStatistics(); std::shared_ptr compressed_data; @@ -1074,7 +1079,7 @@ void ColumnWriterImpl::BuildDataPageV2(int64_t definition_levels_rle_size, auto [page_stats, page_size_stats] = GetPageStatistics(); page_stats.ApplyStatSizeLimits(properties_->max_statistics_size(descr_->path())); - page_stats.set_is_signed(SortOrder::SIGNED == descr_->sort_order()); + page_stats.set_is_signed(CanWriteLegacyStatistics(*descr_)); ResetPageStatistics(); int32_t num_values = static_cast(num_buffered_values_); @@ -1120,7 +1125,7 @@ int64_t ColumnWriterImpl::Close() { auto [chunk_statistics, chunk_size_statistics] = GetChunkStatistics(); chunk_statistics.ApplyStatSizeLimits( properties_->max_statistics_size(descr_->path())); - chunk_statistics.set_is_signed(SortOrder::SIGNED == descr_->sort_order()); + chunk_statistics.set_is_signed(CanWriteLegacyStatistics(*descr_)); // Write stats only if the column has at least one row written if (rows_written_ > 0 && chunk_statistics.is_set()) { @@ -1307,7 +1312,7 @@ class TypedColumnWriterImpl : public ColumnWriterImpl, // SortOrder::UNKNOWN. Currently, the presence of statistics is tied to // having a known sort order and so null counts will be missing. if (properties->statistics_enabled(descr_->path())) { - if (SortOrder::UNKNOWN != descr_->sort_order()) { + if (descr_->can_use_min_max()) { page_statistics_ = MakeStatistics(descr_, allocator_); chunk_statistics_ = MakeStatistics(descr_, allocator_); } diff --git a/cpp/src/parquet/metadata.cc b/cpp/src/parquet/metadata.cc index 656c2d70abb8..a01f8bd6af01 100644 --- a/cpp/src/parquet/metadata.cc +++ b/cpp/src/parquet/metadata.cc @@ -290,6 +290,9 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl { FromThrift(column_metadata_->statistics, GetStatisticsMinMaxField(*descr_))); } } + if (descr_->column_order().get_order() == ColumnOrder::IEEE_754_TOTAL_ORDER) { + return true; + } return writer_version_->HasCorrectStatistics(type(), *possible_encoded_stats_, descr_->sort_order()); } @@ -1566,9 +1569,6 @@ bool ApplicationVersion::VersionEq(const ApplicationVersion& other_version) cons bool ApplicationVersion::HasCorrectStatistics(Type::type col_type, const EncodedStatistics& statistics, SortOrder::type sort_order) const { - if (sort_order == SortOrder::TOTAL_ORDER) { - return true; - } // parquet-cpp version 1.3.0 and parquet-mr 1.10.0 onwards stats are computed // correctly for all types if ((application_ == "parquet-cpp" && VersionLt(PARQUET_CPP_FIXED_STATS_VERSION())) || diff --git a/cpp/src/parquet/schema.cc b/cpp/src/parquet/schema.cc index e3ec6fd083f8..ec2aaf5e00cb 100644 --- a/cpp/src/parquet/schema.cc +++ b/cpp/src/parquet/schema.cc @@ -622,8 +622,7 @@ void ToParquet(const GroupNode* schema, std::vector* out) bool IsFloatingPoint(const ColumnDescriptor& descr) { return descr.physical_type() == Type::FLOAT || descr.physical_type() == Type::DOUBLE || - (descr.physical_type() == Type::FIXED_LEN_BYTE_ARRAY && descr.logical_type() && - descr.logical_type()->type() == LogicalType::Type::FLOAT16); + descr.logical_type()->type() == LogicalType::Type::FLOAT16; } // ---------------------------------------------------------------------- @@ -953,21 +952,6 @@ std::string ColumnDescriptor::ToString() const { return ss.str(); } -SortOrder::type ColumnDescriptor::sort_order() const { - const auto& la = logical_type(); - const auto pt = physical_type(); - switch (column_order().get_order()) { - case ColumnOrder::IEEE_754_TOTAL_ORDER: - return schema::IsFloatingPoint(*this) ? SortOrder::TOTAL_ORDER : SortOrder::UNKNOWN; - case ColumnOrder::UNKNOWN: - return SortOrder::UNKNOWN; - case ColumnOrder::TYPE_DEFINED_ORDER: - case ColumnOrder::UNDEFINED: - return la ? GetSortOrder(la, pt) : GetSortOrder(converted_type(), pt); - } - return SortOrder::UNKNOWN; -} - int ColumnDescriptor::type_scale() const { return primitive_node_->decimal_metadata().scale; } diff --git a/cpp/src/parquet/schema.h b/cpp/src/parquet/schema.h index e9208e206f19..857bc8a5280f 100644 --- a/cpp/src/parquet/schema.h +++ b/cpp/src/parquet/schema.h @@ -376,7 +376,11 @@ class PARQUET_EXPORT ColumnDescriptor { ColumnOrder column_order() const { return primitive_node_->column_order(); } - SortOrder::type sort_order() const; + SortOrder::type sort_order() const { + const auto& la = logical_type(); + auto pt = physical_type(); + return la ? GetSortOrder(la, pt) : GetSortOrder(converted_type(), pt); + } // Whether ColumnOrder-governed min/max values have a supported ordering. bool can_use_min_max() const { @@ -384,7 +388,7 @@ class PARQUET_EXPORT ColumnDescriptor { case ColumnOrder::TYPE_DEFINED_ORDER: return sort_order() != SortOrder::UNKNOWN; case ColumnOrder::IEEE_754_TOTAL_ORDER: - return sort_order() == SortOrder::TOTAL_ORDER; + return true; case ColumnOrder::UNDEFINED: // If there is no defined column order, the obsolete min and max fields // in the Statistics object are to be used, and they are always sorted diff --git a/cpp/src/parquet/schema_test.cc b/cpp/src/parquet/schema_test.cc index 55085b657bc6..6dea60f4175b 100644 --- a/cpp/src/parquet/schema_test.cc +++ b/cpp/src/parquet/schema_test.cc @@ -688,14 +688,7 @@ TEST(TestColumnDescriptor, CanUseStats) { node = Float("name"); primitive_node = std::static_pointer_cast(node); primitive_node->SetColumnOrder(ColumnOrder::ieee_754_total_order_); - EXPECT_EQ(SortOrder::TOTAL_ORDER, ColumnDescriptor(node, 0, 0).sort_order()); EXPECT_TRUE(ColumnDescriptor(node, 0, 0).can_use_min_max()); - - node = Int32("name"); - primitive_node = std::static_pointer_cast(node); - primitive_node->SetColumnOrder(ColumnOrder::ieee_754_total_order_); - EXPECT_EQ(SortOrder::UNKNOWN, ColumnDescriptor(node, 0, 0).sort_order()); - EXPECT_FALSE(ColumnDescriptor(node, 0, 0).can_use_min_max()); } class TestSchemaDescriptor : public ::testing::Test { diff --git a/cpp/src/parquet/statistics.cc b/cpp/src/parquet/statistics.cc index 288193363116..d2c0e3fdd008 100644 --- a/cpp/src/parquet/statistics.cc +++ b/cpp/src/parquet/statistics.cc @@ -714,14 +714,15 @@ bool IsNaNValue(Float16 value) { return value.is_nan(); } -template - requires(sort_order == SortOrder::SIGNED || sort_order == SortOrder::TOTAL_ORDER) +template + requires(column_order == ColumnOrder::TYPE_DEFINED_ORDER || + column_order == ColumnOrder::IEEE_754_TOTAL_ORDER) class FloatingValueSummary { public: void Add(const T& value) { if (IsNaNValue(value)) { ++nan_count_; - if constexpr (sort_order == SortOrder::TOTAL_ORDER) { + if constexpr (column_order == ColumnOrder::IEEE_754_TOTAL_ORDER) { if (is_all_nan_) { if (bounds_.has_value()) { UpdateBounds(value); @@ -746,7 +747,7 @@ class FloatingValueSummary { private: static bool Less(const T& lhs, const T& rhs) { - if constexpr (sort_order == SortOrder::TOTAL_ORDER) { + if constexpr (column_order == ColumnOrder::IEEE_754_TOTAL_ORDER) { return std::is_lt(TotalOrderCompare(lhs, rhs)); } else { return lhs < rhs; @@ -782,7 +783,7 @@ class TypedStatisticsImpl : public TypedStatistics { max_buffer_(AllocateBuffer(pool_, 0)), logical_type_(LogicalTypeId(descr_)), is_half_float_(logical_type_ == LogicalType::Type::FLOAT16) { - if (descr->sort_order() != SortOrder::UNKNOWN) { + if (descr->can_use_min_max()) { comparator_ = MakeComparator(descr); } TypedStatisticsImpl::Reset(); @@ -1122,11 +1123,11 @@ class TypedStatisticsImpl : public TypedStatistics { } } - template + template void UpdateFloatingBoundsWithOrder(VisitValues&& visit_values, bool update_nan_count) { using ArrowFloat = decltype(ToArrowFloat(std::declval())); - FloatingValueSummary summary; + FloatingValueSummary summary; std::invoke(std::forward(visit_values), [&](const auto& value) { summary.Add(value); }); if (has_nan_count_ && update_nan_count) { @@ -1147,21 +1148,22 @@ class TypedStatisticsImpl : public TypedStatistics { template void UpdateFloatingBounds(VisitValues&& visit_values, bool update_nan_count) { - if (descr_->sort_order() == SortOrder::TOTAL_ORDER) { - UpdateFloatingBoundsWithOrder( + if (descr_->column_order().get_order() == ColumnOrder::IEEE_754_TOTAL_ORDER) { + UpdateFloatingBoundsWithOrder( std::forward(visit_values), update_nan_count); } else { - DCHECK_EQ(descr_->sort_order(), SortOrder::SIGNED); - UpdateFloatingBoundsWithOrder( + DCHECK(descr_->can_use_min_max()); + UpdateFloatingBoundsWithOrder( std::forward(visit_values), update_nan_count); } } void SetMinMaxPair(std::pair min_max) { if (comparator_ == nullptr) return; - auto maybe_min_max = descr_->sort_order() == SortOrder::TOTAL_ORDER - ? std::optional>(min_max) - : CleanStatistic(min_max, logical_type_); + auto maybe_min_max = + descr_->column_order().get_order() == ColumnOrder::IEEE_754_TOTAL_ORDER + ? std::optional>(min_max) + : CleanStatistic(min_max, logical_type_); if (!maybe_min_max) return; auto min = maybe_min_max.value().first; @@ -1170,7 +1172,7 @@ class TypedStatisticsImpl : public TypedStatistics { bool replace_all_nan_bounds = false; if constexpr (std::same_as || std::same_as || std::same_as) { - if (descr_->sort_order() == SortOrder::TOTAL_ORDER) { + if (descr_->column_order().get_order() == ColumnOrder::IEEE_754_TOTAL_ORDER) { DCHECK((!std::same_as) || is_half_float_); const bool min_is_nan = IsNaNValue(ToArrowFloat(min)); @@ -1213,8 +1215,8 @@ template bool TypedStatisticsImpl::MinMaxEqual( const TypedStatisticsImpl& other) const { if constexpr (std::same_as || std::same_as) { - if (descr_->sort_order() == SortOrder::TOTAL_ORDER && - other.descr_->sort_order() == SortOrder::TOTAL_ORDER) { + if (descr_->column_order().get_order() == ColumnOrder::IEEE_754_TOTAL_ORDER && + other.descr_->column_order().get_order() == ColumnOrder::IEEE_754_TOTAL_ORDER) { return std::is_eq(TotalOrderCompare(min_, other.min_)) && std::is_eq(TotalOrderCompare(max_, other.max_)); } @@ -1339,66 +1341,69 @@ std::shared_ptr DoMakeComparator(Type::type physical_type, LogicalType::Type::type logical_type, SortOrder::type sort_order, int type_length) { - switch (sort_order) { - case SortOrder::SIGNED: - switch (physical_type) { - case Type::BOOLEAN: - return std::make_shared>(); - case Type::INT32: - return std::make_shared>(); - case Type::INT64: - return std::make_shared>(); - case Type::INT96: - return std::make_shared>(); - case Type::FLOAT: - return std::make_shared>(); - case Type::DOUBLE: - return std::make_shared>(); - case Type::BYTE_ARRAY: - return std::make_shared>(); - case Type::FIXED_LEN_BYTE_ARRAY: - if (logical_type == LogicalType::Type::FLOAT16) { - return std::make_shared>( - type_length); - } - return std::make_shared>(type_length); - default: - ParquetException::NYI("Signed Compare not implemented"); - } - case SortOrder::UNSIGNED: - switch (physical_type) { - case Type::INT32: - return std::make_shared>(); - case Type::INT64: - return std::make_shared>(); - case Type::INT96: - return std::make_shared>(); - case Type::BYTE_ARRAY: - return std::make_shared>(); - case Type::FIXED_LEN_BYTE_ARRAY: - return std::make_shared>(type_length); - default: - ParquetException::NYI("Unsigned Compare not implemented"); - } - case SortOrder::TOTAL_ORDER: - switch (physical_type) { - case Type::FLOAT: - return std::make_shared>(); - case Type::DOUBLE: - return std::make_shared>(); - case Type::FIXED_LEN_BYTE_ARRAY: - if (logical_type == LogicalType::Type::FLOAT16) { - return std::make_shared>(); - } - break; - default: - break; + if (SortOrder::SIGNED == sort_order) { + switch (physical_type) { + case Type::BOOLEAN: + return std::make_shared>(); + case Type::INT32: + return std::make_shared>(); + case Type::INT64: + return std::make_shared>(); + case Type::INT96: + return std::make_shared>(); + case Type::FLOAT: + return std::make_shared>(); + case Type::DOUBLE: + return std::make_shared>(); + case Type::BYTE_ARRAY: + return std::make_shared>(); + case Type::FIXED_LEN_BYTE_ARRAY: + if (logical_type == LogicalType::Type::FLOAT16) { + return std::make_shared>( + type_length); + } + return std::make_shared>(type_length); + default: + ParquetException::NYI("Signed Compare not implemented"); + } + } else if (SortOrder::UNSIGNED == sort_order) { + switch (physical_type) { + case Type::INT32: + return std::make_shared>(); + case Type::INT64: + return std::make_shared>(); + case Type::INT96: + return std::make_shared>(); + case Type::BYTE_ARRAY: + return std::make_shared>(); + case Type::FIXED_LEN_BYTE_ARRAY: + return std::make_shared>(type_length); + default: + ParquetException::NYI("Unsigned Compare not implemented"); + } + } else { + throw ParquetException("UNKNOWN Sort Order"); + } + return nullptr; +} + +std::shared_ptr DoMakeTotalOrderComparator( + Type::type physical_type, LogicalType::Type::type logical_type) { + switch (physical_type) { + case Type::FLOAT: + return std::make_shared>(); + case Type::DOUBLE: + return std::make_shared>(); + case Type::FIXED_LEN_BYTE_ARRAY: + if (logical_type == LogicalType::Type::FLOAT16) { + return std::make_shared>(); } - throw ParquetException( - "Total order comparison is only supported for floating-point types"); + break; default: - throw ParquetException("UNKNOWN Sort Order"); + break; } + throw ParquetException( + "Total order comparison is only supported for floating-point types"); } } // namespace @@ -1414,6 +1419,12 @@ std::shared_ptr Comparator::Make(Type::type physical_type, } std::shared_ptr Comparator::Make(const ColumnDescriptor* descr) { + if (!descr->can_use_min_max()) { + throw ParquetException("Column order does not define a supported comparison"); + } + if (descr->column_order().get_order() == ColumnOrder::IEEE_754_TOTAL_ORDER) { + return DoMakeTotalOrderComparator(descr->physical_type(), LogicalTypeId(descr)); + } return DoMakeComparator(descr->physical_type(), LogicalTypeId(descr), descr->sort_order(), descr->type_length()); } diff --git a/cpp/src/parquet/statistics.h b/cpp/src/parquet/statistics.h index d0d7cb5f8f84..c63d4441bfc7 100644 --- a/cpp/src/parquet/statistics.h +++ b/cpp/src/parquet/statistics.h @@ -59,8 +59,7 @@ class PARQUET_EXPORT Comparator { SortOrder::type sort_order, int type_length = -1); - /// \brief Create typed comparator inferring default sort order from - /// ColumnDescriptor + /// \brief Create typed comparator using the column order from ColumnDescriptor /// \param[in] descr the Parquet column schema static std::shared_ptr Make(const ColumnDescriptor* descr); }; diff --git a/cpp/src/parquet/thrift_internal.h b/cpp/src/parquet/thrift_internal.h index d3f9978d0904..545a93f1eb5b 100644 --- a/cpp/src/parquet/thrift_internal.h +++ b/cpp/src/parquet/thrift_internal.h @@ -268,10 +268,11 @@ static inline StatisticsMinMaxField GetStatisticsMinMaxField( const ColumnDescriptor& descr) { switch (descr.column_order().get_order()) { case ColumnOrder::TYPE_DEFINED_ORDER: - case ColumnOrder::IEEE_754_TOTAL_ORDER: return descr.sort_order() != SortOrder::UNKNOWN ? StatisticsMinMaxField::kMinValueMaxValue : StatisticsMinMaxField::kInvalid; + case ColumnOrder::IEEE_754_TOTAL_ORDER: + return StatisticsMinMaxField::kMinValueMaxValue; case ColumnOrder::UNDEFINED: return descr.sort_order() == SortOrder::SIGNED ? StatisticsMinMaxField::kLegacyMinMax diff --git a/cpp/src/parquet/types.h b/cpp/src/parquet/types.h index 790273623f8e..0024b82e5f16 100644 --- a/cpp/src/parquet/types.h +++ b/cpp/src/parquet/types.h @@ -125,7 +125,7 @@ struct Repetition { // Parquet file. These stats are discarded for types that need unsigned. // See PARQUET-686. struct SortOrder { - enum type { SIGNED, UNSIGNED, TOTAL_ORDER, UNKNOWN }; + enum type { SIGNED, UNSIGNED, UNKNOWN }; }; namespace schema { diff --git a/docs/source/python/parquet.rst b/docs/source/python/parquet.rst index 5c2c1bfb6c8b..4b6e3e835a05 100644 --- a/docs/source/python/parquet.rst +++ b/docs/source/python/parquet.rst @@ -330,8 +330,8 @@ such as the row groups and column chunk metadata and statistics: has_dictionary_page: True dictionary_page_offset: 4 data_page_offset: 36 - total_compressed_size: 106 - total_uncompressed_size: 102 + total_compressed_size: 88 + total_uncompressed_size: 84 bloom_filter_offset: None bloom_filter_length: None