diff --git a/be/src/format_v2/parquet/parquet_statistics.cpp b/be/src/format_v2/parquet/parquet_statistics.cpp index d0692a80205f3f..c5d561004c9f38 100644 --- a/be/src/format_v2/parquet/parquet_statistics.cpp +++ b/be/src/format_v2/parquet/parquet_statistics.cpp @@ -25,6 +25,7 @@ #include #include +#include #include #include #include @@ -34,6 +35,7 @@ #include #include #include +#include #include #include @@ -118,6 +120,15 @@ bool set_decoded_field(const ParquetColumnSchema& column_schema, DecodedValueKin return read_decoded_field(column_schema, view, field, timezone).ok(); } +template +bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { + if constexpr (std::is_floating_point_v) { + // Parquet requires readers to ignore min/max statistics if either bound is NaN. + return !std::isnan(min_value) && !std::isnan(max_value); + } + return true; +} + template bool set_decoded_min_max(const std::shared_ptr<::parquet::Statistics>& statistics, const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, @@ -125,10 +136,13 @@ bool set_decoded_min_max(const std::shared_ptr<::parquet::Statistics>& statistic const cctz::time_zone* timezone) { auto typed_statistics = std::static_pointer_cast<::parquet::TypedStatistics>(statistics); - if (!set_decoded_field(column_schema, value_kind, typed_statistics->min(), - &column_statistics->min_value, timezone) || - !set_decoded_field(column_schema, value_kind, typed_statistics->max(), - &column_statistics->max_value, timezone)) { + const auto& min_value = typed_statistics->min(); + const auto& max_value = typed_statistics->max(); + if (!valid_min_max(min_value, max_value) || + !set_decoded_field(column_schema, value_kind, min_value, &column_statistics->min_value, + timezone) || + !set_decoded_field(column_schema, value_kind, max_value, &column_statistics->max_value, + timezone)) { return false; } return true; @@ -666,7 +680,11 @@ std::shared_ptr make_zonemap_from_statistics( return std::make_shared(std::move(zone_map)); } if (!statistics.has_min_max) { - return nullptr; + // Null counts remain trustworthy when min/max decoding fails (for example, because a + // floating-point bound is NaN). pass_all prevents range pruning without discarding the + // has_null/has_not_null flags needed by IS NULL and IS NOT NULL predicates. + zone_map.pass_all = true; + return std::make_shared(std::move(zone_map)); } zone_map.min_value = statistics.min_value; zone_map.max_value = statistics.max_value; @@ -693,6 +711,11 @@ void accumulate_zonemap_stats(const ZoneMapEvalContext& ctx, ParquetPruningStats } // namespace +std::shared_ptr ParquetStatisticsUtils::MakeZoneMap( + const ParquetColumnStatistics& statistics) { + return make_zonemap_from_statistics(statistics); +} + ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics( const ParquetColumnSchema& column_schema, const std::shared_ptr<::parquet::Statistics>& statistics, const cctz::time_zone* timezone) { @@ -861,8 +884,8 @@ bool check_statistics(const ::parquet::RowGroupMetaData& row_group, DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns()); auto column_chunk = row_group.ColumnChunk(column_schema->leaf_column_id); if (column_chunk != nullptr) { - zone_map = - make_zonemap_from_statistics(ParquetStatisticsUtils::TransformColumnStatistics( + zone_map = ParquetStatisticsUtils::MakeZoneMap( + ParquetStatisticsUtils::TransformColumnStatistics( *column_schema, column_chunk->statistics(), timezone)); } add_slot_zonemap(&ctx, slot_index, column_schema->type, std::move(zone_map)); @@ -969,10 +992,18 @@ bool set_page_decoded_min_max(const std::shared_ptr<::parquet::ColumnIndex>& col page_idx >= typed_index->max_values().size()) { return false; } - if (!set_decoded_field(column_schema, value_kind, typed_index->min_values()[page_idx], - &page_statistics->min_value, timezone) || - !set_decoded_field(column_schema, value_kind, typed_index->max_values()[page_idx], - &page_statistics->max_value, timezone)) { + const auto& min_value = typed_index->min_values()[page_idx]; + const auto& max_value = typed_index->max_values()[page_idx]; + if (!valid_min_max(min_value, max_value)) { + // A NaN invalidates only this page's bounds, not the ColumnIndex itself. Keep the page + // conservatively by returning usable null-count statistics with has_min_max=false, while + // allowing later pages with finite bounds to remain eligible for pruning. + return true; + } + if (!set_decoded_field(column_schema, value_kind, min_value, &page_statistics->min_value, + timezone) || + !set_decoded_field(column_schema, value_kind, max_value, &page_statistics->max_value, + timezone)) { return false; } page_statistics->has_min_max = true; @@ -1206,15 +1237,15 @@ bool select_ranges_for_expr_zonemap( const auto page_count = offset_index->page_locations().size(); for (size_t page_idx = 0; page_idx < page_count; ++page_idx) { ParquetColumnStatistics page_statistics; - if (!build_page_statistics(column_index, *column_schema, page_idx, &page_statistics, - timezone)) { + if (!ParquetStatisticsUtils::TransformColumnIndexStatistics( + column_index, *column_schema, page_idx, &page_statistics, timezone)) { ranges->clear(); return false; } ZoneMapEvalContext ctx; add_slot_zonemap(&ctx, slot_index, column_schema->type, - make_zonemap_from_statistics(page_statistics)); + ParquetStatisticsUtils::MakeZoneMap(page_statistics)); const auto result = VExprContext::evaluate_zonemap_filter(conjuncts, ctx); page_stats.merge_page_eval_stats(ctx.stats); if (result == ZoneMapFilterResult::kNoMatch) { @@ -1356,6 +1387,13 @@ void build_page_skip_plans(const std::shared_ptr<::parquet::RowGroupPageIndexRea } // namespace +bool ParquetStatisticsUtils::TransformColumnIndexStatistics( + const std::shared_ptr<::parquet::ColumnIndex>& column_index, + const ParquetColumnSchema& column_schema, size_t page_idx, + ParquetColumnStatistics* page_statistics, const cctz::time_zone* timezone) { + return build_page_statistics(column_index, column_schema, page_idx, page_statistics, timezone); +} + Status select_row_group_ranges_by_page_index( ::parquet::ParquetFileReader* file_reader, const std::vector>& file_schema, diff --git a/be/src/format_v2/parquet/parquet_statistics.h b/be/src/format_v2/parquet/parquet_statistics.h index 818d645052ca5d..9e94562bf2d37b 100644 --- a/be/src/format_v2/parquet/parquet_statistics.h +++ b/be/src/format_v2/parquet/parquet_statistics.h @@ -15,6 +15,7 @@ #pragma once +#include #include #include #include @@ -30,6 +31,7 @@ namespace parquet { class BloomFilter; +class ColumnIndex; class FileMetaData; class ParquetFileReader; class Statistics; @@ -41,6 +43,9 @@ class time_zone; namespace doris { class RuntimeState; +namespace segment_v2 { +struct ZoneMap; +} // namespace segment_v2 } // namespace doris namespace doris::format::parquet { @@ -119,11 +124,19 @@ struct ParquetColumnStatistics { // -> bloom filter(evaluate_bloom_filter) // ============================================================================ struct ParquetStatisticsUtils { + static std::shared_ptr MakeZoneMap( + const ParquetColumnStatistics& statistics); + static ParquetColumnStatistics TransformColumnStatistics( const ParquetColumnSchema& column_schema, const std::shared_ptr<::parquet::Statistics>& statistics, const cctz::time_zone* timezone = nullptr); + static bool TransformColumnIndexStatistics( + const std::shared_ptr<::parquet::ColumnIndex>& column_index, + const ParquetColumnSchema& column_schema, size_t page_idx, + ParquetColumnStatistics* page_statistics, const cctz::time_zone* timezone = nullptr); + static bool BloomFilterExcludes(const ParquetColumnSchema& column_schema, int slot_index, const VExprContextSPtrs& conjuncts, const ::parquet::BloomFilter& bloom_filter); diff --git a/be/test/format_v2/parquet/parquet_statistics_test.cpp b/be/test/format_v2/parquet/parquet_statistics_test.cpp index e5cdd8a87457e0..5fe0c4f150db93 100644 --- a/be/test/format_v2/parquet/parquet_statistics_test.cpp +++ b/be/test/format_v2/parquet/parquet_statistics_test.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -32,11 +33,13 @@ #include #include #include +#include #include #include "core/data_type/data_type_number.h" #include "core/data_type/data_type_string.h" #include "core/field.h" +#include "exprs/expr_zonemap_filter.h" #include "exprs/vexpr.h" #include "exprs/vexpr_context.h" #include "exprs/vslot_ref.h" @@ -75,6 +78,27 @@ std::shared_ptr uint32_array(const std::vector& values) return finish_array(&builder); } +std::shared_ptr float_array(const std::vector& values) { + arrow::FloatBuilder builder; + for (const auto value : values) { + EXPECT_TRUE(builder.Append(value).ok()); + } + return finish_array(&builder); +} + +std::shared_ptr double_array(const std::vector& values) { + arrow::DoubleBuilder builder; + for (const auto value : values) { + EXPECT_TRUE(builder.Append(value).ok()); + } + return finish_array(&builder); +} + +template +std::string encoded_value(const NativeType& value) { + return {reinterpret_cast(&value), sizeof(value)}; +} + std::shared_ptr string_array(const std::vector& values) { arrow::StringBuilder builder; for (const auto& value : values) { @@ -128,6 +152,49 @@ std::vector> build_file_sc return file_schema; } +template +class TestColumnIndex final : public ::parquet::TypedColumnIndex { +public: + using NativeType = typename ParquetDType::c_type; + + TestColumnIndex(NativeType min_value, NativeType max_value) + : TestColumnIndex(std::vector {min_value}, + std::vector {max_value}) {} + + TestColumnIndex(std::vector min_values, std::vector max_values) + : _null_pages(min_values.size(), false), + _null_counts(min_values.size(), 0), + _min_values(std::move(min_values)), + _max_values(std::move(max_values)) { + EXPECT_EQ(_min_values.size(), _max_values.size()); + for (size_t page_idx = 0; page_idx < _min_values.size(); ++page_idx) { + _non_null_page_indices.push_back(static_cast(page_idx)); + } + } + + const std::vector& null_pages() const override { return _null_pages; } + const std::vector& encoded_min_values() const override { return _encoded_values; } + const std::vector& encoded_max_values() const override { return _encoded_values; } + ::parquet::BoundaryOrder::type boundary_order() const override { + return ::parquet::BoundaryOrder::Unordered; + } + bool has_null_counts() const override { return true; } + const std::vector& null_counts() const override { return _null_counts; } + const std::vector& non_null_page_indices() const override { + return _non_null_page_indices; + } + const std::vector& min_values() const override { return _min_values; } + const std::vector& max_values() const override { return _max_values; } + +private: + const std::vector _null_pages; + const std::vector _encoded_values; + const std::vector _null_counts; + std::vector _non_null_page_indices; + const std::vector _min_values; + const std::vector _max_values; +}; + class Int32ZoneMapExpr final : public VExpr { public: enum class Op { GE, GT, IS_NULL, IS_NOT_NULL }; @@ -416,6 +483,133 @@ TEST(ParquetStatisticsTransformTest, HandlesMissingStatisticsAndAllNullChunks) { EXPECT_FALSE(all_null_stats.has_min_max); } +TEST(ParquetStatisticsTransformTest, IgnoresNaNFloatAndDoubleMinMax) { + auto table = arrow::Table::Make(arrow::schema({arrow::field("f", arrow::float32(), false), + arrow::field("d", arrow::float64(), false)}), + {float_array({1.0F, 2.0F}), double_array({1.0, 2.0})}); + auto reader = make_reader(table, 2, false, true); + auto schema = build_file_schema(*reader); + + const float float_nan = std::numeric_limits::quiet_NaN(); + const float float_max = 2.0F; + auto float_stats = ::parquet::MakeStatistics<::parquet::FloatType>( + schema[0]->descriptor, encoded_value(float_nan), encoded_value(float_max), 2, 0, 0, + true, true, false); + const auto converted_float = format::parquet::ParquetStatisticsUtils::TransformColumnStatistics( + *schema[0], float_stats); + EXPECT_FALSE(converted_float.has_min_max); + EXPECT_TRUE(converted_float.has_not_null); + + const double double_nan = std::numeric_limits::quiet_NaN(); + const double double_min = 1.0; + auto double_stats = ::parquet::MakeStatistics<::parquet::DoubleType>( + schema[1]->descriptor, encoded_value(double_min), encoded_value(double_nan), 2, 0, 0, + true, true, false); + const auto converted_double = + format::parquet::ParquetStatisticsUtils::TransformColumnStatistics(*schema[1], + double_stats); + EXPECT_FALSE(converted_double.has_min_max); + EXPECT_TRUE(converted_double.has_not_null); + + const double double_max = 2.0; + auto finite_stats = ::parquet::MakeStatistics<::parquet::DoubleType>( + schema[1]->descriptor, encoded_value(double_min), encoded_value(double_max), 2, 0, 0, + true, true, false); + const auto converted_finite = + format::parquet::ParquetStatisticsUtils::TransformColumnStatistics(*schema[1], + finite_stats); + EXPECT_TRUE(converted_finite.has_min_max); +} + +TEST(ParquetStatisticsTransformTest, IgnoresNaNFloatAndDoubleColumnIndexMinMax) { + auto table = arrow::Table::Make(arrow::schema({arrow::field("f", arrow::float32(), false), + arrow::field("d", arrow::float64(), false)}), + {float_array({1.0F, 2.0F}), double_array({1.0, 2.0})}); + auto reader = make_reader(table, 2, false, true); + auto schema = build_file_schema(*reader); + + auto float_index = std::make_shared>( + 1.0F, std::numeric_limits::quiet_NaN()); + format::parquet::ParquetColumnStatistics float_page_stats; + EXPECT_TRUE(format::parquet::ParquetStatisticsUtils::TransformColumnIndexStatistics( + float_index, *schema[0], 0, &float_page_stats)); + EXPECT_FALSE(float_page_stats.has_min_max); + EXPECT_TRUE(float_page_stats.has_not_null); + + auto double_index = std::make_shared>( + std::numeric_limits::quiet_NaN(), 2.0); + format::parquet::ParquetColumnStatistics double_page_stats; + EXPECT_TRUE(format::parquet::ParquetStatisticsUtils::TransformColumnIndexStatistics( + double_index, *schema[1], 0, &double_page_stats)); + EXPECT_FALSE(double_page_stats.has_min_max); + EXPECT_TRUE(double_page_stats.has_not_null); + + auto finite_index = std::make_shared>(1.0, 2.0); + format::parquet::ParquetColumnStatistics finite_page_stats; + EXPECT_TRUE(format::parquet::ParquetStatisticsUtils::TransformColumnIndexStatistics( + finite_index, *schema[1], 0, &finite_page_stats)); + EXPECT_TRUE(finite_page_stats.has_min_max); + + auto mixed_index = std::make_shared>( + std::vector {std::numeric_limits::quiet_NaN(), 1.0}, + std::vector {std::numeric_limits::quiet_NaN(), 2.0}); + format::parquet::ParquetColumnStatistics nan_page_stats; + EXPECT_TRUE(format::parquet::ParquetStatisticsUtils::TransformColumnIndexStatistics( + mixed_index, *schema[1], 0, &nan_page_stats)); + EXPECT_FALSE(nan_page_stats.has_min_max); + + format::parquet::ParquetColumnStatistics following_page_stats; + EXPECT_TRUE(format::parquet::ParquetStatisticsUtils::TransformColumnIndexStatistics( + mixed_index, *schema[1], 1, &following_page_stats)); + EXPECT_TRUE(following_page_stats.has_min_max); + EXPECT_EQ(following_page_stats.min_value.get(), 1.0); + EXPECT_EQ(following_page_stats.max_value.get(), 2.0); +} + +TEST(ParquetStatisticsTransformTest, PreservesNullCountWhenNaNInvalidatesMinMax) { + auto table = arrow::Table::Make(arrow::schema({arrow::field("f", arrow::float64(), false)}), + {double_array({1.0, 2.0})}); + auto reader = make_reader(table, 2, false, true); + auto schema = build_file_schema(*reader); + + const double nan = std::numeric_limits::quiet_NaN(); + const double max_value = 2.0; + auto footer_stats = ::parquet::MakeStatistics<::parquet::DoubleType>( + schema[0]->descriptor, encoded_value(nan), encoded_value(max_value), 2, 0, 0, true, + true, false); + const auto converted_footer = + format::parquet::ParquetStatisticsUtils::TransformColumnStatistics(*schema[0], + footer_stats); + auto footer_zone_map = format::parquet::ParquetStatisticsUtils::MakeZoneMap(converted_footer); + ASSERT_NE(footer_zone_map, nullptr); + EXPECT_TRUE(footer_zone_map->pass_all); + EXPECT_FALSE(footer_zone_map->has_null); + EXPECT_TRUE(footer_zone_map->has_not_null); + EXPECT_FALSE(expr_zonemap::range_stats_usable_for_zonemap(*footer_zone_map, schema[0]->type)); + + ZoneMapEvalContext footer_ctx; + footer_ctx.slots.emplace(0, ZoneMapEvalContext::SlotZoneMap {.data_type = schema[0]->type, + .zone_map = footer_zone_map}); + Int32ZoneMapExpr is_null_expr(0, Int32ZoneMapExpr::Op::IS_NULL); + EXPECT_EQ(is_null_expr.evaluate_zonemap_filter(footer_ctx), ZoneMapFilterResult::kNoMatch); + + auto column_index = std::make_shared>(nan, max_value); + format::parquet::ParquetColumnStatistics page_stats; + ASSERT_TRUE(format::parquet::ParquetStatisticsUtils::TransformColumnIndexStatistics( + column_index, *schema[0], 0, &page_stats)); + auto page_zone_map = format::parquet::ParquetStatisticsUtils::MakeZoneMap(page_stats); + ASSERT_NE(page_zone_map, nullptr); + EXPECT_TRUE(page_zone_map->pass_all); + EXPECT_FALSE(page_zone_map->has_null); + EXPECT_TRUE(page_zone_map->has_not_null); + EXPECT_FALSE(expr_zonemap::range_stats_usable_for_zonemap(*page_zone_map, schema[0]->type)); + + ZoneMapEvalContext page_ctx; + page_ctx.slots.emplace(0, ZoneMapEvalContext::SlotZoneMap {.data_type = schema[0]->type, + .zone_map = page_zone_map}); + EXPECT_EQ(is_null_expr.evaluate_zonemap_filter(page_ctx), ZoneMapFilterResult::kNoMatch); +} + TEST(ParquetStatisticsPruningTest, ExprZonemapPredicatesAndNullPredicatesPruneRowGroups) { auto table = arrow::Table::Make(arrow::schema({arrow::field("i", arrow::int32(), true)}), {int32_array({std::nullopt, std::nullopt, 3, 4, 5, 6})});