This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 e925e6ae1b2 [feature](binlog) Add per-row LSN column for detail (DUP)
tables to support IVM (#66889)
e925e6ae1b2 is described below
commit e925e6ae1b2abe5d8d2c2ef1e1677ec14d014d58
Author: Userwhite <[email protected]>
AuthorDate: Thu Aug 27 12:07:42 2026 +0800
[feature](binlog) Add per-row LSN column for detail (DUP) tables to
support IVM (#66889)
Problem Summary:
Incremental View Maintenance (IVM) needs a stable per-row identity for
every row in the base table. For MOW/unique tables this can be derived
from the unique key, but a detail (DUP_KEYS) table has no stable unique
key — two rows can be byte-identical, and the commit TSO alone is not
row-unique within one commit.
Row Binlog already allocates a monotonic LSN per row so that binlog rows
can be ordered/deduplicated. This PR surfaces that same LSN as a
first-class hidden column on the DUP base table, and makes Stream return
the LSN alongside the existing sequence (TSO) column, so IVM has a
stable (TSO, LSN) identity for every base row.
Scope (intentional):
- Only the DUP base index with row-binlog enabled gets the base
__DORIS_ROW_LSN_COL__ column.
- MOW does not expose or store a base LSN column (a MOW row already has
a stable unique key; adding an aggregated per-row LSN there is
meaningless and would complicate merge semantics). MOW still uses the
LSN internally for the row-binlog side-channel.
- Rollup / MV indexes are not covered; only the base index stores the
row LSN.
Design overview：
<img width="2860" height="2480" alt="image"
src="https://github.com/user-attachments/assets/889af47f-a495-441c-8d47-29eb23fcb0e9";
/>
---
be/src/cloud/cloud_rowset_builder.cpp | 1 +
be/src/cloud/cloud_rowset_writer.cpp | 4 +
be/src/cloud/pb_convert.cpp | 12 +++
be/src/exec/sink/writer/vtablet_writer.cpp | 22 +++---
be/src/exec/sink/writer/vtablet_writer.h | 6 +-
be/src/load/channel/tablets_channel.cpp | 28 +++----
be/src/load/delta_writer/delta_writer_context.h | 2 +-
be/src/load/memtable/memtable.cpp | 86 ++++++++++++----------
be/src/load/memtable/memtable.h | 25 ++++---
be/src/load/memtable/memtable_flush_executor.cpp | 25 ++++---
be/src/load/memtable/memtable_flush_executor.h | 3 +
be/src/load/memtable/memtable_writer.cpp | 23 +++---
be/src/load/memtable/memtable_writer.h | 2 +-
be/src/storage/binlog.h | 45 ++++-------
be/src/storage/rowset/beta_rowset_writer.cpp | 3 +
be/src/storage/rowset/beta_rowset_writer_v2.cpp | 4 +
be/src/storage/rowset/group_rowset_writer.cpp | 18 +++++
be/src/storage/rowset/group_rowset_writer.h | 5 +-
be/src/storage/rowset/rowset_fwd.h | 3 +
be/src/storage/rowset/rowset_writer_context.h | 26 +++++++
be/src/storage/rowset_builder.cpp | 1 +
be/src/storage/tablet/base_tablet.cpp | 4 +-
be/src/storage/tablet/tablet_meta.cpp | 6 ++
be/src/storage/tablet/tablet_schema.cpp | 9 +++
be/src/storage/tablet/tablet_schema.h | 2 +
be/src/storage/transform/row_binlog_derive.cpp | 4 +-
be/src/storage/transform/row_binlog_derive.h | 3 +-
be/src/storage/utils.h | 1 +
.../load/memtable/memtable_flush_executor_test.cpp | 18 ++++-
.../cloud_group_rowset_builder_writer_test.cpp | 11 ++-
be/test/olap/rowset/group_rowset_writer_test.cpp | 22 ++++--
be/test/storage/mow/mow_transform_test_base.h | 3 +-
.../storage/rowset/segment_flusher_format_test.cpp | 3 +-
.../storage/transform/row_binlog_derive_test.cpp | 41 ++++++-----
.../main/java/org/apache/doris/catalog/Column.java | 8 ++
.../java/org/apache/doris/catalog/OlapTable.java | 5 ++
.../catalog/stream/TableStreamBuildFactory.java | 8 ++
.../cloud/datasource/CloudInternalCatalog.java | 5 ++
.../apache/doris/datasource/InternalCatalog.java | 2 +-
.../rewrite/NormalizeOlapTableStreamScan.java | 71 +++++++++---------
.../plans/commands/info/ColumnDefinition.java | 12 +++
.../trees/plans/commands/info/CreateTableInfo.java | 19 ++---
.../plans/logical/LogicalOlapTableStreamScan.java | 26 +++++--
.../org/apache/doris/task/CreateReplicaTask.java | 5 ++
gensrc/proto/internal_service.proto | 5 +-
gensrc/proto/olap_file.proto | 18 +++--
gensrc/thrift/AgentService.thrift | 7 +-
.../row_binlog_p0/test_row_binlog_basic.groovy | 27 +++++++
.../test_olap_table_stream_history_query.groovy | 25 +++++--
.../test_olap_table_stream_snapshot.groovy | 8 ++
.../test_table_stream_query_comprehensive.groovy | 17 +++++
51 files changed, 495 insertions(+), 244 deletions(-)
diff --git a/be/src/cloud/cloud_rowset_builder.cpp
b/be/src/cloud/cloud_rowset_builder.cpp
index 0ab26eef6a5..8730b86a555 100644
--- a/be/src/cloud/cloud_rowset_builder.cpp
+++ b/be/src/cloud/cloud_rowset_builder.cpp
@@ -139,6 +139,7 @@ Status CloudGroupRowsetBuilder::init() {
RETURN_IF_ERROR(RowsetFactory::create_empty_group_rowset_writer(&group_writer));
group_writer->set_data_writer(_data_builder->rowset_writer());
group_writer->set_row_binlog_writer(_row_binlog_builder->rowset_writer());
+
RETURN_IF_ERROR(group_writer->init(_data_builder->rowset_writer()->context()));
{
const auto& data_ctx = _data_builder->rowset_writer()->context();
diff --git a/be/src/cloud/cloud_rowset_writer.cpp
b/be/src/cloud/cloud_rowset_writer.cpp
index d214a9ff0b4..a0ef7977c59 100644
--- a/be/src/cloud/cloud_rowset_writer.cpp
+++ b/be/src/cloud/cloud_rowset_writer.cpp
@@ -41,6 +41,10 @@ CloudRowsetWriter::~CloudRowsetWriter() {
Status CloudRowsetWriter::init(const RowsetWriterContext&
rowset_writer_context) {
_context = rowset_writer_context;
+ // Row-binlog writer or a schema carrying ROW_LSN_COL needs allocated LSN.
+ _context._need_allocate_lsn =
+ _context.write_binlog_opt().enable ||
+ (_context.tablet_schema != nullptr &&
_context.tablet_schema->row_lsn_col_idx() >= 0);
_rowset_meta = std::make_shared<RowsetMeta>();
if (_context.is_local_rowset()) {
diff --git a/be/src/cloud/pb_convert.cpp b/be/src/cloud/pb_convert.cpp
index 900aca4c7fc..b7a48e0dbb6 100644
--- a/be/src/cloud/pb_convert.cpp
+++ b/be/src/cloud/pb_convert.cpp
@@ -504,6 +504,9 @@ void doris_tablet_schema_to_cloud(TabletSchemaCloudPB* out,
const TabletSchemaPB
if (in.has_commit_tso_col_idx()) {
out->set_commit_tso_col_idx(in.commit_tso_col_idx());
}
+ if (in.has_row_lsn_col_idx()) {
+ out->set_row_lsn_col_idx(in.row_lsn_col_idx());
+ }
if (in.has_binlog_tso_col_idx()) {
out->set_binlog_tso_col_idx(in.binlog_tso_col_idx());
}
@@ -565,6 +568,9 @@ void doris_tablet_schema_to_cloud(TabletSchemaCloudPB* out,
TabletSchemaPB&& in)
if (in.has_commit_tso_col_idx()) {
out->set_commit_tso_col_idx(in.commit_tso_col_idx());
}
+ if (in.has_row_lsn_col_idx()) {
+ out->set_row_lsn_col_idx(in.row_lsn_col_idx());
+ }
if (in.has_binlog_tso_col_idx()) {
out->set_binlog_tso_col_idx(in.binlog_tso_col_idx());
}
@@ -639,6 +645,9 @@ void cloud_tablet_schema_to_doris(TabletSchemaPB* out,
const TabletSchemaCloudPB
if (in.has_commit_tso_col_idx()) {
out->set_commit_tso_col_idx(in.commit_tso_col_idx());
}
+ if (in.has_row_lsn_col_idx()) {
+ out->set_row_lsn_col_idx(in.row_lsn_col_idx());
+ }
if (in.has_binlog_tso_col_idx()) {
out->set_binlog_tso_col_idx(in.binlog_tso_col_idx());
}
@@ -701,6 +710,9 @@ void cloud_tablet_schema_to_doris(TabletSchemaPB* out,
TabletSchemaCloudPB&& in)
if (in.has_commit_tso_col_idx()) {
out->set_commit_tso_col_idx(in.commit_tso_col_idx());
}
+ if (in.has_row_lsn_col_idx()) {
+ out->set_row_lsn_col_idx(in.row_lsn_col_idx());
+ }
if (in.has_binlog_tso_col_idx()) {
out->set_binlog_tso_col_idx(in.binlog_tso_col_idx());
}
diff --git a/be/src/exec/sink/writer/vtablet_writer.cpp
b/be/src/exec/sink/writer/vtablet_writer.cpp
index e1a3a29914d..4fac6ee4e38 100644
--- a/be/src/exec/sink/writer/vtablet_writer.cpp
+++ b/be/src/exec/sink/writer/vtablet_writer.cpp
@@ -937,8 +937,8 @@ Status VNodeChannel::add_block(Block* block, const Payload*
payload) {
_cur_add_block_request->add_tablet_ids(row_part_tablet_ids->tablet_ids[route_idx]);
}
}
- for (auto row_binlog_lsn : payload->row_binlog_lsns) {
- _cur_add_block_request->add_row_binlog_lsns(row_binlog_lsn);
+ for (auto allocated_lsn : payload->allocated_lsns) {
+ _cur_add_block_request->add_allocated_lsns(allocated_lsn);
}
_write_bytes.fetch_add(_cur_mutable_block->bytes());
@@ -964,7 +964,7 @@ Status VNodeChannel::add_block(Block* block, const Payload*
payload) {
_cur_mutable_block = MutableBlock::create_unique(block->clone_empty());
_cur_add_block_request->clear_tablet_ids();
_cur_add_block_request->clear_partition_ids();
- _cur_add_block_request->clear_row_binlog_lsns();
+ _cur_add_block_request->clear_allocated_lsns();
}
return Status::OK();
@@ -1644,7 +1644,7 @@ Status VTabletWriter::_init(RuntimeState* state,
RuntimeProfile* profile) {
bool has_row_binlog = std::any_of(_schema->indexes().begin(),
_schema->indexes().end(),
[](const auto* index) { return
index->row_binlog_id > 0; });
if (has_row_binlog) {
- _row_binlog_lsn_buffer =
GlobalAutoIncBuffers::GetInstance()->get_auto_inc_buffer(
+ _allocated_lsn_buffer =
GlobalAutoIncBuffers::GetInstance()->get_auto_inc_buffer(
_schema->db_id(), _schema->table_id(), kBinlogLsnAutoIncId);
}
_schema->set_timestamp_ms(state->timestamp_ms());
@@ -2160,10 +2160,10 @@ Status
VTabletWriter::_generate_one_index_channel_payload(
size_t row_cnt = row_ids.size();
bool has_row_binlog = _schema->indexes()[index_idx]->row_binlog_id > 0;
- std::vector<int64_t> row_binlog_lsns;
+ std::vector<int64_t> allocated_lsns;
if (has_row_binlog && row_cnt > 0) {
- DCHECK(_row_binlog_lsn_buffer != nullptr);
- RETURN_IF_ERROR(allocate_binlog_lsn(_row_binlog_lsn_buffer, row_cnt,
row_binlog_lsns));
+ DCHECK(_allocated_lsn_buffer != nullptr);
+ RETURN_IF_ERROR(allocate_lsn(_allocated_lsn_buffer, row_cnt,
allocated_lsns));
}
for (size_t i = 0; i < row_ids.size(); i++) {
@@ -2185,13 +2185,13 @@ Status
VTabletWriter::_generate_one_index_channel_payload(
payload_it->second.row_ids->reserve(row_cnt);
payload_it->second.route_idxs.reserve(row_cnt);
if (has_row_binlog) {
- payload_it->second.row_binlog_lsns.reserve(row_cnt);
+ payload_it->second.allocated_lsns.reserve(row_cnt);
}
}
payload_it->second.row_ids->push_back(row_ids[i]);
payload_it->second.route_idxs.push_back(cast_set<uint32_t>(i));
if (has_row_binlog) {
-
payload_it->second.row_binlog_lsns.push_back(row_binlog_lsns[i]);
+ payload_it->second.allocated_lsns.push_back(allocated_lsns[i]);
}
continue;
}
@@ -2216,13 +2216,13 @@ Status
VTabletWriter::_generate_one_index_channel_payload(
payload_it->second.row_ids->reserve(row_cnt);
payload_it->second.route_idxs.reserve(row_cnt);
if (has_row_binlog) {
- payload_it->second.row_binlog_lsns.reserve(row_cnt);
+ payload_it->second.allocated_lsns.reserve(row_cnt);
}
}
payload_it->second.row_ids->push_back(row_ids[i]);
payload_it->second.route_idxs.push_back(cast_set<uint32_t>(i));
if (has_row_binlog) {
-
payload_it->second.row_binlog_lsns.push_back(row_binlog_lsns[i]);
+ payload_it->second.allocated_lsns.push_back(allocated_lsns[i]);
}
}
}
diff --git a/be/src/exec/sink/writer/vtablet_writer.h
b/be/src/exec/sink/writer/vtablet_writer.h
index 1605faba306..932e8b2b56c 100644
--- a/be/src/exec/sink/writer/vtablet_writer.h
+++ b/be/src/exec/sink/writer/vtablet_writer.h
@@ -226,7 +226,7 @@ struct Payload {
std::unique_ptr<IColumn::Selector> row_ids;
RowPartTabletIds* row_part_tablet_ids = nullptr;
std::vector<uint32_t> route_idxs;
- std::vector<int64_t> row_binlog_lsns;
+ std::vector<int64_t> allocated_lsns;
};
// every NodeChannel keeps a data transmission channel with one BE. for
multiple times open, it has a dozen of requests and corresponding closures.
@@ -723,8 +723,8 @@ private:
bthread::Mutex _stop_check_channel;
std::vector<std::shared_ptr<IndexChannel>> _channels;
std::unordered_map<int64_t, std::shared_ptr<IndexChannel>>
_index_id_to_channel;
- // Table-level row-binlog LSN buffer
- std::shared_ptr<AutoIncIDBuffer> _row_binlog_lsn_buffer;
+ // Table-level LSN buffer
+ std::shared_ptr<AutoIncIDBuffer> _allocated_lsn_buffer;
std::unique_ptr<ThreadPoolToken> _send_batch_thread_pool_token;
diff --git a/be/src/load/channel/tablets_channel.cpp
b/be/src/load/channel/tablets_channel.cpp
index d35b260076e..a589b8ba13e 100644
--- a/be/src/load/channel/tablets_channel.cpp
+++ b/be/src/load/channel/tablets_channel.cpp
@@ -627,14 +627,14 @@ Status BaseTabletsChannel::_write_block_data(
print_id(_load_id), _index_id, request.packet_seq(),
send_data.rows(),
request.tablet_ids_size());
}
- bool has_row_binlog_lsn = request.row_binlog_lsns_size() > 0;
- if (has_row_binlog_lsn) {
- if (send_data.rows() != request.row_binlog_lsns_size()) {
+ bool has_allocated_lsn = request.allocated_lsns_size() > 0;
+ if (has_allocated_lsn) {
+ if (send_data.rows() != request.allocated_lsns_size()) {
return Status::InternalError(
- "invalid add block request row-binlog lsn count,
load_id={}, index_id={}, "
- "packet_seq={}, block_rows={}, row_binlog_lsns_size={}",
+ "invalid add block request allocated lsn count,
load_id={}, index_id={}, "
+ "packet_seq={}, block_rows={}, allocated_lsns_size={}",
print_id(_load_id), _index_id, request.packet_seq(),
send_data.rows(),
- request.row_binlog_lsns_size());
+ request.allocated_lsns_size());
}
}
@@ -791,10 +791,10 @@ Status
BaseTabletsChannel::_write_block_data_for_adaptive_random_bucket(
RETURN_IF_ERROR(_prepare_adaptive_random_bucket_writer(tablet_writer));
TabletAddRowsPayload rows {.row_idxs = row_idxs};
- if (request.row_binlog_lsns_size() > 0) {
- rows.row_binlog_lsns.reserve(row_idxs.size());
+ if (request.allocated_lsns_size() > 0) {
+ rows.allocated_lsns.reserve(row_idxs.size());
for (auto row_idx : row_idxs) {
-
rows.row_binlog_lsns.emplace_back(request.row_binlog_lsns(row_idx));
+
rows.allocated_lsns.emplace_back(request.allocated_lsns(row_idx));
}
}
bool memtable_flushed = false;
@@ -920,14 +920,14 @@ void BaseTabletsChannel::_build_tablet_to_rows(
// tests show that a relatively coarse-grained read lock here performs
better under multicore scenario
// see: https://github.com/apache/doris/pull/28552
std::shared_lock<std::shared_mutex> rlock(_broken_tablets_lock);
- bool has_row_binlog_lsn = request.row_binlog_lsns_size() > 0;
+ bool has_allocated_lsn = request.allocated_lsns_size() > 0;
if (request.is_single_tablet_block()) {
// The cloud mode need the tablet ids to prepare rowsets.
int64_t tablet_id = request.tablet_ids(0);
auto& rows = (*tablet_to_rows)[tablet_id];
rows.row_idxs.emplace_back(0);
- if (has_row_binlog_lsn) {
- rows.row_binlog_lsns.emplace_back(request.row_binlog_lsns(0));
+ if (has_allocated_lsn) {
+ rows.allocated_lsns.emplace_back(request.allocated_lsns(0));
}
return;
}
@@ -940,8 +940,8 @@ void BaseTabletsChannel::_build_tablet_to_rows(
}
auto& rows = (*tablet_to_rows)[tablet_id];
rows.row_idxs.emplace_back(i);
- if (has_row_binlog_lsn) {
- rows.row_binlog_lsns.emplace_back(request.row_binlog_lsns(i));
+ if (has_allocated_lsn) {
+ rows.allocated_lsns.emplace_back(request.allocated_lsns(i));
}
}
}
diff --git a/be/src/load/delta_writer/delta_writer_context.h
b/be/src/load/delta_writer/delta_writer_context.h
index 856601c0be6..06b4659c79b 100644
--- a/be/src/load/delta_writer/delta_writer_context.h
+++ b/be/src/load/delta_writer/delta_writer_context.h
@@ -60,7 +60,7 @@ struct WriteRequest {
struct TabletAddRowsPayload {
DorisVector<uint32_t> row_idxs;
- DorisVector<int64_t> row_binlog_lsns = {};
+ DorisVector<int64_t> allocated_lsns = {};
};
} // namespace doris
diff --git a/be/src/load/memtable/memtable.cpp
b/be/src/load/memtable/memtable.cpp
index b9b31a34a23..2a9ac792f3b 100644
--- a/be/src/load/memtable/memtable.cpp
+++ b/be/src/load/memtable/memtable.cpp
@@ -29,6 +29,7 @@
#include "bvar/bvar.h"
#include "common/config.h"
#include "core/column/column.h"
+#include "core/column/column_vector.h"
#include "exprs/aggregate/aggregate_function_reader.h"
#include "exprs/aggregate/aggregate_function_simple_factory.h"
#include "load/delta_writer/delta_writer_context.h"
@@ -53,14 +54,14 @@ using namespace ErrorCode;
MemTable::MemTable(int64_t tablet_id, std::shared_ptr<TabletSchema>
tablet_schema,
const std::vector<SlotDescriptor*>* slot_descs,
TupleDescriptor* tuple_desc,
bool enable_unique_key_mow, PartialUpdateInfo*
partial_update_info,
- const std::shared_ptr<ResourceContext>& resource_ctx, bool
need_row_binlog_lsn)
+ const std::shared_ptr<ResourceContext>& resource_ctx, bool
need_lsn)
: _mem_type(MemType::ACTIVE),
_tablet_id(tablet_id),
_enable_unique_key_mow(enable_unique_key_mow),
_keys_type(tablet_schema->keys_type()),
_tablet_schema(tablet_schema),
_resource_ctx(resource_ctx),
- _need_row_binlog_lsn(need_row_binlog_lsn),
+ _need_lsn(need_lsn),
_is_first_insertion(true),
_agg_functions(tablet_schema->num_columns()),
_offsets_of_aggregate_states(tablet_schema->num_columns()),
@@ -81,6 +82,7 @@ MemTable::MemTable(int64_t tablet_id,
std::shared_ptr<TabletSchema> tablet_schem
}
}
_init_columns_offset_by_slot_descs(slot_descs, tuple_desc);
+ _row_lsn_col_pos = tablet_schema->row_lsn_col_idx();
// TODO: Support ZOrderComparator in the future
_row_in_blocks =
std::make_unique<DorisVector<std::shared_ptr<RowInBlock>>>();
_load_mem_limit = MemInfo::mem_limit() *
config::load_process_max_memory_limit_percent / 100;
@@ -187,7 +189,7 @@ MemTable::~MemTable() {
_input_mutable_block.clear();
_output_mutable_block.clear();
// Reset the LSN sidecar to release its capacity tracked by the
memtable tracker.
- _output_row_binlog_lsns = DorisVector<int64_t>();
+ _output_allocated_lsns = std::make_shared<std::vector<int64_t>>();
}
if (_is_flush_success) {
// If the memtable is flush success, then its memtracker's consumption
should be 0
@@ -208,19 +210,19 @@ Status MemTable::insert(const Block* input_block, const
TabletAddRowsPayload& ro
_resource_ctx->memory_context()->mem_tracker()->write_tracker());
SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);
const auto& row_idxs = rows.row_idxs;
- const auto& row_binlog_lsns = rows.row_binlog_lsns;
+ const auto& allocated_lsns = rows.allocated_lsns;
- if (_need_row_binlog_lsn) {
- if (row_binlog_lsns.empty()) {
+ if (_need_lsn) {
+ if (allocated_lsns.empty()) {
return Status::InternalError(
- "row binlog lsn is missing for memtable insert, "
+ "allocated lsn is missing for memtable insert, "
"tablet_id={}",
_tablet_id);
}
- DCHECK_EQ(row_binlog_lsns.size(), row_idxs.size());
- } else if (!row_binlog_lsns.empty()) {
+ DCHECK_EQ(allocated_lsns.size(), row_idxs.size());
+ } else if (!allocated_lsns.empty()) {
return Status::InternalError(
- "row binlog lsn is unexpectedly provided for memtable insert,
tablet_id={}",
+ "allocated lsn is unexpectedly provided for memtable insert,
tablet_id={}",
_tablet_id);
}
@@ -266,30 +268,36 @@ Status MemTable::insert(const Block* input_block, const
TabletAddRowsPayload& ro
size_t cursor_in_mutableblock = _input_mutable_block.rows();
RETURN_IF_ERROR(_input_mutable_block.add_rows(input_block, row_idxs.data(),
row_idxs.data() + num_rows,
&_column_offset));
+ if (_need_lsn && _row_lsn_col_pos >= 0) {
+ auto lsn_column = ColumnInt64::create();
+ lsn_column->get_data().assign(allocated_lsns.begin(),
allocated_lsns.end());
+ _input_mutable_block.get_column_by_position(_row_lsn_col_pos)
+ ->replace_column_data_range(*lsn_column, 0, num_rows,
cursor_in_mutableblock);
+ }
for (int i = 0; i < num_rows; i++) {
_row_in_blocks->emplace_back(std::make_shared<RowInBlock>(
- cursor_in_mutableblock + i, _need_row_binlog_lsn ?
row_binlog_lsns[i] : 0));
+ cursor_in_mutableblock + i, _need_lsn ? allocated_lsns[i] :
0));
}
_stat.raw_rows += num_rows;
return Status::OK();
}
-void MemTable::_merge_row_binlog_lsn(RowInBlock* src_row, RowInBlock* dst_row)
{
- if (_need_row_binlog_lsn) {
- dst_row->_row_binlog_lsn = std::max(dst_row->_row_binlog_lsn,
src_row->_row_binlog_lsn);
+void MemTable::_merge_allocated_lsn(RowInBlock* src_row, RowInBlock* dst_row) {
+ if (_need_lsn) {
+ dst_row->_allocated_lsn = std::max(dst_row->_allocated_lsn,
src_row->_allocated_lsn);
}
}
-void MemTable::_append_output_row_binlog_lsn(RowInBlock* row) {
- if (_need_row_binlog_lsn) {
- _output_row_binlog_lsns.emplace_back(row->_row_binlog_lsn);
+void MemTable::_append_output_allocated_lsn(RowInBlock* row) {
+ if (_need_lsn) {
+ _output_allocated_lsns->emplace_back(row->_allocated_lsn);
}
}
void MemTable::_aggregate_two_row_with_sequence_map(MutableBlock&
mutable_block,
RowInBlock* src_row,
RowInBlock* dst_row) {
- _merge_row_binlog_lsn(src_row, dst_row);
+ _merge_allocated_lsn(src_row, dst_row);
// for each mapping replace value columns according to the sequence column
compare result
// for example: a b c d s1 s2 (key:a , s1=>[b,c], s2=>[d])
// src row: 1 4 5 6 8 9
@@ -329,7 +337,7 @@ void
MemTable::_aggregate_two_row_with_sequence_map(MutableBlock& mutable_block,
template <bool has_skip_bitmap_col>
void MemTable::_aggregate_two_row_in_block(MutableBlock& mutable_block,
RowInBlock* src_row,
RowInBlock* dst_row) {
- _merge_row_binlog_lsn(src_row, dst_row);
+ _merge_allocated_lsn(src_row, dst_row);
// for flexible partial update, the caller must guarantees that either
src_row and dst_row
// both specify the sequence column, or src_row and dst_row both don't
specify the
// sequence column
@@ -378,12 +386,12 @@ Status MemTable::_put_into_output(Block& in_block) {
DorisVector<uint32_t> row_pos_vec;
DCHECK(in_block.rows() <= std::numeric_limits<int>::max());
row_pos_vec.reserve(in_block.rows());
- if (_need_row_binlog_lsn) {
- _output_row_binlog_lsns.reserve(_output_row_binlog_lsns.size() +
in_block.rows());
+ if (_need_lsn) {
+ _output_allocated_lsns->reserve(_output_allocated_lsns->size() +
in_block.rows());
}
for (int i = 0; i < _row_in_blocks->size(); i++) {
row_pos_vec.emplace_back((*_row_in_blocks)[i]->_row_pos);
- _append_output_row_binlog_lsn((*_row_in_blocks)[i].get());
+ _append_output_allocated_lsn((*_row_in_blocks)[i].get());
}
return _output_mutable_block.add_rows(&in_block, row_pos_vec.data(),
row_pos_vec.data() +
in_block.rows());
@@ -442,17 +450,17 @@ Status MemTable::_sort_by_cluster_keys() {
DorisVector<std::shared_ptr<RowInBlock>> row_in_blocks;
row_in_blocks.reserve(mutable_block.rows());
- if (_need_row_binlog_lsn) {
- DCHECK_EQ(_output_row_binlog_lsns.size(), mutable_block.rows());
+ if (_need_lsn) {
+ DCHECK_EQ(_output_allocated_lsns->size(), mutable_block.rows());
}
for (size_t i = 0; i < mutable_block.rows(); i++) {
row_in_blocks.emplace_back(
- _need_row_binlog_lsn ? std::make_shared<RowInBlock>(i,
_output_row_binlog_lsns[i])
- : std::make_shared<RowInBlock>(i));
+ _need_lsn ? std::make_shared<RowInBlock>(i,
(*_output_allocated_lsns)[i])
+ : std::make_shared<RowInBlock>(i));
}
- if (_need_row_binlog_lsn) {
- _output_row_binlog_lsns.clear();
- _output_row_binlog_lsns.reserve(mutable_block.rows());
+ if (_need_lsn) {
+ _output_allocated_lsns = std::make_shared<std::vector<int64_t>>();
+ _output_allocated_lsns->reserve(mutable_block.rows());
}
Tie tie = Tie(0, mutable_block.rows());
@@ -484,7 +492,7 @@ Status MemTable::_sort_by_cluster_keys() {
row_pos_vec.reserve(in_block.rows());
for (int i = 0; i < row_in_blocks.size(); i++) {
row_pos_vec.emplace_back(row_in_blocks[i]->_row_pos);
- _append_output_row_binlog_lsn(row_in_blocks[i].get());
+ _append_output_allocated_lsn(row_in_blocks[i].get());
}
std::vector<int> column_offset;
for (int i = 0; i < _column_offset.size(); ++i) {
@@ -548,7 +556,7 @@ void MemTable::_finalize_one_row(RowInBlock* row,
MutableBlock& mutable_block, i
*mutable_block.get_column_by_position(i), row->_row_pos);
}
}
- _append_output_row_binlog_lsn(row);
+ _append_output_allocated_lsn(row);
if constexpr (!is_final) {
row->_row_pos = row_pos;
}
@@ -641,7 +649,7 @@ void MemTable::_aggregate() {
//TODO(weixang):opt here.
_output_mutable_block =
MutableBlock::build_mutable_block(std::move(*empty_input_block));
_output_mutable_block.clear_column_data();
- _output_row_binlog_lsns.clear();
+ _output_allocated_lsns = std::make_shared<std::vector<int64_t>>();
*_row_in_blocks = temp_row_in_blocks;
_last_sorted_pos = _row_in_blocks->size();
}
@@ -695,7 +703,7 @@ void
MemTable::_aggregate_for_flexible_partial_update_without_seq_col(
if (cur_row_has_delete_sign) {
if (row_without_delete_sign != nullptr) {
// if there exits row without delete sign, remove it first
- _merge_row_binlog_lsn(row_without_delete_sign.get(),
cur_row);
+ _merge_allocated_lsn(row_without_delete_sign.get(),
cur_row);
_clear_row_agg(row_without_delete_sign.get());
_stat.merged_rows++;
row_without_delete_sign = nullptr;
@@ -806,15 +814,15 @@ size_t MemTable::get_flush_reserve_memory_size() const {
}
Status MemTable::_to_block(std::unique_ptr<Block>* res) {
- _output_row_binlog_lsns.clear();
+ _output_allocated_lsns = std::make_shared<std::vector<int64_t>>();
size_t same_keys_num = _sort();
if (_keys_type == KeysType::DUP_KEYS || same_keys_num == 0) {
if (_keys_type == KeysType::DUP_KEYS &&
_tablet_schema->num_key_columns() == 0) {
_output_mutable_block.swap(_input_mutable_block);
- if (_need_row_binlog_lsn) {
- _output_row_binlog_lsns.reserve(_row_in_blocks->size());
+ if (_need_lsn) {
+ _output_allocated_lsns->reserve(_row_in_blocks->size());
for (const auto& row : *_row_in_blocks) {
- _append_output_row_binlog_lsn(row.get());
+ _append_output_allocated_lsn(row.get());
}
}
} else {
@@ -832,8 +840,8 @@ Status MemTable::_to_block(std::unique_ptr<Block>* res) {
}
RETURN_IF_ERROR(_sort_by_cluster_keys());
}
- if (_need_row_binlog_lsn) {
- DCHECK_EQ(_output_row_binlog_lsns.size(),
_output_mutable_block.rows());
+ if (_need_lsn) {
+ DCHECK_EQ(_output_allocated_lsns->size(),
_output_mutable_block.rows());
}
_input_mutable_block.clear();
*res = Block::create_unique(_output_mutable_block.to_block());
diff --git a/be/src/load/memtable/memtable.h b/be/src/load/memtable/memtable.h
index 5e7c60a31dd..4228da6b928 100644
--- a/be/src/load/memtable/memtable.h
+++ b/be/src/load/memtable/memtable.h
@@ -34,6 +34,7 @@
#include "runtime/memory/mem_tracker.h"
#include "runtime/thread_context.h"
#include "storage/partial_update_info.h"
+#include "storage/rowset/rowset_fwd.h"
#include "storage/tablet/tablet_schema.h"
namespace doris {
@@ -52,14 +53,14 @@ enum MemType { ACTIVE = 0, WRITE_FINISHED = 1, FLUSH = 2 };
// row pos in _input_mutable_block
struct RowInBlock {
size_t _row_pos;
- int64_t _row_binlog_lsn = 0;
+ int64_t _allocated_lsn = 0;
char* _agg_mem = nullptr;
size_t* _agg_state_offset = nullptr;
bool _has_init_agg;
RowInBlock(size_t row) : _row_pos(row), _has_init_agg(false) {}
- RowInBlock(size_t row, int64_t row_binlog_lsn)
- : _row_pos(row), _row_binlog_lsn(row_binlog_lsn),
_has_init_agg(false) {}
+ RowInBlock(size_t row, int64_t allocated_lsn)
+ : _row_pos(row), _allocated_lsn(allocated_lsn),
_has_init_agg(false) {}
void init_agg_places(char* agg_mem, size_t* agg_state_offset) {
_has_init_agg = true;
@@ -177,8 +178,7 @@ public:
MemTable(int64_t tablet_id, std::shared_ptr<TabletSchema> tablet_schema,
const std::vector<SlotDescriptor*>* slot_descs, TupleDescriptor*
tuple_desc,
bool enable_unique_key_mow, PartialUpdateInfo*
partial_update_info,
- const std::shared_ptr<ResourceContext>& resource_ctx,
- bool need_row_binlog_lsn = false);
+ const std::shared_ptr<ResourceContext>& resource_ctx, bool
need_lsn = false);
~MemTable();
int64_t tablet_id() const { return _tablet_id; }
@@ -195,7 +195,7 @@ public:
Status to_block(std::unique_ptr<Block>* res);
- const DorisVector<int64_t>& row_binlog_lsns() const { return
_output_row_binlog_lsns; }
+ ConstAllocatedLsnVectorSharedPtr allocated_lsns() const { return
_output_allocated_lsns; }
bool empty() const { return _input_mutable_block.rows() == 0; }
@@ -219,12 +219,12 @@ private:
void _aggregate_two_row_in_block(MutableBlock& mutable_block, RowInBlock*
new_row,
RowInBlock* row_in_skiplist);
- // Merge row-binlog LSN sidecar only when MemTable merges two RowInBlock
objects.
+ // Merge allocated LSN sidecar only when MemTable merges two RowInBlock
objects.
// Table models that require complex merge semantics, such as AGG tables
and unique key
- // merge-on-read tables, do not support row-binlog LSN now and are
rejected in insert().
- void _merge_row_binlog_lsn(RowInBlock* src_row, RowInBlock* dst_row);
+ // merge-on-read tables, do not support allocated LSN now and are rejected
in insert().
+ void _merge_allocated_lsn(RowInBlock* src_row, RowInBlock* dst_row);
- void _append_output_row_binlog_lsn(RowInBlock* row);
+ void _append_output_allocated_lsn(RowInBlock* row);
void _aggregate_two_row_with_sequence_map(MutableBlock& mutable_block,
RowInBlock* new_row,
RowInBlock* row_in_skiplist);
@@ -257,6 +257,7 @@ private:
void _init_columns_offset_by_slot_descs(const
std::vector<SlotDescriptor*>* slot_descs,
const TupleDescriptor* tuple_desc);
std::vector<int> _column_offset;
+ int32_t _row_lsn_col_pos = -1;
// Number of rows inserted to this memtable.
// This is not the rows in this memtable, because rows may be merged
@@ -266,8 +267,8 @@ private:
//for vectorized
MutableBlock _input_mutable_block;
MutableBlock _output_mutable_block;
- DorisVector<int64_t> _output_row_binlog_lsns;
- bool _need_row_binlog_lsn = false;
+ AllocatedLsnVectorSharedPtr _output_allocated_lsns =
std::make_shared<std::vector<int64_t>>();
+ bool _need_lsn = false;
size_t _last_sorted_pos = 0;
size_t _last_agg_pos = 0;
diff --git a/be/src/load/memtable/memtable_flush_executor.cpp
b/be/src/load/memtable/memtable_flush_executor.cpp
index 7a3c7cf82be..1970d074a5e 100644
--- a/be/src/load/memtable/memtable_flush_executor.cpp
+++ b/be/src/load/memtable/memtable_flush_executor.cpp
@@ -119,6 +119,10 @@ SharedMemtable::~SharedMemtable() {
if (block == nullptr) {
return;
}
+ if (has_allocated_lsns) {
+ DCHECK(rowset_ctx != nullptr);
+ rowset_ctx->remove_segment_allocated_lsns(segment_id);
+ }
DCHECK(memtable != nullptr);
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(
memtable->resource_ctx()->memory_context()->mem_tracker()->write_tracker());
@@ -183,6 +187,8 @@ Status FlushToken::submit(std::shared_ptr<MemTable>
mem_table) {
shared_memtable = std::make_shared<SharedMemtable>();
shared_memtable->memtable = mem_table;
+ shared_memtable->rowset_ctx =
+
const_cast<RowsetWriterContext*>(&group_rowset_writer->context());
// Keep data/binlog segment_id allocators in sync.
auto segment_id = DORIS_TRY(data_writer->allocate_segment_id());
auto binlog_segment_id =
DORIS_TRY(binlog_writer->allocate_segment_id());
@@ -308,6 +314,15 @@ Status FlushToken::_memtable2block(MemTable* memtable,
SharedMemtable* shared_me
shared_memtable->block_status = memtable->to_block(&block);
if (shared_memtable->block_status.ok()) {
shared_memtable->block.reset(block.release());
+ auto* rowset_ctx = shared_memtable->rowset_ctx;
+ DCHECK(rowset_ctx != nullptr);
+ if (rowset_ctx->need_allocated_lsn() &&
shared_memtable->block->rows() > 0) {
+ auto memtable_lsns = memtable->allocated_lsns();
+ DCHECK_EQ(memtable_lsns->size(),
shared_memtable->block->rows());
+
rowset_ctx->insert_segment_allocated_lsns(shared_memtable->segment_id,
+ memtable_lsns);
+ shared_memtable->has_allocated_lsns = true;
+ }
}
});
if (!shared_memtable->block_status.ok()) {
@@ -392,16 +407,6 @@ void FlushToken::_flush_memtable_impl(RowsetWriter*
flush_writer, MemTable* memt
// }};
std::shared_ptr<Block> flush_block;
RETURN_IF_ERROR(_memtable2block(memtable, shared_memtable,
flush_block));
- if (flush_writer->context().write_binlog_opt().enable &&
flush_block->rows() > 0) {
- const auto& memtable_lsns = memtable->row_binlog_lsns();
- DCHECK_EQ(memtable_lsns.size(), flush_block->rows());
- auto lsn_ids =
std::make_shared<std::vector<int64_t>>(memtable_lsns.begin(),
-
memtable_lsns.end());
- const_cast<RowsetWriterContext&>(flush_writer->context())
- .write_binlog_opt()
- .write_binlog_config()
- .insert_seg_lsn(segment_id, std::move(lsn_ids));
- }
RETURN_IF_ERROR(
flush_writer->flush_memtable(flush_block.get(),
segment_id, &flush_size));
memtable->set_flush_success();
diff --git a/be/src/load/memtable/memtable_flush_executor.h
b/be/src/load/memtable/memtable_flush_executor.h
index dbe997cfceb..c66051cc5f7 100644
--- a/be/src/load/memtable/memtable_flush_executor.h
+++ b/be/src/load/memtable/memtable_flush_executor.h
@@ -38,6 +38,7 @@ class MemTableMemoryLimiter;
class Block;
class GroupRowsetWriter;
class OlapTableSchemaParam;
+struct RowsetWriterContext;
class RowsetWriter;
class SystemMetrics;
class WorkloadGroup;
@@ -63,6 +64,8 @@ struct SharedMemtable {
std::once_flag block_once;
Status block_status;
std::shared_ptr<Block> block;
+ RowsetWriterContext* rowset_ctx = nullptr;
+ bool has_allocated_lsns = false;
std::atomic<int> finished_sub_task_count {0};
// data + binlog
diff --git a/be/src/load/memtable/memtable_writer.cpp
b/be/src/load/memtable/memtable_writer.cpp
index 5a6eb301cf1..63f399fba01 100644
--- a/be/src/load/memtable/memtable_writer.cpp
+++ b/be/src/load/memtable/memtable_writer.cpp
@@ -83,25 +83,26 @@ Status MemTableWriter::init(std::shared_ptr<RowsetWriter>
rowset_writer,
_unique_key_mow = unique_key_mow;
_partial_update_info = partial_update_info;
_resource_ctx = thread_context()->resource_ctx();
- _need_row_binlog_lsn = false;
+ // Only the row-binlog base index needs LSN, aligned with the sink
(row_binlog_id > 0).
+ _need_lsn = false;
if (_req.table_schema_param != nullptr) {
for (const auto* index_schema : _req.table_schema_param->indexes()) {
if (index_schema->index_id == _req.index_id) {
- _need_row_binlog_lsn = index_schema->row_binlog_id > 0;
+ _need_lsn = index_schema->row_binlog_id > 0;
break;
}
}
}
- if (_need_row_binlog_lsn) {
+ if (_need_lsn) {
const auto keys_type = _tablet_schema->keys_type();
if (keys_type == KeysType::AGG_KEYS ||
(keys_type == KeysType::UNIQUE_KEYS && !_unique_key_mow)) {
// Row-binlog LSN sidecar does not support MemTable aggregation
now. For AGG tables
// and unique key merge-on-read tables, multiple input rows can be
merged into one
// output row in MemTable, so their output LSN semantics should be
implemented
- // explicitly before enabling row-binlog LSNs on these table types.
+ // explicitly before enabling allocated LSNs on these table types.
return Status::NotSupported(
- "row binlog lsn does not support AGG table or unique key
merge-on-read table");
+ "allocated lsn does not support AGG table or unique key
merge-on-read table");
}
}
@@ -125,7 +126,7 @@ Status MemTableWriter::write(const Block* block, const
TabletAddRowsPayload& row
*memtable_flushed = false;
}
if (UNLIKELY(rows.row_idxs.empty())) {
- DCHECK(rows.row_binlog_lsns.empty());
+ DCHECK(rows.allocated_lsns.empty());
return Status::OK();
}
_lock_watch.start();
@@ -142,13 +143,13 @@ Status MemTableWriter::write(const Block* block, const
TabletAddRowsPayload& row
_req.tablet_id,
_req.load_id.hi(), _req.load_id.lo());
}
- if (_need_row_binlog_lsn) {
- if (rows.row_binlog_lsns.empty()) {
+ if (_need_lsn) {
+ if (rows.allocated_lsns.empty()) {
return Status::InternalError(
- "row binlog lsn is missing for tablet_id={}, index_id={},
load_id={}-{}",
+ "allocated lsn is missing for tablet_id={}, index_id={},
load_id={}-{}",
_req.tablet_id, _req.index_id, _req.load_id.hi(),
_req.load_id.lo());
}
- DCHECK(rows.row_binlog_lsns.size() == rows.row_idxs.size());
+ DCHECK(rows.allocated_lsns.size() == rows.row_idxs.size());
}
// Flush and reset memtable if it is raw rows great than int32_t.
@@ -269,7 +270,7 @@ void MemTableWriter::_reset_mem_table() {
std::lock_guard<std::mutex> l(_mem_table_ptr_lock);
_mem_table.reset(new MemTable(_req.tablet_id, _tablet_schema,
_req.slots, _req.tuple_desc,
_unique_key_mow,
_partial_update_info.get(), _resource_ctx,
- _need_row_binlog_lsn));
+ _need_lsn));
}
_segment_num++;
diff --git a/be/src/load/memtable/memtable_writer.h
b/be/src/load/memtable/memtable_writer.h
index aa59a7fae6e..90e9ba564fa 100644
--- a/be/src/load/memtable/memtable_writer.h
+++ b/be/src/load/memtable/memtable_writer.h
@@ -124,7 +124,7 @@ private:
std::shared_ptr<MemTable> _mem_table;
TabletSchemaSPtr _tablet_schema;
bool _unique_key_mow = false;
- bool _need_row_binlog_lsn = false;
+ bool _need_lsn = false;
// This variable is accessed from writer thread and token flush thread
// use a shared ptr to avoid use after free problem.
diff --git a/be/src/storage/binlog.h b/be/src/storage/binlog.h
index 097b1e6a5ca..b59eded10ce 100644
--- a/be/src/storage/binlog.h
+++ b/be/src/storage/binlog.h
@@ -32,7 +32,8 @@
#include "common/status.h"
#include "exec/sink/autoinc_buffer.h"
#include "storage/olap_common.h"
-#include "storage/olap_define.h" // DataWriteType
+#include "storage/olap_define.h" // DataWriteType
+#include "storage/rowset/rowset_fwd.h"
#include "storage/tablet/tablet_fwd.h" // BaseTabletSPtr
#include "storage/tablet/tablet_schema.h" // TabletSchemaSPtr
#include "storage/utils.h" // BINLOG_*_COL
@@ -136,12 +137,12 @@ inline std::string
get_binlog_data_key_from_meta_key(const std::string_view meta
return fmt::format("{}data_{}", kBinlogPrefix,
meta_key.substr(kBinlogMetaPrefix.length()));
}
-// Allocate per-row LSNs for row-binlog data.
+// Allocate per-row LSNs.
// The caller must provide a valid auto-inc buffer (typically from
GlobalAutoIncBuffers).
-inline Status allocate_binlog_lsn(const std::shared_ptr<AutoIncIDBuffer>&
lsn_buffer,
- size_t num_rows, std::vector<int64_t>&
lsn_ids) {
+inline Status allocate_lsn(const std::shared_ptr<AutoIncIDBuffer>& lsn_buffer,
size_t num_rows,
+ std::vector<int64_t>& lsn_ids) {
if (lsn_buffer == nullptr) {
- return Status::InternalError("binlog<row> try to get lsn buffer but
null");
+ return Status::InternalError("try to get lsn buffer but null");
}
DCHECK(num_rows > 0);
@@ -168,23 +169,23 @@ inline int64_t extract_tso_physical_time(int64_t tso) {
namespace segment_v2 {
-class SegmentWriteBinlogLsnMap {
+class SegmentAllocatedLsnMap {
public:
- void insert_seg_lsn(int64_t seg_id, std::shared_ptr<std::vector<int64_t>>
lsn_ids) {
+ void insert_segment_allocated_lsns(int64_t seg_id,
ConstAllocatedLsnVectorSharedPtr lsn_ids) {
std::lock_guard<std::mutex> l(_mutex);
_seg_id_to_lsn_ids.emplace(seg_id, std::move(lsn_ids));
}
- void remove_seg(int64_t seg_id) {
+ void remove_segment(int64_t seg_id) {
std::lock_guard<std::mutex> l(_mutex);
_seg_id_to_lsn_ids.erase(seg_id);
}
- std::shared_ptr<const std::vector<int64_t>> get_seg_lsn(int64_t seg_id)
const {
+ ConstAllocatedLsnVectorSharedPtr get_segment_allocated_lsns(int64_t
seg_id) const {
std::lock_guard<std::mutex> l(_mutex);
auto it = _seg_id_to_lsn_ids.find(seg_id);
- CHECK(it != _seg_id_to_lsn_ids.end())
- << "SegmentWriteBinlogLsnMap::get_seg_lsn missing seg_id=" <<
seg_id
+ DCHECK(it != _seg_id_to_lsn_ids.end())
+ << "SegmentAllocatedLsnMap::get_segment_allocated_lsns missing
seg_id=" << seg_id
<< ", existing_seg_ids=[" << ([&] {
std::string s;
for (const auto& [id, _] : _seg_id_to_lsn_ids) {
@@ -201,7 +202,7 @@ public:
private:
mutable std::mutex _mutex;
- std::map<int64_t, std::shared_ptr<std::vector<int64_t>>>
_seg_id_to_lsn_ids;
+ std::map<int64_t, ConstAllocatedLsnVectorSharedPtr> _seg_id_to_lsn_ids;
};
struct SegmentWriteBinlogOptions {
@@ -219,26 +220,6 @@ public:
// input rowset's row-binlog, read LSN from it at publish time
RowsetSharedPtr row_binlog_rowset = nullptr;
} source;
-
- void insert_seg_lsn(int64_t seg_id, std::shared_ptr<std::vector<int64_t>>
lsn_ids) {
- DCHECK(lsn_map != nullptr);
- lsn_map->insert_seg_lsn(seg_id, std::move(lsn_ids));
- }
-
- void remove_seg(int64_t seg_id) {
- DCHECK(lsn_map != nullptr);
- lsn_map->remove_seg(seg_id);
- }
-
- std::shared_ptr<const std::vector<int64_t>> get_seg_lsn(int64_t seg_id)
const {
- DCHECK(lsn_map != nullptr);
- return lsn_map->get_seg_lsn(seg_id);
- }
-
- // Shared LSN storage for row-binlog writers.
- // Keep it as a pointer so SegmentWriteBinlogOptions stays copyable.
- std::shared_ptr<SegmentWriteBinlogLsnMap> lsn_map =
- std::make_shared<SegmentWriteBinlogLsnMap>();
};
} // namespace segment_v2
diff --git a/be/src/storage/rowset/beta_rowset_writer.cpp
b/be/src/storage/rowset/beta_rowset_writer.cpp
index 9a19e30fe7e..c2033c7661d 100644
--- a/be/src/storage/rowset/beta_rowset_writer.cpp
+++ b/be/src/storage/rowset/beta_rowset_writer.cpp
@@ -368,6 +368,9 @@ BetaRowsetWriter::~BetaRowsetWriter() {
Status BaseBetaRowsetWriter::init(const RowsetWriterContext&
rowset_writer_context) {
_context = rowset_writer_context;
DCHECK(_context.tablet_schema != nullptr);
+ // Row-binlog writer or a schema carrying ROW_LSN_COL needs allocated LSN.
+ _context._need_allocate_lsn =
+ _context.write_binlog_opt().enable ||
_context.tablet_schema->row_lsn_col_idx() >= 0;
_rowset_meta.reset(new RowsetMeta);
if (_context.storage_resource) {
_rowset_meta->set_remote_storage_resource(*_context.storage_resource);
diff --git a/be/src/storage/rowset/beta_rowset_writer_v2.cpp
b/be/src/storage/rowset/beta_rowset_writer_v2.cpp
index 0f626e2f5da..0f767f9128e 100644
--- a/be/src/storage/rowset/beta_rowset_writer_v2.cpp
+++ b/be/src/storage/rowset/beta_rowset_writer_v2.cpp
@@ -61,6 +61,10 @@ BetaRowsetWriterV2::~BetaRowsetWriterV2() = default;
Status BetaRowsetWriterV2::init(const RowsetWriterContext&
rowset_writer_context) {
_context = rowset_writer_context;
+ // Row-binlog writer or a schema carrying ROW_LSN_COL needs allocated LSN.
+ _context._need_allocate_lsn =
+ _context.write_binlog_opt().enable ||
+ (_context.tablet_schema != nullptr &&
_context.tablet_schema->row_lsn_col_idx() >= 0);
_context.segment_collector =
std::make_shared<SegmentCollectorT<BetaRowsetWriterV2>>(this);
_context.file_writer_creator =
std::make_shared<FileWriterCreatorT<BetaRowsetWriterV2>>(this);
return Status::OK();
diff --git a/be/src/storage/rowset/group_rowset_writer.cpp
b/be/src/storage/rowset/group_rowset_writer.cpp
index a02a9d81272..f905665811d 100644
--- a/be/src/storage/rowset/group_rowset_writer.cpp
+++ b/be/src/storage/rowset/group_rowset_writer.cpp
@@ -32,6 +32,24 @@ void GroupRowsetWriter::set_row_binlog_writer(
_row_binlog_rowset_writer = row_binlog_rowset_writer;
}
+Status GroupRowsetWriter::init(const RowsetWriterContext&
rowset_writer_context) {
+ DCHECK(_txn_rowset_writer != nullptr);
+ DCHECK(_row_binlog_rowset_writer != nullptr);
+
+ _context = rowset_writer_context;
+ auto& data_ctx =
const_cast<RowsetWriterContext&>(_txn_rowset_writer->context());
+ auto& row_binlog_ctx =
const_cast<RowsetWriterContext&>(_row_binlog_rowset_writer->context());
+ _context._need_allocate_lsn =
+ data_ctx.need_allocated_lsn() ||
row_binlog_ctx.need_allocated_lsn();
+ if (_context.need_allocated_lsn()) {
+ // Share segment LSNs between data and row-binlog writers.
+ _context.allocated_lsn_map =
std::make_shared<segment_v2::SegmentAllocatedLsnMap>();
+ data_ctx.allocated_lsn_map = _context.allocated_lsn_map;
+ row_binlog_ctx.allocated_lsn_map = _context.allocated_lsn_map;
+ }
+ return Status::OK();
+}
+
Status GroupRowsetWriter::flush_rowsets() {
RETURN_IF_ERROR(_txn_rowset_writer->flush());
RETURN_IF_ERROR(_row_binlog_rowset_writer->flush());
diff --git a/be/src/storage/rowset/group_rowset_writer.h
b/be/src/storage/rowset/group_rowset_writer.h
index 8aa2672b7ec..4c0520067ff 100644
--- a/be/src/storage/rowset/group_rowset_writer.h
+++ b/be/src/storage/rowset/group_rowset_writer.h
@@ -39,10 +39,7 @@ public:
RowsetWriterSharedPtr data_writer() { return _txn_rowset_writer; }
- Status init(const RowsetWriterContext& rowset_writer_context) override {
- _context = rowset_writer_context;
- return Status::OK();
- }
+ Status init(const RowsetWriterContext& rowset_writer_context) override;
Status add_block(const Block* block) override {
return Status::Error<ErrorCode::NOT_IMPLEMENTED_ERROR>(
diff --git a/be/src/storage/rowset/rowset_fwd.h
b/be/src/storage/rowset/rowset_fwd.h
index 09778025e96..91f4b54df9d 100644
--- a/be/src/storage/rowset/rowset_fwd.h
+++ b/be/src/storage/rowset/rowset_fwd.h
@@ -19,6 +19,7 @@
#include <memory>
#include <unordered_map>
+#include <vector>
namespace doris {
@@ -36,5 +37,7 @@ class RowsetWriter;
using RowsetWriterSharedPtr = std::shared_ptr<RowsetWriter>;
class RowsetBuilder;
using RowsetBuilderSharedPtr = std::shared_ptr<RowsetBuilder>;
+using AllocatedLsnVectorSharedPtr = std::shared_ptr<std::vector<int64_t>>;
+using ConstAllocatedLsnVectorSharedPtr = std::shared_ptr<const
std::vector<int64_t>>;
} // namespace doris
diff --git a/be/src/storage/rowset/rowset_writer_context.h
b/be/src/storage/rowset/rowset_writer_context.h
index 2390aadeb89..6c36c69ca07 100644
--- a/be/src/storage/rowset/rowset_writer_context.h
+++ b/be/src/storage/rowset/rowset_writer_context.h
@@ -21,9 +21,12 @@
#include <glog/logging.h>
#include <functional>
+#include <memory>
+#include <mutex>
#include <optional>
#include <string_view>
#include <unordered_map>
+#include <vector>
#include "cloud/config.h"
#include "common/status.h"
@@ -180,6 +183,29 @@ struct RowsetWriterContext {
std::string job_id;
+ // Per-segment LSNs allocated before memtable flush. The same storage feeds
+ // both the base row LSN column and row-binlog LSN column.
+ std::shared_ptr<segment_v2::SegmentAllocatedLsnMap> allocated_lsn_map =
nullptr;
+ bool _need_allocate_lsn = false;
+
+ void insert_segment_allocated_lsns(int64_t segment_id,
+ ConstAllocatedLsnVectorSharedPtr
allocated_lsns) {
+ DCHECK(allocated_lsn_map != nullptr);
+ allocated_lsn_map->insert_segment_allocated_lsns(segment_id,
std::move(allocated_lsns));
+ }
+
+ void remove_segment_allocated_lsns(int64_t segment_id) {
+ DCHECK(allocated_lsn_map != nullptr);
+ allocated_lsn_map->remove_segment(segment_id);
+ }
+
+ ConstAllocatedLsnVectorSharedPtr get_segment_allocated_lsns(int64_t
segment_id) const {
+ DCHECK(allocated_lsn_map != nullptr);
+ return allocated_lsn_map->get_segment_allocated_lsns(segment_id);
+ }
+
+ bool need_allocated_lsn() const { return _need_allocate_lsn; }
+
bool is_local_rowset() const { return !storage_resource; }
std::string segment_path(int seg_id) const {
diff --git a/be/src/storage/rowset_builder.cpp
b/be/src/storage/rowset_builder.cpp
index 5c825c3806c..f8fb243494f 100644
--- a/be/src/storage/rowset_builder.cpp
+++ b/be/src/storage/rowset_builder.cpp
@@ -530,6 +530,7 @@ Status GroupRowsetBuilder::init() {
RETURN_IF_ERROR(RowsetFactory::create_empty_group_rowset_writer(&group_writer));
group_writer->set_data_writer(_txn_rs_builder->rowset_writer());
group_writer->set_row_binlog_writer(_row_binlog_rowset_builder->rowset_writer());
+
RETURN_IF_ERROR(group_writer->init(_txn_rs_builder->rowset_writer()->context()));
{
const auto& data_ctx = _txn_rs_builder->rowset_writer()->context();
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index 06232a389fd..c4a487f80ce 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -890,8 +890,7 @@ Status
BaseTablet::calc_segment_delete_bitmap(RowsetSharedPtr rowset,
for (auto p : sort_perm) {
lsn_ids->emplace_back(src.get_data()[p]);
}
-
binlog_ctx.write_binlog_opt().write_binlog_config().insert_seg_lsn(
- segment_id, std::move(lsn_ids));
+ binlog_ctx.insert_segment_allocated_lsns(segment_id,
std::move(lsn_ids));
}
}
RETURN_IF_ERROR(rowset_writer->flush_single_block(&ordered_block,
segment_id));
@@ -1677,6 +1676,7 @@ Status BaseTablet::update_delete_bitmap(const
BaseTabletSPtr& self, TabletTxnInf
RETURN_IF_ERROR(RowsetFactory::create_empty_group_rowset_writer(&group_writer));
group_writer->set_data_writer(data_writer_sp);
group_writer->set_row_binlog_writer(row_binlog_writer_sp);
+
RETURN_IF_ERROR(group_writer->init(group_writer->data_writer()->context()));
transient_rs_writer = std::move(group_writer);
}
diff --git a/be/src/storage/tablet/tablet_meta.cpp
b/be/src/storage/tablet/tablet_meta.cpp
index acaae738d82..21ab367809f 100644
--- a/be/src/storage/tablet/tablet_meta.cpp
+++ b/be/src/storage/tablet/tablet_meta.cpp
@@ -395,6 +395,9 @@ void TabletMeta::init_schema_from_thrift(const
TTabletSchema& tablet_schema,
tablet_schema_pb->set_num_short_key_columns(tablet_schema.short_key_column_count);
tablet_schema_pb->set_num_rows_per_row_block(config::default_num_rows_per_column_file_block);
tablet_schema_pb->set_sequence_col_idx(tablet_schema.sequence_col_idx);
+ if (tablet_schema.__isset.row_lsn_col_idx) {
+ tablet_schema_pb->set_row_lsn_col_idx(tablet_schema.row_lsn_col_idx);
+ }
if (tablet_schema.__isset.binlog_tso_idx) {
tablet_schema_pb->set_binlog_tso_col_idx(tablet_schema.binlog_tso_idx);
}
@@ -587,6 +590,9 @@ void TabletMeta::init_schema_from_thrift(const
TTabletSchema& tablet_schema,
if (tablet_schema.__isset.commit_tso_col_idx) {
tablet_schema_pb->set_commit_tso_col_idx(tablet_schema.commit_tso_col_idx);
}
+ if (tablet_schema.__isset.row_lsn_col_idx) {
+ tablet_schema_pb->set_row_lsn_col_idx(tablet_schema.row_lsn_col_idx);
+ }
if (tablet_schema.__isset.store_row_column) {
tablet_schema_pb->set_store_row_column(tablet_schema.store_row_column);
}
diff --git a/be/src/storage/tablet/tablet_schema.cpp
b/be/src/storage/tablet/tablet_schema.cpp
index 1ac5586aaa8..9e612842a16 100644
--- a/be/src/storage/tablet/tablet_schema.cpp
+++ b/be/src/storage/tablet/tablet_schema.cpp
@@ -874,6 +874,8 @@ void TabletSchema::append_column(TabletColumn column,
ColumnType col_type) {
_skip_bitmap_col_idx = _num_columns;
} else if (UNLIKELY(column.name() == COMMIT_TSO_COL)) {
_commit_tso_col_idx = _num_columns;
+ } else if (UNLIKELY(column.name() == ROW_LSN_COL)) {
+ _row_lsn_col_idx = _num_columns;
} else if (UNLIKELY(column.name() == BINLOG_TSO_COL)) {
_binlog_tso_col_idx = _num_columns;
} else if (UNLIKELY(column.name() == BINLOG_LSN_COL)) {
@@ -1083,6 +1085,7 @@ void TabletSchema::init_from_pb(const TabletSchemaPB&
schema, bool ignore_extrac
_version_col_idx = schema.version_col_idx();
_skip_bitmap_col_idx = schema.skip_bitmap_col_idx();
_commit_tso_col_idx = schema.commit_tso_col_idx();
+ _row_lsn_col_idx = schema.row_lsn_col_idx();
_binlog_tso_col_idx = schema.binlog_tso_col_idx();
_binlog_lsn_col_idx = schema.binlog_lsn_col_idx();
_binlog_op_col_idx = schema.binlog_op_col_idx();
@@ -1190,6 +1193,7 @@ void TabletSchema::shawdow_copy_without_columns(const
TabletSchema& tablet_schem
_version_col_idx = -1;
_skip_bitmap_col_idx = -1;
_commit_tso_col_idx = -1;
+ _row_lsn_col_idx = -1;
_binlog_tso_col_idx = -1;
_binlog_lsn_col_idx = -1;
_binlog_op_col_idx = -1;
@@ -1258,6 +1262,7 @@ void TabletSchema::build_current_tablet_schema(int64_t
index_id, int32_t version
_version_col_idx = -1;
_skip_bitmap_col_idx = -1;
_commit_tso_col_idx = -1;
+ _row_lsn_col_idx = -1;
_binlog_tso_col_idx = -1;
_binlog_lsn_col_idx = -1;
_binlog_op_col_idx = -1;
@@ -1288,6 +1293,8 @@ void TabletSchema::build_current_tablet_schema(int64_t
index_id, int32_t version
_skip_bitmap_col_idx = _num_columns;
} else if (UNLIKELY(column->name() == COMMIT_TSO_COL)) {
_commit_tso_col_idx = _num_columns;
+ } else if (UNLIKELY(column->name() == ROW_LSN_COL)) {
+ _row_lsn_col_idx = _num_columns;
} else if (UNLIKELY(column->name() == BINLOG_TSO_COL)) {
_binlog_tso_col_idx = _num_columns;
} else if (UNLIKELY(column->name() == BINLOG_LSN_COL)) {
@@ -1418,6 +1425,7 @@ void TabletSchema::to_schema_pb(TabletSchemaPB*
tablet_schema_pb) const {
tablet_schema_pb->set_version_col_idx(_version_col_idx);
tablet_schema_pb->set_skip_bitmap_col_idx(_skip_bitmap_col_idx);
tablet_schema_pb->set_commit_tso_col_idx(_commit_tso_col_idx);
+ tablet_schema_pb->set_row_lsn_col_idx(_row_lsn_col_idx);
tablet_schema_pb->set_binlog_tso_col_idx(_binlog_tso_col_idx);
tablet_schema_pb->set_binlog_lsn_col_idx(_binlog_lsn_col_idx);
tablet_schema_pb->set_binlog_op_col_idx(_binlog_op_col_idx);
@@ -1803,6 +1811,7 @@ bool operator==(const TabletSchema& a, const
TabletSchema& b) {
if (a._version_col_idx != b._version_col_idx) return false;
if (a._skip_bitmap_col_idx != b._skip_bitmap_col_idx) return false;
if (a._commit_tso_col_idx != b._commit_tso_col_idx) return false;
+ if (a._row_lsn_col_idx != b._row_lsn_col_idx) return false;
if (a._binlog_tso_col_idx != b._binlog_tso_col_idx) return false;
if (a._binlog_lsn_col_idx != b._binlog_lsn_col_idx) return false;
if (a._binlog_op_col_idx != b._binlog_op_col_idx) return false;
diff --git a/be/src/storage/tablet/tablet_schema.h
b/be/src/storage/tablet/tablet_schema.h
index 9d797d4e6be..ed7646263a7 100644
--- a/be/src/storage/tablet/tablet_schema.h
+++ b/be/src/storage/tablet/tablet_schema.h
@@ -522,6 +522,7 @@ public:
int32_t skip_bitmap_col_idx() const { return _skip_bitmap_col_idx; }
bool is_tso_enabled() const { return _commit_tso_col_idx != -1 ||
_binlog_tso_col_idx != -1; }
int32_t commit_tso_col_idx() const { return _commit_tso_col_idx; }
+ int32_t row_lsn_col_idx() const { return _row_lsn_col_idx; }
int32_t binlog_tso_col_idx() const { return _binlog_tso_col_idx; }
int32_t binlog_lsn_col_idx() const { return _binlog_lsn_col_idx; }
int32_t binlog_op_col_idx() const { return _binlog_op_col_idx; }
@@ -823,6 +824,7 @@ private:
int32_t _version_col_idx = -1;
int32_t _skip_bitmap_col_idx = -1;
int32_t _commit_tso_col_idx = -1;
+ int32_t _row_lsn_col_idx = -1;
int32_t _binlog_tso_col_idx = -1;
int32_t _binlog_lsn_col_idx = -1;
int32_t _binlog_op_col_idx = -1;
diff --git a/be/src/storage/transform/row_binlog_derive.cpp
b/be/src/storage/transform/row_binlog_derive.cpp
index b06db536f5e..3d11f169459 100644
--- a/be/src/storage/transform/row_binlog_derive.cpp
+++ b/be/src/storage/transform/row_binlog_derive.cpp
@@ -191,8 +191,8 @@ Status resolve_binlog_context(TransformExecContext& ctx,
const Block* block,
return Status::InternalError<false>(
"binlog<row> blocks must be flushed through
flush_single_block");
}
- c->lsn_ids = cfg.get_seg_lsn(ctx.segment_id);
- cfg.remove_seg(ctx.segment_id);
+ c->lsn_ids = ctx.rowset_ctx->get_segment_allocated_lsns(ctx.segment_id);
+ ctx.rowset_ctx->remove_segment_allocated_lsns(ctx.segment_id);
CHECK(c->lsn_ids->size() >= c->num_rows) << c->lsn_ids->size() << " vs "
<< c->num_rows;
// Preserve the source writer's layout: system columns may be a prefix or
diff --git a/be/src/storage/transform/row_binlog_derive.h
b/be/src/storage/transform/row_binlog_derive.h
index 2d735cb9075..b7e174e1a74 100644
--- a/be/src/storage/transform/row_binlog_derive.h
+++ b/be/src/storage/transform/row_binlog_derive.h
@@ -17,6 +17,7 @@
#pragma once
+#include "storage/rowset/rowset_fwd.h"
#include "storage/transform/block_transform.h"
namespace doris {
@@ -40,7 +41,7 @@ bool binlog_needs_historical_lookup(const
RowsetWriterContext& context);
struct BinlogDeriveContext {
TabletSchemaSPtr binlog_schema;
TabletSchemaSPtr source_schema;
- std::shared_ptr<const std::vector<int64_t>> lsn_ids;
+ ConstAllocatedLsnVectorSharedPtr lsn_ids;
size_t num_rows = 0;
uint32_t binlog_tso_cid = 0;
uint32_t binlog_lsn_cid = 0;
diff --git a/be/src/storage/utils.h b/be/src/storage/utils.h
index a6ea7a1874a..ee6f71fb6f8 100644
--- a/be/src/storage/utils.h
+++ b/be/src/storage/utils.h
@@ -40,6 +40,7 @@ static const std::string VERSION_COL =
"__DORIS_VERSION_COL__";
static const std::string SKIP_BITMAP_COL = "__DORIS_SKIP_BITMAP_COL__";
static const std::string SEQUENCE_COL = "__DORIS_SEQUENCE_COL__";
static const std::string COMMIT_TSO_COL = "__DORIS_COMMIT_TSO_COL__";
+static const std::string ROW_LSN_COL = "__DORIS_ROW_LSN_COL__";
static const std::string BINLOG_TSO_COL = "__DORIS_BINLOG_TSO__";
static const std::string BINLOG_LSN_COL = "__DORIS_BINLOG_LSN__";
static const std::string BINLOG_OP_COL = "__DORIS_BINLOG_OP__";
diff --git a/be/test/load/memtable/memtable_flush_executor_test.cpp
b/be/test/load/memtable/memtable_flush_executor_test.cpp
index 24bb7dd99e1..b53d0891f8c 100644
--- a/be/test/load/memtable/memtable_flush_executor_test.cpp
+++ b/be/test/load/memtable/memtable_flush_executor_test.cpp
@@ -65,6 +65,9 @@ public:
Status init(const RowsetWriterContext& ctx) override {
_context = ctx;
+ _context._need_allocate_lsn = _context.write_binlog_opt().enable ||
+ (_context.tablet_schema != nullptr &&
+
_context.tablet_schema->row_lsn_col_idx() >= 0);
return Status::OK();
}
@@ -80,6 +83,12 @@ public:
std::this_thread::sleep_for(std::chrono::milliseconds(_flush_delay_ms));
}
_last_segment_id = segment_id;
+ // Capture LSN during flush, or it is released after flush.
+ if (_context.need_allocated_lsn()) {
+ _last_seg_lsn = _context.get_segment_allocated_lsns(segment_id);
+ EXPECT_NE(_last_seg_lsn, nullptr);
+ EXPECT_EQ(_last_seg_lsn->size(), block->rows());
+ }
++(*_flush_cnt);
*flush_size = 1;
if (_fail_on_flush) {
@@ -113,6 +122,8 @@ public:
int32_t last_segment_id() const { return _last_segment_id; }
+ ConstAllocatedLsnVectorSharedPtr last_seg_lsn() const { return
_last_seg_lsn; }
+
private:
std::atomic<int>* _flush_cnt;
bool _fail_on_flush;
@@ -120,6 +131,7 @@ private:
int _flush_delay_ms;
int32_t _next_segment_id = 0;
int32_t _last_segment_id = -1;
+ ConstAllocatedLsnVectorSharedPtr _last_seg_lsn = nullptr;
};
struct GroupFlushTestContext {
@@ -265,7 +277,7 @@ protected:
TabletAddRowsPayload rows;
for (size_t i = 0; i < lsns.size(); ++i) {
rows.row_idxs.emplace_back(i);
- rows.row_binlog_lsns.emplace_back(lsns[i]);
+ rows.allocated_lsns.emplace_back(lsns[i]);
}
ASSERT_TRUE(ctx->memtable->insert(&block, rows).ok());
}
@@ -467,9 +479,7 @@ TEST_F(MemTableFlushExecutorGroupFlushTest,
TestGroupFlushToken) {
EXPECT_EQ(1, data_flush_cnt.load());
EXPECT_EQ(1, binlog_flush_cnt.load());
EXPECT_EQ(data_writer->last_segment_id(),
binlog_writer->last_segment_id());
- auto seg_lsn =
-
binlog_writer->context().write_binlog_opt().write_binlog_config().get_seg_lsn(
- binlog_writer->last_segment_id());
+ auto seg_lsn = binlog_writer->last_seg_lsn();
ASSERT_NE(seg_lsn, nullptr);
ASSERT_EQ(ctx.memtable->raw_rows(), seg_lsn->size());
EXPECT_EQ(1000, (*seg_lsn)[0]);
diff --git a/be/test/olap/rowset/cloud_group_rowset_builder_writer_test.cpp
b/be/test/olap/rowset/cloud_group_rowset_builder_writer_test.cpp
index 7f337a5ded6..bc6572d1a4a 100644
--- a/be/test/olap/rowset/cloud_group_rowset_builder_writer_test.cpp
+++ b/be/test/olap/rowset/cloud_group_rowset_builder_writer_test.cpp
@@ -296,9 +296,10 @@ protected:
cfg.source.source_write_type = DataWriteType::TYPE_DIRECT;
auto lsn_buffer = AutoIncIDBuffer::create_shared(1, 1,
kBinlogLsnAutoIncId);
lsn_buffer->append_range_for_test(1000, num_rows);
- auto lsn_ids = std::make_shared<std::vector<int64_t>>();
- RETURN_IF_ERROR(allocate_binlog_lsn(lsn_buffer, num_rows, *lsn_ids));
- cfg.insert_seg_lsn(0, lsn_ids);
+ std::vector<int64_t> allocated_lsns;
+ RETURN_IF_ERROR(allocate_lsn(lsn_buffer, num_rows, allocated_lsns));
+ auto lsn_ids =
std::make_shared<std::vector<int64_t>>(allocated_lsns.begin(),
+
allocated_lsns.end());
auto row_binlog_writer_res =
_row_binlog_tablet->create_rowset_writer(row_binlog_context,
false);
@@ -313,6 +314,10 @@ protected:
(*group_writer)
->set_row_binlog_writer(
std::shared_ptr<RowsetWriter>(std::move(row_binlog_writer_res.value())));
+
RETURN_IF_ERROR((*group_writer)->init((*group_writer)->data_writer()->context()));
+ auto& group_binlog_ctx =
+
const_cast<RowsetWriterContext&>((*group_writer)->row_binlog_writer()->context());
+ group_binlog_ctx.insert_segment_allocated_lsns(0, lsn_ids);
return Status::OK();
}
diff --git a/be/test/olap/rowset/group_rowset_writer_test.cpp
b/be/test/olap/rowset/group_rowset_writer_test.cpp
index 4047dc53b22..604c8366c41 100644
--- a/be/test/olap/rowset/group_rowset_writer_test.cpp
+++ b/be/test/olap/rowset/group_rowset_writer_test.cpp
@@ -187,8 +187,10 @@ protected:
auto lsn_buffer = AutoIncIDBuffer::create_shared(1, 1,
kBinlogLsnAutoIncId);
lsn_buffer->append_range_for_test(1000, num_rows);
auto lsn_ids = std::make_shared<std::vector<int64_t>>();
- RETURN_IF_ERROR_RESULT(allocate_binlog_lsn(lsn_buffer, num_rows,
*lsn_ids));
- binlog_options.insert_seg_lsn(0, lsn_ids);
+ RETURN_IF_ERROR_RESULT(allocate_lsn(lsn_buffer, num_rows, *lsn_ids));
+ row_binlog_context.allocated_lsn_map =
+ std::make_shared<segment_v2::SegmentAllocatedLsnMap>();
+ row_binlog_context.insert_segment_allocated_lsns(0, lsn_ids);
return _row_binlog_tablet->create_rowset_writer(row_binlog_context,
false);
}
@@ -220,6 +222,11 @@ protected:
std::shared_ptr<RowsetWriter>(std::move(data_writer_result.value())));
group_writer->set_row_binlog_writer(
std::shared_ptr<RowsetWriter>(std::move(row_binlog_writer_result.value())));
+ auto& row_binlog_context =
+
const_cast<RowsetWriterContext&>(group_writer->row_binlog_writer()->context());
+ auto lsn_ids = row_binlog_context.get_segment_allocated_lsns(0);
+
RETURN_IF_ERROR_RESULT(group_writer->init(group_writer->data_writer()->context()));
+ row_binlog_context.insert_segment_allocated_lsns(0, lsn_ids);
return group_writer;
}
@@ -256,9 +263,10 @@ protected:
cfg.source.source_write_type = DataWriteType::TYPE_DIRECT;
auto lsn_buffer = AutoIncIDBuffer::create_shared(1, 1,
kBinlogLsnAutoIncId);
lsn_buffer->append_range_for_test(1000, num_rows);
- auto lsn_ids = std::make_shared<std::vector<int64_t>>();
- RETURN_IF_ERROR(allocate_binlog_lsn(lsn_buffer, num_rows, *lsn_ids));
- cfg.insert_seg_lsn(0, lsn_ids);
+ std::vector<int64_t> allocated_lsns;
+ RETURN_IF_ERROR(allocate_lsn(lsn_buffer, num_rows, allocated_lsns));
+ auto lsn_ids =
std::make_shared<std::vector<int64_t>>(allocated_lsns.begin(),
+
allocated_lsns.end());
auto row_binlog_writer_res =
_row_binlog_tablet->create_rowset_writer(row_binlog_context,
false);
if (!row_binlog_writer_res.has_value()) {
@@ -272,6 +280,10 @@ protected:
(*group_writer)
->set_row_binlog_writer(
std::shared_ptr<RowsetWriter>(std::move(row_binlog_writer_res.value())));
+
RETURN_IF_ERROR((*group_writer)->init((*group_writer)->data_writer()->context()));
+ auto& group_binlog_ctx =
+
const_cast<RowsetWriterContext&>((*group_writer)->row_binlog_writer()->context());
+ group_binlog_ctx.insert_segment_allocated_lsns(0, lsn_ids);
return Status::OK();
}
diff --git a/be/test/storage/mow/mow_transform_test_base.h
b/be/test/storage/mow/mow_transform_test_base.h
index 3dccf20ffe1..803f13f838c 100644
--- a/be/test/storage/mow/mow_transform_test_base.h
+++ b/be/test/storage/mow/mow_transform_test_base.h
@@ -475,8 +475,7 @@ protected:
}
// The per-segment LSN range the binlog derive consumes (one LSN per row).
- static std::shared_ptr<std::vector<int64_t>> make_seg_lsn(size_t num_rows,
- int64_t start =
1000) {
+ static AllocatedLsnVectorSharedPtr make_seg_lsn(size_t num_rows, int64_t
start = 1000) {
auto lsn_ids = std::make_shared<std::vector<int64_t>>();
for (size_t i = 0; i < num_rows; ++i) {
lsn_ids->push_back(start + static_cast<int64_t>(i));
diff --git a/be/test/storage/rowset/segment_flusher_format_test.cpp
b/be/test/storage/rowset/segment_flusher_format_test.cpp
index 92ff3bdfb7b..a29164599cc 100644
--- a/be/test/storage/rowset/segment_flusher_format_test.cpp
+++ b/be/test/storage/rowset/segment_flusher_format_test.cpp
@@ -3104,6 +3104,7 @@ protected:
source_mow_context = make_mow_context(source_tablet->tablet_id(),
history);
}
context.write_binlog_opt().enable = true;
+ context.allocated_lsn_map =
std::make_shared<segment_v2::SegmentAllocatedLsnMap>();
context.write_binlog_opt().set_need_before(need_before);
auto& options = context.write_binlog_opt().write_binlog_config();
options.source.tablet_schema = source_tablet->tablet_schema();
@@ -3117,7 +3118,7 @@ protected:
for (int64_t row = 0; row < 3; ++row) {
lsn_ids->push_back(1000 + segment_id * 100 + row);
}
- options.insert_seg_lsn(segment_id, std::move(lsn_ids));
+ context.insert_segment_allocated_lsns(segment_id,
std::move(lsn_ids));
}
}
diff --git a/be/test/storage/transform/row_binlog_derive_test.cpp
b/be/test/storage/transform/row_binlog_derive_test.cpp
index b9df3aec1d5..4ea6bb7ab17 100644
--- a/be/test/storage/transform/row_binlog_derive_test.cpp
+++ b/be/test/storage/transform/row_binlog_derive_test.cpp
@@ -308,6 +308,11 @@ protected:
ctx.segment_id = 0;
return ctx;
}
+
+ void register_segment_lsns(RowsetWriterContext& rwc, size_t num_rows) {
+ rwc.allocated_lsn_map =
std::make_shared<segment_v2::SegmentAllocatedLsnMap>();
+ rwc.insert_segment_allocated_lsns(0, make_seg_lsn(num_rows));
+ }
};
// ===========================================================================
@@ -334,7 +339,7 @@ TEST_F(RowBinlogDeriveTest,
PlainAppendAndDeleteWithAfterAndLsn) {
cfg.source.source_write_type = DataWriteType::TYPE_DIRECT;
cfg.source.is_transient_rowset_writer = false;
cfg.write_before = false; // no PU, no BEFORE -> plain derive
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"PlainRowBinlogDerive"}));
@@ -400,7 +405,7 @@ TEST_F(RowBinlogDeriveTest,
PlainNoDeleteSignColumnAllAppend) {
cfg.source.source_write_type = DataWriteType::TYPE_DIRECT;
cfg.source.is_transient_rowset_writer = false;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"PlainRowBinlogDerive"}));
@@ -491,7 +496,7 @@ TEST_F(RowBinlogDeriveTest,
PlainMapsHiddenKeysAndSkipsHiddenNonKeys) {
cfg.source.tablet_schema = source_schema;
cfg.source.source_write_type = DataWriteType::TYPE_DIRECT;
cfg.source.is_transient_rowset_writer = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
Block block = source_schema->create_storage_block();
{
@@ -578,7 +583,7 @@ TEST_F(RowBinlogDeriveTest,
MowUpdateAndAppendTakesHistoryV2) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"MowRowBinlogDerive"}));
@@ -722,7 +727,7 @@ TEST_F(RowBinlogDeriveTest,
MowHiddenKeyColumnIsPartOfTheProbeKey) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(1));
+ register_segment_lsns(rwc, 1);
auto chain = build_transform_chain(rwc);
TransformExecContext ctx = exec_ctx(binlog_schema, &rwc);
@@ -800,7 +805,7 @@ TEST_F(RowBinlogDeriveTest,
MowPartialUpdateWithBeforeImage) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = true;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"MowRowBinlogDerive"}));
@@ -893,7 +898,7 @@ TEST_F(RowBinlogDeriveTest,
MowInsertAfterDeletedHistoryRowIsAppend) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
TransformExecContext ctx = exec_ctx(binlog_schema, &rwc);
@@ -979,7 +984,7 @@ TEST_F(RowBinlogDeriveTest,
MowLooksUpHistoryInBaseTabletNotBinlogTablet) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(1));
+ register_segment_lsns(rwc, 1);
auto chain = build_transform_chain(rwc);
TransformExecContext ctx = exec_ctx(binlog_schema, &rwc);
@@ -1053,7 +1058,7 @@ TEST_F(RowBinlogDeriveTest, MowDeleteExistingAndNewKey) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"MowRowBinlogDerive"}));
@@ -1136,7 +1141,7 @@ TEST_F(RowBinlogDeriveTest, MowBeforeImageMirrorsHistory)
{
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = true;
- cfg.insert_seg_lsn(0, make_seg_lsn(3));
+ register_segment_lsns(rwc, 3);
// setup_retriever_and_lookup requires partial_update_info on the retriever
// context (build_after_block / retrieve_historical_row DCHECK it). For a
@@ -1258,7 +1263,7 @@ TEST_F(RowBinlogDeriveTest,
MowBeforeNoopZeroValueColumns) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = true;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"MowRowBinlogDerive"}));
@@ -1346,7 +1351,7 @@ TEST_F(RowBinlogDeriveTest,
MowSeqSourceSeqLosesStillReadsHistory) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"MowRowBinlogDerive"}));
@@ -1420,7 +1425,7 @@ TEST_F(RowBinlogDeriveTest,
MowRejectsFlexiblePartialUpdate) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(2));
+ register_segment_lsns(rwc, 2);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"MowRowBinlogDerive"}));
@@ -1490,7 +1495,7 @@ TEST_F(RowBinlogDeriveTest,
MowFixedPartialUpdateRejectsBadWidth) {
// too narrow: no columns (0 < num_key_columns 1)
{
- cfg.insert_seg_lsn(0, make_seg_lsn(1));
+ register_segment_lsns(rwc, 1);
TransformExecContext ctx = make_ctx();
Block block;
auto st = chain.apply(ctx, &block);
@@ -1500,7 +1505,7 @@ TEST_F(RowBinlogDeriveTest,
MowFixedPartialUpdateRejectsBadWidth) {
}
// too wide: full width including hidden delete_sign (4 >= num_columns 4)
{
- cfg.insert_seg_lsn(0, make_seg_lsn(1));
+ register_segment_lsns(rwc, 1);
TransformExecContext ctx = make_ctx();
Block block = source_schema->create_storage_block(); // 4 columns
block.get_by_position(0).column->assert_mutable()->insert_default();
@@ -1531,7 +1536,7 @@ TEST_F(RowBinlogDeriveTest, RejectsMissingSourceSchema) {
cfg.source.source_write_type = DataWriteType::TYPE_DIRECT;
cfg.source.is_transient_rowset_writer = false;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(1));
+ register_segment_lsns(rwc, 1);
auto chain = build_transform_chain(rwc); // no PU, no BEFORE -> plain
derive
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"PlainRowBinlogDerive"}));
@@ -1568,7 +1573,7 @@ TEST_F(RowBinlogDeriveTest, RejectsNegativeSegmentId) {
cfg.source.source_write_type = DataWriteType::TYPE_DIRECT;
cfg.source.is_transient_rowset_writer = false;
cfg.write_before = false;
- cfg.insert_seg_lsn(0, make_seg_lsn(1));
+ register_segment_lsns(rwc, 1);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"PlainRowBinlogDerive"}));
@@ -1607,7 +1612,7 @@ TEST_F(RowBinlogDeriveTest,
RejectsSchemaAndWriteBeforeDisagreement) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = make_mow_context(100, {});
cfg.write_before = true; // the schema has no BEFORE columns
- cfg.insert_seg_lsn(0, make_seg_lsn(1));
+ register_segment_lsns(rwc, 1);
auto chain = build_transform_chain(rwc);
TransformExecContext ctx = exec_ctx(binlog_schema, &rwc);
diff --git a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
index 91289f09e17..7ce5d2a3f9d 100644
--- a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
+++ b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
@@ -62,9 +62,11 @@ public class Column implements GsonPostProcessable {
public static final String ICEBERG_ROWID_COL =
"__DORIS_ICEBERG_ROWID_COL__";
// For time-travel (FOR VERSION/TIME AS OF) on duplicate / mow tables with
row binlog enabled.
public static final String COMMIT_TSO_COL = "__DORIS_COMMIT_TSO_COL__";
+ public static final String ROW_LSN_COL = "__DORIS_ROW_LSN_COL__";
// table stream columns
public static final String STREAM_CHANGE_TYPE_COL =
"__DORIS_STREAM_CHANGE_TYPE_COL__";
public static final String STREAM_SEQ_COL =
"__DORIS_STREAM_SEQUENCE_COL__";
+ public static final String STREAM_LSN_COL = "__DORIS_STREAM_LSN_COL__";
// NOTE: you should name hidden column start with '__DORIS_'
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
private static final String COLUMN_ARRAY_CHILDREN = "item";
@@ -75,6 +77,8 @@ public class Column implements GsonPostProcessable {
private static final String COLUMN_MAP_VALUE = "value";
public static final Column STREAM_SEQ_VIRTUAL_COLUMN =
new Column(STREAM_SEQ_COL, Type.BIGINT, false, null, true, null,
false);
+ public static final Column STREAM_LSN_VIRTUAL_COLUMN =
+ new Column(STREAM_LSN_COL, Type.BIGINT, false, null, true, null,
false);
public static final Column STREAM_CHANGE_TYPE_VIRTUAL_COLUMN =
new Column(STREAM_CHANGE_TYPE_COL, Type.STRING, false, null, true,
null, false);
@@ -531,6 +535,10 @@ public class Column implements GsonPostProcessable {
return !visible && aggregationType == AggregateType.NONE &&
nameEquals(COMMIT_TSO_COL, true);
}
+ public boolean isRowLsnColumn() {
+ return !visible && aggregationType == AggregateType.NONE &&
nameEquals(ROW_LSN_COL, true);
+ }
+
// now we only support BloomFilter on (same behavior with BE):
// smallint/int/bigint/largeint
// string/varchar/char/variant
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
index 0aee3ea4f8e..eec6e523ad1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
@@ -2389,6 +2389,11 @@ public class OlapTable extends Table implements
MTMVRelatedTableIf, GsonPostProc
return getBinlogConfig().isEnableForStreaming();
}
+ // Whether the base table physically stores the row LSN column (dup table
with row binlog).
+ public boolean hasRowLsnColumn() {
+ return getBaseSchema(true).stream().anyMatch(Column::isRowLsnColumn);
+ }
+
public void createNewRowBinlogMeta(IdGeneratorBuffer idGeneratorBuffer,
long dbId)
throws DdlException {
writeLock();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
index 105015626df..619b1e60a9f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java
@@ -19,6 +19,7 @@ package org.apache.doris.catalog.stream;
import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.Type;
import org.apache.doris.common.DdlException;
@@ -61,6 +62,13 @@ public class TableStreamBuildFactory {
Column sequenceColumn = new Column(Column.STREAM_SEQ_COL, Type.BIGINT);
sequenceColumn.setIsVisible(false);
schema.add(sequenceColumn);
+ // Only expose stream LSN when the base table stores row LSN, e.g. dup
table with binlog.
+ if (params.baseTable instanceof OlapTable
+ && ((OlapTable) params.baseTable).hasRowLsnColumn()) {
+ Column lsnColumn = new Column(Column.STREAM_LSN_COL, Type.BIGINT);
+ lsnColumn.setIsVisible(false);
+ schema.add(lsnColumn);
+ }
Column changeTypeColumn = new Column(Column.STREAM_CHANGE_TYPE_COL,
Type.VARCHAR);
changeTypeColumn.setIsVisible(false);
schema.add(changeTypeColumn);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
index 5c604c94677..59dd30ca988 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
@@ -508,6 +508,7 @@ public class CloudInternalCatalog extends InternalCatalog {
int deleteSign = -1;
int sequenceCol = -1;
int commitTsoCol = -1;
+ int rowLsnCol = -1;
for (int i = 0; i < schemaColumns.size(); i++) {
Column column = schemaColumns.get(i);
if (column.isDeleteSignColumn()) {
@@ -519,10 +520,14 @@ public class CloudInternalCatalog extends InternalCatalog
{
if (column.isCommitTsoColumn()) {
commitTsoCol = i;
}
+ if (column.isRowLsnColumn()) {
+ rowLsnCol = i;
+ }
}
schemaBuilder.setDeleteSignIdx(deleteSign);
schemaBuilder.setSequenceColIdx(sequenceCol);
schemaBuilder.setCommitTsoColIdx(commitTsoCol);
+ schemaBuilder.setRowLsnColIdx(rowLsnCol);
schemaBuilder.setStoreRowColumn(storeRowColumn);
if (dataSortInfo.getSortType() == TSortType.LEXICAL) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index 65a426da75f..b921b5ec970 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -2404,7 +2404,7 @@ public class InternalCatalog implements
CatalogIf<Database> {
throw new DdlException("Cannot create temporary table with binlog
enable");
}
createTableInfo.getProperties().putAll(createTableBinlogConfig.toProperties());
-
createTableInfo.createCommitTSOColumnIfNecessary(createTableBinlogConfig);
+
createTableInfo.createRowBinlogHiddenColumnsIfNecessary(createTableBinlogConfig);
// get keys type
KeysDesc keysDesc = createTableInfo.getKeysDesc();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
index 2a2d0e9b848..7bebf5b0d78 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
@@ -69,8 +69,8 @@ import java.util.Set;
import java.util.stream.Collectors;
/**
- * 1. remove STREAM_CHANGE_TYPE_VIRTUAL_COLUMN & STREAM_SEQ_VIRTUAL_COLUMN
from olap table stream scan output
- * with alias projection
+ * 1. remove STREAM_CHANGE_TYPE_VIRTUAL_COLUMN / STREAM_SEQ_VIRTUAL_COLUMN /
STREAM_LSN_VIRTUAL_COLUMN
+ * from olap table stream scan output with alias projection
* 2. add delete sign column if unique base table
*/
public class NormalizeOlapTableStreamScan extends OneRewriteRuleFactory {
@@ -93,6 +93,12 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
new VarcharLiteral("UPDATE_AFTER"))), new
VarcharLiteral("UNKNOWN"));
}
+ private static boolean isStreamVirtualSlot(Slot slot, Column
virtualColumn) {
+ return slot instanceof SlotReference
+ && ((SlotReference) slot).getOriginalColumn().isPresent()
+ && ((SlotReference)
slot).getOriginalColumn().get().equals(virtualColumn);
+ }
+
private Plan normalize(LogicalOlapTableStreamScan scan, CascadesContext
cascadesContext) {
// short-cut for empty partition
if (scan.getSelectedPartitionIds().isEmpty()) {
@@ -152,8 +158,8 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
*
* <p>{@code isIncremental} distinguishes the two callers:
* <ul>
- * <li>true — normal stream consumption: map the binlog op/timestamp
columns into the stream
- * virtual columns STREAM_CHANGE_TYPE_COL /
STREAM_SEQ_COL.</li>
+ * <li>true — normal stream consumption: map the binlog
op/timestamp/lsn columns into the
+ * stream virtual columns STREAM_CHANGE_TYPE_COL /
STREAM_SEQ_COL / STREAM_LSN_COL.</li>
* <li>false — snapshot rebuild: only keep DELETE & UPDATE_BEFORE
rows so they can be added
* back to reconstruct the "before" image at the snapshot
point.</li>
* </ul>
@@ -180,9 +186,12 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
// project stream virtual slot from binlog
Slot opSlot = null;
Slot seqSlot = null;
+ Slot lsnSlot = null;
for (int i = 0; i < binlogOutputSlots.size(); i++) {
if
(binlogOutputSlots.get(i).getName().equals(Column.BINLOG_TSO_COL)) {
seqSlot = binlogOutputSlots.get(i);
+ } else if
(binlogOutputSlots.get(i).getName().equals(Column.BINLOG_LSN_COL)) {
+ lsnSlot = binlogOutputSlots.get(i);
} else if
(binlogOutputSlots.get(i).getName().equals(Column.BINLOG_OPERATION_COL)) {
opSlot = binlogOutputSlots.get(i);
}
@@ -197,17 +206,15 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
if (isIncremental) {
// replace stream virtual column with alias slot reference
for (Slot slot : originSlots) {
- if (slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN)) {
+ if (isStreamVirtualSlot(slot,
Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN)) {
project.add(new
Alias(StatementScopeIdGenerator.newExprId(), buildChangeTypeExpr(opSlot),
Column.STREAM_CHANGE_TYPE_COL));
- } else if (slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_SEQ_VIRTUAL_COLUMN)) {
+ } else if (isStreamVirtualSlot(slot,
Column.STREAM_SEQ_VIRTUAL_COLUMN)) {
+ Preconditions.checkArgument(seqSlot != null, "Commit tso
column not found in binlog output");
project.add(new
Alias(StatementScopeIdGenerator.newExprId(), seqSlot, Column.STREAM_SEQ_COL));
+ } else if (isStreamVirtualSlot(slot,
Column.STREAM_LSN_VIRTUAL_COLUMN)) {
+ Preconditions.checkArgument(lsnSlot != null, "Row lsn
column not found in binlog output");
+ project.add(new
Alias(StatementScopeIdGenerator.newExprId(), lsnSlot, Column.STREAM_LSN_COL));
}
}
} else {
@@ -337,12 +344,12 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
* <p>Selected partitions are split into:
* <ul>
* <li>historical partitions — never consumed history data; scanned from
the base table and all
- * rows are treated as APPEND (change type = "APPEND", seq = commit
tso);</li>
+ * rows are treated as APPEND (change type = "APPEND", seq = commit
tso, lsn = row lsn);</li>
* <li>incremental partitions — read row-level changes from binlog via
* {@link #makeIncrementalScanFromBinlog}.</li>
* </ul>
- * The two plans are unioned. {@code notVirtualSlots} are the origin
output slots excluding the
- * two stream virtual columns (STREAM_CHANGE_TYPE / STREAM_SEQ), which are
filled separately.
+ * The two plans are unioned. {@code notVirtualSlots} are the origin
output slots excluding
+ * stream virtual columns (STREAM_CHANGE_TYPE / STREAM_SEQ / STREAM_LSN),
which are filled separately.
*/
private Plan makeTableStreamScan(LogicalOlapTableStreamScan scan,
CascadesContext cascadesContext) {
OlapTableStreamWrapper streamWrapper = scan.getTable();
@@ -353,16 +360,11 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
Plan historyPlan = null;
Plan incrementalPlan = null;
List<Slot> originSlots = scan.getOutput();
- // notVirtualSlots = originSlots - (STREAM_CHANGE_TYPE_VIRTUAL_COLUMN
+ STREAM_SEQ_VIRTUAL_COLUMN)
+ // notVirtualSlots = originSlots - stream virtual columns
List<Slot> notVirtualSlots = originSlots.stream()
- .filter(slot -> !(slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN)))
- .filter(slot -> !(slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_SEQ_VIRTUAL_COLUMN)))
+ .filter(slot -> !isStreamVirtualSlot(slot,
Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN))
+ .filter(slot -> !isStreamVirtualSlot(slot,
Column.STREAM_SEQ_VIRTUAL_COLUMN))
+ .filter(slot -> !isStreamVirtualSlot(slot,
Column.STREAM_LSN_VIRTUAL_COLUMN))
.collect(ImmutableList.toImmutableList());
// history plan
@@ -371,30 +373,33 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
Plan plan = makeOlapScanOnBaseTable(scan, cascadesContext,
baseTable, historicalPartitionIds);
List<Slot> baseOutputSlots = plan.getOutput();
Slot tsoSlot = null;
+ Slot lsnSlot = null;
for (Slot slot : baseOutputSlots) {
if (slot.getName().equals(Column.COMMIT_TSO_COL)) {
tsoSlot = slot;
+ } else if (slot.getName().equals(Column.ROW_LSN_COL)) {
+ lsnSlot = slot;
}
- if (tsoSlot != null) {
+ if (tsoSlot != null && lsnSlot != null) {
break;
}
}
Preconditions.checkArgument(tsoSlot != null, "Commit tso column
not found in base table output");
+ if (baseTable.hasRowLsnColumn()) {
+ Preconditions.checkArgument(lsnSlot != null, "Row lsn column
not found in base table output");
+ }
List<NamedExpression> project =
mapOriginOutputFromChild(notVirtualSlots, baseOutputSlots, false);
for (Slot slot : originSlots) {
- if (slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN)) {
+ if (isStreamVirtualSlot(slot,
Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN)) {
project.add(new
Alias(StatementScopeIdGenerator.newExprId(), new VarcharLiteral("APPEND"),
Column.STREAM_CHANGE_TYPE_COL));
}
- if (slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_SEQ_VIRTUAL_COLUMN)) {
+ if (isStreamVirtualSlot(slot,
Column.STREAM_SEQ_VIRTUAL_COLUMN)) {
project.add(new
Alias(StatementScopeIdGenerator.newExprId(), tsoSlot, Column.STREAM_SEQ_COL));
}
+ if (isStreamVirtualSlot(slot,
Column.STREAM_LSN_VIRTUAL_COLUMN)) {
+ project.add(new
Alias(StatementScopeIdGenerator.newExprId(), lsnSlot, Column.STREAM_LSN_COL));
+ }
}
historyPlan = new LogicalProject<>(project, plan);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java
index 77c12f8714b..c828b7e3176 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java
@@ -724,6 +724,18 @@ public class ColumnDefinition {
return columnDefinition;
}
+ /**
+ * add hidden column __DORIS_ROW_LSN_COL__ for stable row identity on
row-binlog tables.
+ */
+ public static ColumnDefinition newRowLsnColumnDefinition(AggregateType
aggregateType) {
+ ColumnDefinition columnDefinition = new
ColumnDefinition(Column.ROW_LSN_COL, BigIntType.INSTANCE, false,
+ aggregateType, false, Optional.of(new
DefaultValue(DefaultValue.ZERO_NUMBER)),
+ "doris row lsn hidden column", false);
+ columnDefinition.setEnableAddHiddenColumn(true);
+
+ return columnDefinition;
+ }
+
/**
* used in CreateTableInfo.validate(), specify the default value as
DefaultValue.NULL_DEFAULT_VALUE
* becasue ColumnDefinition.validate() will check that bitmap type column
don't set default value
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
index 32b0d1c3869..993cf820c64 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
@@ -1628,16 +1628,17 @@ public class CreateTableInfo {
}
/**
- * check if add Commit TSO Column
+ * Add hidden columns required by row binlog.
*/
- public void createCommitTSOColumnIfNecessary(BinlogConfig binlogConfig) {
- // __DORIS_COMMIT_TSO_COL__ injection for time-travel:
- // only on dup / mow tables with row binlog enabled
(binlog.enable=true && binlog.format=ROW).
- if (keysType.equals(KeysType.DUP_KEYS)
- || (keysType.equals(KeysType.UNIQUE_KEYS) &&
isEnableMergeOnWrite)) {
- if (binlogConfig.isRowFormat()) {
-
columns.add(ColumnDefinition.newCommitTsoColumnDefinition(AggregateType.NONE));
- }
+ public void createRowBinlogHiddenColumnsIfNecessary(BinlogConfig
binlogConfig) {
+ if (!binlogConfig.isRowFormat()) {
+ return;
+ }
+ if (keysType.equals(KeysType.DUP_KEYS)) {
+
columns.add(ColumnDefinition.newCommitTsoColumnDefinition(AggregateType.NONE));
+
columns.add(ColumnDefinition.newRowLsnColumnDefinition(AggregateType.NONE));
+ } else if (keysType.equals(KeysType.UNIQUE_KEYS) &&
isEnableMergeOnWrite) {
+
columns.add(ColumnDefinition.newCommitTsoColumnDefinition(AggregateType.NONE));
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
index b490684e858..48c40ce192a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
@@ -19,6 +19,7 @@ package org.apache.doris.nereids.trees.plans.logical;
import org.apache.doris.analysis.TableScanParams;
import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.constraint.TableIdentifier;
@@ -129,9 +130,13 @@ public class LogicalOlapTableStreamScan extends
LogicalOlapScan {
if (cachedOutput.isPresent()) {
return cachedOutput.get();
}
- // for reset, we could use get full schema of base table;
- // otherwise, we only need to get the schema without hidden columns
- List<Column> baseSchema = table.getBaseSchema(readMode ==
StreamReadMode.RESET);
+ // RESET and DUP_KEYS SNAPSHOT are rebuilt from the base table
directly (no binlog union),
+ // so use the full schema to expose hidden columns like ROW_LSN_COL.
Others only need visible.
+ boolean useFullSchemaScan = readMode == StreamReadMode.RESET
+ || (readMode == StreamReadMode.SNAPSHOT
+ && table instanceof OlapTable
+ && ((OlapTable) table).getKeysType() ==
KeysType.DUP_KEYS);
+ List<Column> baseSchema = table.getBaseSchema(useFullSchemaScan);
List<SlotReference> slotFromColumn = createSlotsVectorized(baseSchema);
ImmutableList.Builder<Slot> slots = ImmutableList.builder();
@@ -144,16 +149,16 @@ public class LogicalOlapTableStreamScan extends
LogicalOlapScan {
continue;
}
Pair<Long, String> key = Pair.of(selectedIndexId, col.getName());
- // For INCREMENTAL / SNAPSHOT reads, non-key value columns are
materialized from the
- // base table row-binlog whose after/before value columns are
always nullable (see
+ // For INCREMENTAL / SNAPSHOT(MOW) reads, non-key value columns
are materialized from
+ // the base table row-binlog whose after/before value columns are
always nullable (see
// Column.generateAfterValueColumn / generateBeforeValueColumn).
Declare these value
// columns as nullable here so the stream scan output stays
consistent with the plan
// expanded in NormalizeOlapTableStreamScan, otherwise
AdjustNullable reports a
- // not-nullable -> nullable conflict. RESET does a full base-table
scan, so keep its
- // original nullability.
+ // not-nullable -> nullable conflict. Full base scans (RESET /
SNAPSHOT(DUP)) do a full
+ // base-table scan, so keep their original nullability.
Slot slot = cacheSlotWithSlotName.computeIfAbsent(key, k -> {
SlotReference slotRef = slotFromColumn.get(index);
- boolean forceNullable = readMode != StreamReadMode.RESET &&
!baseSchema.get(index).isKey();
+ boolean forceNullable = !useFullSchemaScan &&
!baseSchema.get(index).isKey();
return forceNullable ? slotRef.withNullable(true) : slotRef;
});
slots.add(slot);
@@ -174,6 +179,11 @@ public class LogicalOlapTableStreamScan extends
LogicalOlapScan {
// add stream exclusive virtual columns.
slots.add(SlotReference.fromColumn(
exprIdGenerator.getNextId(), table,
Column.STREAM_SEQ_VIRTUAL_COLUMN, qualified()));
+ // Only expose stream LSN when the base table stores row LSN, e.g.
dup table with binlog.
+ if (table instanceof OlapTable && ((OlapTable)
table).hasRowLsnColumn()) {
+ slots.add(SlotReference.fromColumn(
+ exprIdGenerator.getNextId(), table,
Column.STREAM_LSN_VIRTUAL_COLUMN, qualified()));
+ }
slots.add(SlotReference.fromColumn(
exprIdGenerator.getNextId(), table,
Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN, qualified()));
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/task/CreateReplicaTask.java
b/fe/fe-core/src/main/java/org/apache/doris/task/CreateReplicaTask.java
index 61c14853588..4a93c82eea6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/CreateReplicaTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/CreateReplicaTask.java
@@ -316,6 +316,7 @@ public class CreateReplicaTask extends AgentTask {
int sequenceCol = -1;
int versionCol = -1;
int commitTsoCol = -1;
+ int rowLsnCol = -1;
List<TColumn> tColumns = null;
Object tCols = objectPool.get(columns);
if (tCols != null) {
@@ -357,12 +358,16 @@ public class CreateReplicaTask extends AgentTask {
if (column.isCommitTsoColumn()) {
commitTsoCol = i;
}
+ if (column.isRowLsnColumn()) {
+ rowLsnCol = i;
+ }
}
tSchema.setColumns(tColumns);
tSchema.setDeleteSignIdx(deleteSign);
tSchema.setSequenceColIdx(sequenceCol);
tSchema.setVersionColIdx(versionCol);
tSchema.setCommitTsoColIdx(commitTsoCol);
+ tSchema.setRowLsnColIdx(rowLsnCol);
tSchema.setRowStoreColCids(rowStoreColumnUniqueIds);
if (!CollectionUtils.isEmpty(clusterKeyUids)) {
tSchema.setClusterKeyUids(clusterKeyUids);
diff --git a/gensrc/proto/internal_service.proto
b/gensrc/proto/internal_service.proto
index b29608b15a4..00cd171dcc0 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -189,9 +189,8 @@ message PTabletWriterAddBlockRequest {
// for auto-partition first stage close, we should hang.
optional bool hang_wait = 15 [default = false];
optional bool is_adaptive_random_bucket = 16 [default = false];
- // Per-row row-binlog LSNs allocated by olap table sink. The order is the
same
- // as rows in block/tablet_ids.
- repeated int64 row_binlog_lsns = 17;
+ // Per-row LSNs allocated by olap table sink. The order is the same as
rows in block/tablet_ids.
+ repeated int64 allocated_lsns = 17;
};
message PSlaveTabletNodes {
diff --git a/gensrc/proto/olap_file.proto b/gensrc/proto/olap_file.proto
index 7ad1ff7135a..e2e503cac90 100644
--- a/gensrc/proto/olap_file.proto
+++ b/gensrc/proto/olap_file.proto
@@ -562,14 +562,17 @@ message TabletSchemaPB {
// cid of the time-travel commit tso column (__DORIS_COMMIT_TSO_COL__).
optional int32 commit_tso_col_idx = 39 [default = -1];
+ // cid of the base-table row lsn column (__DORIS_ROW_LSN_COL__).
+ optional int32 row_lsn_col_idx = 40 [default = -1];
+
// cid of the binlog tso column (__DORIS_BINLOG_TSO__).
- optional int32 binlog_tso_col_idx = 40 [default = -1];
+ optional int32 binlog_tso_col_idx = 41 [default = -1];
// cid of the binlog lsn column (__DORIS_BINLOG_LSN__).
- optional int32 binlog_lsn_col_idx = 41 [default = -1];
+ optional int32 binlog_lsn_col_idx = 42 [default = -1];
// cid of the binlog op column (__DORIS_BINLOG_OP__).
- optional int32 binlog_op_col_idx = 42 [default = -1];
+ optional int32 binlog_op_col_idx = 43 [default = -1];
optional SplitSchemaPB __split_schema = 1000; // A special field, DO NOT
change it.
}
@@ -621,14 +624,17 @@ message TabletSchemaCloudPB {
// cid of the time-travel commit tso column (__DORIS_COMMIT_TSO_COL__).
optional int32 commit_tso_col_idx = 40 [default = -1];
+ // cid of the base-table row lsn column (__DORIS_ROW_LSN_COL__).
+ optional int32 row_lsn_col_idx = 41 [default = -1];
+
// cid of the binlog tso column (__DORIS_BINLOG_TSO__).
- optional int32 binlog_tso_col_idx = 41 [default = -1];
+ optional int32 binlog_tso_col_idx = 42 [default = -1];
// cid of the binlog lsn column (__DORIS_BINLOG_LSN__).
- optional int32 binlog_lsn_col_idx = 42 [default = -1];
+ optional int32 binlog_lsn_col_idx = 43 [default = -1];
// cid of the binlog op column (__DORIS_BINLOG_OP__).
- optional int32 binlog_op_col_idx = 43 [default = -1];
+ optional int32 binlog_op_col_idx = 44 [default = -1];
optional bool is_dynamic_schema = 100 [default=false];
diff --git a/gensrc/thrift/AgentService.thrift
b/gensrc/thrift/AgentService.thrift
index fb689e12976..fb5c1a54931 100644
--- a/gensrc/thrift/AgentService.thrift
+++ b/gensrc/thrift/AgentService.thrift
@@ -53,9 +53,10 @@ struct TTabletSchema {
24: optional i64 storage_dict_page_size = 262144
25: optional list<Types.TColumnGroup> seq_map
26: optional i32 commit_tso_col_idx = -1
- 27: optional i32 binlog_tso_idx = -1
- 28: optional i32 binlog_lsn_idx = -1
- 29: optional i32 binlog_op_idx = -1
+ 27: optional i32 row_lsn_col_idx = -1
+ 28: optional i32 binlog_tso_idx = -1
+ 29: optional i32 binlog_lsn_idx = -1
+ 30: optional i32 binlog_op_idx = -1
}
// this enum stands for different storage format in src_backends
diff --git a/regression-test/suites/row_binlog_p0/test_row_binlog_basic.groovy
b/regression-test/suites/row_binlog_p0/test_row_binlog_basic.groovy
index 4bae6b52aef..1b7936bd592 100644
--- a/regression-test/suites/row_binlog_p0/test_row_binlog_basic.groovy
+++ b/regression-test/suites/row_binlog_p0/test_row_binlog_basic.groovy
@@ -200,6 +200,25 @@ suite("test_row_binlog_basic", "nonConcurrent") {
ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
+ def dupRawLsnRows = sql """
+ SELECT k1, k2, k3, v1, v2, __DORIS_ROW_LSN_COL__
+ FROM test_dup_with_binlog
+ ORDER BY k1, k2, k3, v1, v2
+ """
+ def dupBinlogLsnRows = sql """
+ SELECT k1, k2, k3, v1, v2, __DORIS_BINLOG_LSN__
+ FROM binlog("table" = "test_dup_with_binlog")
+ ORDER BY k1, k2, k3, v1, v2
+ """
+ assertEquals(dupRawLsnRows.size(), dupBinlogLsnRows.size())
+ def dupRowLsns = dupRawLsnRows.collect { it[5] as long }
+ dupRowLsns.each { lsn -> assertTrue(lsn > 0, "row lsn should be positive
but got ${lsn}") }
+ assertEquals(dupRowLsns.size(), dupRowLsns.toSet().size())
+ for (int i = 0; i < dupRawLsnRows.size(); i++) {
+ assertEquals(dupRawLsnRows[i][0..4], dupBinlogLsnRows[i][0..4])
+ assertEquals(dupRawLsnRows[i][5], dupBinlogLsnRows[i][5])
+ }
+
sql """
INSERT INTO test_mow_with_binlog VALUES
(1, 1, 1, 10, '10'),
@@ -228,6 +247,14 @@ suite("test_row_binlog_basic", "nonConcurrent") {
ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
+ test {
+ sql """
+ SELECT __DORIS_ROW_LSN_COL__
+ FROM test_mow_with_binlog
+ """
+ exception "Unknown column"
+ }
+
sql """
INSERT INTO test_mow_with_before_binlog VALUES
(1, 1, 1, 10, '10'),
diff --git
a/regression-test/suites/table_stream_p0/test_olap_table_stream_history_query.groovy
b/regression-test/suites/table_stream_p0/test_olap_table_stream_history_query.groovy
index e76bad400ed..6b88f29ec37 100644
---
a/regression-test/suites/table_stream_p0/test_olap_table_stream_history_query.groovy
+++
b/regression-test/suites/table_stream_p0/test_olap_table_stream_history_query.groovy
@@ -131,17 +131,32 @@ suite("test_olap_table_stream_history_query") {
sql "SET show_hidden_columns=true;"
- // verify select * exposes the hidden stream columns with the same real
sequence value
- def checkStreamHistoryWithHiddenColumns = { streamName ->
+ // verify hidden stream columns with the same real sequence value
+ def checkMowStreamHistoryWithHiddenColumns = { streamName ->
long seq = fetchHistorySeq(streamName)
assertEquals([[1, "s1", seq, "APPEND"], [2, "s2", seq, "APPEND"], [3,
"s3", seq, "APPEND"]],
- sql("select * from ${streamName} order by sid"))
+ sql("""select sid, sname, __DORIS_STREAM_SEQUENCE_COL__,
__DORIS_STREAM_CHANGE_TYPE_COL__
+ from ${streamName} order by sid"""))
+ }
+
+ def checkDupStreamHistoryWithHiddenColumns = { streamName ->
+ long seq = fetchHistorySeq(streamName)
+ def rows = sql("""select sid, sname, __DORIS_STREAM_SEQUENCE_COL__,
+ __DORIS_STREAM_LSN_COL__,
__DORIS_STREAM_CHANGE_TYPE_COL__
+ from ${streamName} order by sid""")
+ assertEquals([[1, "s1", seq, "APPEND"], [2, "s2", seq, "APPEND"], [3,
"s3", seq, "APPEND"]],
+ rows.collect { [it[0], it[1], it[2], it[4]] })
+ def lsns = rows.collect { it[3] as long }
+ lsns.each { lsn -> assertTrue(lsn > 0, "stream lsn should be positive
but got ${lsn}") }
+ for (int i = 1; i < lsns.size(); i++) {
+ assertTrue(lsns[i - 1] < lsns[i], "stream lsn should be increasing
but got ${lsns}")
+ }
}
checkStreamHistory("s1")
checkStreamHistory("s2")
- checkStreamHistoryWithHiddenColumns("s1")
- checkStreamHistoryWithHiddenColumns("s2")
+ checkMowStreamHistoryWithHiddenColumns("s1")
+ checkDupStreamHistoryWithHiddenColumns("s2")
sql "DROP DATABASE IF EXISTS test_olap_table_stream_history_query_db"
}
diff --git
a/regression-test/suites/table_stream_p0/test_olap_table_stream_snapshot.groovy
b/regression-test/suites/table_stream_p0/test_olap_table_stream_snapshot.groovy
index 511f338438e..6d13d7904c4 100644
---
a/regression-test/suites/table_stream_p0/test_olap_table_stream_snapshot.groovy
+++
b/regression-test/suites/table_stream_p0/test_olap_table_stream_snapshot.groovy
@@ -62,6 +62,13 @@ suite("test_olap_table_stream_snapshot", "nonConcurrent") {
exception "__DORIS_STREAM_SEQUENCE_COL__"
}
}
+ // DUP snapshot exposes base row LSN.
+ def checkDupSnapshotLsn = { streamName ->
+ def lsns = sql("SELECT __DORIS_ROW_LSN_COL__ FROM
${streamName}@snapshot() ORDER BY id")
+ .collect { it[0] as long }
+ assertTrue(lsns.size() > 0)
+ lsns.each { lsn -> assertTrue(lsn > 0, "snapshot row lsn should be
positive but got ${lsn}") }
+ }
// 1) DUP + append_only + show_initial_rows=true + non-partitioned table.
// snapshot reads the stream creation snapshot and does not advance the
stream offset.
@@ -136,6 +143,7 @@ suite("test_olap_table_stream_snapshot", "nonConcurrent") {
waitVisible()
checkRows([["1", "10"], ["2", "20"]],
"SELECT id, v FROM s_dup_np_false@snapshot() ORDER BY id")
+ checkDupSnapshotLsn("s_dup_np_false")
checkRows([["3", "30"]],
"SELECT id, v FROM s_dup_np_false ORDER BY id")
sql """
diff --git
a/regression-test/suites/table_stream_p0/test_table_stream_query_comprehensive.groovy
b/regression-test/suites/table_stream_p0/test_table_stream_query_comprehensive.groovy
index 9273e709ffc..32277b32af3 100644
---
a/regression-test/suites/table_stream_p0/test_table_stream_query_comprehensive.groovy
+++
b/regression-test/suites/table_stream_p0/test_table_stream_query_comprehensive.groovy
@@ -172,6 +172,23 @@ suite("test_table_stream_query_comprehensive",
"nonConcurrent") {
assertEquals(0, sql("SELECT id, v FROM dup_ao_stream").size())
order_qt_dup_ao_target "SELECT id, v FROM dup_ao_target ORDER BY id"
+ // A.5 incremental DUP stream exposes positive and unique per-row LSNs.
+ sql "INSERT INTO dup_ao_base VALUES (6, 'f'), (7, 'g')"
+ sql "INSERT INTO dup_ao_base VALUES (8, 'h'), (9, 'i')"
+ sql "INSERT INTO dup_ao_base VALUES (10, 'j'), (11, 'k')"
+ sql "sync"
+ sleep(1200)
+ def streamLsnRows = sql """
+ SELECT __DORIS_STREAM_SEQUENCE_COL__, __DORIS_STREAM_LSN_COL__
+ FROM dup_ao_stream
+ ORDER BY __DORIS_STREAM_SEQUENCE_COL__, __DORIS_STREAM_LSN_COL__
+ """
+ assertEquals(6, streamLsnRows.size())
+ def streamLsns = streamLsnRows.collect { it[1] as long }
+ streamLsns.each { lsn -> assertTrue(lsn > 0, "stream lsn should be
positive but got ${lsn}") }
+ assertEquals(streamLsns.size(), streamLsns.toSet().size())
+ assertEquals([2, 2, 2], streamLsnRows.groupBy { it[0]
}.values().collect { it.size() }.sort())
+
// ============================================================
// Section B. DUP, non-partitioned, append_only, initial=true.
// Focus: historical seed rows are returned first (APPEND), then
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
