From 3c43f4cd96d8041713391701fe6746c40fba6708 Mon Sep 17 00:00:00 2001 From: csun5285 Date: Wed, 8 Jul 2026 21:42:49 +0800 Subject: [PATCH] [fix](storage) align pruned complex column types with sub-column iterators on merge/agg read paths When a query prunes nested sub-columns of a complex column (EXPLAIN shows "pruned type"), the scan's struct sub-column iterators are filtered down to the pruned subset (set_access_paths + remove_pruned_sub_iterators), while the FE-pruned data type is recorded only in TabletSchema::_pruned_columns_data_type and honored only by TabletSchema::create_block*. Read paths that materialize column shapes from other sources still used the FULL storage type, so the positional pairing in StructFileColumnIterator::next_batch worked on two different ordinal spaces: 1. VMergeIteratorContext::block_reset built per-segment merge blocks from the storage Schema (full type). A topn ordered read over a rowset with multiple OVERLAPPING segments then decoded pruned children into the wrong dst columns: - "[E-3110] Method insert_many_dict_data is not supported for TINYINT" (dict-encoded text child decoded into a tinyint child) - "... is not supported for INT" (single-child pruning variant) - BE SIGSEGV when the shifted child types happen to match and the loop runs past the pruned iterator vector (same family as DORIS-26595) This is the root cause of DORIS-26889: the CI's s3 insert naturally produces overlapping multi-segment rowsets, while single-segment local loads never take the merge path. On master the default plan usually shields this path via TopN lazy materialization (#63736 + #64242), but e.g. raising topn_opt_limit_threshold, disabling topn_lazy_materialization_threshold, or light_schema_change=false tables expose it; branch-4.1 hits it by default. 2. BlockReader::_init_agg_state built the reader_replace argument type from the raw TabletColumn (full type) while the stored block uses the pruned type. On AGG tables any pruned read that goes through the aggregate merge path (>= 2 data rowsets, no special session vars needed) failed with "Aggregate function reader_replace argument 0 type check failed: ColumnStruct has 2 elements, but DataTypeStruct has 3 elements". Fix: make the shape-materializing read paths use the same FE-pruned type as the sub-column iterators, so dst blocks and iterators stay in one ordinal space by construction. - VMergeIteratorContext::init reads the FE-pruned complex types from StorageReadOptions::tablet_schema (TabletSchema::pruned_columns_data_type(), already set on the query read path), and block_reset then builds each merge block with the pruned type for pruned columns and the full storage type for the rest, so the block's struct children align with the pruned sub-column iterators. No new plumbing through the merge iterator or the storage Schema. - TabletColumn::get_aggregate_function now takes the column's actual data type explicitly. BlockReader::_init_agg_state passes the stored block's column type (already the pruned type for pruned complex columns), while the load/compaction/schema-change paths pass the full column type. The new regression case constructs an OVERLAPPING multi-segment rowset deterministically (MemTable.need_flush debug point + NULL keys in every batch, segment count kept below the segcompaction threshold), self-checks the layout via tablet meta, and covers all variants: the dict-error and SIGSEGV topn shapes on the inline path, the AGG default-session shape across two rowsets, and the VMaterializeNode default path as a guard. All variants fail before this fix and pass after it. Co-Authored-By: Claude Opus 4.8 (1M context) --- be/src/load/memtable/memtable.cpp | 2 + be/src/storage/iterator/block_reader.cpp | 5 +- .../iterator/vertical_block_reader.cpp | 7 +- .../storage/iterator/vgeneric_iterators.cpp | 7 +- be/src/storage/iterator/vgeneric_iterators.h | 18 ++ .../storage/schema_change/schema_change.cpp | 2 + be/src/storage/tablet/tablet_schema.cpp | 3 +- be/src/storage/tablet/tablet_schema.h | 8 +- be/test/exec/scan/vgeneric_iterators_test.cpp | 150 ++++++++++++++ ...opn_pruned_struct_overlapping_segments.out | 73 +++++++ ..._pruned_struct_overlapping_segments.groovy | 190 ++++++++++++++++++ 11 files changed, 457 insertions(+), 8 deletions(-) create mode 100644 regression-test/data/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.out create mode 100644 regression-test/suites/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.groovy diff --git a/be/src/load/memtable/memtable.cpp b/be/src/load/memtable/memtable.cpp index 8f085d5f14e54b..d4dedaeb18a900 100644 --- a/be/src/load/memtable/memtable.cpp +++ b/be/src/load/memtable/memtable.cpp @@ -29,6 +29,7 @@ #include "bvar/bvar.h" #include "common/config.h" #include "core/column/column.h" +#include "core/data_type/data_type_factory.hpp" #include "exprs/aggregate/aggregate_function_reader.h" #include "exprs/aggregate/aggregate_function_simple_factory.h" #include "load/memtable/memtable_memory_limiter.h" @@ -123,6 +124,7 @@ void MemTable::_init_agg_functions(const Block* block) { } } else { function = _tablet_schema->column(cid).get_aggregate_function( + DataTypeFactory::instance().create_data_type(_tablet_schema->column(cid)), AGG_LOAD_SUFFIX, _tablet_schema->column(cid).get_be_exec_version()); if (function == nullptr) { LOG(WARNING) << "column get aggregate function failed, column=" diff --git a/be/src/storage/iterator/block_reader.cpp b/be/src/storage/iterator/block_reader.cpp index b3d6e97074a32c..d7e4662971e9f2 100644 --- a/be/src/storage/iterator/block_reader.cpp +++ b/be/src/storage/iterator/block_reader.cpp @@ -501,8 +501,11 @@ Status BlockReader::_init_agg_state(const ReaderParams& read_params) { for (auto idx : _agg_columns_idx) { auto column = tablet_schema.column( read_params.origin_return_columns->at(_return_columns_loc[idx])); + // Use the stored block's column type (already the FE-pruned type for a pruned + // complex column) so the aggregate function's argument type matches. AggregateFunctionPtr function = - column.get_aggregate_function(AGG_READER_SUFFIX, read_params.get_be_exec_version()); + column.get_aggregate_function(_next_row.block->get_by_position(idx).type, + AGG_READER_SUFFIX, read_params.get_be_exec_version()); // to avoid coredump when something goes wrong(i.e. column missmatch) if (!function) { diff --git a/be/src/storage/iterator/vertical_block_reader.cpp b/be/src/storage/iterator/vertical_block_reader.cpp index 703776ccb708b9..eb38aee03bd6aa 100644 --- a/be/src/storage/iterator/vertical_block_reader.cpp +++ b/be/src/storage/iterator/vertical_block_reader.cpp @@ -28,6 +28,7 @@ #include "core/block/column_with_type_and_name.h" #include "core/column/column_nullable.h" #include "core/column/column_vector.h" +#include "core/data_type/data_type_factory.hpp" #include "core/data_type/data_type_number.h" #include "exprs/aggregate/aggregate_function_reader.h" #include "storage/compaction/compaction.h" @@ -194,10 +195,10 @@ void VerticalBlockReader::_init_agg_state(const ReaderParams& read_params) { auto& tablet_schema = *_tablet_schema; for (size_t idx = 0; idx < _return_columns.size(); ++idx) { + const auto& column = tablet_schema.column(_return_columns.at(idx)); AggregateFunctionPtr function = - tablet_schema.column(_return_columns.at(idx)) - .get_aggregate_function(AGG_READER_SUFFIX, - read_params.get_be_exec_version()); + column.get_aggregate_function(DataTypeFactory::instance().create_data_type(column), + AGG_READER_SUFFIX, read_params.get_be_exec_version()); DCHECK(function != nullptr); const auto* column_ptr = _stored_data_columns[idx].get(); const IColumn* columns[] = {column_ptr}; diff --git a/be/src/storage/iterator/vgeneric_iterators.cpp b/be/src/storage/iterator/vgeneric_iterators.cpp index 742781ab01e0f5..c094ea876ae32c 100644 --- a/be/src/storage/iterator/vgeneric_iterators.cpp +++ b/be/src/storage/iterator/vgeneric_iterators.cpp @@ -107,7 +107,7 @@ Status VMergeIteratorContext::block_reset(const std::shared_ptr& block) { const auto& column_ids = _output_schema->column_ids(); for (size_t i = 0; i < _output_schema->num_column_ids(); ++i) { auto column_desc = _output_schema->column(column_ids[i]); - auto data_type = Schema::get_data_type_ptr(*column_desc); + auto data_type = _data_type_maybe_pruned(*column_desc); if (data_type == nullptr) { return Status::RuntimeError("invalid data type"); } @@ -294,6 +294,11 @@ Status VAutoIncrementIterator::init(const StorageReadOptions& opts) { Status VMergeIteratorContext::init(const StorageReadOptions& opts) { _block_row_max = opts.block_row_max; _record_rowids = opts.record_rowids; + // For query reads that prune nested sub-columns, block_reset() must build the + // merge block with the FE-pruned complex types recorded on the tablet schema. + if (opts.tablet_schema != nullptr && opts.tablet_schema->has_pruned_columns()) { + _pruned_columns_data_type = &opts.tablet_schema->pruned_columns_data_type(); + } RETURN_IF_ERROR(_load_next_block()); if (valid()) { RETURN_IF_ERROR(advance()); diff --git a/be/src/storage/iterator/vgeneric_iterators.h b/be/src/storage/iterator/vgeneric_iterators.h index 0a4242ebb82adb..5e422f27e94135 100644 --- a/be/src/storage/iterator/vgeneric_iterators.h +++ b/be/src/storage/iterator/vgeneric_iterators.h @@ -185,6 +185,20 @@ class VMergeIteratorContext { // Load next block into _block Status _load_next_block(); + // Data type used to materialize `column` in the merge block: the FE-pruned + // type when the query pruned this complex column's sub-columns (so the block's + // struct children line up with the pruned sub-column iterators), otherwise the + // full storage type. + DataTypePtr _data_type_maybe_pruned(const TabletColumn& column) const { + if (_pruned_columns_data_type) { + auto it = _pruned_columns_data_type->find(column.unique_id()); + if (it != _pruned_columns_data_type->end()) { + return it->second; + } + } + return Schema::get_data_type_ptr(column); + } + RowwiseIteratorUPtr _iter; int _sequence_id_idx = -1; @@ -209,6 +223,10 @@ class VMergeIteratorContext { // block_reset() uses _output_schema to build _block, and copy_rows() iterates over // _output_schema->num_column_ids() columns to copy from _block to the destination. const SchemaSPtr _output_schema; + // col unique id -> FE-pruned type for pruned complex columns; null otherwise. Set in + // init() from opts.tablet_schema (outlives this context); block_reset() uses it so the + // merge block's struct children line up with the pruned sub-column iterators. + const std::map* _pruned_columns_data_type = nullptr; int _num_key_columns; std::vector* _compare_columns; std::vector _block_row_locations; diff --git a/be/src/storage/schema_change/schema_change.cpp b/be/src/storage/schema_change/schema_change.cpp index b705d75a4da205..80ecc5d16f6e95 100644 --- a/be/src/storage/schema_change/schema_change.cpp +++ b/be/src/storage/schema_change/schema_change.cpp @@ -44,6 +44,7 @@ #include "core/block/column_with_type_and_name.h" #include "core/column/column.h" #include "core/column/column_nullable.h" +#include "core/data_type/data_type_factory.hpp" #include "exec/common/util.hpp" #include "exec/common/variant_util.h" #include "exprs/aggregate/aggregate_function.h" @@ -174,6 +175,7 @@ class MultiBlockMerger { for (int i = key_number; i < columns; i++) { try { AggregateFunctionPtr function = tablet_schema->column(i).get_aggregate_function( + DataTypeFactory::instance().create_data_type(tablet_schema->column(i)), AGG_LOAD_SUFFIX, tablet_schema->column(i).get_be_exec_version()); if (!function) { return Status::InternalError( diff --git a/be/src/storage/tablet/tablet_schema.cpp b/be/src/storage/tablet/tablet_schema.cpp index b7ac5757edc3f8..fb945b430d1e90 100644 --- a/be/src/storage/tablet/tablet_schema.cpp +++ b/be/src/storage/tablet/tablet_schema.cpp @@ -788,11 +788,10 @@ AggregateFunctionPtr TabletColumn::get_aggregate_function_union(DataTypePtr type return AggregateStateUnion::create(state_type->get_nested_function(), {type}, type); } -AggregateFunctionPtr TabletColumn::get_aggregate_function(std::string suffix, +AggregateFunctionPtr TabletColumn::get_aggregate_function(DataTypePtr type, std::string suffix, int current_be_exec_version) const { AggregateFunctionPtr function = nullptr; - auto type = DataTypeFactory::instance().create_data_type(*this); if (type && type->get_primitive_type() == PrimitiveType::TYPE_AGG_STATE) { function = get_aggregate_function_union(type, current_be_exec_version); } else { diff --git a/be/src/storage/tablet/tablet_schema.h b/be/src/storage/tablet/tablet_schema.h index 600099f952f1ec..53020b1a84f2f5 100644 --- a/be/src/storage/tablet/tablet_schema.h +++ b/be/src/storage/tablet/tablet_schema.h @@ -153,7 +153,9 @@ class TabletColumn : public MetadataAdder { FieldAggregationMethod aggregation() const { return _aggregation; } AggregateFunctionPtr get_aggregate_function_union(DataTypePtr type, int current_be_exec_version) const; - AggregateFunctionPtr get_aggregate_function(std::string suffix, + // `type` is the data type of the column as actually read (the FE-pruned type + // when the query prunes nested sub-columns, otherwise the full column type). + AggregateFunctionPtr get_aggregate_function(DataTypePtr type, std::string suffix, int current_be_exec_version) const; int precision() const { return _precision; } int frac() const { return _frac; } @@ -753,6 +755,10 @@ class TabletSchema : public MetadataAdder { bool has_pruned_columns() const { return !_pruned_columns_data_type.empty(); } + const std::map& pruned_columns_data_type() const { + return _pruned_columns_data_type; + } + TabletStorageFormatPB storage_format() const { return _storage_format; } void set_storage_format(TabletStorageFormatPB v) { _storage_format = v; } diff --git a/be/test/exec/scan/vgeneric_iterators_test.cpp b/be/test/exec/scan/vgeneric_iterators_test.cpp index f8e3e09401a96d..7d9aa2f125087f 100644 --- a/be/test/exec/scan/vgeneric_iterators_test.cpp +++ b/be/test/exec/scan/vgeneric_iterators_test.cpp @@ -359,6 +359,156 @@ TEST(VGenericIteratorsTest, MergeWithSeqColumn) { EXPECT_EQ(seg_iter_num - 1, actual_value); } +// Schema with a smallint key c1 and a struct column c2 = struct. +static Schema create_struct_schema(int32_t struct_unique_id) { + std::vector col_schemas; + auto c1 = std::make_shared(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_SMALLINT, true); + c1->set_is_key(true); + col_schemas.emplace_back(c1); + + // The (agg, type, nullable) constructor does not support complex types, so + // build the struct column with the default constructor plus setters. + auto c2 = std::make_shared(); + c2->set_type(FieldType::OLAP_FIELD_TYPE_STRUCT); + c2->set_aggregation_method(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE); + c2->set_is_nullable(true); + c2->set_name("c2"); + c2->set_unique_id(struct_unique_id); + for (const auto* sub_name : {"f1", "f2", "f3"}) { + TabletColumn sub(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_INT, true); + sub.set_name(sub_name); + c2->add_sub_column(sub); + } + col_schemas.emplace_back(c2); + + std::vector column_ids(col_schemas.size()); + for (uint32_t cid = 0; cid < column_ids.size(); ++cid) { + column_ids[cid] = cid; + } + + Schema schema(col_schemas, column_ids); + return schema; +} + +// Fills whatever block it is given: ascending keys (key_start + i * key_stride) so +// several inputs interleave under merge, and default values for the struct column. +// Also checks that the merge context built its block with the expected struct type. +class PrunedStructUtIterator : public RowwiseIterator { +public: + PrunedStructUtIterator(const Schema& schema, size_t num_rows, int16_t key_start, + int16_t key_stride, DataTypePtr expected_struct_type) + : _schema(schema), + _num_rows(num_rows), + _key(key_start), + _key_stride(key_stride), + _expected_struct_type(std::move(expected_struct_type)) {} + ~PrunedStructUtIterator() override = default; + + Status init(const StorageReadOptions& opts) override { return Status::OK(); } + + Status next_batch(Block* block) override { + if (_rows_returned >= _num_rows) { + return Status::EndOfFile("End of PrunedStructUtIterator"); + } + // The merge block must carry the FE-pruned struct type (not the full storage + // type) so that its struct children line up with the pruned sub-column readers. + EXPECT_TRUE(block->get_by_position(1).type->equals(*_expected_struct_type)); + while (_rows_returned < _num_rows) { + ColumnWithTypeAndName& key_col = block->get_by_position(0); + ((IColumn&)(*key_col.column)).insert_data((const char*)&_key, sizeof(_key)); + ColumnWithTypeAndName& struct_col = block->get_by_position(1); + ((IColumn&)(*struct_col.column)).insert_default(); + _key = (int16_t)(_key + _key_stride); + ++_rows_returned; + } + return Status::OK(); + } + + const Schema& schema() const override { return _schema; } + +private: + const Schema& _schema; + size_t _num_rows; + size_t _rows_returned = 0; + int16_t _key; + int16_t _key_stride; + DataTypePtr _expected_struct_type; +}; + +// When the query prunes struct sub-columns, StorageReadOptions carries a tablet schema +// whose pruned-columns map records the FE-pruned type for the column's unique id. The +// merge iterator must build its internal blocks with that pruned type (see +// VMergeIteratorContext::_data_type_maybe_pruned); otherwise the block's full struct +// children would be paired with the pruned sub-column iterators by bare index. +TEST(VGenericIteratorsTest, MergeWithPrunedStructColumn) { + constexpr int32_t struct_unique_id = 10; + auto schema = create_struct_schema(struct_unique_id); + auto output_schema = std::make_shared(schema); + + // The FE-pruned type of c2 keeps only {f1, f3} out of {f1, f2, f3}. + TabletColumn pruned_c2; + pruned_c2.set_type(FieldType::OLAP_FIELD_TYPE_STRUCT); + pruned_c2.set_aggregation_method(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE); + pruned_c2.set_is_nullable(true); + pruned_c2.set_name("c2"); + pruned_c2.set_unique_id(struct_unique_id); + for (const auto* sub_name : {"f1", "f3"}) { + TabletColumn sub(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_INT, true); + sub.set_name(sub_name); + pruned_c2.add_sub_column(sub); + } + auto pruned_type = Schema::get_data_type_ptr(pruned_c2); + auto full_type = Schema::get_data_type_ptr(*schema.column(1)); + EXPECT_FALSE(pruned_type->equals(*full_type)); + + auto tablet_schema = std::make_shared(); + tablet_schema->add_pruned_columns_data_type(struct_unique_id, pruned_type); + EXPECT_TRUE(tablet_schema->has_pruned_columns()); + EXPECT_EQ(tablet_schema->pruned_columns_data_type().size(), 1); + EXPECT_TRUE( + tablet_schema->pruned_columns_data_type().at(struct_unique_id)->equals(*pruned_type)); + + std::vector inputs; + inputs.push_back(std::make_unique(schema, 100, 0, 2, pruned_type)); + inputs.push_back(std::make_unique(schema, 100, 1, 2, pruned_type)); + + auto iter = new_merge_iterator(std::move(inputs), -1, false, false, nullptr, output_schema); + StorageReadOptions opts; + opts.tablet_schema = tablet_schema; + auto st = iter->init(opts); + EXPECT_TRUE(st.ok()); + + // The destination block mirrors a query-level block: the struct column is created + // with the pruned type, matching what the merge context copies into it. + Block block; + auto key_type = Schema::get_data_type_ptr(*schema.column(0)); + auto key_column = key_type->create_column(); + block.insert(ColumnWithTypeAndName(std::move(key_column), key_type, "c1")); + auto struct_column = pruned_type->create_column(); + block.insert(ColumnWithTypeAndName(std::move(struct_column), pruned_type, "c2")); + std::vector row_is_same; + BlockWithSameBit block_with_same_bit {.block = &block, .same_bit = row_is_same}; + + do { + st = iter->next_batch(&block_with_same_bit); + } while (st.ok()); + + EXPECT_TRUE(st.is()); + EXPECT_EQ(block.rows(), 200); + + // Keys 0..199 in order: the two inputs (0,2,4,... and 1,3,5,...) interleaved. + auto c0 = block.get_by_position(0).column; + for (size_t i = 0; i < block.rows(); ++i) { + EXPECT_EQ(i, (*c0)[i].get()); + } + // The struct column kept the pruned 2-child layout end to end. + EXPECT_TRUE(block.get_by_position(1).type->equals(*pruned_type)); + EXPECT_EQ(block.get_by_position(1).column->size(), 200); +} + // Mirror of MergeWithSeqColumn but with small_seq_first=true. // Same key across all segments, seq values are 0..seg_iter_num-1; the merge // iterator should keep exactly one row whose seq value is the smallest (0). diff --git a/regression-test/data/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.out b/regression-test/data/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.out new file mode 100644 index 00000000000000..65e98091bdab12 --- /dev/null +++ b/regression-test/data/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.out @@ -0,0 +1,73 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !dict_default_session -- +1 s1 +2 s2 +3 s3 +4 s4 +5 s5 +6 s6 +7 s7 +8 s8 +9 s9 +10 s10 + +-- !agg_default_session -- +1 s1 +2 s2 +3 s3 +4 s4 +5 s5 +6 s6 +7 s7 +8 s8 +9 s9 +10 s10 + +-- !agg_second_rowset -- +s1501 +s1502 +s1503 +s1504 +s1505 +s1506 +s1507 +s1508 +s1509 +s1510 + +-- !dict_inline_topn -- +1 s1 +2 s2 +3 s3 +4 s4 +5 s5 +6 s6 +7 s7 +8 s8 +9 s9 +10 s10 + +-- !dict_inline_topn_single_child -- +s1 +s2 +s3 +s4 +s5 +s6 +s7 +s8 +s9 +s10 + +-- !int_inline_topn_last_child -- +3000001 +3000002 +3000003 +3000004 +3000005 +3000006 +3000007 +3000008 +3000009 +3000010 + diff --git a/regression-test/suites/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.groovy b/regression-test/suites/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.groovy new file mode 100644 index 00000000000000..e00662c46eb095 --- /dev/null +++ b/regression-test/suites/fault_injection_p0/test_topn_pruned_struct_overlapping_segments.groovy @@ -0,0 +1,190 @@ +// 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. + +// Regression case for the pruned-struct read misalignment on the inline topn scan path. +// +// Root cause: when a topn ordered read (order by key limit N) hits a rowset with +// multiple OVERLAPPING segments, BetaRowsetReader picks VMergeIterator, whose +// VMergeIteratorContext::init() eagerly loads the first block of every segment. +// That block is created from the storage Schema (FULL struct type, all children), +// while the struct sub-column iterators have been filtered down to the pruned +// subset by set_access_paths()/remove_pruned_sub_iterators(). The pairing loop in +// StructFileColumnIterator::next_batch() indexes _sub_column_iterators[i] by the +// dst tuple position, so the two ordinal spaces disagree: +// - text/dict child data decoded into a TINYINT child column +// -> [E-3110] insert_many_dict_data is not supported for TINYINT +// - same-typed children shift silently, and the loop runs past the pruned +// iterator vector -> out-of-bounds -> BE SIGSEGV +// +// The FE TopN lazy materialization (VMaterializeNode) usually shields this path +// on master; setting topn_lazy_materialization_threshold=-1 (or any state where +// the rule bails, e.g. light_schema_change=false tables) exposes it. +// +// Layout construction: MemTable.need_flush debug point forces one segment per +// inserted block (batch_size=500, 2000 rows -> 4 segments, below the +// segcompaction_batch_size=10 threshold so segcompaction never merges them), and +// the NULL k1 rows in every block make every segment's key lower bound NULL, so +// the segments overlap and the rowset is marked OVERLAPPING. +suite("test_topn_pruned_struct_overlapping_segments", "nonConcurrent") { + def dictTable = "test_topn_pruned_struct_ovlp_dict" + def intTable = "test_topn_pruned_struct_ovlp_int" + def aggTable = "test_topn_pruned_struct_ovlp_agg" + + sql "DROP TABLE IF EXISTS ${dictTable}" + sql "DROP TABLE IF EXISTS ${intTable}" + sql "DROP TABLE IF EXISTS ${aggTable}" + sql """ + CREATE TABLE ${dictTable} ( + k1 BIGINT, + c ARRAY>> + ) DUPLICATE KEY(k1) + DISTRIBUTED BY HASH(k1) BUCKETS 1 + PROPERTIES('replication_num' = '1') + """ + sql """ + CREATE TABLE ${intTable} ( + k1 BIGINT, + c ARRAY>> + ) DUPLICATE KEY(k1) + DISTRIBUTED BY HASH(k1) BUCKETS 1 + PROPERTIES('replication_num' = '1') + """ + + try { + GetDebugPoint().enableDebugPointForAllBEs("MemTable.need_flush") + // 4 blocks of 500 rows -> 4 segments per tablet; k1 NULL every 20 rows so + // every segment starts at NULL -> overlapping key ranges -> OVERLAPPING rowset. + sql "SET batch_size = 500" + sql """ + INSERT INTO ${dictTable} + SELECT if(number % 20 = 0, NULL, number), + array(array(named_struct( + 'col1', CAST(number AS INT), + 'col2', CAST(number % 127 AS TINYINT), + 'col17', concat('s', number)))) + FROM numbers('number' = '2000') + """ + sql """ + INSERT INTO ${intTable} + SELECT if(number % 20 = 0, NULL, number), + array(array(named_struct( + 'col1', CAST(number + 1000000 AS INT), + 'col2', CAST(number + 2000000 AS INT), + 'col3', CAST(number + 3000000 AS INT)))) + FROM numbers('number' = '2000') + """ + } finally { + GetDebugPoint().disableDebugPointForAllBEs("MemTable.need_flush") + } + + // Self-check the storage layout: the test is void unless the rowset really is + // a multi-segment OVERLAPPING one. BetaRowsetReader::is_merge_iterator() needs + // both the OVERLAPPING flag and num_segments > 1, so assert both. + for (def tbl : [dictTable, intTable]) { + def tablets = sql_return_maparray("SHOW TABLETS FROM ${tbl}") + def (code, out, err) = curl("GET", tablets[0].MetaUrl) + assertEquals(0, code) + def jsonMeta = parseJson(out.trim()) + def hasMergeLayout = jsonMeta.rs_metas.any { meta -> + meta.segments_overlap_pb == "OVERLAPPING" && meta.num_segments > 1 + } + assertTrue(hasMergeLayout, + "expected an OVERLAPPING rowset with num_segments > 1 in ${tbl}, " + + "check the MemTable.need_flush debug point") + } + + // AGG-table variant: the same ordinal-space mismatch also hits the aggregate + // merge path (BlockReader::_init_agg_state used to build the reader_replace + // argument type from the full TabletColumn while the stored block uses the + // pruned type). It needs no debug point, no topn and no special session vars: + // two ordinary inserts (two rowsets) + a pruned read used to fail with + // "Aggregate function reader_replace argument 0 type check failed". + sql """ + CREATE TABLE ${aggTable} ( + k1 BIGINT, + c ARRAY>> REPLACE + ) AGGREGATE KEY(k1) + DISTRIBUTED BY HASH(k1) BUCKETS 1 + PROPERTIES('replication_num' = '1') + """ + sql """ + INSERT INTO ${aggTable} + SELECT number, array(array(named_struct( + 'col1', CAST(number AS INT), + 'col2', CAST(number % 127 AS TINYINT), + 'col17', concat('s', number)))) + FROM numbers('number' = '1000') + """ + sql """ + INSERT INTO ${aggTable} + SELECT number + 1000, array(array(named_struct( + 'col1', CAST(number + 1000 AS INT), + 'col2', CAST(number % 127 AS TINYINT), + 'col17', concat('s', number + 1000)))) + FROM numbers('number' = '1000') + """ + + // 1. Default session: on master this plans VMaterializeNode (TopN lazy + // materialization + rowid fetch); guards that path stays correct too. + // Expected: k1 1..10 -> col1 = k1, col17 = 's' + k1. + qt_dict_default_session """ + SELECT element_at(c[1][1], 1), element_at(c[1][1], 'col17') + FROM ${dictTable} WHERE k1 IS NOT NULL ORDER BY k1 LIMIT 10 + """ + + // AGG merge path, pure default session (no topn involved at all). + qt_agg_default_session """ + SELECT element_at(c[1][1], 1), element_at(c[1][1], 'col17') + FROM ${aggTable} WHERE k1 >= 1 AND k1 <= 10 ORDER BY k1 + """ + // Rows from the second rowset must merge and read correctly too. + qt_agg_second_rowset """ + SELECT element_at(c[1][1], 'col17') + FROM ${aggTable} WHERE k1 >= 1501 AND k1 <= 1510 ORDER BY k1 + """ + + // 2. Force the inline topn scan path (no VMaterializeNode). This is the exact + // path that used to fail with + // "[E-3110] Method insert_many_dict_data is not supported for TINYINT". + sql "SET topn_lazy_materialization_threshold = -1" + // The storage ordered-key read (read_orderby_key -> need_ordered_result -> + // VMergeIterator) is additionally gated by enable_segment_limit_pushdown in + // OlapScanner. The variable is randomized in fuzzy regression runs; pin it so + // this section always exercises the merge path this fix targets. + sql "SET enable_segment_limit_pushdown = true" + + qt_dict_inline_topn """ + SELECT element_at(c[1][1], 1), element_at(c[1][1], 'col17') + FROM ${dictTable} WHERE k1 IS NOT NULL ORDER BY k1 LIMIT 10 + """ + + // Single pruned child: the misalignment shifts col17 data onto col1 (INT), + // which used to fail with "... is not supported for INT". + qt_dict_inline_topn_single_child """ + SELECT element_at(c[1][1], 'col17') + FROM ${dictTable} WHERE k1 IS NOT NULL ORDER BY k1 LIMIT 10 + """ + + // Same-typed children: the shifted decode does not error out, so the pairing + // loop used to run past the pruned iterator vector and SIGSEGV the BE; a + // partial regression could also return col1's values (1000001..) instead of + // col3's (3000001..) here. + qt_int_inline_topn_last_child """ + SELECT element_at(c[1][1], 'col3') + FROM ${intTable} WHERE k1 IS NOT NULL ORDER BY k1 LIMIT 10 + """ +}