Skip to content

Commit 6363e04

Browse files
HappenLeeHappenLee
committed
[improvement](be) Prune always-true logical commit TSO predicates
### What problem does this PR solve? Issue Number: N/A Related PR: #68097 Problem Summary: On a single-version rowset whose logical commit TSO is 42, TSO > 41 survives segment filtering even though every row satisfies it. Always-true predicate elimination only visits cached physical readers, so it misses an uncached logical TSO reader. Its reader helper also accesses physical zone-map metadata directly, bypassing constant readers. Retain the first-stage TSO reader within new_iterator and reuse it during predicate elimination, independently of physical cache state. Obtain the zone map through the existing reader interface and reuse is_always_true. Keep per-segment predicate copies and existing column-read dependency checks, so projected TSO values are still read. Report null constants as nullable zone maps to keep comparisons conservative. Add comments and unit tests for cache state, predicate removal, data output and physical reader fallback. No performance benchmark was run. ### Release note Avoid redundant row-level commit TSO filters when the segment's logical TSO proves them always true. ### Check List (For Author) - Test: Unit Test - ASAN: 43 tests passed across ConstantColumnIteratorTest, SegmentIteratorExprZonemapTest and ColumnReaderCacheTest - clang-format 16, build hygiene and git diff --check passed - clang-tidy attempted: blocked by the pre-existing unmatched NOLINTEND in core/types.h:576; no diagnostics on changed lines - Behavior changed: Yes (remove redundant filters; query results unchanged) - Does this need documentation: No
1 parent 865b431 commit 6363e04

4 files changed

Lines changed: 197 additions & 8 deletions

File tree

‎be/src/storage/segment/column_reader.cpp‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -756,12 +756,14 @@ Status ColumnReader::prune_predicates_by_zone_map(
756756
std::vector<std::shared_ptr<ColumnPredicate>>& predicates, const int column_id,
757757
bool* pruned) const {
758758
*pruned = false;
759-
if (_zone_map_index == nullptr) {
759+
if (!has_zone_map()) {
760760
return Status::OK();
761761
}
762762

763+
// Read-time constants expose a logical [value, value] zone map without a physical index.
764+
// Use the reader interface so predicate removal agrees with the values returned by reads.
763765
ZoneMap zone_map;
764-
RETURN_IF_ERROR(ZoneMap::from_proto(*_segment_zone_map, _data_type, zone_map));
766+
RETURN_IF_ERROR(get_segment_zone_map(&zone_map));
765767
if (zone_map.pass_all) {
766768
return Status::OK();
767769
}
@@ -901,6 +903,7 @@ Status ColumnReader::get_segment_zone_map(segment_v2::ZoneMap* zone_map) const {
901903
Status ConstantColumnReader::get_segment_zone_map(segment_v2::ZoneMap* zone_map) const {
902904
zone_map->min_value = _value;
903905
zone_map->max_value = _value;
906+
zone_map->has_null = _value.is_null();
904907
zone_map->has_not_null = !_value.is_null();
905908
return Status::OK();
906909
}

‎be/src/storage/segment/segment.cpp‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -410,6 +410,11 @@ Status Segment::new_iterator(ReadSchemaSPtr schema, const StorageReadOptions& re
410410
RETURN_IF_ERROR(_create_column_meta_once(read_options.stats, &read_options.io_ctx));
411411

412412
read_options.stats->total_segment_number++;
413+
// Logical TSO values depend on this scan's rowset/read options, while the shared cache
414+
// stores physical readers by column identity. Retain the reader locally for predicate
415+
// simplification below: cache enumeration would miss a cold logical TSO column, and
416+
// resolving it again would allocate another constant reader.
417+
std::shared_ptr<ColumnReader> commit_tso_reader;
413418
// trying to prune the current segment by segment-level zone map
414419
for (const auto& entry : read_options.col_id_to_predicates) {
415420
// col_id_to_predicates is keyed by read-schema ordinal.
@@ -424,6 +429,9 @@ Status Segment::new_iterator(ReadSchemaSPtr schema, const StorageReadOptions& re
424429
RETURN_IF_ERROR(st);
425430
// should be OK
426431
DCHECK(reader != nullptr);
432+
if (column_id == schema->commit_tso_ordinal()) {
433+
commit_tso_reader = reader;
434+
}
427435
if (!reader->has_zone_map()) {
428436
continue;
429437
}
@@ -514,17 +522,22 @@ Status Segment::new_iterator(ReadSchemaSPtr schema, const StorageReadOptions& re
514522
if (ordinal < 0) {
515523
continue;
516524
}
517-
// This enumeration contains physical readers only. A cached TSO placeholder must
518-
// not make a logical predicate look unconditionally true.
519-
auto reader = it.second;
525+
// TSO is handled once with the reader retained above, regardless of cache warmup.
526+
// Its cached physical placeholder must never prove a logical predicate always true.
520527
if (ordinal == schema->commit_tso_ordinal()) {
521-
RETURN_IF_ERROR(get_column_reader(*schema->column(ordinal), &reader, read_options));
528+
continue;
522529
}
523530
bool tmp_pruned = false;
524-
RETURN_IF_ERROR(
525-
reader->prune_predicates_by_zone_map(pruned_predicates, ordinal, &tmp_pruned));
531+
RETURN_IF_ERROR(it.second->prune_predicates_by_zone_map(pruned_predicates, ordinal,
532+
&tmp_pruned));
526533
pruned |= tmp_pruned;
527534
}
535+
if (commit_tso_reader) {
536+
bool tso_pruned = false;
537+
RETURN_IF_ERROR(commit_tso_reader->prune_predicates_by_zone_map(
538+
pruned_predicates, schema->commit_tso_ordinal(), &tso_pruned));
539+
pruned |= tso_pruned;
540+
}
528541

529542
if (pruned) {
530543
auto options_with_pruned_predicates = read_options;

‎be/test/storage/segment/constant_column_iterator_test.cpp‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
#include <gtest/gtest.h>
1919

2020
#include <memory>
21+
#include <vector>
2122

2223
#include "common/status.h"
2324
#include "core/assert_cast.h"
@@ -92,6 +93,51 @@ TEST_F(ConstantColumnIteratorTest, MatchConditionUsesConstantZoneMap) {
9293
EXPECT_FALSE(matched);
9394
}
9495

96+
TEST_F(ConstantColumnIteratorTest, PrunePredicatesUsesLogicalZoneMap) {
97+
// Call through the base interface: the constant reader has no physical zone-map index.
98+
ConstantColumnReader constant_reader(Field::create_field<TYPE_BIGINT>(int64_t {100}));
99+
ColumnReader& reader = constant_reader;
100+
auto make_gt_predicate = [](int column_id, int64_t value) {
101+
return std::make_shared<ComparisonPredicateBase<TYPE_BIGINT, PredicateType::GT>>(
102+
column_id, "", Field::create_field<TYPE_BIGINT>(value));
103+
};
104+
auto always_true = make_gt_predicate(0, 99);
105+
auto not_always_true = make_gt_predicate(0, 100);
106+
auto other_column = make_gt_predicate(1, 99);
107+
// EQ does not implement is_always_true(); preserve its existing conservative behavior.
108+
auto unsupported = std::make_shared<ComparisonPredicateBase<TYPE_BIGINT, PredicateType::EQ>>(
109+
0, "", Field::create_field<TYPE_BIGINT>(int64_t {100}));
110+
std::vector<std::shared_ptr<ColumnPredicate>> predicates {always_true, not_always_true,
111+
other_column, unsupported};
112+
bool pruned = false;
113+
ASSERT_TRUE(reader.prune_predicates_by_zone_map(predicates, 0, &pruned).ok());
114+
EXPECT_TRUE(pruned);
115+
ASSERT_EQ(3, predicates.size());
116+
EXPECT_EQ(not_always_true, predicates[0]);
117+
EXPECT_EQ(other_column, predicates[1]);
118+
EXPECT_EQ(unsupported, predicates[2]);
119+
120+
ASSERT_TRUE(reader.prune_predicates_by_zone_map(predicates, 0, &pruned).ok());
121+
EXPECT_FALSE(pruned);
122+
EXPECT_EQ(3, predicates.size());
123+
}
124+
125+
TEST_F(ConstantColumnIteratorTest, NullConstantDoesNotProveComparisonAlwaysTrue) {
126+
ConstantColumnReader reader {Field()};
127+
segment_v2::ZoneMap zone_map;
128+
ASSERT_TRUE(reader.get_segment_zone_map(&zone_map).ok());
129+
EXPECT_TRUE(zone_map.has_null);
130+
EXPECT_FALSE(zone_map.has_not_null);
131+
auto predicate = std::make_shared<ComparisonPredicateBase<TYPE_BIGINT, PredicateType::GT>>(
132+
0, "", Field::create_field<TYPE_BIGINT>(int64_t {0}));
133+
std::vector<std::shared_ptr<ColumnPredicate>> predicates {predicate};
134+
bool pruned = true;
135+
ASSERT_TRUE(reader.prune_predicates_by_zone_map(predicates, 0, &pruned).ok());
136+
EXPECT_FALSE(pruned);
137+
ASSERT_EQ(1, predicates.size());
138+
EXPECT_EQ(predicate, predicates[0]);
139+
}
140+
95141
// A constant reader has no physical index to open, so index-iterator creation is a no-op rather
96142
// than an attempt to load one from state this reader never initializes.
97143
TEST_F(ConstantColumnIteratorTest, NewIndexIteratorIsANoOp) {

‎be/test/storage/segment/segment_iterator_expr_zonemap_test.cpp‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,15 @@ std::shared_ptr<AndBlockColumnPredicate> make_commit_tso_gt_predicate(int32_t co
131131
return predicates;
132132
}
133133

134+
void add_commit_tso_gt_predicate(StorageReadOptions* opts, int64_t value) {
135+
auto predicate = std::make_shared<ComparisonPredicateBase<TYPE_BIGINT, PredicateType::GT>>(
136+
1, COMMIT_TSO_COL, Field::create_field<TYPE_BIGINT>(value));
137+
opts->column_predicates.push_back(predicate);
138+
auto predicates = AndBlockColumnPredicate::create_shared();
139+
predicates->add_column_predicate(SingleColumnBlockPredicate::create_unique(predicate));
140+
opts->col_id_to_predicates.emplace(1, std::move(predicates));
141+
}
142+
134143
// Read schema covers all tablet columns in order, so ordinal == tablet cid.
135144
ReadSchemaSPtr make_read_schema(const TabletSchemaSPtr& tablet_schema) {
136145
return std::make_shared<ReadSchema>(tablet_schema->columns());
@@ -266,6 +275,54 @@ class SegmentIteratorExprZonemapTest : public testing::Test {
266275
}
267276
}
268277

278+
void expect_commit_tso_block(RowwiseIterator* iter, const ReadSchemaSPtr& schema,
279+
bool output_tso) {
280+
auto block = schema->create_read_block();
281+
auto st = iter->next_batch(&block);
282+
ASSERT_TRUE(st.ok()) << st;
283+
ASSERT_EQ(kCommitTsoRows, block.rows());
284+
const auto& keys = assert_cast<const ColumnInt32&>(*block.get_by_position(0).column);
285+
for (int rid = 0; rid < kCommitTsoRows; ++rid) {
286+
EXPECT_EQ(rid, keys.get_element(rid));
287+
if (output_tso) {
288+
const auto& tsos =
289+
assert_cast<const ColumnInt64&>(*block.get_by_position(1).column);
290+
EXPECT_EQ(42, tsos.get_element(rid));
291+
}
292+
}
293+
}
294+
295+
void expect_always_true_commit_tso_pruning(const std::shared_ptr<Segment>& segment,
296+
bool output_tso) {
297+
auto schema = make_read_schema(_tablet_schema);
298+
auto opts = commit_tso_read_options();
299+
opts.io_ctx.reader_type = ReaderType::READER_QUERY;
300+
std::set<int32_t> output_columns {0};
301+
if (output_tso) {
302+
output_columns.insert(1);
303+
}
304+
opts.output_columns = &output_columns;
305+
// Physical TSO is 0, but every logical row satisfies 42 > 41.
306+
add_commit_tso_gt_predicate(&opts, 41);
307+
std::unique_ptr<RowwiseIterator> iter;
308+
ASSERT_TRUE(segment->new_iterator(schema, opts, &iter).ok());
309+
ASSERT_FALSE(iter->empty());
310+
auto* scan = assert_cast<SegmentIterator*>(iter.get());
311+
EXPECT_TRUE(scan->_opts.column_predicates.empty());
312+
EXPECT_TRUE(scan->_opts.col_id_to_predicates.empty());
313+
EXPECT_TRUE(scan->_opts.zonemap_always_true_pred_cols.contains(1));
314+
EXPECT_EQ(output_tso, scan->_need_read_data(1));
315+
316+
// Simplification belongs to this segment; the caller's predicates are reusable.
317+
EXPECT_EQ(1, opts.column_predicates.size());
318+
319+
ASSERT_NO_FATAL_FAILURE(expect_commit_tso_block(iter.get(), schema, output_tso));
320+
321+
// Other expression dependencies must still force the real TSO to be read.
322+
scan->_column_states[1].has_common_expr = true;
323+
EXPECT_TRUE(scan->_need_read_data(1));
324+
}
325+
269326
void prepare_expr_context(const VExprContextSPtr& expr_ctx) {
270327
RowDescriptor row_desc;
271328
auto st = expr_ctx->prepare(&_runtime_state, row_desc);
@@ -411,6 +468,76 @@ TEST_F(SegmentIteratorExprZonemapTest, NewIteratorPrunesCommitTsoByReadOptionVal
411468
EXPECT_EQ(1, _stats.filtered_segment_number);
412469
}
413470

471+
TEST_F(SegmentIteratorExprZonemapTest, CommitTsoAlwaysTruePruningIgnoresPhysicalCacheState) {
472+
_tablet_schema = make_commit_tso_tablet_schema();
473+
for (bool physical_first : {false, true}) {
474+
SCOPED_TRACE(physical_first);
475+
std::shared_ptr<Segment> segment;
476+
ASSERT_NO_FATAL_FAILURE(build_commit_tso_segment(&segment));
477+
if (physical_first) {
478+
std::shared_ptr<ColumnReader> physical;
479+
ASSERT_TRUE(segment->get_physical_column_reader(_tablet_schema->column(1), &physical,
480+
&_stats)
481+
.ok());
482+
}
483+
for (bool output_tso : {false, true}) {
484+
SCOPED_TRACE(output_tso);
485+
ASSERT_NO_FATAL_FAILURE(expect_always_true_commit_tso_pruning(segment, output_tso));
486+
}
487+
}
488+
}
489+
490+
TEST_F(SegmentIteratorExprZonemapTest, MultiVersionCommitTsoPrunesOnlyAlwaysTruePredicates) {
491+
_tablet_schema = make_commit_tso_tablet_schema();
492+
std::shared_ptr<Segment> segment;
493+
ASSERT_NO_FATAL_FAILURE(build_commit_tso_segment(&segment, {42, 42, 42, 42, 84, 84, 84, 84}));
494+
auto schema = make_read_schema(_tablet_schema);
495+
for (int64_t threshold : {41, 42}) {
496+
SCOPED_TRACE(threshold);
497+
auto opts = commit_tso_read_options();
498+
opts.io_ctx.reader_type = ReaderType::READER_QUERY;
499+
opts.version = Version(7, 8);
500+
opts.commit_tso = TsoRange(42, 84);
501+
std::set<int32_t> output_columns {0, 1};
502+
opts.output_columns = &output_columns;
503+
add_commit_tso_gt_predicate(&opts, threshold);
504+
std::unique_ptr<RowwiseIterator> iter;
505+
ASSERT_TRUE(segment->new_iterator(schema, opts, &iter).ok());
506+
ASSERT_FALSE(iter->empty());
507+
auto* scan = assert_cast<SegmentIterator*>(iter.get());
508+
const bool always_true = threshold < 42;
509+
EXPECT_EQ(always_true, scan->_opts.column_predicates.empty());
510+
EXPECT_EQ(always_true, scan->_opts.col_id_to_predicates.empty());
511+
EXPECT_EQ(always_true, scan->_opts.zonemap_always_true_pred_cols.contains(1));
512+
auto block = schema->create_read_block();
513+
auto st = iter->next_batch(&block);
514+
ASSERT_TRUE(st.ok()) << st;
515+
ASSERT_EQ(always_true ? 8 : 4, block.rows());
516+
const auto& tsos = assert_cast<const ColumnInt64&>(*block.get_by_position(1).column);
517+
for (size_t rid = 0; rid < block.rows(); ++rid) {
518+
EXPECT_EQ(always_true && rid < 4 ? 42 : 84, tsos.get_element(rid));
519+
}
520+
}
521+
}
522+
523+
TEST_F(SegmentIteratorExprZonemapTest, UnpublishedCommitTsoPruningUsesPhysicalValue) {
524+
_tablet_schema = make_commit_tso_tablet_schema();
525+
std::shared_ptr<Segment> segment;
526+
ASSERT_NO_FATAL_FAILURE(build_commit_tso_segment(&segment));
527+
auto schema = make_read_schema(_tablet_schema);
528+
auto opts = commit_tso_read_options(-1);
529+
opts.io_ctx.reader_type = ReaderType::READER_QUERY;
530+
// No commit TSO is assigned yet: use the physical 0, so 0 > -1 is always true.
531+
add_commit_tso_gt_predicate(&opts, -1);
532+
std::unique_ptr<RowwiseIterator> iter;
533+
ASSERT_TRUE(segment->new_iterator(schema, opts, &iter).ok());
534+
ASSERT_FALSE(iter->empty());
535+
auto* scan = assert_cast<SegmentIterator*>(iter.get());
536+
EXPECT_TRUE(scan->_opts.column_predicates.empty());
537+
EXPECT_TRUE(scan->_opts.zonemap_always_true_pred_cols.contains(1));
538+
EXPECT_FALSE(scan->_need_read_data(1));
539+
}
540+
414541
// Internal unpublished reads and logical reads share a Segment, but must never change each
415542
// other's values through cache state. Exercise both orders using a real physical placeholder.
416543
TEST_F(SegmentIteratorExprZonemapTest, CommitTsoReadsAreIndependentOfPhysicalCacheOrder) {

0 commit comments

Comments
 (0)