[fix](be) guard timestamp pruning across DST rollback

Disable unsafe ORC timestamp SARG pushdown in civil years with backward timezone transitions, while preserving Parquet min/max pruning for TIMESTAMPTZ values that retain UTC ordering. Add ORC, Parquet, and timezone transition regression coverage.

Author: Gabriel39

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

@@ -141,7 +141,10 @@ int64_t floor_timestamp_seconds(int64_t value, ParquetTimeUnit time_unit) {
 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) {
+        !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(

be/src/format_v2/timestamp_statistics.h

@@ -51,7 +51,10 @@ inline bool utc_timestamp_range_is_monotonic(int64_t min_seconds, int64_t max_se
         if (transition.to < transition.from) {
             return false;
         }
-        current = transition_time;
+        // 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;
 }

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();

be/test/format_v2/parquet/parquet_statistics_test.cpp

@@ -36,6 +36,7 @@
 
 #include "core/data_type/data_type_number.h"
 #include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_timestamptz.h"
 #include "core/field.h"
 #include "exprs/vexpr.h"
 #include "exprs/vexpr_context.h"
@@ -420,6 +421,32 @@ TEST(ParquetStatisticsTransformTest, DisablesUtcTimestampMinMaxAcrossDstRollback
     EXPECT_LT(utc_stats.min_value, utc_stats.max_value);
 }
 
+TEST(ParquetStatisticsTransformTest, KeepsTimestampTzMinMaxAcrossDstRollback) {
+    constexpr int64_t MICROS_PER_SECOND = 1000000;
+    auto table = arrow::Table::Make(
+            arrow::schema(
+                    {arrow::field("ts", arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"), false)}),
+            {timestamp_array({1636263000 * MICROS_PER_SECOND, 1636266600 * MICROS_PER_SECOND})});
+    auto reader = make_reader(table, 2, false, true);
+    auto schema = build_file_schema(*reader);
+    auto statistics = reader->metadata()->RowGroup(0)->ColumnChunk(0)->statistics();
+
+    // This is the effective type produced by enable_mapping_timestamp_tz. The physical timestamp
+    // flags intentionally remain adjusted-to-UTC so decoding can preserve the source semantics.
+    schema[0]->type = std::make_shared<DataTypeTimeStampTz>(6);
+    schema[0]->type_descriptor.doris_type = schema[0]->type;
+
+    cctz::time_zone new_york;
+    ASSERT_TRUE(cctz::load_time_zone("America/New_York", &new_york));
+    const auto timestamp_tz_stats =
+            format::parquet::ParquetStatisticsUtils::TransformColumnStatistics(
+                    *schema[0], statistics, &new_york);
+    EXPECT_TRUE(timestamp_tz_stats.has_min_max);
+    EXPECT_EQ(timestamp_tz_stats.min_value.get_type(), TYPE_TIMESTAMPTZ);
+    EXPECT_EQ(timestamp_tz_stats.max_value.get_type(), TYPE_TIMESTAMPTZ);
+    EXPECT_LT(timestamp_tz_stats.min_value, timestamp_tz_stats.max_value);
+}
+
 TEST(ParquetStatisticsTransformTest, HandlesMissingStatisticsAndAllNullChunks) {
     auto no_stats_table = arrow::Table::Make(
             arrow::schema({arrow::field("i", arrow::int32(), true)}), {int32_array({1, 2, 3})});

be/test/format_v2/timestamp_statistics_test.cpp

@@ -31,6 +31,10 @@ TEST(TimestampStatisticsTest, DetectsBackwardTimezoneTransitionsInUtcRange) {
     EXPECT_FALSE(utc_timestamp_range_is_monotonic(1636263000, 1636264800, new_york));
     EXPECT_TRUE(utc_timestamp_range_is_monotonic(1636264800, 1636266600, new_york));
 
+    // A range beginning before the spring-forward transition must continue scanning and find the
+    // later rollback in the same year.
+    EXPECT_FALSE(utc_timestamp_range_is_monotonic(1609477200, 1641013200, new_york));
+
     // The 2021 spring transition at 07:00 UTC skips civil values but preserves ordering.
     EXPECT_TRUE(utc_timestamp_range_is_monotonic(1615703400, 1615707000, new_york));
 }