Userwhite commented on code in PR #63110:
URL: https://github.com/apache/doris/pull/63110#discussion_r3234533101
##########
be/src/storage/segment/segment_iterator.cpp:
##########
@@ -2388,6 +2389,120 @@ void
SegmentIterator::_replace_version_col_if_needed(const std::vector<ColumnId>
VLOG_DEBUG << "replaced version column in segment iterator,
version_col_idx:" << version_idx;
}
+void SegmentIterator::_update_lsn_col_if_needed(const std::vector<ColumnId>&
column_ids,
+ size_t num_rows) {
+ // | real version(64) | auto-inc row_id(64) |
+ if (_opts.version.first != _opts.version.second) {
+ return;
+ }
+
+ if (_opts.io_ctx.reader_type != ReaderType::READER_BINLOG &&
+ _opts.io_ctx.reader_type != ReaderType::READER_BINLOG_COMPACTION) {
+ return;
+ }
+
+ int32_t lsn_col_idx = _schema->lsn_col_idx();
+ if (lsn_col_idx < 0 || std::ranges::find(column_ids, lsn_col_idx) ==
column_ids.end()) {
+ return;
+ }
+
+ if (_is_pred_column[lsn_col_idx]) {
+ auto* lsn_column = assert_cast<PredicateColumnType<TYPE_LARGEINT>*>(
+ _current_return_columns[lsn_col_idx].get());
+ std::vector<Int128> binlog_lsns;
+ binlog_lsns.reserve(num_rows);
+ static constexpr Int128 kLow64Mask = (static_cast<Int128>(1) << 64) -
1;
+ for (size_t j = 0; j < num_rows; j++) {
+ const Int128 row_id = lsn_column->get_data()[j];
+
binlog_lsns.emplace_back((static_cast<Int128>(_opts.version.second) << 64) |
+ (row_id & kLow64Mask));
+ }
+ _current_return_columns[lsn_col_idx]->clear();
+ for (const auto& binlog_lsn : binlog_lsns) {
+ lsn_column->insert_data(reinterpret_cast<const
char*>(&binlog_lsn), 0);
+ }
+ return;
+ }
+
+ auto* lsn_column =
assert_cast<ColumnInt128*>(_current_return_columns[lsn_col_idx].get());
+ const auto* column_desc = _schema->column(lsn_col_idx);
+ auto column = Schema::get_data_type_ptr(*column_desc)->create_column();
+ DCHECK(column_desc->type() == FieldType::OLAP_FIELD_TYPE_LARGEINT);
+ auto* col_ptr = assert_cast<ColumnInt128*>(column.get());
+
+ static constexpr Int128 kLow64Mask = (static_cast<Int128>(1) << 64) - 1;
+ for (size_t j = 0; j < num_rows; j++) {
+ const Int128 row_id = lsn_column->get_element(j);
+ const Int128 binlog_lsn =
+ (static_cast<Int128>(_opts.version.second) << 64) | (row_id &
kLow64Mask);
+ col_ptr->insert_value(binlog_lsn);
+ }
+ _current_return_columns[lsn_col_idx] = std::move(column);
+}
+
+void SegmentIterator::_update_tso_col_if_needed(const std::vector<ColumnId>&
column_ids,
+ size_t num_rows) {
+ // use commit tso to replace timestamp col
+ if (_opts.version.first != _opts.version.second) {
+ return;
+ }
+
+ if (_opts.io_ctx.reader_type != ReaderType::READER_BINLOG &&
+ _opts.io_ctx.reader_type != ReaderType::READER_BINLOG_COMPACTION) {
+ return;
+ }
+
+ int32_t tso_col_idx = _schema->tso_col_idx();
+ if (tso_col_idx < 0 || std::ranges::find(column_ids, tso_col_idx) ==
column_ids.end()) {
+ return;
+ }
+
+ DCHECK_EQ(_opts.commit_tso.start_tso(), _opts.commit_tso.end_tso());
+ Int64 commit_tso = _opts.commit_tso.end_tso() == -1 ? 0 :
_opts.commit_tso.end_tso();
Review Comment:
can we ensure order just by only physical timestamp
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
