[fix](be) Prevent DST timestamp statistics mispruning

be/src/format_v2/orc/orc_reader.cpp

@@ -69,6 +69,7 @@
 #include "exprs/vslot_ref.h"
 #include "format_v2/column_mapper.h"
 #include "format_v2/orc/orc_search_argument.h"
+#include "format_v2/timestamp_statistics.h"
 #include "io/fs/file_reader.h"
 #include "runtime/exec_env.h"
 #include "runtime/runtime_profile.h"
@@ -531,6 +532,11 @@ bool set_timestamp_zone_map(const ::orc::ColumnStatistics& statistics,
                 timestamp_statistics->getMaximum(), timestamp_statistics->getMaximumNanos()));
         return true;
     }
+    if (!format::utc_timestamp_range_is_monotonic(
+                format::floor_epoch_seconds(timestamp_statistics->getMinimum(), 1000),
+                format::floor_epoch_seconds(timestamp_statistics->getMaximum(), 1000), timezone)) {
+        return false;
+    }
     zone_map->min_value = Field::create_field<TYPE_DATETIMEV2>(datetime_v2_from_orc_millis(
             timestamp_statistics->getMinimum(), timestamp_statistics->getMinimumNanos(), timezone));
     zone_map->max_value = Field::create_field<TYPE_DATETIMEV2>(datetime_v2_from_orc_millis(

be/src/format_v2/orc/orc_search_argument.cpp

@@ -45,6 +45,7 @@
 #include "exprs/vliteral.h"
 #include "exprs/vslot_ref.h"
 #include "exprs/vtopn_pred.h"
+#include "format_v2/timestamp_statistics.h"
 
 namespace doris::format::orc {
 namespace {
@@ -597,20 +598,47 @@ std::optional<DateTimeLiteralParts> date_time_literal_parts(const Field& field)
 
 std::optional<::orc::Literal> make_timestamp_literal(const Field& field,
                                                      const cctz::time_zone& timezone) {
+    const auto civil_year_is_monotonic = [&](const cctz::civil_second& civil_seconds) {
+        // Localized file predicates may already have selected one side of a repeated civil time.
+        // Checking the whole civil year catches that lossy representation while retaining SARG
+        // pruning for years and zones without a backward transition.
+        const auto year_start =
+                cctz::convert(cctz::civil_second(civil_seconds.year(), 1, 1), timezone);
+        const auto next_year_start =
+                cctz::convert(cctz::civil_second(civil_seconds.year() + 1, 1, 1), timezone);
+        return format::utc_timestamp_range_is_monotonic(year_start.time_since_epoch().count(),
+                                                        next_year_start.time_since_epoch().count(),
+                                                        timezone);
+    };
     switch (field.get_type()) {
     case TYPE_DATETIME: {
         const auto& datetime = field.get<TYPE_DATETIME>();
         const cctz::civil_second civil_seconds(datetime.year(), datetime.month(), datetime.day(),
                                                datetime.hour(), datetime.minute(),
                                                datetime.second());
-        return ::orc::Literal(cctz::convert(civil_seconds, timezone).time_since_epoch().count(), 0);
+        const auto lookup = timezone.lookup(civil_seconds);
+        // ORC SearchArguments accept one UTC timestamp literal. A repeated or skipped local civil
+        // time has no unique UTC representation, so pushing it down could prune a stripe that
+        // contains another valid interpretation. Keep such predicates for row-level evaluation.
+        if (!civil_year_is_monotonic(civil_seconds) ||
+            lookup.kind != cctz::time_zone::civil_lookup::UNIQUE) {
+            return std::nullopt;
+        }
+        return ::orc::Literal(lookup.pre.time_since_epoch().count(), 0);
     }
     case TYPE_DATETIMEV2: {
         const auto& datetime = field.get<TYPE_DATETIMEV2>();
         const cctz::civil_second civil_seconds(datetime.year(), datetime.month(), datetime.day(),
                                                datetime.hour(), datetime.minute(),
                                                datetime.second());
-        const auto seconds = cctz::convert(civil_seconds, timezone).time_since_epoch().count();
+        const auto lookup = timezone.lookup(civil_seconds);
+        // See the DATETIME path above. The fractional part does not disambiguate a civil time
+        // repeated by a backward timezone transition.
+        if (!civil_year_is_monotonic(civil_seconds) ||
+            lookup.kind != cctz::time_zone::civil_lookup::UNIQUE) {
+            return std::nullopt;
+        }
+        const auto seconds = lookup.pre.time_since_epoch().count();
         const auto nanos = cast_set<int32_t>(datetime.microsecond() * 1000);
         return ::orc::Literal(seconds, nanos);
     }

be/src/format_v2/parquet/parquet_statistics.cpp

@@ -34,6 +34,7 @@
 #include <optional>
 #include <set>
 #include <string>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -45,6 +46,7 @@
 #include "exprs/expr_zonemap_filter.h"
 #include "exprs/vexpr_context.h"
 #include "format_v2/parquet/parquet_column_schema.h"
+#include "format_v2/timestamp_statistics.h"
 #include "runtime/runtime_profile.h"
 #include "storage/index/zone_map/zone_map_index.h"
 #include "storage/index/zone_map/zonemap_eval_context.h"
@@ -118,13 +120,51 @@ bool set_decoded_field(const ParquetColumnSchema& column_schema, DecodedValueKin
     return read_decoded_field(column_schema, view, field, timezone).ok();
 }
 
+int64_t floor_timestamp_seconds(int64_t value, ParquetTimeUnit time_unit) {
+    int64_t units_per_second = 1;
+    switch (time_unit) {
+    case ParquetTimeUnit::MILLIS:
+        units_per_second = 1000;
+        break;
+    case ParquetTimeUnit::MICROS:
+        units_per_second = 1000000;
+        break;
+    case ParquetTimeUnit::NANOS:
+        units_per_second = 1000000000;
+        break;
+    default:
+        DORIS_CHECK(false);
+    }
+    return format::floor_epoch_seconds(value, units_per_second);
+}
+
+bool timestamp_min_max_is_safe(const ParquetColumnSchema& column_schema, int64_t min_value,
+                               int64_t max_value, const cctz::time_zone* timezone) {
+    if (!column_schema.type_descriptor.is_timestamp ||
+        !column_schema.type_descriptor.timestamp_is_adjusted_to_utc || timezone == nullptr ||
+        remove_nullable(column_schema.type)->get_primitive_type() == TYPE_TIMESTAMPTZ) {
+        // TIMESTAMPTZ keeps the original UTC ordering, so local civil-time rollback does not make
+        // its converted min/max non-monotonic.
+        return true;
+    }
+    return format::utc_timestamp_range_is_monotonic(
+            floor_timestamp_seconds(min_value, column_schema.type_descriptor.time_unit),
+            floor_timestamp_seconds(max_value, column_schema.type_descriptor.time_unit), *timezone);
+}
+
 template <typename ParquetDType>
 bool set_decoded_min_max(const std::shared_ptr<::parquet::Statistics>& statistics,
                          const ParquetColumnSchema& column_schema, DecodedValueKind value_kind,
                          ParquetColumnStatistics* column_statistics,
                          const cctz::time_zone* timezone) {
     auto typed_statistics =
             std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics);
+    if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) {
+        if (!timestamp_min_max_is_safe(column_schema, typed_statistics->min(),
+                                       typed_statistics->max(), timezone)) {
+            return false;
+        }
+    }
     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(),
@@ -969,6 +1009,12 @@ bool set_page_decoded_min_max(const std::shared_ptr<::parquet::ColumnIndex>& col
         page_idx >= typed_index->max_values().size()) {
         return false;
     }
+    if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) {
+        if (!timestamp_min_max_is_safe(column_schema, typed_index->min_values()[page_idx],
+                                       typed_index->max_values()[page_idx], timezone)) {
+            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],

be/src/format_v2/timestamp_statistics.h

@@ -0,0 +1,62 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cctz/time_zone.h>
+
+#include <cstdint>
+
+#include "common/check.h"
+
+namespace doris::format {
+
+inline int64_t floor_epoch_seconds(int64_t value, int64_t units_per_second) {
+    DORIS_CHECK(units_per_second > 0);
+    auto seconds = value / units_per_second;
+    if (value < 0 && value % units_per_second != 0) {
+        --seconds;
+    }
+    return seconds;
+}
+
+// UTC instants are ordered, but converting them to civil DATETIMEV2 values is not monotonic when
+// a timezone moves its clock backward. Metadata min/max converted across such a transition cannot
+// safely form a ZoneMap: the true civil minimum or maximum may occur inside the UTC interval.
+inline bool utc_timestamp_range_is_monotonic(int64_t min_seconds, int64_t max_seconds,
+                                             const cctz::time_zone& timezone) {
+    DORIS_CHECK(min_seconds <= max_seconds);
+    auto current = cctz::time_point<cctz::seconds>(cctz::seconds(min_seconds));
+    const auto range_end = cctz::time_point<cctz::seconds>(cctz::seconds(max_seconds));
+    cctz::time_zone::civil_transition transition;
+    while (timezone.next_transition(current, &transition)) {
+        const auto transition_time = timezone.lookup(transition.to).trans;
+        if (transition_time > range_end) {
+            return true;
+        }
+        if (transition.to < transition.from) {
+            return false;
+        }
+        // Move past the transition that was just inspected. Some cctz implementations return the
+        // same transition again when queried at its exact instant, which would otherwise prevent
+        // us from seeing a later rollback in the requested range.
+        current = transition_time + cctz::seconds(1);
+    }
+    return true;
+}
+
+} // namespace doris::format

be/test/format_v2/orc/orc_reader_test.cpp

@@ -74,6 +74,7 @@
 #include "runtime/runtime_state.h"
 #include "storage/segment/condition_cache.h"
 #include "storage/utils.h"
+#include "util/timezone_utils.h"
 
 namespace doris {
 namespace {
@@ -2455,6 +2456,49 @@ class NullableGreaterThanExpr final : public VExpr {
     const std::string _expr_name;
 };
 
+template <PrimitiveType Primitive>
+class NullableEqualsExpr final : public VExpr {
+public:
+    using ColumnType = typename PrimitiveTypeTraits<Primitive>::ColumnType;
+    using ValueType = typename PrimitiveTypeTraits<Primitive>::CppType;
+
+    NullableEqualsExpr(int column_id, DataTypePtr type, const Field& value, std::string column_name)
+            : VExpr(std::make_shared<DataTypeUInt8>(), false),
+              _column_id(column_id),
+              _value(value.get<Primitive>()),
+              _expr_name("NullableEqualsExpr") {
+        _node_type = TExprNodeType::BINARY_PRED;
+        _opcode = TExprOpcode::EQ;
+        add_child(TableSlotRef::create_shared(column_id, column_id, -1, make_nullable(type),
+                                              std::move(column_name)));
+        add_child(TableLiteral::create_shared(std::move(type), value));
+    }
+
+    Status execute_column_impl(VExprContext* context, const Block* block, const Selector* selector,
+                               size_t count, ColumnPtr& result_column) const override {
+        const auto& nullable_column =
+                assert_cast<const ColumnNullable&>(*block->get_by_position(_column_id).column);
+        const auto& input = assert_cast<const ColumnType&>(nullable_column.get_nested_column());
+        auto result = ColumnUInt8::create();
+        auto& result_data = result->get_data();
+        result_data.resize(count);
+        for (size_t row = 0; row < count; ++row) {
+            const size_t input_row = selector == nullptr ? row : (*selector)[row];
+            result_data[row] = !nullable_column.is_null_at(input_row) &&
+                               input.get_element(input_row) == _value;
+        }
+        result_column = std::move(result);
+        return Status::OK();
+    }
+
+    const std::string& expr_name() const override { return _expr_name; }
+
+private:
+    const int _column_id;
+    const ValueType _value;
+    const std::string _expr_name;
+};
+
 template <PrimitiveType Primitive>
 class NullableInExpr final : public VExpr {
 public:
@@ -4036,7 +4080,9 @@ void write_multi_stripe_orc_sarg_types_file(const std::string& file_path) {
     out.write(memory_stream.getData(), static_cast<std::streamsize>(memory_stream.getLength()));
 }
 
-void write_multi_stripe_orc_timestamp_instant_sarg_file(const std::string& file_path) {
+void write_multi_stripe_orc_timestamp_instant_sarg_file(
+        const std::string& file_path, int64_t first_timestamp_second = 0,
+        int64_t second_timestamp_second = 1609459200) {
     auto type = std::unique_ptr<::orc::Type>(::orc::Type::buildTypeFromString(
             "struct<timestamp_instant_col:timestamp with local time zone,payload:string>"));
 
@@ -4068,8 +4114,8 @@ void write_multi_stripe_orc_timestamp_instant_sarg_file(const std::string& file_
         writer->add(*batch);
     };
 
-    add_batch(0);
-    add_batch(1609459200);
+    add_batch(first_timestamp_second);
+    add_batch(second_timestamp_second);
     writer->close();
 
     std::ofstream out(file_path, std::ios::binary);
@@ -8327,6 +8373,49 @@ TEST_F(NewOrcReaderTest, SargTimestampInstantConjunctUsesSessionTimezone) {
     EXPECT_EQ(reader->reader_statistics().filtered_group_rows, 200);
 }
 
+TEST_F(NewOrcReaderTest, SargTimestampInstantRepeatedCivilTimeDoesNotPruneStripes) {
+    const auto multi_stripe_file_path =
+            (_test_dir / "sarg_timestamp_instant_dst_rollback.orc").string();
+    // These UTC instants both decode to 2021-11-07 01:30:00.123 in America/New_York,
+    // once before and once after the UTC-04:00 to UTC-05:00 rollback.
+    write_multi_stripe_orc_timestamp_instant_sarg_file(multi_stripe_file_path, 1636263000,
+                                                       1636266600);
+    ASSERT_EQ(get_orc_stripe_count(multi_stripe_file_path), 2);
+
+    auto reader = create_reader_for_path(multi_stripe_file_path);
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    TimezoneUtils::load_timezones_to_cache();
+    state.set_timezone("America/New_York");
+    ASSERT_TRUE(reader->init(&state).ok());
+
+    std::vector<format::ColumnDefinition> schema;
+    ASSERT_TRUE(reader->get_schema(&schema).ok());
+    ASSERT_EQ(schema.size(), 2);
+
+    const auto literal =
+            Field::create_field<TYPE_DATETIMEV2>(make_datetime_v2(2021, 11, 7, 1, 30, 0, 123000));
+    auto request = std::make_shared<format::FileScanRequest>();
+    request->predicate_columns = {field_projection(0)};
+    request->conjuncts.push_back(
+            VExprContext::create_shared(std::make_shared<NullableEqualsExpr<TYPE_DATETIMEV2>>(
+                    0, remove_nullable(schema[0].type), literal, "timestamp_instant_col")));
+    ASSERT_TRUE(reader->open(request).ok());
+
+    bool eof = false;
+    size_t result_rows = 0;
+    while (!eof) {
+        Block block = build_file_block({schema[0]});
+        size_t rows = 0;
+        ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+        result_rows += rows;
+    }
+
+    EXPECT_EQ(result_rows, 2);
+    EXPECT_EQ(reader->reader_statistics().filtered_row_groups, 0);
+    EXPECT_EQ(reader->reader_statistics().filtered_row_groups_by_min_max, 0);
+    EXPECT_EQ(reader->reader_statistics().filtered_group_rows, 0);
+}
+
 TEST_F(NewOrcReaderTest, SargTimestampLowerPrecisionCastDoesNotPruneStripes) {
     const auto multi_stripe_file_path =
             (_test_dir / "sarg_timestamp_lower_precision_cast.orc").string();