This is an automated email from the ASF dual-hosted git repository.
csun5285 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new b8dc27e263b [refactor](storage) Replace record_rowids with a
compaction rowid column (#69195)
b8dc27e263b is described below
commit b8dc27e263bf3ab791eae3310cec8d76a599e62d
Author: Chenyang Sun <[email protected]>
AuthorDate: Wed Oct 21 10:15:44 2026 +0800
[refactor](storage) Replace record_rowids with a compaction rowid column
(#69195)
Replace `record_rowids` and separate row-location buffers with a
temporary compaction rowid column.
The temporary column keeps source row locations aligned with data rows
during compaction and is removed before writing output segments.
---
be/src/common/consts.h | 2 +
be/src/storage/iterator/block_reader.cpp | 47 +-------
be/src/storage/iterator/block_reader.h | 5 -
be/src/storage/iterator/vcollect_iterator.cpp | 51 --------
be/src/storage/iterator/vcollect_iterator.h | 22 ----
be/src/storage/iterator/vertical_block_reader.cpp | 45 ++-----
be/src/storage/iterator/vertical_block_reader.h | 10 +-
.../storage/iterator/vertical_merge_iterator.cpp | 64 +++-------
be/src/storage/iterator/vertical_merge_iterator.h | 45 ++-----
be/src/storage/iterator/vgeneric_iterators.cpp | 16 ---
be/src/storage/iterator/vgeneric_iterators.h | 28 -----
be/src/storage/iterators.h | 6 -
be/src/storage/merger.cpp | 131 ++++++++++++++++-----
be/src/storage/rowset/beta_rowset_reader.cpp | 4 +-
be/src/storage/rowset/beta_rowset_reader.h | 5 -
be/src/storage/rowset/rowset_reader.h | 3 -
be/src/storage/rowset/rowset_reader_context.h | 1 -
be/src/storage/rowset/segcompaction.cpp | 9 +-
be/src/storage/schema.cpp | 14 +++
be/src/storage/schema.h | 9 +-
be/src/storage/segment/column_reader.h | 43 +++++++
be/src/storage/segment/count_on_index_fastpath.h | 16 +--
.../storage/segment/lazy_init_segment_iterator.h | 4 -
be/src/storage/segment/segment_iterator.cpp | 26 ++--
be/src/storage/segment/segment_iterator.h | 10 +-
be/src/storage/tablet/tablet_reader.cpp | 1 -
be/src/storage/tablet/tablet_reader.h | 3 +-
be/test/exec/scan/vgeneric_iterators_test.cpp | 12 ++
.../compaction/vertical_compaction_test.cpp | 2 -
.../block_reader_change_next_block_test.cpp | 4 -
be/test/storage/read_schema_test.cpp | 18 +++
.../segment/count_on_index_fastpath_test.cpp | 24 +---
.../segment_iterator_count_emit_shortcut_test.cpp | 8 --
33 files changed, 258 insertions(+), 430 deletions(-)
diff --git a/be/src/common/consts.h b/be/src/common/consts.h
index b6fe2c8788b..8cb9e42d6db 100644
--- a/be/src/common/consts.h
+++ b/be/src/common/consts.h
@@ -25,6 +25,8 @@ const std::string CSV = "csv";
const std::string CSV_WITH_NAMES = "csv_with_names";
const std::string CSV_WITH_NAMES_AND_TYPES = "csv_with_names_and_types";
const std::string GLOBAL_ROWID_COL = "__DORIS_GLOBAL_ROWID_COL__";
+// Synthetic row-id column for compaction reads, containing each source row's
physical RowLocation.
+const std::string COMPACTION_ROWID_COL = "__DORIS_COMPACTION_ROWID_COL__";
const std::string ROW_STORE_COL = "__DORIS_ROW_STORE_COL__";
const std::string DYNAMIC_COLUMN_NAME = "__DORIS_DYNAMIC_COL__";
const std::string PARTIAL_UPDATE_AUTO_INC_COL =
"__PARTIAL_UPDATE_AUTO_INC_COLUMN__";
diff --git a/be/src/storage/iterator/block_reader.cpp
b/be/src/storage/iterator/block_reader.cpp
index 0012fea97c7..285cee951f2 100644
--- a/be/src/storage/iterator/block_reader.cpp
+++ b/be/src/storage/iterator/block_reader.cpp
@@ -639,13 +639,6 @@ Status BlockReader::_direct_next_block(Block* block, bool*
eof) {
}
*eof = res.is<END_OF_FILE>();
_eof = *eof;
- if (UNLIKELY(_reader_context.record_rowids)) {
- res =
_vcollect_iter.current_block_row_locations(&_block_row_locations);
- if (UNLIKELY(!res.ok() && res != Status::Error<END_OF_FILE>(""))) {
- return res;
- }
- DCHECK_EQ(_block_row_locations.size(), block->rows());
- }
return Status::OK();
}
@@ -658,17 +651,9 @@ Status BlockReader::_replace_key_next_block(Block* block,
bool* eof) {
auto target_block_row = 0;
auto target_columns_guard = block->mutate_columns_scoped();
auto& target_columns = target_columns_guard.mutable_columns();
- // currently seq mapping only support mor table
- // so this will not be executed for the time being
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations.resize(batch_max_rows());
- }
auto merged_row = 0;
while (target_block_row < batch_max_rows() && !_eof) {
RETURN_IF_ERROR(_insert_data_normal(target_columns));
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations[target_block_row] =
_vcollect_iter.current_row_location();
- }
target_block_row++;
while (!_eof) {
@@ -678,9 +663,6 @@ Status BlockReader::_replace_key_next_block(Block* block,
bool* eof) {
if (UNLIKELY(res.is<END_OF_FILE>())) {
_eof = true;
*eof = true;
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations.resize(target_block_row);
- }
break;
}
@@ -700,9 +682,6 @@ Status BlockReader::_replace_key_next_block(Block* block,
bool* eof) {
// key, so it is safe to stop accumulating here without repeating any
row.
if (target_block_row % BLOCK_SIZE_CHECK_INTERVAL_ROWS == 0 &&
_reached_byte_budget(target_columns)) {
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations.resize(target_block_row);
- }
break;
}
}
@@ -808,16 +787,9 @@ Status BlockReader::_unique_key_next_block(Block* block,
bool* eof) {
auto target_block_row = 0;
auto target_columns_guard = block->mutate_columns_scoped();
auto& target_columns = target_columns_guard.mutable_columns();
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations.resize(batch_max_rows());
- }
do {
RETURN_IF_ERROR(_insert_data_normal(target_columns));
-
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations[target_block_row] =
_vcollect_iter.current_row_location();
- }
target_block_row++;
// the version is in reverse order, the first row is the highest
version,
@@ -827,9 +799,6 @@ Status BlockReader::_unique_key_next_block(Block* block,
bool* eof) {
if (UNLIKELY(res.is<END_OF_FILE>())) {
_eof = true;
*eof = true;
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations.resize(target_block_row);
- }
break;
}
@@ -840,9 +809,6 @@ Status BlockReader::_unique_key_next_block(Block* block,
bool* eof) {
// Byte-budget check: _next_row is already saved so stopping here is
safe.
if (target_block_row % BLOCK_SIZE_CHECK_INTERVAL_ROWS == 0 &&
_reached_byte_budget(target_columns)) {
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations.resize(target_block_row);
- }
break;
}
} while (target_block_row < batch_max_rows());
@@ -860,16 +826,8 @@ Status BlockReader::_unique_key_next_block(Block* block,
bool* eof) {
reinterpret_cast<ColumnInt8*>(target_columns[delete_sign_idx].get())
->get_data()
.data();
- int delete_count = 0;
for (int i = 0; i < target_block_row; ++i) {
- bool sign = (delete_data[i] == 0);
- filter_data[i] = sign;
- if (UNLIKELY(!sign)) {
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations[i].row_id = -1;
- delete_count++;
- }
- }
+ filter_data[i] = delete_data[i] == 0;
}
auto target_columns_size = target_columns.size();
_delete_filter_column = std::move(delete_filter_column);
@@ -880,9 +838,6 @@ Status BlockReader::_unique_key_next_block(Block* block,
bool* eof) {
block->insert(column_with_type_and_name);
RETURN_IF_ERROR(Block::filter_block(block, target_columns_size,
target_columns_size));
_stats.rows_del_filtered += target_block_row - block->rows();
- if (UNLIKELY(_reader_context.record_rowids)) {
- DCHECK_EQ(_block_row_locations.size(), block->rows() +
delete_count);
- }
}
return Status::OK();
}
diff --git a/be/src/storage/iterator/block_reader.h
b/be/src/storage/iterator/block_reader.h
index de1ad68fdcd..d3615369454 100644
--- a/be/src/storage/iterator/block_reader.h
+++ b/be/src/storage/iterator/block_reader.h
@@ -34,7 +34,6 @@
#include "storage/iterator/vcollect_iterator.h"
#include "storage/rowset/rowset_reader.h"
#include "storage/tablet/tablet_reader.h"
-#include "storage/utils.h"
namespace doris {
class ColumnPredicate;
@@ -49,8 +48,6 @@ public:
Status next_block_with_aggregation(Block* block, bool* eof) override;
- std::vector<RowLocation> current_block_row_locations() { return
_block_row_locations; }
-
void update_profile(RuntimeProfile* profile) override {
return _vcollect_iter.update_profile(profile);
}
@@ -149,8 +146,6 @@ private:
Status (BlockReader::*_next_block_func)(Block* block, bool* eof) = nullptr;
- std::vector<RowLocation> _block_row_locations;
-
ColumnPtr _delete_filter_column;
bool _is_rowsets_overlapping = true;
diff --git a/be/src/storage/iterator/vcollect_iterator.cpp
b/be/src/storage/iterator/vcollect_iterator.cpp
index 2ccc4ff06d2..f761fe84639 100644
--- a/be/src/storage/iterator/vcollect_iterator.cpp
+++ b/be/src/storage/iterator/vcollect_iterator.cpp
@@ -542,10 +542,6 @@ Status
VCollectIterator::Level0Iterator::refresh_current_row() {
if (res.is<END_OF_FILE>() && _is_empty()) {
break;
}
-
- if (UNLIKELY(_reader->_reader_context.record_rowids)) {
-
RETURN_IF_ERROR(_rs_reader->current_block_row_locations(&_block_row_locations));
- }
}
} while (!_is_empty());
_ref.row_pos = -1;
@@ -591,30 +587,10 @@ Status VCollectIterator::Level0Iterator::next(Block*
block) {
if (res.is<END_OF_FILE>() && block->rows() == 0) {
return Status::Error<END_OF_FILE>("");
}
- if (UNLIKELY(_reader->_reader_context.record_rowids)) {
-
RETURN_IF_ERROR(_rs_reader->current_block_row_locations(&_block_row_locations));
- }
return Status::OK();
}
}
-RowLocation VCollectIterator::Level0Iterator::current_row_location() {
- RowLocation& segment_row_id = _block_row_locations[_get_data_by_ref ?
_current : _ref.row_pos];
- return RowLocation(_rs_reader->rowset()->rowset_id(),
segment_row_id.segment_id,
- segment_row_id.row_id);
-}
-
-Status VCollectIterator::Level0Iterator::current_block_row_locations(
- std::vector<RowLocation>* block_row_locations) {
- block_row_locations->resize(_block_row_locations.size());
- for (auto i = 0; i < _block_row_locations.size(); i++) {
- RowLocation& row_location = _block_row_locations[i];
- (*block_row_locations)[i] =
RowLocation(_rs_reader->rowset()->rowset_id(),
- row_location.segment_id,
row_location.row_id);
- }
- return Status::OK();
-}
-
VCollectIterator::Level1Iterator::Level1Iterator(
std::list<std::unique_ptr<VCollectIterator::LevelIterator>> children,
TabletReader* reader,
bool merge, bool is_reverse, bool skip_same)
@@ -915,14 +891,8 @@ Status
VCollectIterator::Level1Iterator::_merge_next(Block* block) {
size_t column_count = target_columns.size();
auto batch_size = _reader->batch_max_rows();
- if (UNLIKELY(_reader->_reader_context.record_rowids)) {
- _block_row_locations.resize(batch_size);
- }
int continuous_row_in_block = 0;
do {
- if (UNLIKELY(_reader->_reader_context.record_rowids)) {
- _block_row_locations[target_block_row] =
_cur_child->current_row_location();
- }
++target_block_row;
++continuous_row_in_block;
// cur block finished, copy before merge_next cause merge_next will
@@ -941,9 +911,6 @@ Status VCollectIterator::Level1Iterator::_merge_next(Block*
block) {
}
auto res = _merge_next(&cur_row);
if (UNLIKELY(res.is<END_OF_FILE>())) {
- if (UNLIKELY(_reader->_reader_context.record_rowids)) {
- _block_row_locations.resize(target_block_row);
- }
return res;
}
@@ -988,9 +955,6 @@ Status VCollectIterator::Level1Iterator::_merge_next(Block*
block) {
continuous_row_in_block);
}
}
- if (UNLIKELY(_reader->_reader_context.record_rowids)) {
- _block_row_locations.resize(target_block_row);
- }
return Status::OK();
}
} while (true);
@@ -1018,19 +982,4 @@ Status
VCollectIterator::Level1Iterator::_normal_next(Block* block) {
}
}
-Status VCollectIterator::Level1Iterator::current_block_row_locations(
- std::vector<RowLocation>* block_row_locations) {
- if (!_merge) {
- if (UNLIKELY(_cur_child == nullptr)) {
- block_row_locations->clear();
- return Status::Error<END_OF_FILE>("");
- }
- return _cur_child->current_block_row_locations(block_row_locations);
- } else {
- DCHECK(_reader->_reader_context.record_rowids);
- *block_row_locations = _block_row_locations;
- return Status::OK();
- }
-}
-
} // namespace doris
diff --git a/be/src/storage/iterator/vcollect_iterator.h
b/be/src/storage/iterator/vcollect_iterator.h
index 8518aa5bc2d..6f0f9d1a6f4 100644
--- a/be/src/storage/iterator/vcollect_iterator.h
+++ b/be/src/storage/iterator/vcollect_iterator.h
@@ -27,7 +27,6 @@
#include "common/status.h"
#include "storage/iterators.h"
#include "storage/rowset/rowset_reader_context.h"
-#include "storage/utils.h"
#ifdef USE_LIBCPP
#include <queue>
#else
@@ -76,12 +75,6 @@ public:
bool is_merge() const { return _merge; }
- RowLocation current_row_location() { return
_inner_iter->current_row_location(); }
-
- Status current_block_row_locations(std::vector<RowLocation>*
block_row_locations) {
- return _inner_iter->current_block_row_locations(block_row_locations);
- }
-
void update_profile(RuntimeProfile* profile) {
if (_inner_iter != nullptr) {
_inner_iter->update_profile(profile);
@@ -147,10 +140,6 @@ private:
const inline std::vector<uint32_t>* compare_columns() const { return
_compare_columns; }
- virtual RowLocation current_row_location() = 0;
-
- virtual Status current_block_row_locations(std::vector<RowLocation>*
row_location) = 0;
-
[[nodiscard]] virtual Status ensure_first_row_ref() = 0;
virtual void update_profile(RuntimeProfile* profile) = 0;
@@ -221,10 +210,6 @@ private:
Status next(Block* block) override;
- RowLocation current_row_location() override;
-
- Status current_block_row_locations(std::vector<RowLocation>*
block_row_locations) override;
-
void update_profile(RuntimeProfile* profile) override {
if (_rs_reader != nullptr) {
_rs_reader->update_profile(profile);
@@ -286,7 +271,6 @@ private:
int _current;
BlockView _block_view;
- std::vector<RowLocation> _block_row_locations;
std::vector<bool> _row_is_same;
bool _is_merge_iterator = false;
bool _get_data_by_ref = false;
@@ -306,10 +290,6 @@ private:
Status next(Block* block) override;
- RowLocation current_row_location() override { return
_cur_child->current_row_location(); }
-
- Status current_block_row_locations(std::vector<RowLocation>*
block_row_locations) override;
-
[[nodiscard]] Status ensure_first_row_ref() override;
~Level1Iterator() override;
@@ -356,8 +336,6 @@ private:
bool _skip_same;
// used when `_merge == true`
std::unique_ptr<MergeHeap> _heap;
-
- std::vector<RowLocation> _block_row_locations;
};
std::unique_ptr<LevelIterator> _inner_iter;
diff --git a/be/src/storage/iterator/vertical_block_reader.cpp
b/be/src/storage/iterator/vertical_block_reader.cpp
index e966ad34042..659e9494db5 100644
--- a/be/src/storage/iterator/vertical_block_reader.cpp
+++ b/be/src/storage/iterator/vertical_block_reader.cpp
@@ -102,8 +102,7 @@ Status
VerticalBlockReader::next_block_with_aggregation(Block* block, bool* eof)
Status VerticalBlockReader::_get_segment_iterators(const ReaderParams&
read_params,
std::vector<RowwiseIteratorUPtr>* segment_iters,
- std::vector<bool>*
iterator_init_flag,
- std::vector<RowsetId>*
rowset_ids) {
+ std::vector<bool>*
iterator_init_flag) {
auto res = _capture_rs_readers(read_params);
if (!res.ok()) {
LOG(WARNING) << "fail to init reader when _capture_rs_readers. res:"
<< res
@@ -154,9 +153,6 @@ Status VerticalBlockReader::_get_segment_iterators(const
ReaderParams& read_para
iterator_init_flag->push_back(false);
}
}
- for (size_t i = 0; i < added_iterators; ++i) {
- rowset_ids->push_back(rowset->rowset_id());
- }
rs_split.rs_reader->reset_read_options();
}
return Status::OK();
@@ -165,27 +161,22 @@ Status VerticalBlockReader::_get_segment_iterators(const
ReaderParams& read_para
Status VerticalBlockReader::_init_collect_iter(const ReaderParams& read_params,
CompactionSampleInfo*
sample_info) {
std::vector<bool> iterator_init_flag;
- std::vector<RowsetId> rowset_ids;
std::vector<RowwiseIteratorUPtr>* segment_iters_ptr =
read_params.segment_iters_ptr;
std::vector<RowwiseIteratorUPtr> iter_ptr_vector;
if (!segment_iters_ptr) {
- RETURN_IF_ERROR(_get_segment_iterators(read_params, &iter_ptr_vector,
&iterator_init_flag,
- &rowset_ids));
+ RETURN_IF_ERROR(_get_segment_iterators(read_params, &iter_ptr_vector,
&iterator_init_flag));
CHECK(iter_ptr_vector.size() == iterator_init_flag.size());
segment_iters_ptr = &iter_ptr_vector;
} else {
for (int i = 0; i < segment_iters_ptr->size(); ++i) {
iterator_init_flag.push_back(true);
- RowsetId rowsetid;
- rowset_ids.push_back(rowsetid); // TODO: _record_rowids need it
}
// TODO(zhangzhengyu): is it enough for a context?
_reader_context.reader_type = read_params.reader_type;
_reader_context.need_ordered_result = true; // TODO: should it be?
_reader_context.is_unique = _tablet_schema->keys_type() == UNIQUE_KEYS;
_reader_context.is_key_column_group = read_params.is_key_column_group;
- _reader_context.record_rowids = read_params.record_rowids;
}
// build heap if key column iterator or build vertical merge iterator if
value column
@@ -197,12 +188,12 @@ Status VerticalBlockReader::_init_collect_iter(const
ReaderParams& read_params,
}
if (_tablet_schema->num_key_columns() == 0) {
_vcollect_iter = new_vertical_fifo_merge_iterator(
- std::move(*segment_iters_ptr), iterator_init_flag,
rowset_ids,
- _tablet_schema->keys_type(), seq_col_idx,
_row_sources_buffer, _context_stats);
+ std::move(*segment_iters_ptr), iterator_init_flag,
_tablet_schema->keys_type(),
+ seq_col_idx, _row_sources_buffer, _context_stats);
} else {
_vcollect_iter = new_vertical_heap_merge_iterator(
- std::move(*segment_iters_ptr), iterator_init_flag,
rowset_ids,
- _tablet_schema->keys_type(), seq_col_idx,
_row_sources_buffer, _context_stats,
+ std::move(*segment_iters_ptr), iterator_init_flag,
_tablet_schema->keys_type(),
+ seq_col_idx, _row_sources_buffer, _context_stats,
read_params.key_group_cluster_key_idxes);
}
} else {
@@ -211,7 +202,7 @@ Status VerticalBlockReader::_init_collect_iter(const
ReaderParams& read_params,
}
// init collect iterator
StorageReadOptions opts(_stats);
- opts.record_rowids = read_params.record_rowids;
+ opts.rowset_id = _reader_context.rowset_id;
opts.use_insert_order_when_same =
_reader_context.use_insert_order_when_same;
if (read_params.batch_size > 0) {
opts.block_row_max = cast_set<int>(read_params.batch_size);
@@ -331,13 +322,6 @@ Status VerticalBlockReader::_direct_next_block(Block*
block, bool* eof) {
}
*eof = (res.is<END_OF_FILE>());
_eof = *eof;
- if (_reader_context.is_key_column_group &&
UNLIKELY(_reader_context.record_rowids)) {
- res =
_vcollect_iter->current_block_row_locations(&_block_row_locations);
- if (UNLIKELY(!res.ok() && res != Status::Error<END_OF_FILE>(""))) {
- return res;
- }
- DCHECK_EQ(_block_row_locations.size(), block->rows());
- }
return Status::OK();
}
@@ -506,13 +490,6 @@ Status VerticalBlockReader::_unique_key_next_block(Block*
block, bool* eof) {
if (UNLIKELY(!res.ok() && !res.is<END_OF_FILE>())) {
return res;
}
- if (UNLIKELY(_reader_context.record_rowids)) {
- auto ret =
_vcollect_iter->current_block_row_locations(&_block_row_locations);
- if (UNLIKELY(!ret.ok() && !ret.is<END_OF_FILE>())) {
- return res;
- }
- DCHECK_EQ(_block_row_locations.size(), block->rows());
- }
if (_row_sources_buffer->buffered_size() < row_buffer_size_start) {
row_buffer_size_start = 0;
@@ -549,7 +526,6 @@ Status VerticalBlockReader::_unique_key_next_block(Block*
block, bool* eof) {
.data();
int cur_row = 0;
- int delete_count = 0;
while (cur_row < block_rows) {
if (_row_sources_buffer->get_agg_flag(row_source_idx)) {
row_source_idx++;
@@ -559,10 +535,6 @@ Status VerticalBlockReader::_unique_key_next_block(Block*
block, bool* eof) {
filter_data[cur_row] = sign;
if (UNLIKELY(!sign)) {
_row_sources_buffer->set_agg_flag(row_source_idx, true);
- if (UNLIKELY(_reader_context.record_rowids)) {
- _block_row_locations[cur_row].row_id = -1;
- delete_count++;
- }
}
cur_row++;
row_source_idx++;
@@ -580,9 +552,6 @@ Status VerticalBlockReader::_unique_key_next_block(Block*
block, bool* eof) {
block->insert(column_with_type_and_name);
RETURN_IF_ERROR(Block::filter_block(block, column_to_keep,
column_to_keep));
_stats.rows_del_filtered += block_rows - block->rows();
- if (UNLIKELY(_reader_context.record_rowids)) {
- DCHECK_EQ(_block_row_locations.size(), block->rows() +
delete_count);
- }
}
size_t filtered_rows_in_rs_buffer = 0;
diff --git a/be/src/storage/iterator/vertical_block_reader.h
b/be/src/storage/iterator/vertical_block_reader.h
index 543f3f734d2..b849f2bbe24 100644
--- a/be/src/storage/iterator/vertical_block_reader.h
+++ b/be/src/storage/iterator/vertical_block_reader.h
@@ -35,13 +35,10 @@
#include "storage/iterators.h"
#include "storage/tablet/tablet.h"
#include "storage/tablet/tablet_reader.h"
-#include "storage/utils.h"
#pragma once
namespace doris {
-struct RowsetId;
-
class RowSourcesBuffer;
class RowsetMeta;
struct RowBatch;
@@ -69,8 +66,6 @@ public:
return _vcollect_iter->merged_rows();
}
- std::vector<RowLocation> current_block_row_locations() { return
_block_row_locations; }
-
static uint64_t nextId;
private:
@@ -95,8 +90,7 @@ private:
Status _get_segment_iterators(const ReaderParams& read_params,
std::vector<RowwiseIteratorUPtr>*
segment_iters,
- std::vector<bool>* iterator_init_flag,
- std::vector<RowsetId>* rowset_ids);
+ std::vector<bool>* iterator_init_flag);
void _init_agg_state(const ReaderParams& read_params);
void _append_agg_data(MutableColumns& columns);
@@ -143,8 +137,6 @@ private:
phmap::flat_hash_map<const Block*, std::vector<std::pair<int, int>>>
_temp_ref_map;
- std::vector<RowLocation> _block_row_locations;
-
// For sparse column compaction optimization
// Set from reader_params.enable_sparse_optimization (calculated in
Merger::vertical_merge_rowsets)
bool _enable_sparse_optimization = false;
diff --git a/be/src/storage/iterator/vertical_merge_iterator.cpp
b/be/src/storage/iterator/vertical_merge_iterator.cpp
index 2adbbb754b4..22faa1adb62 100644
--- a/be/src/storage/iterator/vertical_merge_iterator.cpp
+++ b/be/src/storage/iterator/vertical_merge_iterator.cpp
@@ -417,7 +417,6 @@ Status VerticalMergeIteratorContext::init(const
StorageReadOptions& opts,
return Status::OK();
}
_block_row_max = opts.block_row_max;
- _record_rowids = opts.record_rowids;
RETURN_IF_ERROR(_load_next_block());
if (sample_info != nullptr) {
sample_info->bytes += bytes();
@@ -517,14 +516,6 @@ Status VerticalMergeIteratorContext::_load_next_block() {
return st;
}
}
- if (UNLIKELY(_record_rowids)) {
-
RETURN_IF_ERROR(_iter->current_block_row_locations(&_block_row_locations));
- for (auto i = 0; i < _block_row_locations.size(); i++) {
- RowLocation& row_location = _block_row_locations[i];
- _block_row_locations[i] =
- RowLocation(_rowset_id, row_location.segment_id,
row_location.row_id);
- }
- }
} while (_block->rows() == 0);
_index_in_block = -1;
_valid = true;
@@ -536,9 +527,6 @@ Status VerticalHeapMergeIterator::next_batch(Block* block) {
size_t row_idx = 0;
VerticalMergeIteratorContext* pre_ctx = nullptr;
std::vector<RowSource> tmp_row_sources;
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations.resize(_block_row_max);
- }
while (_get_size(block) < _block_row_max) {
if (_merge_heap.empty()) {
VLOG_NOTICE << "_merge_heap empty";
@@ -569,9 +557,6 @@ Status VerticalHeapMergeIterator::next_batch(Block* block) {
}
pre_ctx = ctx;
}
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations[row_idx] = ctx->current_row_location();
- }
row_idx++;
if (ctx->is_cur_block_finished() || row_idx >= _block_row_max) {
// current block finished, ctx not advance
@@ -608,9 +593,6 @@ Status VerticalHeapMergeIterator::next_batch(Block* block) {
if (!_merge_heap.empty()) {
return Status::OK();
}
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations.resize(row_idx);
- }
return Status::EndOfFile("no more data in segment");
}
@@ -618,7 +600,6 @@ Status VerticalHeapMergeIterator::init(const
StorageReadOptions& opts,
CompactionSampleInfo* sample_info) {
DCHECK(_origin_iters.size() == _iterator_init_flags.size());
_opts.emplace(opts);
- _record_rowids = opts.record_rowids;
if (_origin_iters.empty()) {
return Status::OK();
}
@@ -628,7 +609,7 @@ Status VerticalHeapMergeIterator::init(const
StorageReadOptions& opts,
for (size_t seg_order = 0; seg_order < num_iters; ++seg_order) {
auto& iter = _origin_iters[seg_order];
auto ctx = std::make_unique<VerticalMergeIteratorContext>(
- std::move(iter), _rowset_ids[seg_order], seg_order,
_seq_col_idx, _context_stats,
+ std::move(iter), seg_order, _seq_col_idx, _context_stats,
opts.use_insert_order_when_same, _key_group_cluster_key_idxes);
_ori_iter_ctx.push_back(std::move(ctx));
}
@@ -662,9 +643,6 @@ Status VerticalHeapMergeIterator::init(const
StorageReadOptions& opts,
Status VerticalFifoMergeIterator::next_batch(Block* block) {
size_t row_idx = 0;
std::vector<RowSource> tmp_row_sources;
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations.resize(_block_row_max);
- }
while (_get_size(block) < _block_row_max) {
if (_cur_iter_ctx == nullptr) {
VLOG_NOTICE << "_merge_list empty";
@@ -675,9 +653,6 @@ Status VerticalFifoMergeIterator::next_batch(Block* block) {
// Fifo only for duplicate no key
_cur_iter_ctx->add_cur_batch();
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations[row_idx] =
_cur_iter_ctx->current_row_location();
- }
row_idx++;
if (_cur_iter_ctx->is_cur_block_finished() || row_idx >=
_block_row_max) {
// current block finished, ctx not advance
@@ -694,9 +669,9 @@ Status VerticalFifoMergeIterator::next_batch(Block* block) {
cur_order++) {
auto& next_iter = _origin_iters[cur_order];
std::unique_ptr<VerticalMergeIteratorContext> next_ctx(
- new VerticalMergeIteratorContext(
- std::move(next_iter), _rowset_ids[cur_order],
cur_order,
- _seq_col_idx, _context_stats,
_opts->use_insert_order_when_same));
+ new VerticalMergeIteratorContext(std::move(next_iter),
cur_order,
+ _seq_col_idx,
_context_stats,
+
_opts->use_insert_order_when_same));
RETURN_IF_ERROR(next_ctx->init(*_opts));
if (next_ctx->valid()) {
_cur_iter_ctx.swap(next_ctx);
@@ -710,9 +685,6 @@ Status VerticalFifoMergeIterator::next_batch(Block* block) {
if (_cur_iter_ctx) {
return Status::OK();
}
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations.resize(row_idx);
- }
return Status::EndOfFile("no more data in segment");
}
@@ -721,7 +693,6 @@ Status VerticalFifoMergeIterator::init(const
StorageReadOptions& opts,
DCHECK(_origin_iters.size() == _iterator_init_flags.size());
DCHECK(_keys_type == KeysType::DUP_KEYS);
_opts.emplace(opts);
- _record_rowids = opts.record_rowids;
if (_origin_iters.empty()) {
return Status::OK();
}
@@ -735,9 +706,9 @@ Status VerticalFifoMergeIterator::init(const
StorageReadOptions& opts,
// will not be pushed into heap, we should init next one util we find a
valid iter
// so this rowset can work in heap
for (auto& iter : _origin_iters) {
- std::unique_ptr<VerticalMergeIteratorContext> ctx(new
VerticalMergeIteratorContext(
- std::move(iter), _rowset_ids[seg_order], seg_order,
_seq_col_idx, _context_stats,
- opts.use_insert_order_when_same));
+ std::unique_ptr<VerticalMergeIteratorContext> ctx(
+ new VerticalMergeIteratorContext(std::move(iter), seg_order,
_seq_col_idx,
+ _context_stats,
opts.use_insert_order_when_same));
RETURN_IF_ERROR(ctx->init(opts, sample_info));
if (!ctx->valid()) {
++seg_order;
@@ -993,9 +964,8 @@ Status VerticalMaskMergeIterator::init(const
StorageReadOptions& opts,
return Status::OK();
}
_schema = &(*_origin_iters.begin())->schema();
- RowsetId rs_id;
for (auto& iter : _origin_iters) {
- auto ctx =
std::make_unique<VerticalMergeIteratorContext>(std::move(iter), rs_id, -1, -1,
+ auto ctx =
std::make_unique<VerticalMergeIteratorContext>(std::move(iter), -1, -1,
_context_stats);
_origin_iter_ctx.push_back(std::move(ctx));
}
@@ -1009,21 +979,21 @@ Status VerticalMaskMergeIterator::init(const
StorageReadOptions& opts,
// interfaces to create vertical merge iterator
std::shared_ptr<RowwiseIterator> new_vertical_heap_merge_iterator(
std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>&
iterator_init_flag,
- const std::vector<RowsetId>& rowset_ids, KeysType keys_type, int32_t
seq_col_idx,
- RowSourcesBuffer* row_sources, VerticalCompactionContextStats*
context_stats,
+ KeysType keys_type, int32_t seq_col_idx, RowSourcesBuffer* row_sources,
+ VerticalCompactionContextStats* context_stats,
std::vector<uint32_t> key_group_cluster_key_idxes) {
- return std::make_shared<VerticalHeapMergeIterator>(
- std::move(inputs), iterator_init_flag, rowset_ids, keys_type,
seq_col_idx, row_sources,
- context_stats, key_group_cluster_key_idxes);
+ return std::make_shared<VerticalHeapMergeIterator>(std::move(inputs),
iterator_init_flag,
+ keys_type, seq_col_idx,
row_sources,
+ context_stats,
key_group_cluster_key_idxes);
}
std::shared_ptr<RowwiseIterator> new_vertical_fifo_merge_iterator(
std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>&
iterator_init_flag,
- const std::vector<RowsetId>& rowset_ids, KeysType keys_type, int32_t
seq_col_idx,
- RowSourcesBuffer* row_sources, VerticalCompactionContextStats*
context_stats) {
+ KeysType keys_type, int32_t seq_col_idx, RowSourcesBuffer* row_sources,
+ VerticalCompactionContextStats* context_stats) {
return std::make_shared<VerticalFifoMergeIterator>(std::move(inputs),
iterator_init_flag,
- rowset_ids, keys_type,
seq_col_idx,
- row_sources,
context_stats);
+ keys_type, seq_col_idx,
row_sources,
+ context_stats);
}
std::shared_ptr<RowwiseIterator> new_vertical_mask_merge_iterator(
diff --git a/be/src/storage/iterator/vertical_merge_iterator.h
b/be/src/storage/iterator/vertical_merge_iterator.h
index c3b378ae3e6..2a8fb8289f7 100644
--- a/be/src/storage/iterator/vertical_merge_iterator.h
+++ b/be/src/storage/iterator/vertical_merge_iterator.h
@@ -35,7 +35,6 @@
#include "storage/iterators.h"
#include "storage/olap_common.h"
#include "storage/schema.h"
-#include "storage/utils.h"
#pragma once
@@ -162,12 +161,11 @@ private:
// takes ownership of rowwise iterator
class VerticalMergeIteratorContext {
public:
- VerticalMergeIteratorContext(RowwiseIteratorUPtr&& iter, RowsetId
rowset_id, uint32_t order,
- int32_t seq_col_idx,
VerticalCompactionContextStats* context_stats,
+ VerticalMergeIteratorContext(RowwiseIteratorUPtr&& iter, uint32_t order,
int32_t seq_col_idx,
+ VerticalCompactionContextStats* context_stats,
bool use_insert_order_when_same = false,
std::vector<uint32_t>
key_group_cluster_key_idxes = {})
: _iter(std::move(iter)),
- _rowset_id(rowset_id),
_order(order),
_seq_col_idx(seq_col_idx),
_num_key_columns(_iter->schema().num_key_columns()),
@@ -217,10 +215,6 @@ public:
ref->row_pos = _index_in_block;
}
bool inited() const { return _inited; }
- RowLocation current_row_location() {
- DCHECK(_record_rowids);
- return _block_row_locations[_index_in_block];
- }
size_t bytes() {
if (_block) {
@@ -257,7 +251,6 @@ private:
void _mark_inactive();
RowwiseIteratorUPtr _iter;
- RowsetId _rowset_id;
// segment order, used to compare key
const uint32_t _order = 0;
@@ -282,8 +275,6 @@ private:
// use to identify whether it's first block load from RowwiseIterator
bool _is_first_row = true;
bool _is_active_context_counted = false;
- bool _record_rowids = false;
- std::vector<RowLocation> _block_row_locations;
};
// --------------- VerticalHeapMergeIterator ------------- //
@@ -291,14 +282,12 @@ class VerticalHeapMergeIterator : public RowwiseIterator {
public:
// VerticalMergeIterator takes the ownership of input iterators
VerticalHeapMergeIterator(std::vector<RowwiseIteratorUPtr>&& iters,
- std::vector<bool> iterator_init_flags,
- std::vector<RowsetId> rowset_ids, KeysType
keys_type,
+ std::vector<bool> iterator_init_flags, KeysType
keys_type,
int32_t seq_col_idx, RowSourcesBuffer*
row_sources_buf,
VerticalCompactionContextStats* context_stats,
std::vector<uint32_t>
key_group_cluster_key_idxes)
: _origin_iters(std::move(iters)),
_iterator_init_flags(std::move(iterator_init_flags)),
- _rowset_ids(std::move(rowset_ids)),
_keys_type(keys_type),
_seq_col_idx(seq_col_idx),
_row_sources_buf(row_sources_buf),
@@ -313,11 +302,6 @@ public:
Status next_batch(Block* block) override;
const ReadSchema& schema() const override { return *_schema; }
uint64_t merged_rows() const override { return _merged_rows; }
- Status current_block_row_locations(std::vector<RowLocation>*
block_row_locations) override {
- DCHECK(_record_rowids);
- *block_row_locations = _block_row_locations;
- return Status::OK();
- }
private:
int64_t _get_size(Block* block) { return block->rows(); }
@@ -325,7 +309,6 @@ private:
// It will be released after '_merge_heap' has been built.
std::vector<RowwiseIteratorUPtr> _origin_iters;
std::vector<bool> _iterator_init_flags;
- std::vector<RowsetId> _rowset_ids;
const ReadSchema* _schema = nullptr;
@@ -351,8 +334,6 @@ private:
// StorageReadOptions is supplied by init(), after this iterator is
constructed, and must be
// retained for lazily initialized segment contexts.
std::optional<StorageReadOptions> _opts;
- bool _record_rowids = false;
- std::vector<RowLocation> _block_row_locations;
std::vector<uint32_t> _key_group_cluster_key_idxes;
};
@@ -361,13 +342,11 @@ class VerticalFifoMergeIterator : public RowwiseIterator {
public:
// VerticalFifoMergeIterator takes the ownership of input iterators
VerticalFifoMergeIterator(std::vector<RowwiseIteratorUPtr>&& iters,
- std::vector<bool> iterator_init_flags,
- std::vector<RowsetId> rowset_ids, KeysType
keys_type,
+ std::vector<bool> iterator_init_flags, KeysType
keys_type,
int32_t seq_col_idx, RowSourcesBuffer*
row_sources_buf,
VerticalCompactionContextStats* context_stats)
: _origin_iters(std::move(iters)),
_iterator_init_flags(std::move(iterator_init_flags)),
- _rowset_ids(std::move(rowset_ids)),
_keys_type(keys_type),
_seq_col_idx(seq_col_idx),
_row_sources_buf(row_sources_buf),
@@ -381,11 +360,6 @@ public:
Status next_batch(Block* block) override;
const ReadSchema& schema() const override { return *_schema; }
uint64_t merged_rows() const override { return _merged_rows; }
- Status current_block_row_locations(std::vector<RowLocation>*
block_row_locations) override {
- DCHECK(_record_rowids);
- *block_row_locations = _block_row_locations;
- return Status::OK();
- }
private:
int64_t _get_size(Block* block) { return block->rows(); }
@@ -393,7 +367,6 @@ private:
// It will be released after '_merge_heap' has been built.
std::vector<RowwiseIteratorUPtr> _origin_iters;
std::vector<bool> _iterator_init_flags;
- std::vector<RowsetId> _rowset_ids;
const ReadSchema* _schema = nullptr;
@@ -407,8 +380,6 @@ private:
// StorageReadOptions is supplied by init(), after this iterator is
constructed, and must be
// retained for lazily initialized segment contexts.
std::optional<StorageReadOptions> _opts;
- bool _record_rowids = false;
- std::vector<RowLocation> _block_row_locations;
};
// --------------- RowBatch for batch optimization ------------- //
@@ -481,14 +452,14 @@ private:
// segment merge iterator
std::shared_ptr<RowwiseIterator> new_vertical_heap_merge_iterator(
std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>&
iterator_init_flag,
- const std::vector<RowsetId>& rowset_ids, KeysType key_type, int32_t
seq_col_idx,
- RowSourcesBuffer* row_sources_buf, VerticalCompactionContextStats*
context_stats,
+ KeysType key_type, int32_t seq_col_idx, RowSourcesBuffer*
row_sources_buf,
+ VerticalCompactionContextStats* context_stats,
std::vector<uint32_t> key_group_cluster_key_idxes);
std::shared_ptr<RowwiseIterator> new_vertical_fifo_merge_iterator(
std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>&
iterator_init_flag,
- const std::vector<RowsetId>& rowset_ids, KeysType key_type, int32_t
seq_col_idx,
- RowSourcesBuffer* row_sources_buf, VerticalCompactionContextStats*
context_stats);
+ KeysType key_type, int32_t seq_col_idx, RowSourcesBuffer*
row_sources_buf,
+ VerticalCompactionContextStats* context_stats);
std::shared_ptr<RowwiseIterator> new_vertical_mask_merge_iterator(
std::vector<RowwiseIteratorUPtr>&& inputs, RowSourcesBuffer*
row_sources_buf,
diff --git a/be/src/storage/iterator/vgeneric_iterators.cpp
b/be/src/storage/iterator/vgeneric_iterators.cpp
index fa75c4bb638..b3c7358f418 100644
--- a/be/src/storage/iterator/vgeneric_iterators.cpp
+++ b/be/src/storage/iterator/vgeneric_iterators.cpp
@@ -277,7 +277,6 @@ Status VAutoIncrementIterator::init(const
StorageReadOptions& opts) {
Status VMergeIteratorContext::init(const StorageReadOptions& opts) {
_block_row_max = opts.block_row_max;
- _record_rowids = opts.record_rowids;
RETURN_IF_ERROR(_load_next_block());
if (valid()) {
RETURN_IF_ERROR(_validate_compare_contract(opts));
@@ -370,9 +369,6 @@ Status VMergeIteratorContext::_load_next_block() {
return st;
}
}
- if (UNLIKELY(_record_rowids)) {
-
RETURN_IF_ERROR(_iter->current_block_row_locations(&_block_row_locations));
- }
} while (_block->rows() == 0);
_index_in_block = -1;
_valid = true;
@@ -383,8 +379,6 @@ Status VMergeIterator::init(const StorageReadOptions& opts)
{
if (_origin_iters.empty()) {
return Status::OK();
}
- _record_rowids = opts.record_rowids;
-
for (auto& iter : _origin_iters) {
auto ctx = std::make_shared<VMergeIteratorContext>(
std::move(iter), _sequence_id_idx, _is_unique, _is_reverse,
@@ -423,8 +417,6 @@ public:
const ReadSchema& schema() const override { return *_read_schema; }
- Status current_block_row_locations(std::vector<RowLocation>* locations)
override;
-
void update_profile(RuntimeProfile* profile) override {
if (_cur_iter != nullptr) {
_cur_iter->update_profile(profile);
@@ -471,14 +463,6 @@ Status VUnionIterator::next_batch(Block* block) {
return Status::EndOfFile("End of VUnionIterator");
}
-Status VUnionIterator::current_block_row_locations(std::vector<RowLocation>*
locations) {
- if (!_cur_iter) {
- locations->clear();
- return Status::EndOfFile("End of VUnionIterator");
- }
- return _cur_iter->current_block_row_locations(locations);
-}
-
RowwiseIteratorUPtr new_merge_iterator(std::vector<RowwiseIteratorUPtr>&&
inputs,
int sequence_id_idx, bool is_unique,
bool is_reverse,
uint64_t* merged_rows, ReadSchemaSPtr
read_schema,
diff --git a/be/src/storage/iterator/vgeneric_iterators.h
b/be/src/storage/iterator/vgeneric_iterators.h
index 22012471067..9176f1daf9d 100644
--- a/be/src/storage/iterator/vgeneric_iterators.h
+++ b/be/src/storage/iterator/vgeneric_iterators.h
@@ -34,7 +34,6 @@
#include "storage/iterators.h"
#include "storage/schema.h"
#include "storage/segment/column_reader.h"
-#include "storage/utils.h"
namespace doris {
class RuntimeProfile;
@@ -123,11 +122,6 @@ public:
Status copy_rows(BlockWithSameBit* block, bool advanced = true);
Status copy_rows(BlockView* view, bool advanced = true);
- RowLocation current_row_location() {
- DCHECK(_record_rowids);
- return _block_row_locations[_index_in_block];
- }
-
// Advance internal row index to next valid row
// Return error if error happens
// Don't call this when valid() is false, action is undefined
@@ -189,8 +183,6 @@ private:
const ReadSchemaSPtr _read_schema;
int _num_key_columns;
std::vector<uint32_t>* _compare_columns;
- std::vector<RowLocation> _block_row_locations;
- bool _record_rowids = false;
size_t _cur_batch_num = 0;
// used to store data load from iterator->next_batch(Block*)
@@ -226,12 +218,6 @@ public:
const ReadSchema& schema() const override { return *_read_schema; }
- Status current_block_row_locations(std::vector<RowLocation>*
block_row_locations) override {
- DCHECK(_record_rowids);
- *block_row_locations = _block_row_locations;
- return Status::OK();
- }
-
void update_profile(RuntimeProfile* profile) override {
if (!_origin_iters.empty()) {
_origin_iters[0]->update_profile(profile);
@@ -247,9 +233,6 @@ private:
template <typename T>
Status _next_batch(T* block) {
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations.resize(_block_row_max);
- }
size_t row_idx = 0;
std::shared_ptr<VMergeIteratorContext> pre_ctx;
while (_get_size(block) < _block_row_max) {
@@ -269,9 +252,6 @@ private:
pre_ctx = ctx;
}
pre_ctx->set_pre_ctx_same(ctx.get());
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations[row_idx] =
ctx->current_row_location();
- }
row_idx++;
if (ctx->is_cur_block_finished() || row_idx >= _block_row_max)
{
// current block finished, ctx not advance
@@ -296,12 +276,6 @@ private:
if (!_merge_heap.empty()) {
return Status::OK();
}
- // Still last batch needs to be processed
-
- if (UNLIKELY(_record_rowids)) {
- _block_row_locations.resize(row_idx);
- }
-
return Status::EndOfFile("no more data in segment");
}
@@ -331,8 +305,6 @@ private:
// See VMergeIteratorContext::_small_seq_first; forwarded to every context
on init().
bool _small_seq_first = false;
uint64_t* _merged_rows = nullptr;
- bool _record_rowids = false;
- std::vector<RowLocation> _block_row_locations;
};
// Create a merge iterator for input iterators. Merge iterator will merge
diff --git a/be/src/storage/iterators.h b/be/src/storage/iterators.h
index 3c7cc40a1d0..5c369b685b2 100644
--- a/be/src/storage/iterators.h
+++ b/be/src/storage/iterators.h
@@ -130,7 +130,6 @@ public:
// The compaction output schema's variant path layout; null outside a
variant compaction.
VariantCompactionPathsSPtr variant_compaction_paths;
bool enable_unique_key_merge_on_write = false;
- bool record_rowids = false;
std::vector<int> topn_filter_source_node_ids;
// used for special optimization for query : ORDER BY key DESC LIMIT n
bool read_orderby_key_reverse = false;
@@ -243,11 +242,6 @@ public:
virtual bool is_merge_iterator() const { return false; }
- virtual Status current_block_row_locations(std::vector<RowLocation>*
block_row_locations) {
- return Status::InternalError("should not reach here, current class: " +
- demangle(typeid(*this).name()));
- }
-
// return schema for this Iterator
virtual const ReadSchema& schema() const = 0;
diff --git a/be/src/storage/merger.cpp b/be/src/storage/merger.cpp
index a743e90467e..b89ec47b064 100644
--- a/be/src/storage/merger.cpp
+++ b/be/src/storage/merger.cpp
@@ -23,6 +23,7 @@
#include <unistd.h>
#include <algorithm>
+#include <cstring>
#include <iterator>
#include <memory>
#include <mutex>
@@ -37,7 +38,9 @@
#include "common/config.h"
#include "common/logging.h"
#include "common/status.h"
+#include "core/assert_cast.h"
#include "core/block/block.h"
+#include "core/column/column_string.h"
#include "runtime/runtime_state.h"
#include "storage/iterator/block_reader.h"
#include "storage/iterator/vertical_block_reader.h"
@@ -96,6 +99,30 @@ Status set_key_range(TabletReader::ReaderParams*
reader_params, const TabletSche
return Status::OK();
}
+// Extract row locations from the rowid column and remove it from the output
block.
+Block prepare_compaction_output_block(const Block& block, int32_t
rowid_ordinal,
+ std::vector<RowLocation>* row_locations)
{
+ DORIS_CHECK_GE(rowid_ordinal, 0);
+ auto ordinal = static_cast<size_t>(rowid_ordinal);
+ DORIS_CHECK_LT(ordinal, block.columns());
+ const auto& rowid_column =
+ assert_cast<const
ColumnString&>(*block.get_by_position(ordinal).column);
+ DORIS_CHECK_EQ(rowid_column.size(), block.rows());
+ row_locations->clear();
+ row_locations->reserve(rowid_column.size());
+ for (size_t i = 0; i < rowid_column.size(); ++i) {
+ const auto rowid = rowid_column.get_data_at(i);
+ DORIS_CHECK_EQ(rowid.size, sizeof(RowLocation));
+ RowLocation location;
+ std::memcpy(&location, rowid.data, sizeof(RowLocation));
+ row_locations->emplace_back(location);
+ }
+
+ Block output_block = block;
+ output_block.erase(ordinal);
+ return output_block;
+}
+
} // namespace
Status Merger::vmerge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type,
@@ -148,16 +175,24 @@ Status Merger::vmerge_rowsets(BaseTabletSPtr tablet,
ReaderType reader_type,
reader_params.delete_bitmap =
tablet->tablet_meta()->delete_bitmap_ptr();
}
- if (stats_output && stats_output->rowid_conversion) {
- reader_params.record_rowids = true;
+ const bool convert_rowids =
+ stats_output != nullptr && stats_output->rowid_conversion !=
nullptr;
+ if (convert_rowids) {
+ DORIS_CHECK_NE(cur_tablet_schema.keys_type(), KeysType::AGG_KEYS);
+ DORIS_CHECK(!cur_tablet_schema.has_seq_map());
reader_params.rowid_conversion = stats_output->rowid_conversion;
stats_output->rowid_conversion->set_dst_rowset_id(dst_rowset_writer->rowset_id());
}
reader_params.read_schema =
std::make_shared<ReadSchema>(cur_tablet_schema.columns());
+ if (convert_rowids) {
+ reader_params.read_schema->append_compaction_rowid_column();
+ }
RETURN_IF_ERROR(reader.init(reader_params));
Block block = reader_params.read_schema->create_read_block();
+ const auto rowid_ordinal = reader_params.read_schema->rowid_ordinal();
+ std::vector<RowLocation> row_locations;
size_t output_rows = 0;
bool eof = false;
while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {
@@ -175,15 +210,23 @@ Status Merger::vmerge_rowsets(BaseTabletSPtr tablet,
ReaderType reader_type,
RETURN_NOT_OK_STATUS_WITH_WARN(reader.next_block_with_aggregation(&block, &eof),
"failed to read next block when merging
rowsets of tablet " +
std::to_string(tablet->tablet_id()));
- RETURN_NOT_OK_STATUS_WITH_WARN(dst_rowset_writer->add_block(&block),
- "failed to write block when merging
rowsets of tablet " +
-
std::to_string(tablet->tablet_id()));
-
- if (reader_params.record_rowids && block.rows() > 0) {
- std::vector<uint32_t> segment_num_rows;
-
RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));
- RETURN_IF_ERROR(stats_output->rowid_conversion->add(
- reader.current_block_row_locations(), segment_num_rows));
+ if (convert_rowids) {
+ Block output_block =
+ prepare_compaction_output_block(block, rowid_ordinal,
&row_locations);
+
RETURN_NOT_OK_STATUS_WITH_WARN(dst_rowset_writer->add_block(&output_block),
+ "failed to write block when merging
rowsets of tablet " +
+
std::to_string(tablet->tablet_id()));
+
+ if (block.rows() > 0) {
+ std::vector<uint32_t> segment_num_rows;
+
RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));
+ RETURN_IF_ERROR(
+ stats_output->rowid_conversion->add(row_locations,
segment_num_rows));
+ }
+ } else {
+
RETURN_NOT_OK_STATUS_WITH_WARN(dst_rowset_writer->add_block(&block),
+ "failed to write block when merging
rowsets of tablet " +
+
std::to_string(tablet->tablet_id()));
}
output_rows += block.rows();
@@ -344,18 +387,26 @@ Status Merger::vertical_compact_one_group(
reader_params.delete_bitmap =
tablet->tablet_meta()->delete_bitmap_ptr();
}
- if (is_key && stats_output && stats_output->rowid_conversion) {
- reader_params.record_rowids = true;
+ const bool convert_rowids =
+ is_key && stats_output != nullptr &&
stats_output->rowid_conversion != nullptr;
+ if (convert_rowids) {
+ DORIS_CHECK_NE(tablet_schema.keys_type(), KeysType::AGG_KEYS);
+ DORIS_CHECK(!tablet_schema.has_seq_map());
reader_params.rowid_conversion = stats_output->rowid_conversion;
stats_output->rowid_conversion->set_dst_rowset_id(dst_rowset_writer->rowset_id());
}
reader_params.read_schema = std::make_shared<ReadSchema>(
project_columns_by_ordinal(tablet_schema.columns(), column_group));
+ if (convert_rowids) {
+ reader_params.read_schema->append_compaction_rowid_column();
+ }
reader_params.batch_size = batch_size;
RETURN_IF_ERROR(reader.init(reader_params, sample_info));
Block block = reader_params.read_schema->create_read_block();
+ const auto rowid_ordinal = reader_params.read_schema->rowid_ordinal();
+ std::vector<RowLocation> row_locations;
size_t output_rows = 0;
bool eof = false;
while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {
@@ -372,17 +423,27 @@ Status Merger::vertical_compact_one_group(
RETURN_NOT_OK_STATUS_WITH_WARN(reader.next_block_with_aggregation(&block, &eof),
"failed to read next block when merging
rowsets of tablet " +
std::to_string(tablet->tablet_id()));
- RETURN_NOT_OK_STATUS_WITH_WARN(
- dst_rowset_writer->add_columns(&block, column_group, is_key,
max_rows_per_segment,
- has_cluster_key),
- "failed to write block when merging rowsets of tablet " +
- std::to_string(tablet->tablet_id()));
-
- if (is_key && reader_params.record_rowids && block.rows() > 0) {
- std::vector<uint32_t> segment_num_rows;
-
RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));
- RETURN_IF_ERROR(stats_output->rowid_conversion->add(
- reader.current_block_row_locations(), segment_num_rows));
+ if (convert_rowids) {
+ Block output_block =
+ prepare_compaction_output_block(block, rowid_ordinal,
&row_locations);
+ RETURN_NOT_OK_STATUS_WITH_WARN(
+ dst_rowset_writer->add_columns(&output_block,
column_group, is_key,
+ max_rows_per_segment,
has_cluster_key),
+ "failed to write block when merging rowsets of tablet " +
+ std::to_string(tablet->tablet_id()));
+
+ if (block.rows() > 0) {
+ std::vector<uint32_t> segment_num_rows;
+
RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));
+ RETURN_IF_ERROR(
+ stats_output->rowid_conversion->add(row_locations,
segment_num_rows));
+ }
+ } else {
+ RETURN_NOT_OK_STATUS_WITH_WARN(
+ dst_rowset_writer->add_columns(&block, column_group,
is_key,
+ max_rows_per_segment,
has_cluster_key),
+ "failed to write block when merging rowsets of tablet " +
+ std::to_string(tablet->tablet_id()));
}
output_rows += block.rows();
block.clear_column_data();
@@ -420,8 +481,10 @@ Status Merger::vertical_compact_one_group(
RowSourcesBuffer* row_source_buf, VerticalBlockReader&
src_block_reader,
segment_v2::VerticalSegmentWriter& dst_segment_writer, Statistics*
stats_output,
uint64_t* index_size, KeyBoundsPB& key_bounds, SimpleRowIdConversion*
rowid_conversion) {
- // TODO: record_rowids
Block block = read_schema.create_read_block();
+ const auto rowid_ordinal = read_schema.rowid_ordinal();
+ const bool convert_rowids = is_key && rowid_conversion != nullptr;
+ std::vector<RowLocation> row_locations;
size_t output_rows = 0;
bool eof = false;
while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {
@@ -432,12 +495,18 @@ Status Merger::vertical_compact_one_group(
if (!block.rows()) {
break;
}
- RETURN_NOT_OK_STATUS_WITH_WARN(dst_segment_writer.append_block(&block,
0, block.rows()),
- "failed to write block when merging
rowsets of tablet " +
- std::to_string(tablet_id));
-
- if (is_key && rowid_conversion != nullptr) {
-
rowid_conversion->add(src_block_reader.current_block_row_locations());
+ if (convert_rowids) {
+ Block output_block =
+ prepare_compaction_output_block(block, rowid_ordinal,
&row_locations);
+ RETURN_NOT_OK_STATUS_WITH_WARN(
+ dst_segment_writer.append_block(&output_block, 0,
output_block.rows()),
+ "failed to write block when merging rowsets of tablet " +
+ std::to_string(tablet_id));
+ rowid_conversion->add(row_locations);
+ } else {
+
RETURN_NOT_OK_STATUS_WITH_WARN(dst_segment_writer.append_block(&block, 0,
block.rows()),
+ "failed to write block when merging
rowsets of tablet " +
+ std::to_string(tablet_id));
}
output_rows += block.rows();
block.clear_column_data();
diff --git a/be/src/storage/rowset/beta_rowset_reader.cpp
b/be/src/storage/rowset/beta_rowset_reader.cpp
index 4bc3bce0b05..bed808d461c 100644
--- a/be/src/storage/rowset/beta_rowset_reader.cpp
+++ b/be/src/storage/rowset/beta_rowset_reader.cpp
@@ -203,7 +203,6 @@ Status
BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
_read_options.variant_compaction_paths =
_read_context->variant_compaction_paths;
_read_options.enable_unique_key_merge_on_write =
_read_context->enable_unique_key_merge_on_write;
- _read_options.record_rowids = _read_context->record_rowids;
_read_options.topn_filter_source_node_ids =
_read_context->topn_filter_source_node_ids;
_read_options.read_orderby_key_reverse =
_read_context->read_orderby_key_reverse;
_read_options.use_insert_order_when_same =
_read_context->use_insert_order_when_same;
@@ -268,7 +267,8 @@ Status
BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
seg_start = 0;
seg_end = segment_count;
}
- if (_read_context->record_rowids && _read_context->rowid_conversion) {
+ if (_read_context->rowid_conversion != nullptr) {
+ DORIS_CHECK_GE(read_schema->rowid_ordinal(), 0);
// init segment rowid map for rowid conversion
std::vector<uint32_t> segment_rows;
RETURN_IF_ERROR(_rowset->get_segment_num_rows(&segment_rows,
should_use_cache, _stats));
diff --git a/be/src/storage/rowset/beta_rowset_reader.h
b/be/src/storage/rowset/beta_rowset_reader.h
index c223a62089d..cf490f1c4b9 100644
--- a/be/src/storage/rowset/beta_rowset_reader.h
+++ b/be/src/storage/rowset/beta_rowset_reader.h
@@ -38,7 +38,6 @@
namespace doris {
class RuntimeProfile;
class ReadSchema;
-struct RowLocation;
struct RowsetReaderContext;
class BetaRowsetReader : public RowsetReader {
@@ -85,10 +84,6 @@ public:
RowsetTypePB type() const override { return RowsetTypePB::BETA_ROWSET; }
- Status current_block_row_locations(std::vector<RowLocation>* locations)
override {
- return _iterator->current_block_row_locations(locations);
- }
-
void update_profile(RuntimeProfile* profile) override;
RowsetReaderSharedPtr clone() override;
diff --git a/be/src/storage/rowset/rowset_reader.h
b/be/src/storage/rowset/rowset_reader.h
index 559aec4a52b..8d49278d78d 100644
--- a/be/src/storage/rowset/rowset_reader.h
+++ b/be/src/storage/rowset/rowset_reader.h
@@ -80,9 +80,6 @@ public:
virtual RowsetTypePB type() const = 0;
virtual int64_t newest_write_timestamp() = 0;
- virtual Status current_block_row_locations(std::vector<RowLocation>*
locations) {
- return Status::NotSupported("to be implemented");
- }
virtual void update_profile(RuntimeProfile* profile) = 0;
diff --git a/be/src/storage/rowset/rowset_reader_context.h
b/be/src/storage/rowset/rowset_reader_context.h
index 8bf7dfc2a0f..17b5938a680 100644
--- a/be/src/storage/rowset/rowset_reader_context.h
+++ b/be/src/storage/rowset/rowset_reader_context.h
@@ -87,7 +87,6 @@ struct RowsetReaderContext {
// for unique key merge on write
bool enable_unique_key_merge_on_write = false;
DeleteBitmapPtr delete_bitmap = nullptr;
- bool record_rowids = false;
RowIdConversion* rowid_conversion = nullptr;
bool is_key_column_group = false;
// column unique ids referenced by output/orderby exprs (NOT ordinals)
diff --git a/be/src/storage/rowset/segcompaction.cpp
b/be/src/storage/rowset/segcompaction.cpp
index 6670dd4b34a..116e7b10f2e 100644
--- a/be/src/storage/rowset/segcompaction.cpp
+++ b/be/src/storage/rowset/segcompaction.cpp
@@ -88,10 +88,9 @@ Status SegcompactionWorker::_get_segcompaction_reader(
bool is_key, std::vector<uint32_t>& key_group_cluster_key_idxes,
std::unique_ptr<VerticalBlockReader>* reader) {
const auto& ctx = _writer->_context;
- bool record_rowids = need_convert_delete_bitmap() && is_key;
StorageReadOptions read_options(stat);
read_options.use_page_cache = false;
- read_options.record_rowids = record_rowids;
+ read_options.rowset_id = ctx.rowset_id;
if (!tablet->tablet_schema()->cluster_key_uids().empty()) {
DeleteBitmapPtr delete_bitmap =
std::make_shared<DeleteBitmap>(tablet->tablet_id());
RETURN_IF_ERROR(tablet->calc_delete_bitmap_between_segments(
@@ -117,7 +116,7 @@ Status SegcompactionWorker::_get_segcompaction_reader(
seg_iterators.push_back(std::move(iter));
segment_rows.emplace(seg_ptr->id(), seg_ptr->num_rows());
}
- if (record_rowids && _rowid_conversion != nullptr) {
+ if (is_key && _rowid_conversion != nullptr) {
_rowid_conversion->reset_segment_map(segment_rows);
}
@@ -132,7 +131,6 @@ Status SegcompactionWorker::_get_segcompaction_reader(
reader_params.read_schema = std::move(read_schema);
reader_params.is_key_column_group = is_key;
reader_params.use_page_cache = false;
- reader_params.record_rowids = record_rowids;
reader_params.key_group_cluster_key_idxes = key_group_cluster_key_idxes;
return (*reader)->init(reader_params, nullptr);
}
@@ -303,6 +301,9 @@ Status
SegcompactionWorker::_do_compact_segments(SegCompactionCandidatesSharedPt
RETURN_IF_ERROR(writer->init(column_ids, is_key));
auto schema = std::make_shared<ReadSchema>(
project_columns_by_ordinal(ctx.tablet_schema->columns(),
column_ids));
+ if (is_key && _rowid_conversion != nullptr) {
+ schema->append_compaction_rowid_column();
+ }
OlapReaderStatistics reader_stats;
std::unique_ptr<VerticalBlockReader> reader;
auto s = _get_segcompaction_reader(segments, tablet, schema,
reader_stats, row_sources_buf,
diff --git a/be/src/storage/schema.cpp b/be/src/storage/schema.cpp
index 5c4e0901ff6..642430b0de5 100644
--- a/be/src/storage/schema.cpp
+++ b/be/src/storage/schema.cpp
@@ -17,6 +17,7 @@
#include "storage/schema.h"
+#include <limits>
#include <ranges>
#include <utility>
@@ -181,6 +182,19 @@ void
ReadSchema::append_dropped_columns(std::vector<TabletColumn> columns) {
}
}
+void ReadSchema::append_compaction_rowid_column() {
+ DORIS_CHECK_EQ(_num_block_columns, _read_columns.size());
+ DORIS_CHECK_EQ(_rowid_ordinal, -1);
+ auto column = std::make_shared<TabletColumn>(
+ FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
FieldType::OLAP_FIELD_TYPE_STRING,
+ false, std::numeric_limits<int32_t>::max(), sizeof(RowLocation));
+ column->set_name(BeConsts::COMPACTION_ROWID_COL);
+ _read_columns.emplace_back(column);
+ _read_types.emplace_back(column->get_vec_type());
+ ++_num_block_columns;
+ _init_descriptors();
+}
+
void ReadSchema::_init_before_column_ordinals() {
std::unordered_map<std::string_view, ColumnId> name_to_ordinal;
name_to_ordinal.reserve(_num_block_columns);
diff --git a/be/src/storage/schema.h b/be/src/storage/schema.h
index e5f9397eb2f..21b6a614b2c 100644
--- a/be/src/storage/schema.h
+++ b/be/src/storage/schema.h
@@ -85,6 +85,10 @@ public:
// Append private historical delete-predicate columns without extending
the Block prefix.
void append_dropped_columns(std::vector<TabletColumn> columns);
+ // Append COMPACTION_ROWID_COL to the Block prefix before any historical
delete columns.
+ // It carries each source row's physical location through compaction
filtering and merge.
+ void append_compaction_rowid_column();
+
// Create caller-visible Blocks from the FE-slot prefix.
Block create_read_block() const;
@@ -143,7 +147,7 @@ public:
int32_t delete_sign_ordinal() const { return _delete_sign_ordinal; }
// Sequence column used to choose the winning row during merge.
int32_t sequence_ordinal() const { return _sequence_ordinal; }
- // Synthetic row identifier returned by rowid-producing scans.
+ // Synthetic row identifier returned by queries or carried by compaction
reads.
int32_t rowid_ordinal() const { return _rowid_ordinal; }
// Rowset version synthesized for single-version reads.
int32_t version_ordinal() const { return _version_ordinal; }
@@ -199,7 +203,8 @@ private:
if (col.name() == SEQUENCE_COL) {
_sequence_ordinal = i;
}
- if (col.name().starts_with(BeConsts::GLOBAL_ROWID_COL)) {
+ if (col.name().starts_with(BeConsts::GLOBAL_ROWID_COL) ||
+ col.name() == BeConsts::COMPACTION_ROWID_COL) {
_rowid_ordinal = i;
}
if (col.name() == VERSION_COL) {
diff --git a/be/src/storage/segment/column_reader.h
b/be/src/storage/segment/column_reader.h
index dfbd236c3d3..1b3e0709b39 100644
--- a/be/src/storage/segment/column_reader.h
+++ b/be/src/storage/segment/column_reader.h
@@ -926,6 +926,49 @@ private:
Status _seek_by_offsets(ordinal_t ord);
};
+// Reads COMPACTION_ROWID_COL by synthesizing the physical source RowLocation
during compaction.
+// Its payload is RowLocation, distinct from the query-facing
GlobalRowLoacationV2 format.
+class CompactionRowIdColumnIterator : public ColumnIterator {
+public:
+ CompactionRowIdColumnIterator(RowsetId rowset_id, uint32_t segment_id)
+ : _rowset_id(rowset_id), _segment_id(segment_id) {}
+
+ Status seek_to_ordinal(ordinal_t ord_idx) override {
+ _current_rowid = cast_set<uint32_t>(ord_idx);
+ return Status::OK();
+ }
+
+ Status next_batch(size_t* n, MutableColumnPtr& dst) {
+ bool has_null;
+ return next_batch(n, dst, &has_null);
+ }
+
+ Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null)
override {
+ for (size_t i = 0; i < *n; ++i) {
+ RowLocation location(_rowset_id, _segment_id,
cast_set<uint32_t>(_current_rowid + i));
+ dst->insert_data(reinterpret_cast<const char*>(&location),
sizeof(RowLocation));
+ }
+ _current_rowid += *n;
+ return Status::OK();
+ }
+
+ Status read_by_rowids(const rowid_t* rowids, const size_t count,
+ MutableColumnPtr& dst) override {
+ for (size_t i = 0; i < count; ++i) {
+ RowLocation location(_rowset_id, _segment_id,
cast_set<uint32_t>(rowids[i]));
+ dst->insert_data(reinterpret_cast<const char*>(&location),
sizeof(RowLocation));
+ }
+ return Status::OK();
+ }
+
+ ordinal_t get_current_ordinal() const override { return _current_rowid; }
+
+private:
+ uint32_t _current_rowid = 0;
+ RowsetId _rowset_id;
+ uint32_t _segment_id = 0;
+};
+
class RowIdColumnIteratorV2 : public ColumnIterator {
public:
RowIdColumnIteratorV2(uint8_t version, int64_t backend_id, uint32_t
file_id)
diff --git a/be/src/storage/segment/count_on_index_fastpath.h
b/be/src/storage/segment/count_on_index_fastpath.h
index 0c370d800c2..4e115aee6c3 100644
--- a/be/src/storage/segment/count_on_index_fastpath.h
+++ b/be/src/storage/segment/count_on_index_fastpath.h
@@ -38,7 +38,7 @@
// the per-segment delete bitmap AFTER the index apply, and delete
// predicates filter later -- a fabricated id range cannot participate in
// either subtraction),
-// 4. nothing consumes REAL row ids (rowid recording, ANN topn, BM25
+// 4. nothing consumes REAL row ids (ANN topn, BM25
// scoring, virtual columns), and
// 5. rows are emitted as defaults without reading column data at the
// fabricated ids (the COUNT_ON_INDEX no-read-data contract must be
@@ -66,7 +66,6 @@ struct CountOnIndexFastpathFacts {
bool has_external_row_ranges = false;
bool row_bitmap_is_full = false;
// (4) consumers of real row ids.
- bool record_rowids = false;
bool has_ann_topn = false;
bool has_score_runtime = false;
// (5) rows must be emitted as defaults (no data read at fabricated ids).
@@ -78,9 +77,8 @@ inline bool count_on_index_fastpath_safe(const
CountOnIndexFastpathFacts& f) {
return f.is_count_on_index_agg && !f.has_column_predicates &&
f.common_expr_count == 1 &&
f.single_expr_is_match_pred && !f.has_virtual_column_exprs &&
!f.has_delete_predicates &&
f.segment_delete_bitmap_empty && !f.has_col_id_predicates &&
!f.has_topn_filters &&
- !f.has_external_row_ranges && f.row_bitmap_is_full &&
!f.record_rowids &&
- !f.has_ann_topn && !f.has_score_runtime &&
f.no_need_read_data_opt_enabled &&
- f.keys_type_supported;
+ !f.has_external_row_ranges && f.row_bitmap_is_full &&
!f.has_ann_topn &&
+ !f.has_score_runtime && f.no_need_read_data_opt_enabled &&
f.keys_type_supported;
}
// G03 count-emission shortcut guard (functional core, unit-testable without a
@@ -100,8 +98,7 @@ inline bool count_on_index_fastpath_safe(const
CountOnIndexFastpathFacts& f) {
// 2. no evaluation stage survives (vec/short-circuit/expr eval, leftover
// column predicates or common exprs, delete predicates, lazy
// materialization),
-// 3. nothing consumes real row ids or per-row values (virtual columns,
-// rowid recording),
+// 3. nothing consumes real row ids or per-row values (virtual columns),
// 4. batch accounting is a pure countdown (no read limit, no reverse
// key-ordered read, no condition-cache writes), and
// 5. the block is exactly the read schema and EVERY column would take a
@@ -125,7 +122,6 @@ struct CountEmitShortcutFacts {
bool has_delete_predicates = false;
// (3) consumers of real row ids / per-row values.
bool has_virtual_columns = false;
- bool record_rowids = false;
// (4) batch accounting must be a pure countdown.
bool has_read_limit = false;
bool read_orderby_key_reverse = false;
@@ -138,8 +134,8 @@ struct CountEmitShortcutFacts {
inline bool count_emit_shortcut_safe(const CountEmitShortcutFacts& f) {
return f.count_fastpath_hit && !f.needs_vec_eval && !f.needs_short_eval &&
!f.needs_expr_eval &&
!f.has_remaining_col_predicates && !f.has_remaining_common_exprs &&
- !f.has_delete_predicates && !f.has_virtual_columns &&
!f.record_rowids &&
- !f.has_read_limit && !f.read_orderby_key_reverse &&
!f.has_condition_cache_digest &&
+ !f.has_delete_predicates && !f.has_virtual_columns &&
!f.has_read_limit &&
+ !f.read_orderby_key_reverse && !f.has_condition_cache_digest &&
f.block_shape_matches_schema && f.all_columns_emit_defaults;
}
diff --git a/be/src/storage/segment/lazy_init_segment_iterator.h
b/be/src/storage/segment/lazy_init_segment_iterator.h
index 9d2a89039b0..381f71162ba 100644
--- a/be/src/storage/segment/lazy_init_segment_iterator.h
+++ b/be/src/storage/segment/lazy_init_segment_iterator.h
@@ -50,10 +50,6 @@ public:
const ReadSchema& schema() const override { return *_schema; }
- Status current_block_row_locations(std::vector<RowLocation>* locations)
override {
- return _inner_iterator->current_block_row_locations(locations);
- }
-
void update_profile(RuntimeProfile* profile) override {
if (_inner_iterator != nullptr) {
_inner_iterator->update_profile(profile);
diff --git a/be/src/storage/segment/segment_iterator.cpp
b/be/src/storage/segment/segment_iterator.cpp
index e2ec36a0804..f5895aac905 100644
--- a/be/src/storage/segment/segment_iterator.cpp
+++ b/be/src/storage/segment/segment_iterator.cpp
@@ -1551,7 +1551,6 @@ bool SegmentIterator::_count_on_index_fastpath_safe()
const {
// ranges): the fabricated [0, df) range only counts correctly against a
// full [0, num_rows) bitmap.
facts.row_bitmap_is_full = _row_bitmap.cardinality() ==
uint64_t(num_rows());
- facts.record_rowids = _opts.record_rowids;
facts.has_ann_topn = _opts.ann_topn_runtime != nullptr;
facts.has_score_runtime = _score_runtime != nullptr;
// Mirror of the _need_read_data preamble: rows must be emitted as
defaults,
@@ -1611,7 +1610,6 @@ bool
SegmentIterator::_should_engage_count_emit_shortcut(const Block* block) {
facts.has_delete_predicates = _opts.delete_condition_predicates != nullptr
&&
_opts.delete_condition_predicates->num_of_column_predicate() > 0;
facts.has_virtual_columns = !_virtual_column_exprs.empty();
- facts.record_rowids = _opts.record_rowids || _schema->rowid_ordinal() >= 0;
facts.has_read_limit = _opts.read_limit > 0;
facts.read_orderby_key_reverse = _opts.read_orderby_key_reverse;
facts.has_condition_cache_digest = _opts.condition_cache_digest != 0;
@@ -1889,6 +1887,13 @@ Status SegmentIterator::_init_column_iterators() {
if (cid >= _schema->num_block_columns() &&
!_column_states[cid].has_delete_pred) {
continue;
}
+ if (_schema->column(cid)->name() == BeConsts::COMPACTION_ROWID_COL) {
+ // Compaction reads synthesize source locations rather than
reading a stored column.
+ _column_iterators[cid].reset(
+ new CompactionRowIdColumnIterator(_opts.rowset_id,
_segment->id()));
+ continue;
+ }
+
if
(_schema->column(cid)->name().starts_with(BeConsts::GLOBAL_ROWID_COL)) {
auto& id_file_map = _opts.runtime_state->get_id_file_map();
uint32_t file_id =
id_file_map->get_file_mapping_id(std::make_shared<FileMapping>(
@@ -3404,23 +3409,6 @@ void
SegmentIterator::_convert_dict_code_for_predicate_if_necessary_impl(
}
}
-Status SegmentIterator::current_block_row_locations(std::vector<RowLocation>*
block_row_locations) {
- DCHECK(_opts.record_rowids);
- DCHECK_GE(_block_rowids.size(), _selected_size);
- block_row_locations->resize(_selected_size);
- uint32_t sid = segment_id();
- if (!_is_need_vec_eval && !_is_need_short_eval && !_is_need_expr_eval) {
- for (auto i = 0; i < _selected_size; i++) {
- (*block_row_locations)[i] = RowLocation(sid, _block_rowids[i]);
- }
- } else {
- for (auto i = 0; i < _selected_size; i++) {
- (*block_row_locations)[i] = RowLocation(sid,
_block_rowids[_sel_rowid_idx[i]]);
- }
- }
- return Status::OK();
-}
-
Status SegmentIterator::_construct_compound_expr_context() {
ColumnIteratorOptions iter_opts {
.use_page_cache = _opts.use_page_cache,
diff --git a/be/src/storage/segment/segment_iterator.h
b/be/src/storage/segment/segment_iterator.h
index c693c93dee6..1366dbd07a8 100644
--- a/be/src/storage/segment/segment_iterator.h
+++ b/be/src/storage/segment/segment_iterator.h
@@ -61,7 +61,6 @@ namespace doris {
class VExpr;
class VExprContext;
-struct RowLocation;
namespace segment_v2 {
@@ -81,12 +80,6 @@ public:
[[nodiscard]] Status init(const StorageReadOptions& opts) override;
[[nodiscard]] Status next_batch(Block* block) override;
- // Get current block row locations. This function should be called
- // after the `next_batch` function.
- // Only vectorized version is supported.
- [[nodiscard]] Status current_block_row_locations(
- std::vector<RowLocation>* block_row_locations) override;
-
const ReadSchema& schema() const override { return *_schema; }
uint64_t data_id() const override { return _segment->id(); }
@@ -444,8 +437,9 @@ private:
io::FileReaderSPtr _file_reader;
- // used for compaction, record selectd rowids of current batch
+ // Number of selected rows in the current batch.
uint16_t _selected_size;
+ // Positions of selected rows within the current batch.
std::vector<uint16_t> _sel_rowid_idx;
// Rows already produced by this iterator. Used together with
diff --git a/be/src/storage/tablet/tablet_reader.cpp
b/be/src/storage/tablet/tablet_reader.cpp
index 295c44650d8..1b78fe010e6 100644
--- a/be/src/storage/tablet/tablet_reader.cpp
+++ b/be/src/storage/tablet/tablet_reader.cpp
@@ -184,7 +184,6 @@ Status TabletReader::_capture_rs_readers(const
ReaderParams& read_params) {
_reader_context.enable_unique_key_merge_on_write =
tablet()->enable_unique_key_merge_on_write();
_reader_context.enable_mor_value_predicate_pushdown =
read_params.enable_mor_value_predicate_pushdown;
- _reader_context.record_rowids = read_params.record_rowids;
_reader_context.rowid_conversion = read_params.rowid_conversion;
_reader_context.is_key_column_group = read_params.is_key_column_group;
_reader_context.common_expr_ctxs_push_down =
read_params.common_expr_ctxs_push_down;
diff --git a/be/src/storage/tablet/tablet_reader.h
b/be/src/storage/tablet/tablet_reader.h
index 82f3a707fdb..90434934ea2 100644
--- a/be/src/storage/tablet/tablet_reader.h
+++ b/be/src/storage/tablet/tablet_reader.h
@@ -177,8 +177,7 @@ public:
TPushAggOp::type push_down_agg_type_opt = TPushAggOp::NONE;
VExprContextSPtrs common_expr_ctxs_push_down;
- // used for compaction to record row ids
- bool record_rowids = false;
+ // Used by compaction to initialize source segment row-id mappings.
RowIdConversion* rowid_conversion = nullptr;
std::vector<int> topn_filter_source_node_ids;
// used for special optimization for query : ORDER BY key LIMIT n
diff --git a/be/test/exec/scan/vgeneric_iterators_test.cpp
b/be/test/exec/scan/vgeneric_iterators_test.cpp
index c5d3b819d39..06a59def649 100644
--- a/be/test/exec/scan/vgeneric_iterators_test.cpp
+++ b/be/test/exec/scan/vgeneric_iterators_test.cpp
@@ -20,6 +20,7 @@
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
+#include <cstring>
#include <memory>
#include <vector>
@@ -569,6 +570,10 @@ public:
*(double*)data = j == _seq_col_idx ?
_seq_col_rows_returned : 1;
data_len = sizeof(double);
break;
+ case FieldType::OLAP_FIELD_TYPE_STRING:
+ std::memcpy(data, &_seq_col_rows_returned,
sizeof(_seq_col_rows_returned));
+ data_len = sizeof(_seq_col_rows_returned);
+ break;
default:
break;
}
@@ -597,6 +602,7 @@ public:
TEST(VGenericIteratorsTest, MergeWithSeqColumn) {
EXPECT_TRUE(1);
auto schema = create_schema();
+ schema.append_compaction_rowid_column();
auto output_schema = std::make_shared<ReadSchema>(schema);
std::vector<RowwiseIteratorUPtr> inputs;
@@ -637,6 +643,12 @@ TEST(VGenericIteratorsTest, MergeWithSeqColumn) {
auto seq_col = block.get_by_position(seq_column_id).column;
size_t actual_value = (*seq_col)[0].get<TYPE_BIGINT>();
EXPECT_EQ(seg_iter_num - 1, actual_value);
+
+ const auto rowid =
block.get_by_position(schema.rowid_ordinal()).column->get_data_at(0);
+ ASSERT_EQ(sizeof(int), rowid.size);
+ int rowid_value = 0;
+ std::memcpy(&rowid_value, rowid.data, sizeof(rowid_value));
+ EXPECT_EQ(seg_iter_num - 1, rowid_value);
}
// Mirror of MergeWithSeqColumn but with small_seq_first=true.
diff --git a/be/test/storage/compaction/vertical_compaction_test.cpp
b/be/test/storage/compaction/vertical_compaction_test.cpp
index 70706ebfa5f..353783958c4 100644
--- a/be/test/storage/compaction/vertical_compaction_test.cpp
+++ b/be/test/storage/compaction/vertical_compaction_test.cpp
@@ -1592,8 +1592,6 @@ TEST_F(VerticalCompactionTest, TestAggKeyVerticalMerge) {
// merge input rowset
TabletSharedPtr tablet = create_tablet(*tablet_schema, false);
Merger::Statistics stats;
- RowIdConversion rowid_conversion;
- stats.rowid_conversion = &rowid_conversion;
ASSERT_EQ("0",
bvar::Variable::describe_exposed("vertical_compaction_active_segment_contexts"));
auto s = Merger::vertical_merge_rowsets(tablet,
ReaderType::READER_BASE_COMPACTION,
*tablet_schema, input_rs_readers,
diff --git a/be/test/storage/iterator/block_reader_change_next_block_test.cpp
b/be/test/storage/iterator/block_reader_change_next_block_test.cpp
index 1a9fbcbb1ee..8292848b53b 100644
--- a/be/test/storage/iterator/block_reader_change_next_block_test.cpp
+++ b/be/test/storage/iterator/block_reader_change_next_block_test.cpp
@@ -320,10 +320,6 @@ public:
Status next(Block* /*block*/) override { return
Status::Error<END_OF_FILE>(""); }
- RowLocation current_row_location() override { return {}; }
- Status current_block_row_locations(std::vector<RowLocation>* /*loc*/)
override {
- return Status::OK();
- }
Status ensure_first_row_ref() override { return Status::OK(); }
void update_profile(RuntimeProfile* /*profile*/) override {}
diff --git a/be/test/storage/read_schema_test.cpp
b/be/test/storage/read_schema_test.cpp
index 6cde75ad573..c9ca5428852 100644
--- a/be/test/storage/read_schema_test.cpp
+++ b/be/test/storage/read_schema_test.cpp
@@ -462,4 +462,22 @@ TEST(ReadSchemaTest, TabletColumnFromSlotKeepsTypeDetails)
{
EXPECT_TRUE(variant_column.variant_enable_doc_mode());
}
+TEST(ReadSchemaTest,
AppendedCompactionRowIdExtendsReadBlockBeforeDroppedColumns) {
+ std::vector<TabletColumnPtr> read_columns {create_int_column(10, "k",
true)};
+ auto dropped_column = create_int_column(11, "dropped");
+ ReadSchema read_schema(std::move(read_columns));
+
+ read_schema.append_compaction_rowid_column();
+ read_schema.append_dropped_columns({*dropped_column});
+
+ EXPECT_EQ(1, read_schema.rowid_ordinal());
+ EXPECT_EQ(2, read_schema.num_block_columns());
+ EXPECT_EQ(3, read_schema.num_read_columns());
+ EXPECT_EQ(2, read_schema.ordinal_by_uid(11));
+ Block block = read_schema.create_read_block();
+ ASSERT_EQ(2, block.columns());
+ EXPECT_EQ(BeConsts::COMPACTION_ROWID_COL, block.get_by_position(1).name);
+ EXPECT_EQ(PrimitiveType::TYPE_STRING,
block.get_by_position(1).type->get_primitive_type());
+}
+
} // namespace doris
diff --git a/be/test/storage/segment/count_on_index_fastpath_test.cpp
b/be/test/storage/segment/count_on_index_fastpath_test.cpp
index 9c87ef32b68..1c8b3a748d9 100644
--- a/be/test/storage/segment/count_on_index_fastpath_test.cpp
+++ b/be/test/storage/segment/count_on_index_fastpath_test.cpp
@@ -45,7 +45,6 @@ CountOnIndexFastpathFacts safe_facts() {
f.has_topn_filters = false;
f.has_external_row_ranges = false;
f.row_bitmap_is_full = true;
- f.record_rowids = false;
f.has_ann_topn = false;
f.has_score_runtime = false;
f.no_need_read_data_opt_enabled = true;
@@ -143,11 +142,6 @@ TEST(CountOnIndexFastpath, PrunedRowSpaceVetoes) {
// Consumers of REAL row ids veto.
TEST(CountOnIndexFastpath, RowIdConsumersVeto) {
- {
- auto f = safe_facts();
- f.record_rowids = true;
- EXPECT_FALSE(count_on_index_fastpath_safe(f));
- }
{
auto f = safe_facts();
f.has_ann_topn = true;
@@ -195,7 +189,6 @@ CountEmitShortcutFacts safe_emit_facts() {
f.has_remaining_common_exprs = false;
f.has_delete_predicates = false;
f.has_virtual_columns = false;
- f.record_rowids = false;
f.has_read_limit = false;
f.read_orderby_key_reverse = false;
f.has_condition_cache_digest = false;
@@ -254,18 +247,11 @@ TEST(CountEmitShortcut, SurvivingEvaluationRefuses) {
}
}
-// Consumers of real row ids or per-row values refuse.
-TEST(CountEmitShortcut, RowIdOrValueConsumersRefuse) {
- {
- auto f = safe_emit_facts();
- f.has_virtual_columns = true;
- EXPECT_FALSE(count_emit_shortcut_safe(f));
- }
- {
- auto f = safe_emit_facts();
- f.record_rowids = true;
- EXPECT_FALSE(count_emit_shortcut_safe(f));
- }
+// Virtual columns consume real row ids or per-row values.
+TEST(CountEmitShortcut, VirtualColumnsRefuse) {
+ auto f = safe_emit_facts();
+ f.has_virtual_columns = true;
+ EXPECT_FALSE(count_emit_shortcut_safe(f));
}
// Batch accounting must be a pure countdown: limits, reverse key-ordered
diff --git
a/be/test/storage/segment/segment_iterator_count_emit_shortcut_test.cpp
b/be/test/storage/segment/segment_iterator_count_emit_shortcut_test.cpp
index a60cbbdc266..f48e8c98e3d 100644
--- a/be/test/storage/segment/segment_iterator_count_emit_shortcut_test.cpp
+++ b/be/test/storage/segment/segment_iterator_count_emit_shortcut_test.cpp
@@ -360,14 +360,6 @@ TEST(CountEmitShortcut, EngageRefusalTruthTable) {
Block block = fx.read_schema->create_read_block();
EXPECT_FALSE(fx.iter->_should_engage_count_emit_shortcut(&block));
}
- // Rowid consumers need real row ids.
- {
- Fixture fx;
- fx.set_hit();
- fx.iter->_opts.record_rowids = true;
- Block block = fx.read_schema->create_read_block();
- EXPECT_FALSE(fx.iter->_should_engage_count_emit_shortcut(&block));
- }
// Block shape != read schema (e.g. delete-condition column beyond block).
{
Fixture fx;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
