[branch-4.2][fix](zonemap) Do not trust a cut string bound to prune or to answer MIN/MAX (#67642)

be/src/storage/index/zone_map/zone_map_index.cpp

@@ -38,6 +38,7 @@
 #include "storage/segment/encoding_info.h"
 #include "storage/tablet/tablet_schema.h"
 #include "storage/types.h"
+#include "storage/utils.h"
 #include "util/slice.h"
 #include "util/unaligned.h"
 
@@ -100,6 +101,14 @@ Status ZoneMap::from_proto(const ZoneMapPB& zone_map, const DataTypePtr& data_ty
             parse_bound(zone_map.max(), zone_map_info.max_value);
         }
 
+        // A max of all 0xff carries past its first byte and ends up all zero, which stands above
+        // nothing. Give up the range instead of ruling out rows with it.
+        if (!zone_map_info.pass_all && is_string_type(field_type) &&
+            zone_map.max().size() == MAX_ZONE_MAP_INDEX_SIZE &&
+            zone_map.max().find_first_not_of('\0') == std::string::npos) {
+            zone_map_info.pass_all = true;
+        }
+
         // NaN and infinity only set the flags below, never min/max, so a page holding nothing
         // else leaves both at the values add_values() starts from: min = DBL_MAX and
         // max = -DBL_MAX, neither of which is a value in the page.
@@ -247,11 +256,18 @@ void TypedZoneMapIndexWriter<Type>::modify_index_before_flush(
     // slightly larger than any real string that shares the same 512-byte prefix, ensuring no false negatives —
     // the zone map will never incorrectly skip a page that contains matching data.
     //
-    // In UTF8 encoding, here do not appear 0xff in last byte
+    // A string column holds arbitrary bytes, so the last byte can be 0xff. Adding one to it wraps
+    // to 0x00 and leaves a max below the data, so carry into the byte before it.
     if constexpr (Type == TYPE_CHAR || Type == TYPE_VARCHAR || Type == TYPE_STRING) {
         auto& str = zone_map.max_value.get<Type>();
         if (str.size() == MAX_ZONE_MAP_INDEX_SIZE) {
-            str[str.size() - 1] += 1;
+            for (size_t i = str.size(); i > 0; --i) {
+                auto byte = static_cast<uint8_t>(str[i - 1]) + 1;
+                str[i - 1] = static_cast<char>(byte);
+                if (static_cast<uint8_t>(byte) != 0) {
+                    break;
+                }
+            }
         }
     }
 }

be/src/storage/segment/column_reader.cpp

@@ -450,9 +450,6 @@ Status ColumnReader::next_batch_of_zone_map(size_t* n, MutableColumnPtr& dst) co
     // TODO: this work to get min/max value seems should only do once
     ZoneMap zone_map;
     RETURN_IF_ERROR(ZoneMap::from_proto(*_segment_zone_map, _data_type, zone_map));
-    // Segment::new_iterator does not build this iterator on an invalid zone map, whose min/max
-    // are unset and would be reported below as if they were data.
-    DORIS_CHECK(!zone_map.pass_all);
 
     dst->reserve(*n);
     if (!zone_map.has_not_null) {

be/src/storage/segment/segment.cpp

@@ -148,8 +148,14 @@ Status build_segment_zonemap_context(Segment* segment, const Schema& schema,
     return Status::OK();
 }
 
-// The statistics iterator answers pushed-down aggregates from the segment zone maps alone. An
-// invalid zone map has no min/max to answer with, so the caller has to read the data instead.
+// Whether to force MIN/MAX onto the zone map when its bound is not a value the data holds now: a
+// cut string bound, or one covering rows a delete predicate removed. Statistics collection sets it.
+// MIN/MAX is the only aggregate this can force, because it is the only one that reads the bounds.
+bool pushdown_zonemap_minmax_forced(const StorageReadOptions& read_options) {
+    return read_options.push_down_agg_type_opt == TPushAggOp::MINMAX &&
+           read_options.runtime_state->query_options().force_pushdown_zonemap_minmax;
+}
+
 Status segment_zone_maps_can_answer_agg(Segment* segment, const Schema& schema,
                                         const StorageReadOptions& read_options, bool* usable) {
     *usable = true;
@@ -168,10 +174,26 @@ Status segment_zone_maps_can_answer_agg(Segment* segment, const Schema& schema,
         }
         ZoneMap zone_map;
         RETURN_IF_ERROR(reader->get_segment_zone_map(&zone_map));
+
+        // The zone map gave up its range, so it has no min/max left to answer with.
         if (zone_map.pass_all) {
             *usable = false;
             return Status::OK();
         }
+
+        // Only a string bound is cut at MAX_ZONE_MAP_INDEX_SIZE, and a column of nothing but
+        // nulls stored no bound to look at.
+        if (!is_string_type(schema.column(schema.column_id(i))->type()) || !zone_map.has_not_null) {
+            continue;
+        }
+
+        // A cut bound is not a value the column holds: the min is a prefix of the smallest value
+        // and the max was raised past the largest one. Neither can answer MIN()/MAX().
+        if (zone_map.min_value.as_string_view().size() >= MAX_ZONE_MAP_INDEX_SIZE ||
+            zone_map.max_value.as_string_view().size() >= MAX_ZONE_MAP_INDEX_SIZE) {
+            *usable = false;
+            return Status::OK();
+        }
     }
     return Status::OK();
 }
@@ -463,16 +485,17 @@ Status Segment::new_iterator(SchemaSPtr schema, const StorageReadOptions& read_o
         RETURN_IF_ERROR(load_index(read_options.stats, &read_options.io_ctx));
     }
 
+    // COUNT and MIX report the segment row count, which a delete predicate makes wrong whatever
+    // the zone map bounds hold, so they keep the guard below even when the switch is on.
+    const auto agg = read_options.push_down_agg_type_opt;
+    const bool forced = pushdown_zonemap_minmax_forced(read_options);
     bool use_statistics_iterator =
-            read_options.delete_condition_predicates->num_of_column_predicate() == 0 &&
-            read_options.push_down_agg_type_opt != TPushAggOp::NONE &&
-            read_options.push_down_agg_type_opt != TPushAggOp::COUNT_ON_INDEX;
-    // COUNT only fills defaults, every other pushed-down aggregate reads min/max out of the
-    // segment zone maps.
-    if (use_statistics_iterator && read_options.push_down_agg_type_opt != TPushAggOp::COUNT) {
-        bool usable = false;
-        RETURN_IF_ERROR(segment_zone_maps_can_answer_agg(this, *schema, read_options, &usable));
-        use_statistics_iterator = usable;
+            agg != TPushAggOp::NONE && agg != TPushAggOp::COUNT_ON_INDEX &&
+            (forced || read_options.delete_condition_predicates->num_of_column_predicate() == 0);
+    // COUNT only fills defaults, every other aggregate reads min/max out of the zone maps.
+    if (use_statistics_iterator && !forced && agg != TPushAggOp::COUNT) {
+        RETURN_IF_ERROR(segment_zone_maps_can_answer_agg(this, *schema, read_options,
+                                                         &use_statistics_iterator));
     }
     if (use_statistics_iterator) {
         iter->reset(new_vstatistics_iterator(this->shared_from_this(), *schema));

be/test/exec/scan/vgeneric_iterators_test.cpp

@@ -32,7 +32,12 @@
 #include "gtest/gtest_pred_impl.h"
 #include "io/fs/file_writer.h"
 #include "io/fs/local_file_system.h"
+#include "runtime/runtime_state.h"
 #include "storage/olap_common.h"
+#include "storage/olap_define.h"
+#include "storage/olap_tuple.h"
+#include "storage/predicate/block_column_predicate.h"
+#include "storage/predicate/null_predicate.h"
 #include "storage/row_cursor.h"
 #include "storage/schema.h"
 #include "storage/segment/column_reader.h"
@@ -186,6 +191,201 @@ TEST(VGenericIteratorsTest, StatisticsIteratorPreservesNullForNullableChar) {
     ASSERT_TRUE(fs->delete_directory(test_dir).ok());
 }
 
+// A string zone map bound is cut to 512 bytes, and a cut bound is not a value the column holds:
+// the min is a prefix of the smallest value and the max was raised past the largest one. FE pushes
+// MIN/MAX down for every string column, so the segment is the one that has to notice and hand the
+// query back to a normal read.
+class StatisticsIteratorStringBoundsTest : public testing::Test {
+protected:
+    static constexpr auto kTestDir = "./ut_dir/statistics_string_bounds_test";
+
+    void SetUp() override {
+        _fs = io::global_local_filesystem();
+        ASSERT_TRUE(_fs->delete_directory(kTestDir).ok());
+        ASSERT_TRUE(_fs->create_directory(kTestDir).ok());
+    }
+    void TearDown() override { EXPECT_TRUE(_fs->delete_directory(kTestDir).ok()); }
+
+    static TabletSchemaSPtr make_schema() {
+        auto tablet_schema = std::make_shared<TabletSchema>();
+        TabletColumn key;
+        key.set_name("c1");
+        key.set_unique_id(0);
+        key.set_type(FieldType::OLAP_FIELD_TYPE_INT);
+        key.set_length(4);
+        key.set_index_length(4);
+        key.set_is_key(true);
+        key.set_is_nullable(false);
+        tablet_schema->append_column(key);
+
+        TabletColumn value;
+        value.set_name("c2");
+        value.set_unique_id(1);
+        value.set_type(FieldType::OLAP_FIELD_TYPE_VARCHAR);
+        value.set_length(65535);
+        value.set_is_key(false);
+        value.set_is_nullable(false);
+        value.set_aggregation_method(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE);
+        tablet_schema->append_column(value);
+        tablet_schema->set_storage_page_size(4096);
+        return tablet_schema;
+    }
+
+    // Writes one segment holding `values` in the VARCHAR column and returns the iterator that the
+    // pushed-down `agg` would run on. `accept_cut_bound` is what statistics collection sets:
+    // it takes an inexact min/max as an approximation instead of reading the data.
+    // `with_delete` adds a delete predicate, which leaves the zone map covering removed rows.
+    std::unique_ptr<RowwiseIterator> pushdown_iterator_for(
+            const std::string& name, const std::vector<std::string>& values,
+            bool accept_cut_bound = false, bool with_delete = false,
+            TPushAggOp::type agg = TPushAggOp::MINMAX) {
+        auto tablet_schema = make_schema();
+        const std::string segment_path = std::string(kTestDir) + "/" + name + ".dat";
+
+        io::FileWriterPtr file_writer;
+        EXPECT_TRUE(_fs->create_file(segment_path, &file_writer).ok());
+        SegmentWriterOptions writer_options;
+        writer_options.num_rows_per_block = 1024;
+        TestSegmentWriter writer(file_writer.get(), 0, tablet_schema, nullptr, nullptr, writer_options,
+                                 nullptr);
+        EXPECT_TRUE(writer.init().ok());
+
+        RowCursor row;
+        OlapTuple tuple;
+        for (size_t i = 0; i < tablet_schema->num_columns(); ++i) {
+            tuple.add_null();
+        }
+        EXPECT_EQ(Status::OK(), row.init(tablet_schema, tuple));
+        for (size_t i = 0; i < values.size(); ++i) {
+            row.mutable_field(0) = Field::create_field<TYPE_INT>(static_cast<int32_t>(i));
+            row.mutable_field(1) = Field::create_field<TYPE_STRING>(String(values[i]));
+            EXPECT_TRUE(writer.append_row(row).ok());
+        }
+        uint64_t file_size = 0;
+        uint64_t index_size = 0;
+        EXPECT_TRUE(writer.finalize(&file_size, &index_size).ok());
+        EXPECT_TRUE(file_writer->close().ok());
+
+        std::shared_ptr<segment_v2::Segment> segment;
+        EXPECT_TRUE(segment_v2::Segment::open(_fs, segment_path, 100, 0, RowsetId {.version = 1},
+                                              tablet_schema, io::FileReaderOptions {}, &segment)
+                            .ok());
+
+        std::vector<ColumnId> column_ids {0, 1};
+        // VStatisticsIterator keeps a reference to the schema, so it has to outlive the iterator.
+        auto schema = std::make_shared<Schema>(tablet_schema->columns(), column_ids);
+        StorageReadOptions read_options;
+        read_options.push_down_agg_type_opt = agg;
+        read_options.stats = &_stats;
+        read_options.tablet_schema = tablet_schema;
+
+        if (with_delete) {
+            auto del_pred = NullPredicate::create_shared(0, "c1", true, PrimitiveType::TYPE_INT);
+            read_options.delete_condition_predicates->add_column_predicate(
+                    SingleColumnBlockPredicate::create_unique(del_pred));
+        }
+
+        auto state = std::make_unique<RuntimeState>();
+        TQueryOptions query_options;
+        query_options.__set_force_pushdown_zonemap_minmax(accept_cut_bound);
+        state->set_query_options(query_options);
+        read_options.runtime_state = state.get();
+        // The iterator keeps a copy of read_options, so the state has to outlive it.
+        _states.push_back(std::move(state));
+        _schemas.push_back(schema);
+
+        std::unique_ptr<RowwiseIterator> iter;
+        EXPECT_TRUE(segment->new_iterator(schema, read_options, &iter).ok());
+        return iter;
+    }
+
+    std::shared_ptr<io::FileSystem> _fs;
+    OlapReaderStatistics _stats;
+    std::vector<std::unique_ptr<RuntimeState>> _states;
+    std::vector<SchemaSPtr> _schemas;
+};
+
+TEST_F(StatisticsIteratorStringBoundsTest, ShortBoundsAnswerFromTheZoneMap) {
+    // Every value fits well inside the 512-byte bound, so the stored min/max are the real ones.
+    auto iter = pushdown_iterator_for("short", {"aaa", "bbb", "ccc"});
+    EXPECT_NE(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr)
+            << "exact bounds can answer MIN/MAX without reading the data";
+}
+
+TEST_F(StatisticsIteratorStringBoundsTest, CutBoundsFallBackToReadingTheData) {
+    // The longest value runs past the 512-byte cut, so the stored max is a raised prefix and not a
+    // value in the column. Answering MIN/MAX from it would return a string the table never held.
+    auto iter = pushdown_iterator_for("cut", {"aaa", "bbb", std::string(600, 'c')});
+    EXPECT_EQ(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr)
+            << "a cut bound is not a value from the data, so the query has to read the rows";
+}
+
+// A VARCHAR(512) column full to its declared length was cut too, and FE used to push MIN/MAX down
+// for it because the length is not over 512.
+TEST_F(StatisticsIteratorStringBoundsTest, BoundsCutExactlyAtTheLimitFallBack) {
+    auto iter = pushdown_iterator_for("exact", {"aaa", std::string(MAX_ZONE_MAP_INDEX_SIZE, 'z')});
+    EXPECT_EQ(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr);
+}
+
+// Statistics collection only needs an approximation, and reading the data instead would scan the
+// whole table. It keeps the statistics iterator even when the stored bounds were cut.
+TEST_F(StatisticsIteratorStringBoundsTest, CutBoundsAnswerWhenTheCallerTakesAnApproximation) {
+    auto iter = pushdown_iterator_for("cut_approx", {"aaa", "bbb", std::string(600, 'c')},
+                                      /*accept_cut_bound=*/true);
+    EXPECT_NE(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr)
+            << "statistics collection reads the cut bound rather than scanning the rows";
+}
+
+// A max raised from 0xff wraps to 0x00, so the read side turns pass_all on for that zone. The
+// bounds were parsed before that happened, so statistics collection still reads them.
+TEST_F(StatisticsIteratorStringBoundsTest, PassAllZoneMapAnswersWhenApproximationIsAccepted) {
+    std::string wrapping(MAX_ZONE_MAP_INDEX_SIZE - 1, 'a');
+    wrapping.push_back(static_cast<char>(0xff));
+    auto iter = pushdown_iterator_for("pass_all_approx", {"aaa", wrapping},
+                                      /*accept_cut_bound=*/true);
+    EXPECT_NE(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr)
+            << "a zone map that gave up its range on read still carries the bounds it parsed";
+
+    Block block;
+    for (const auto& column : iter->schema().columns()) {
+        auto data_type = column->get_vec_type();
+        block.insert(ColumnWithTypeAndName(data_type->create_column(), data_type, column->name()));
+    }
+    EXPECT_TRUE(iter->next_batch(&block).ok()) << "reading the bounds must not trip an assertion";
+}
+
+// With the switch off the same zone map sends the query back to the rows.
+TEST_F(StatisticsIteratorStringBoundsTest, PassAllZoneMapFallsBackToReadingTheData) {
+    std::string wrapping(MAX_ZONE_MAP_INDEX_SIZE - 1, 'a');
+    wrapping.push_back(static_cast<char>(0xff));
+    auto iter = pushdown_iterator_for("pass_all_exact", {"aaa", wrapping});
+    EXPECT_EQ(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr);
+}
+
+// A delete predicate leaves the zone map covering rows that are gone, so its min/max may name a
+// value the table no longer holds. That is a real answer for every query but statistics
+// collection, which takes the approximation to avoid scanning the table.
+TEST_F(StatisticsIteratorStringBoundsTest, DeletePredicateFallsBackToReadingTheData) {
+    auto iter = pushdown_iterator_for("del_exact", {"aaa", "bbb"}, /*accept_cut_bound=*/false,
+                                      /*with_delete=*/true);
+    EXPECT_EQ(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr)
+            << "a deleted row may still sit inside the zone map bounds";
+}
+
+TEST_F(StatisticsIteratorStringBoundsTest, DeletePredicateAnswersWhenApproximationIsAccepted) {
+    auto iter = pushdown_iterator_for("del_approx", {"aaa", "bbb"}, /*accept_cut_bound=*/true,
+                                      /*with_delete=*/true);
+    EXPECT_NE(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr)
+            << "statistics collection keeps the zone map even with a delete predicate";
+}
+
+TEST_F(StatisticsIteratorStringBoundsTest, CountKeepsTheDeletePredicateGuardWhenForced) {
+    auto iter = pushdown_iterator_for("count_del", {"aaa", "bbb"}, /*accept_cut_bound=*/true,
+                                      /*with_delete=*/true, TPushAggOp::COUNT);
+    EXPECT_EQ(dynamic_cast<VStatisticsIterator*>(iter.get()), nullptr)
+            << "COUNT reports the segment row count, which still counts the deleted rows";
+}
+
 TEST(VGenericIteratorsTest, Union) {
     auto schema = create_schema();
     auto output_schema = std::make_shared<Schema>(schema);