This is an automated email from the ASF dual-hosted git repository.
Yukang-Lian 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 bd6673b756d [Enhancement](row binlog) Support flexible partial updates
(#66899)
bd6673b756d is described below
commit bd6673b756d3672e1deebf147aa6e262faaaa18a
Author: Jamie <[email protected]>
AuthorDate: Tue Sep 1 15:33:22 2026 +0800
[Enhancement](row binlog) Support flexible partial updates (#66899)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #65810
Problem Summary:
Row Binlog supports fixed-column partial updates, but did not support
flexible partial updates (`UPDATE_FLEXIBLE_COLUMNS`). Flexible updates
carry a per-row skip bitmap and may merge rows before writing, so Row
Binlog must reconstruct complete rows while keeping BEFORE/AFTER values,
operations, row counts, and LSNs aligned.
This PR:
- Routes flexible partial updates through the existing
`MowRowBinlogDeriveStage` transform chain.
- Reuses the existing flexible-update aggregation, a read-only MOW key
probe, and historical-row read plans to materialize complete AFTER rows
and optional complete BEFORE rows.
- Keeps per-row LSN and same-batch DELETE-to-INSERT state aligned
through aggregation. For DELETE-to-INSERT, omitted AFTER cells use their
default or null value, while a live historical row remains the UPDATE
BEFORE image.
- Preserves Base Rowset and Row Binlog `(segment_id, row_id)` alignment
for sequence losers; the synchronized delete bitmap makes those rows
logically invisible.
- Uses a writer-private copy-on-write Block, so mutated columns detach
without changing the Base Rowset writer's input.
- Keeps the existing Row Binlog on-disk format, query protocol, and
APPEND / UPDATE / DELETE semantics.
The implementation is shared by local and Cloud storage. It does not
change FE syntax, Table Stream metadata, Stream Offset handling, or
protobuf definitions.
Scope note: this PR does not lift the existing flexible partial-update
multi-segment load restriction and does not add a zero-row Segment
format.
---
be/src/storage/partial_update_info.cpp | 110 +++++-
be/src/storage/partial_update_info.h | 24 +-
.../storage/segment/historical_row_retriever.cpp | 135 +++++++
be/src/storage/segment/historical_row_retriever.h | 8 +
be/src/storage/transform/row_binlog_derive.cpp | 48 ++-
be/src/storage/transform/row_binlog_derive.h | 10 +-
.../storage/rowset/segment_flusher_format_test.cpp | 406 ++++++++++++++++++++-
.../storage/transform/row_binlog_derive_test.cpp | 50 ++-
.../test_row_binlog_flexible_partial_update.out | 69 ++++
..._binlog_flexible_partial_update_no_sequence.out | 44 +++
.../test_row_binlog_flexible_partial_update.groovy | 176 +++++++++
...nlog_flexible_partial_update_no_sequence.groovy | 101 +++++
12 files changed, 1125 insertions(+), 56 deletions(-)
diff --git a/be/src/storage/partial_update_info.cpp
b/be/src/storage/partial_update_info.cpp
index bc322a97f35..b6a2c9ae8b7 100644
--- a/be/src/storage/partial_update_info.cpp
+++ b/be/src/storage/partial_update_info.cpp
@@ -40,6 +40,7 @@
#include "storage/tablet/tablet_meta.h"
#include "storage/tablet/tablet_schema.h"
#include "storage/utils.h"
+#include "util/defer_op.h"
namespace doris {
@@ -884,12 +885,14 @@ void BlockAggregator::merge_one_row(MutableBlock&
dst_block, Block* src_block, i
rid);
}
}
+ merge_row_lsn(rid);
VLOG_DEBUG << fmt::format("merge a row, after merge,
output_block.rows()={}, state: {}",
dst_block.rows(), _state.to_string());
}
void BlockAggregator::append_one_row(MutableBlock& dst_block, Block*
src_block, int rid) {
dst_block.add_row(src_block, rid);
+ append_row_lsn(rid);
_state.rows++;
VLOG_DEBUG << fmt::format("append a new row, after append,
output_block.rows()={}, state: {}",
dst_block.rows(), _state.to_string());
@@ -901,6 +904,31 @@ void BlockAggregator::remove_last_n_rows(MutableBlock&
dst_block, int n) {
DCHECK_GE(dst_block.mutable_columns()[cid]->size(), n);
dst_block.mutable_columns()[cid]->pop_back(n);
}
+ remove_last_row_lsns(n);
+ }
+}
+
+void BlockAggregator::append_row_lsn(int rid) {
+ if (_output_row_lsns != nullptr) {
+ DCHECK(_input_row_lsns != nullptr);
+ DCHECK_LT(rid, _input_row_lsns->size());
+ _output_row_lsns->push_back(_input_row_lsns->at(rid));
+ }
+}
+
+void BlockAggregator::merge_row_lsn(int rid) {
+ if (_output_row_lsns != nullptr) {
+ DCHECK(_input_row_lsns != nullptr);
+ DCHECK(!_output_row_lsns->empty());
+ DCHECK_LT(rid, _input_row_lsns->size());
+ _output_row_lsns->back() = std::max(_output_row_lsns->back(),
_input_row_lsns->at(rid));
+ }
+}
+
+void BlockAggregator::remove_last_row_lsns(int n) {
+ if (_output_row_lsns != nullptr && n > 0) {
+ DCHECK_GE(_output_row_lsns->size(), n);
+ _output_row_lsns->resize(_output_row_lsns->size() - n);
}
}
@@ -930,6 +958,7 @@ Status BlockAggregator::aggregate_rows(
VLOG_DEBUG << fmt::format("merge rows in range=[{}-{})", start, end);
if (end - start == 1) {
output_block.add_row(block, start);
+ append_row_lsn(start);
VLOG_DEBUG << fmt::format("append a row directly, rid={}", start);
return Status::OK();
}
@@ -1036,8 +1065,10 @@ Status
BlockAggregator::_generate_encoded_default_seq_value(std::string* encoded
Status BlockAggregator::aggregate_for_sequence_column(
Block* block, int num_rows, const
std::vector<IOlapColumnDataAccessor*>& key_columns,
IOlapColumnDataAccessor* seq_column, const
std::vector<RowsetSharedPtr>& specified_rowsets,
- std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) {
+ std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
+ std::vector<int64_t>* row_lsns) {
DCHECK_EQ(block->columns(), _tablet_schema.num_columns());
+ DCHECK(row_lsns == nullptr || row_lsns->size() == num_rows);
// the process logic here is the same as
MemTable::_aggregate_for_flexible_partial_update_without_seq_col()
// after this function, there will be at most 2 rows for a specified key
std::vector<BitmapValue>* skip_bitmaps =
@@ -1047,6 +1078,16 @@ Status BlockAggregator::aggregate_for_sequence_column(
auto filtered_block = _tablet_schema.create_storage_block();
MutableBlock output_block =
MutableBlock::build_mutable_block(std::move(filtered_block));
+ std::vector<int64_t> output_row_lsns;
+ Defer reset_lsn_sidecar {[this] {
+ _input_row_lsns = nullptr;
+ _output_row_lsns = nullptr;
+ }};
+ if (row_lsns != nullptr) {
+ output_row_lsns.reserve(row_lsns->size());
+ _input_row_lsns = row_lsns;
+ _output_row_lsns = &output_row_lsns;
+ }
int same_key_rows {0};
std::string previous_key {};
@@ -1072,6 +1113,10 @@ Status BlockAggregator::aggregate_for_sequence_column(
}
block->swap(output_block.to_block());
+ if (row_lsns != nullptr) {
+ row_lsns->swap(output_row_lsns);
+ DCHECK_EQ(row_lsns->size(), block->rows());
+ }
return Status::OK();
}
@@ -1106,8 +1151,13 @@ Status BlockAggregator::fill_sequence_column(Block*
block, size_t num_rows,
Status BlockAggregator::aggregate_for_insert_after_delete(
Block* block, size_t num_rows, const
std::vector<IOlapColumnDataAccessor*>& key_columns,
const std::vector<RowsetSharedPtr>& specified_rowsets,
- std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) {
+ std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
+ std::vector<int64_t>* row_lsns, std::vector<uint8_t>*
insert_after_delete_flags) {
DCHECK_EQ(block->columns(), _tablet_schema.num_columns());
+ DCHECK(row_lsns == nullptr || row_lsns->size() == num_rows);
+ if (insert_after_delete_flags != nullptr) {
+ insert_after_delete_flags->assign(num_rows, 0);
+ }
// there will be at most 2 rows for a specified key in block when control
flow reaches here
// after this function, there will not be duplicate rows in block
@@ -1130,14 +1180,6 @@ Status
BlockAggregator::aggregate_for_insert_after_delete(
FixedReadPlan read_plan;
// This insert-after-delete pass doesn't report partial-update counts;
discard them.
PartialUpdateStats discarded;
- // the losing delete-sign row is removed from the block below instead of
- // marking itself, so the probe only marks the old row
- segment_v2::MowKeyProbe probe(
- _tablet.get(), &_tablet_schema, _tablet_schema.has_sequence_col(),
_mow_context,
- RowsetId {}, 0,
- segment_v2::MowKeyProbe::Policy {
- .mark_deleted =
segment_v2::MowKeyProbe::MarkDeleted::OLD_ROW,
- });
for (size_t block_pos {0}; block_pos < num_rows; block_pos++) {
size_t delta_pos = block_pos;
auto& skip_bitmap = skip_bitmaps->at(block_pos);
@@ -1152,9 +1194,9 @@ Status BlockAggregator::aggregate_for_insert_after_delete(
DCHECK(previous_has_delete_sign);
DCHECK(!have_delete_sign);
++duplicate_rows;
- auto probe_res = probe.probe(key, /*segment_pos=*/0,
/*key_has_seq_suffix=*/false,
- /*have_delete_sign=*/false,
specified_rowsets,
- segment_caches, discarded);
+ auto probe_res = _probe.probe(key, /*segment_pos=*/0,
/*key_has_seq_suffix=*/false,
+ /*have_delete_sign=*/false,
specified_rowsets,
+ segment_caches, discarded);
DCHECK(probe_res.has_value() ||
probe_res.error().is<ErrorCode::MEM_LIMIT_EXCEEDED>())
<< "[BlockAggregator::aggregate_for_insert_after_delete]
unexpected error "
"status while lookup_row_key:"
@@ -1180,6 +1222,13 @@ Status
BlockAggregator::aggregate_for_insert_after_delete(
}
// and remove the row with delete sign from the current block
filter_map[block_pos - 1] = 0;
+ if (insert_after_delete_flags != nullptr) {
+ insert_after_delete_flags->at(block_pos) = 1;
+ }
+ if (row_lsns != nullptr) {
+ row_lsns->at(block_pos) =
+ std::max(row_lsns->at(block_pos - 1),
row_lsns->at(block_pos));
+ }
}
previous_has_delete_sign = have_delete_sign;
previous_key = std::move(key);
@@ -1189,8 +1238,31 @@ Status
BlockAggregator::aggregate_for_insert_after_delete(
// fill sequence column value for some rows
RETURN_IF_ERROR(fill_sequence_column(block, num_rows, read_plan,
*skip_bitmaps));
}
+ if (row_lsns != nullptr) {
+ std::vector<int64_t> filtered_row_lsns;
+ filtered_row_lsns.reserve(row_lsns->size() - duplicate_rows);
+ for (size_t pos = 0; pos < num_rows; ++pos) {
+ if (filter_map[pos] != 0) {
+ filtered_row_lsns.push_back(row_lsns->at(pos));
+ }
+ }
+ row_lsns->swap(filtered_row_lsns);
+ }
+ if (insert_after_delete_flags != nullptr) {
+ std::vector<uint8_t> filtered_flags;
+ filtered_flags.reserve(insert_after_delete_flags->size() -
duplicate_rows);
+ for (size_t pos = 0; pos < num_rows; ++pos) {
+ if (filter_map[pos] != 0) {
+
filtered_flags.push_back(insert_after_delete_flags->at(pos));
+ }
+ }
+ insert_after_delete_flags->swap(filtered_flags);
+ }
RETURN_IF_ERROR(filter_block(block, num_rows,
std::move(filter_column), duplicate_rows,
"__filter_insert_after_delete_col__"));
+ DCHECK(row_lsns == nullptr || row_lsns->size() == block->rows());
+ DCHECK(insert_after_delete_flags == nullptr ||
+ insert_after_delete_flags->size() == block->rows());
}
return Status::OK();
}
@@ -1246,7 +1318,12 @@ Status BlockAggregator::convert_seq_column(Block* block,
size_t row_pos, size_t
Status BlockAggregator::aggregate_for_flexible_partial_update(
Block* block, size_t num_rows, const std::vector<RowsetSharedPtr>&
specified_rowsets,
- std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches) {
+ std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
+ std::vector<int64_t>* row_lsns, std::vector<uint8_t>*
insert_after_delete_flags) {
+ if (row_lsns != nullptr && row_lsns->size() != num_rows) {
+ return Status::InvalidArgument("row binlog LSN count {} does not match
row count {}",
+ row_lsns->size(), num_rows);
+ }
std::vector<IOlapColumnDataAccessor*> key_columns {};
IOlapColumnDataAccessor* seq_column {nullptr};
@@ -1260,7 +1337,7 @@ Status
BlockAggregator::aggregate_for_flexible_partial_update(
if (_tablet_schema.has_sequence_col()) {
RETURN_IF_ERROR(aggregate_for_sequence_column(block,
static_cast<int>(num_rows),
key_columns, seq_column,
specified_rowsets,
- segment_caches));
+ segment_caches,
row_lsns));
}
// 2. merge duplicate rows and handle insert after delete
@@ -1272,7 +1349,8 @@ Status
BlockAggregator::aggregate_for_flexible_partial_update(
RETURN_IF_ERROR(convert_seq_column(block, 0, num_rows, seq_column));
}
RETURN_IF_ERROR(aggregate_for_insert_after_delete(block, num_rows,
key_columns,
- specified_rowsets,
segment_caches));
+ specified_rowsets,
segment_caches, row_lsns,
+
insert_after_delete_flags));
return Status::OK();
}
diff --git a/be/src/storage/partial_update_info.h
b/be/src/storage/partial_update_info.h
index 01b4bb3f302..4b8ab99840f 100644
--- a/be/src/storage/partial_update_info.h
+++ b/be/src/storage/partial_update_info.h
@@ -217,20 +217,28 @@ public:
std::vector<IOlapColumnDataAccessor*>&
key_columns);
Status convert_seq_column(Block* block, size_t row_pos, size_t num_rows,
IOlapColumnDataAccessor*& seq_column);
+ // Optional sidecars are aggregated and filtered with `block`; on return
each element is
+ // aligned with the corresponding row in the final block. A non-zero
+ // `insert_after_delete_flags` entry marks the surviving INSERT of a
same-batch
+ // DELETE-then-INSERT pair.
Status aggregate_for_flexible_partial_update(
Block* block, size_t num_rows, const std::vector<RowsetSharedPtr>&
specified_rowsets,
- std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
+ std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
+ std::vector<int64_t>* row_lsns = nullptr,
+ std::vector<uint8_t>* insert_after_delete_flags = nullptr);
private:
Status aggregate_for_sequence_column(
Block* block, int num_rows, const
std::vector<IOlapColumnDataAccessor*>& key_columns,
IOlapColumnDataAccessor* seq_column,
const std::vector<RowsetSharedPtr>& specified_rowsets,
- std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
+ std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
+ std::vector<int64_t>* row_lsns);
Status aggregate_for_insert_after_delete(
Block* block, size_t num_rows, const
std::vector<IOlapColumnDataAccessor*>& key_columns,
const std::vector<RowsetSharedPtr>& specified_rowsets,
- std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches);
+ std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
+ std::vector<int64_t>* row_lsns, std::vector<uint8_t>*
insert_after_delete_flags);
Status filter_block(Block* block, size_t num_rows, MutableColumnPtr
filter_column,
int duplicate_rows, std::string col_name);
@@ -244,6 +252,10 @@ private:
void append_one_row(MutableBlock& dst_block, Block* src_block, int rid);
void remove_last_n_rows(MutableBlock& dst_block, int n);
+ void append_row_lsn(int rid);
+ void merge_row_lsn(int rid);
+ void remove_last_row_lsns(int n);
+
// aggregate rows with same keys in range [start, end) from block to
output_block
Status aggregate_rows(MutableBlock& output_block, Block* block, int start,
int end,
std::string key, std::vector<BitmapValue>*
skip_bitmaps,
@@ -262,6 +274,12 @@ private:
const segment_v2::MowKeyProbe& _probe;
HistoricalRowFetcher& _fetcher;
+ // Optional sidecar used by Row Binlog. Flexible aggregation can remove or
merge input rows;
+ // keep the persisted LSN aligned with the surviving row and retain the
largest LSN of all
+ // merged changes.
+ const std::vector<int64_t>* _input_row_lsns = nullptr;
+ std::vector<int64_t>* _output_row_lsns = nullptr;
+
// used to store state when aggregating rows in block
struct AggregateState {
int rows {0};
diff --git a/be/src/storage/segment/historical_row_retriever.cpp
b/be/src/storage/segment/historical_row_retriever.cpp
index dbe5695dea8..56c7389f876 100644
--- a/be/src/storage/segment/historical_row_retriever.cpp
+++ b/be/src/storage/segment/historical_row_retriever.cpp
@@ -143,6 +143,141 @@ Status
PrimaryKeyModelRowRetriever::retrieve_historical_row(const Int8* delete_s
return Status::OK();
}
+//
NOLINTNEXTLINE(readability-function-size,readability-function-cognitive-complexity):
Keep the
+// probe, aligned sidecar aggregation, and history-read plan in one
snapshot-consistent operation.
+Status PrimaryKeyModelRowRetriever::materialize_flexible_partial_update(
+ Block* block, std::shared_ptr<MowContext> mow_context,
+ std::vector<int64_t>* row_lsns) { //
NOLINT(readability-non-const-parameter): in/out sidecar
+ auto& tablet_schema = _context.tablet_schema;
+ if (_context.partial_update_info == nullptr ||
+ !_context.partial_update_info->is_flexible_partial_update()) {
+ return Status::InvalidArgument(
+ "flexible partial update materialization requires flexible
partial update info");
+ }
+ if (block->columns() != tablet_schema->num_columns()) {
+ return Status::InvalidArgument(
+ "illegal flexible partial update block columns: {}, schema
columns: {}",
+ block->columns(), tablet_schema->num_columns());
+ }
+
+ _mow_context = std::move(mow_context);
+ if (_mow_context == nullptr) {
+ return Status::InvalidArgument(
+ "flexible partial update materialization requires MOW
context");
+ }
+
+ std::vector<RowsetSharedPtr> specified_rowsets;
+ {
+ std::shared_lock rlock(_context.tablet->get_header_lock());
+ specified_rowsets = _mow_context->rowset_ptrs;
+ }
+ std::vector<std::unique_ptr<SegmentCacheHandle>>
segment_caches(specified_rowsets.size());
+
+ CHECK(_context.rowset_writer_ctx != nullptr);
+ const bool write_before =
+
_context.rowset_writer_ctx->write_binlog_opt().write_binlog_config().write_before;
+ TabletSchemaSPtr lookup_schema = _context.tablet->tablet_schema();
+ MowKeyProbe probe = MowKeyProbe::for_row_binlog(_context.tablet.get(),
lookup_schema.get(),
+
tablet_schema->has_sequence_col(), _mow_context,
+ write_before);
+ BlockAggregator aggregator(*tablet_schema, _context.tablet, _mow_context,
+ *_context.partial_update_info, *_key_encoder,
probe, *_row_fetcher);
+
+ size_t num_rows = block->rows();
+ std::vector<uint8_t> insert_after_delete_flags;
+ RETURN_IF_ERROR(aggregator.aggregate_for_flexible_partial_update(
+ block, num_rows, specified_rowsets, segment_caches, row_lsns,
+ &insert_after_delete_flags));
+ num_rows = block->rows();
+ if (num_rows == 0) {
+ return Status::OK();
+ }
+ DCHECK_EQ(insert_after_delete_flags.size(), num_rows);
+
+ std::vector<IOlapColumnDataAccessor*> key_columns;
+ RETURN_IF_ERROR(aggregator.convert_pk_columns(block, 0, num_rows,
key_columns));
+ IOlapColumnDataAccessor* seq_column = nullptr;
+ RETURN_IF_ERROR(aggregator.convert_seq_column(block, 0, num_rows,
seq_column));
+
+ auto* skip_bitmaps =
+ &get_mutable_skip_bitmap_column(block,
tablet_schema->skip_bitmap_col_idx())
+ ->get_data();
+ const auto* delete_signs = BaseTablet::get_delete_sign_column_data(*block,
num_rows);
+ DCHECK(delete_signs != nullptr);
+
+ const bool schema_has_seq = tablet_schema->has_sequence_col();
+ const int32_t seq_col_unique_id =
+ schema_has_seq ?
tablet_schema->column(tablet_schema->sequence_col_idx()).unique_id()
+ : -1;
+ const int32_t delete_sign_col_unique_id =
+
tablet_schema->column(tablet_schema->delete_sign_idx()).unique_id();
+ Block full_block = tablet_schema->create_storage_block();
+ for (size_t cid = 0; cid < tablet_schema->num_key_columns(); ++cid) {
+ const auto& input_column = block->get_by_position(cid);
+ auto& full_column = full_block.get_by_position(cid);
+ full_column.column = input_column.column;
+ full_column.type = input_column.type;
+ }
+
+ bool has_default_or_nullable = false;
+ std::vector<bool> use_default_or_null_flag;
+ use_default_or_null_flag.reserve(num_rows);
+ PartialUpdateStats discarded_stats;
+ for (size_t pos = 0; pos < num_rows; ++pos) {
+ const bool row_has_seq =
+ schema_has_seq &&
!skip_bitmaps->at(pos).contains(seq_col_unique_id);
+ std::string key = _key_encoder->full_encode_primary_keys(key_columns,
pos);
+ if (row_has_seq) {
+ _key_encoder->append_seq_suffix(&key, seq_column, pos);
+ }
+ const bool have_delete_sign =
!skip_bitmaps->at(pos).contains(delete_sign_col_unique_id) &&
+ delete_signs[pos] != 0;
+ ProbeOutcome outcome =
+ DORIS_TRY(probe.probe(key, /*segment_pos=*/pos, row_has_seq,
have_delete_sign,
+ specified_rowsets, segment_caches,
discarded_stats));
+ const bool insert_after_delete = insert_after_delete_flags[pos] != 0;
+ DCHECK(!insert_after_delete || !have_delete_sign);
+ if (outcome.result == KeyProbeResult::NOT_FOUND && !have_delete_sign) {
+ RETURN_IF_ERROR(_context.partial_update_info->handle_new_key(
+ *tablet_schema,
+ [&]() -> std::string {
+ return block->dump_one_line(
+ pos,
cast_set<int>(tablet_schema->num_key_columns()));
+ },
+ &skip_bitmaps->at(pos)));
+ }
+
+ const bool use_default_or_null = outcome.use_default_or_null ||
insert_after_delete;
+ has_default_or_nullable |= use_default_or_null;
+ use_default_or_null_flag.emplace_back(use_default_or_null);
+ if (!use_default_or_null) {
+ _row_fetcher->pin_rowset(outcome.rowset);
+ _row_fetcher->plan_flexible_read(outcome.loc, pos,
skip_bitmaps->at(pos));
+ // The same location supplies the optional BEFORE image and the
old delete sign used
+ // to distinguish an append after a tombstone from an update.
+ _row_fetcher->plan_fixed_read(outcome.loc, pos);
+ } else if (insert_after_delete && outcome.result !=
KeyProbeResult::NOT_FOUND) {
+ // The base writer treats this row as a new row for missing-column
fill, but the net
+ // CDC change still replaces the live row that existed before this
load. Keep its
+ // BEFORE image while preventing old values from leaking into
AFTER.
+ _row_fetcher->pin_rowset(outcome.rowset);
+ _row_fetcher->plan_fixed_read(outcome.loc, pos);
+ }
+ _operators.emplace_back(
+ outcome.result == KeyProbeResult::NOT_FOUND
+ ? (have_delete_sign ? ROW_BINLOG_DELETE :
ROW_BINLOG_APPEND)
+ : (have_delete_sign ? ROW_BINLOG_DELETE :
ROW_BINLOG_UPDATE));
+ }
+
+ RETURN_IF_ERROR(_row_fetcher->fill_non_primary_key_columns(
+ *tablet_schema, full_block, use_default_or_null_flag,
has_default_or_nullable,
+ /*segment_start_pos=*/0, /*block_start_pos=*/0, block,
skip_bitmaps));
+ block->swap(full_block);
+ DCHECK_EQ(_operators.size(), block->rows());
+ DCHECK(row_lsns == nullptr || row_lsns->size() == block->rows());
+ return Status::OK();
+}
+
Status PrimaryKeyModelRowRetriever::build_after_block(Block* block, size_t
row_pos,
size_t num_rows) {
DCHECK_EQ(_use_default_or_null_flag.size(), num_rows);
diff --git a/be/src/storage/segment/historical_row_retriever.h
b/be/src/storage/segment/historical_row_retriever.h
index 0f89d6b9dc5..d4042d0e3be 100644
--- a/be/src/storage/segment/historical_row_retriever.h
+++ b/be/src/storage/segment/historical_row_retriever.h
@@ -81,6 +81,14 @@ public:
Status retrieve_historical_row(const Int8* delete_sign_column_data, size_t
row_pos,
size_t num_rows) override;
+ // Row Binlog receives the original flexible-partial-update block, while
the base writer fills
+ // its own copy. Reuse the common flexible aggregator/read plans with a
read-only MOW probe to
+ // build the same full AFTER rows without changing the base tablet's
delete bitmap. The optional
+ // LSN sidecar is merged and filtered together with the rows.
+ Status materialize_flexible_partial_update(Block* block,
+ std::shared_ptr<MowContext>
mow_context,
+ std::vector<int64_t>* row_lsns);
+
Status build_after_block(Block* block, size_t row_pos, size_t num_rows)
override;
Status build_before_block(Block* before_block, const
std::vector<uint32_t>& value_cids,
diff --git a/be/src/storage/transform/row_binlog_derive.cpp
b/be/src/storage/transform/row_binlog_derive.cpp
index 3d11f169459..69dde8a293f 100644
--- a/be/src/storage/transform/row_binlog_derive.cpp
+++ b/be/src/storage/transform/row_binlog_derive.cpp
@@ -301,8 +301,8 @@ bool binlog_needs_historical_lookup(const
RowsetWriterContext& context) {
const bool is_source_direct_write =
cfg.source.source_write_type == DataWriteType::TYPE_DIRECT &&
!cfg.source.is_transient_rowset_writer;
- // A direct partial update (fixed needs AFTER rebuilt; flexible is rejected
- // inside the MoW stage) or a requested BEFORE image needs the probe.
+ // A direct partial update (fixed and flexible both need AFTER rebuilt) or
a
+ // requested BEFORE image needs the probe.
const bool is_partial_update = cfg.source.partial_update_info != nullptr &&
cfg.source.partial_update_info->is_partial_update() &&
is_source_direct_write;
@@ -336,16 +336,38 @@ Status
MowRowBinlogDeriveStage::derive(TransformExecContext& ctx, Block* block,
const BinlogDeriveContext& c) const {
const auto& cfg = ctx.rowset_ctx->write_binlog_opt().write_binlog_config();
- bool is_source_direct_write = cfg.source.source_write_type ==
DataWriteType::TYPE_DIRECT &&
- !cfg.source.is_transient_rowset_writer;
- if (cfg.source.partial_update_info && is_source_direct_write &&
- cfg.source.partial_update_info->is_flexible_partial_update()) {
- return Status::NotSupported("binlog<row> does not support flexible
partial update");
+ const bool is_source_direct_write =
+ cfg.source.source_write_type == DataWriteType::TYPE_DIRECT &&
+ !cfg.source.is_transient_rowset_writer;
+ const bool is_flexible_partial_update =
+ cfg.source.partial_update_info && is_source_direct_write &&
+ cfg.source.partial_update_info->is_flexible_partial_update();
+ const bool is_fixed_partial_update = cfg.source.partial_update_info &&
is_source_direct_write &&
+
cfg.source.partial_update_info->is_fixed_partial_update();
+
+ if (is_flexible_partial_update) {
+ auto retriever = std::make_unique<PrimaryKeyModelRowRetriever>();
+ RETURN_IF_ERROR(retriever->init(HistoricalRowRetrieverContext {
+ .tablet = cfg.source.base_tablet,
+ .tablet_schema = c.source_schema,
+ .rowset_writer_ctx = ctx.rowset_ctx,
+ .partial_update_info = cfg.source.partial_update_info,
+ .is_transient_rowset_writer =
cfg.source.is_transient_rowset_writer,
+ .write_type = cfg.source.source_write_type}));
+
+ std::vector<int64_t> effective_lsn_ids(c.lsn_ids->begin(),
c.lsn_ids->begin() + c.num_rows);
+ RETURN_IF_ERROR(retriever->materialize_flexible_partial_update(
+ block, cfg.source.mow_context, &effective_lsn_ids));
+
+ BinlogDeriveContext effective_context = c;
+ effective_context.num_rows = block->rows();
+ effective_context.lsn_ids =
+ std::make_shared<const
std::vector<int64_t>>(std::move(effective_lsn_ids));
+ return emit_binlog_block(effective_context, block, retriever.get(),
+ /*plain_operators=*/nullptr, block);
}
- bool is_partial_update = cfg.source.partial_update_info &&
-
cfg.source.partial_update_info->is_fixed_partial_update() &&
- is_source_direct_write;
- if (is_partial_update) {
+
+ if (is_fixed_partial_update) {
if (block->columns() < c.source_schema->num_key_columns() ||
block->columns() >= c.source_schema->num_columns()) {
return Status::InvalidArgument(fmt::format(
@@ -360,7 +382,7 @@ Status
MowRowBinlogDeriveStage::derive(TransformExecContext& ctx, Block* block,
// move in the narrow partial-update block)
int32_t delete_sign_pos = c.source_schema->delete_sign_idx();
int32_t seq_pos = c.source_schema->sequence_col_idx();
- if (is_partial_update) {
+ if (is_fixed_partial_update) {
delete_sign_pos = -1;
seq_pos = -1;
int32_t pos = 0;
@@ -390,7 +412,7 @@ Status
MowRowBinlogDeriveStage::derive(TransformExecContext& ctx, Block* block,
// missing columns from history; upserts are already source-schema shaped.
Block full_block;
const Block* after_src = block;
- if (is_partial_update) {
+ if (is_fixed_partial_update) {
full_block = widen_partial_update_block(
*c.source_schema, cfg.source.partial_update_info->update_cids,
*block);
RETURN_IF_ERROR(retriever->build_after_block(&full_block, 0,
c.num_rows));
diff --git a/be/src/storage/transform/row_binlog_derive.h
b/be/src/storage/transform/row_binlog_derive.h
index b7e174e1a74..5e004f23ae0 100644
--- a/be/src/storage/transform/row_binlog_derive.h
+++ b/be/src/storage/transform/row_binlog_derive.h
@@ -31,8 +31,8 @@ namespace segment_v2 {
// it like any DUP_KEYS block. build_transform_chain picks Plain (no historical
// probe) or Mow (with probe) via binlog_needs_historical_lookup().
-// Whether the flush needs the historical key probe: a direct partial update
-// (flexible is rejected later) or a requested BEFORE image. Decided per flush.
+// Whether the flush needs the historical key probe: a direct partial update or
+// a requested BEFORE image. Decided per flush.
bool binlog_needs_historical_lookup(const RowsetWriterContext& context);
// Context for each flush that the base stage works out once and passes to
@@ -75,9 +75,9 @@ protected:
const BinlogDeriveContext& c) const override;
};
-// With a probe: fixed partial update (rebuild AFTER from history) and/or the
-// BEFORE image. It repeats the data chain's key probe on the same source
block,
-// but never marks the delete bitmap.
+// With a probe: fixed/flexible partial update (rebuild AFTER from history)
+// and/or the BEFORE image. It repeats the data chain's key probe on the same
+// source block, but never marks the delete bitmap.
class MowRowBinlogDeriveStage : public RowBinlogDeriveStage {
public:
std::string_view name() const override { return "MowRowBinlogDerive"; }
diff --git a/be/test/storage/rowset/segment_flusher_format_test.cpp
b/be/test/storage/rowset/segment_flusher_format_test.cpp
index a29164599cc..af530741fed 100644
--- a/be/test/storage/rowset/segment_flusher_format_test.cpp
+++ b/be/test/storage/rowset/segment_flusher_format_test.cpp
@@ -1375,6 +1375,115 @@ Result<Block> create_mow_history_block(const
TabletSchemaSPtr& schema) {
return block;
}
+Result<Block> create_flexible_row_binlog_update_block(const TabletSchemaSPtr&
schema) {
+ DORIS_CHECK_GE(schema->skip_bitmap_col_idx(), 0);
+ DORIS_CHECK(schema->has_sequence_col());
+
+ constexpr std::array<int, 9> keys {0, 0, 1, 2, 3, 3, 10, 11, 11};
+ constexpr std::array<int, 9> v1_values {7000, 0, 0, 7002, 0, 0, 0, 0, 0};
+ constexpr std::array<int64_t, 9> v2_values {0, 8000, 8001, 8002, 0, 0, 0,
0, 8011};
+ constexpr std::array<int, 9> sequence_values {6000, 0, 6001, 6002, 6003,
6004, 6010, 6011, 0};
+
+ Block block = schema->create_storage_block();
+ auto* skip_bitmap = assert_cast<ColumnBitmap*>(
+
block.get_by_position(schema->skip_bitmap_col_idx()).column->assert_mutable().get());
+ for (size_t row = 0; row < keys.size(); ++row) {
+ for (size_t column_index = 0; column_index < block.columns();
++column_index) {
+ const auto& name = block.get_by_position(column_index).name;
+ if (name == "k1") {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index,
std::to_string(keys[row])));
+ } else if (name == "v1") {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index,
std::to_string(v1_values[row])));
+ } else if (name == "v2") {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index,
std::to_string(v2_values[row])));
+ } else if (name == SEQUENCE_COL) {
+ RETURN_IF_ERROR_RESULT(append_text_value(&block, column_index,
+
std::to_string(sequence_values[row])));
+ } else if (name == DELETE_SIGN) {
+ const bool is_delete = row == 4 || row == 6 || row == 7;
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index, is_delete ?
"1" : "0"));
+ } else if (name != SKIP_BITMAP_COL) {
+
block.get_by_position(column_index).column->assert_mutable()->insert_default();
+ }
+ }
+
+ BitmapValue skipped_columns;
+ const auto skip = [&](std::string_view name) {
+ const auto column_index = schema->field_index(std::string(name));
+ DORIS_CHECK_GE(column_index, 0);
+ skipped_columns.add(static_cast<uint64_t>(
+
schema->column(static_cast<size_t>(column_index)).unique_id()));
+ };
+ if (row == 0 || row == 4 || row == 5 || row == 6 || row == 7) {
+ skip("v2");
+ }
+ if (row == 1 || row == 2 || row == 4 || row == 5 || row == 6 || row ==
7 || row == 8) {
+ skip("v1");
+ }
+ if (row == 1 || row == 8) {
+ skip(SEQUENCE_COL);
+ }
+ if (row != 4 && row != 6 && row != 7) {
+ skip(DELETE_SIGN);
+ }
+ skip_bitmap->insert_value(std::move(skipped_columns));
+ }
+ return block;
+}
+
+struct FlexibleSequenceRow {
+ int key;
+ int v1;
+ int64_t v2;
+ int sequence;
+ bool is_delete;
+};
+
+Result<Block> create_flexible_sequence_block(const TabletSchemaSPtr& schema,
+ const
std::vector<FlexibleSequenceRow>& rows) {
+ DORIS_CHECK_GE(schema->skip_bitmap_col_idx(), 0);
+ DORIS_CHECK(schema->has_sequence_col());
+
+ Block block = schema->create_storage_block();
+ auto* skip_bitmap = assert_cast<ColumnBitmap*>(
+
block.get_by_position(schema->skip_bitmap_col_idx()).column->assert_mutable().get());
+ const auto delete_sign_unique_id =
schema->column(schema->delete_sign_idx()).unique_id();
+ for (const auto& row : rows) {
+ for (size_t column_index = 0; column_index < block.columns();
++column_index) {
+ const auto& name = block.get_by_position(column_index).name;
+ if (name == "k1") {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index,
std::to_string(row.key)));
+ } else if (name == "v1") {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index,
std::to_string(row.v1)));
+ } else if (name == "v2") {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index,
std::to_string(row.v2)));
+ } else if (name == SEQUENCE_COL) {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index,
std::to_string(row.sequence)));
+ } else if (name == DELETE_SIGN) {
+ RETURN_IF_ERROR_RESULT(
+ append_text_value(&block, column_index, row.is_delete
? "1" : "0"));
+ } else if (name != SKIP_BITMAP_COL) {
+
block.get_by_position(column_index).column->assert_mutable()->insert_default();
+ }
+ }
+
+ BitmapValue skipped_columns;
+ if (!row.is_delete) {
+ skipped_columns.add(static_cast<uint64_t>(delete_sign_unique_id));
+ }
+ skip_bitmap->insert_value(std::move(skipped_columns));
+ }
+ return block;
+}
+
struct ExternalInvertedIndexSignature {
std::string field;
std::map<std::string, std::vector<int32_t>> postings;
@@ -2391,6 +2500,51 @@ Status verify_row_binlog_before_segment(const
TabletSharedPtr& tablet, uint32_t
return Status::OK();
}
+Status verify_flexible_row_binlog_segment(const TabletSharedPtr& tablet,
+ std::string_view case_name) {
+ auto block_result = read_row_binlog_segment(tablet, case_name, 0);
+ if (!block_result.has_value()) {
+ return block_result.error();
+ }
+ const auto& block = block_result.value();
+ if (block.rows() != 6) {
+ return Status::InternalError("{} has {} rows, expected 6", case_name,
block.rows());
+ }
+
+ constexpr std::array<Int32, 6> keys {0, 1, 2, 3, 10, 11};
+ constexpr std::array<Int32, 6> after_v1 {7000, 3001, 7002, 0, 3010, 0};
+ constexpr std::array<Int64, 6> after_v2 {8000, 8001, 8002, 0, 4010, 8011};
+ constexpr std::array<Int64, 6> operations {ROW_BINLOG_UPDATE,
ROW_BINLOG_UPDATE,
+ ROW_BINLOG_APPEND,
ROW_BINLOG_APPEND,
+ ROW_BINLOG_DELETE,
ROW_BINLOG_UPDATE};
+ constexpr std::array<Int64, 6> lsns {1001, 1002, 1003, 1005, 1006, 1008};
+ for (size_t row = 0; row < block.rows(); ++row) {
+ RETURN_IF_ERROR(
+ verify_segment_field(block, "k1", row,
Field::create_field<TYPE_INT>(keys[row])));
+ RETURN_IF_ERROR(verify_segment_field(block, "v1", row,
+
Field::create_field<TYPE_INT>(after_v1[row])));
+ RETURN_IF_ERROR(verify_segment_field(
+ block, "v2", row,
+ row == 3 ? Field {} :
Field::create_field<TYPE_BIGINT>(after_v2[row])));
+ RETURN_IF_ERROR(verify_segment_field(block, BINLOG_OP_COL, row,
+
Field::create_field<TYPE_BIGINT>(operations[row])));
+ RETURN_IF_ERROR(verify_segment_field(block, BINLOG_LSN_COL, row,
+
Field::create_field<TYPE_BIGINT>(lsns[row])));
+ if (row == 2 || row == 3) {
+ RETURN_IF_ERROR(verify_segment_field(block, "__BEFORE__v1__", row,
Field {}));
+ RETURN_IF_ERROR(verify_segment_field(block, "__BEFORE__v2__", row,
Field {}));
+ } else {
+ RETURN_IF_ERROR(verify_segment_field(
+ block, "__BEFORE__v1__", row,
+ Field::create_field<TYPE_INT>(static_cast<Int32>(3000 +
keys[row]))));
+ RETURN_IF_ERROR(verify_segment_field(
+ block, "__BEFORE__v2__", row,
+ Field::create_field<TYPE_BIGINT>(static_cast<Int64>(4000 +
keys[row]))));
+ }
+ }
+ return Status::OK();
+}
+
Status compare_logical_segments(std::string_view case_name, uint32_t
segment_id,
const LogicalSegmentContents& current,
const LogicalSegmentContents& golden) {
@@ -2915,7 +3069,9 @@ protected:
BinlogTabletPair create_binlog_tablets(int64_t tablet_id, bool enable_mow,
bool include_before_columns = false,
- bool with_sequence = false) {
+ bool with_sequence = false,
+ bool flexible_partial_update =
false) {
+ DORIS_CHECK(!flexible_partial_update || enable_mow);
auto request = testutil::create_tablet_request(
tablet_id, 270068390, 10001, 1,
enable_mow ? TKeysType::UNIQUE_KEYS : TKeysType::DUP_KEYS,
@@ -2943,6 +3099,13 @@ protected:
request.tablet_schema.__set_sequence_col_idx(
static_cast<int32_t>(request.tablet_schema.columns.size())
- 1);
}
+ if (flexible_partial_update) {
+ auto skip_bitmap_column = testutil::create_tablet_column(
+ {SKIP_BITMAP_COL, TPrimitiveType::BITMAP, false, true,
TAggregationType::NONE});
+ skip_bitmap_column.__set_visible(false);
+ skip_bitmap_column.__set_default_value(std::string(1, '\0'));
+
request.tablet_schema.columns.push_back(std::move(skip_bitmap_column));
+ }
auto find_source_column = [&](std::string_view name) -> TColumn& {
auto& columns = request.tablet_schema.columns;
const auto it =
@@ -2951,6 +3114,9 @@ protected:
DORIS_CHECK(it != columns.end());
return *it;
};
+ if (flexible_partial_update) {
+ find_source_column("v1").__set_default_value("0");
+ }
find_source_column("v2").__set_is_allow_null(true);
if (enable_mow) {
find_source_column(DELETE_SIGN).__set_visible(false);
@@ -3021,6 +3187,16 @@ protected:
EXPECT_TRUE(_engine->create_tablet(request, &profile).ok());
auto source_tablet = _engine->tablet_manager()->get_tablet(tablet_id);
EXPECT_NE(source_tablet, nullptr);
+ if (flexible_partial_update) {
+ // TTabletSchema has no skip-bitmap ordinal. Production
flexible-update schemas are
+ // persisted as TabletSchemaPB with this field set, so make the
test-created schema
+ // match that representation before writing any rowset.
+ TabletSchemaPB schema_pb;
+ source_tablet->tablet_schema()->to_schema_pb(&schema_pb);
+ schema_pb.set_skip_bitmap_col_idx(
+
source_tablet->tablet_schema()->field_index(SKIP_BITMAP_COL));
+
source_tablet->tablet_meta()->mutable_tablet_schema()->init_from_pb(schema_pb);
+ }
EXPECT_TRUE(_engine->create_tablet(binlog_request, &profile).ok());
auto binlog_tablet =
_engine->tablet_manager()->get_tablet(binlog_request.tablet_id);
EXPECT_NE(binlog_tablet, nullptr);
@@ -3078,6 +3254,66 @@ protected:
return flusher.close();
}
+ Status flush_blocks_without_golden(
+ std::string_view case_name, const TabletSchemaSPtr& schema,
std::vector<Block> blocks,
+ bool enable_vertical_writer, bool enable_unique_key_merge_on_write,
+ const std::function<void(RowsetWriterContext&)>& configure_context,
+ std::shared_ptr<TestSegmentCollector>* output_collector) const {
+ const auto directory = fmt::format("{}/{}", kTestDir, case_name);
+
RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(directory));
+ auto file_writer_creator =
+ std::make_shared<LocalSegmentFileWriterCreator>(directory,
schema);
+ auto segment_collector = std::make_shared<TestSegmentCollector>();
+ RowsetWriterContext context;
+ context.tablet_schema = schema;
+ context.tablet_path = directory;
+ context.tablet_id = 10001;
+ context.rowset_id.init(10002);
+ context.max_rows_per_segment = 1024;
+ context.write_type = DataWriteType::TYPE_DIRECT;
+ context.enable_unique_key_merge_on_write =
enable_unique_key_merge_on_write;
+ context.file_writer_creator = file_writer_creator;
+ context.segment_collector = segment_collector;
+ configure_context(context);
+
+ config::enable_vertical_segment_writer = enable_vertical_writer;
+ auto* sync_point = SyncPoint::get_instance();
+ const bool sync_point_was_enabled = sync_point->get_enable();
+ sync_point->enable_processing();
+ Defer restore_sync_point {[sync_point, sync_point_was_enabled] {
+ if (!sync_point_was_enabled) {
+ sync_point->disable_processing();
+ }
+ }};
+ std::vector<uint32_t> vertical_segment_ids;
+ SyncPoint::CallbackGuard vertical_writer_guard;
+ sync_point->set_call_back(
+ "SegmentFlusher::flush_vertical_segment_writer",
+ [&vertical_segment_ids](auto&& args) {
+
vertical_segment_ids.push_back(*try_any_cast<uint32_t*>(args[0]));
+ },
+ &vertical_writer_guard);
+
+ SegmentFileCollection segment_files;
+ InvertedIndexFileCollection index_files;
+ SegmentFlusher flusher(context, segment_files, index_files);
+ for (size_t segment_id = 0; segment_id < blocks.size(); ++segment_id) {
+ RETURN_IF_ERROR(
+ flusher.flush_single_block(&blocks[segment_id],
cast_set<int32_t>(segment_id)));
+ }
+ RETURN_IF_ERROR(flusher.close());
+ const auto expected_vertical_segment_ids =
+ enable_vertical_writer ? segment_collector->segment_ids :
std::vector<uint32_t> {};
+ if (vertical_segment_ids != expected_vertical_segment_ids) {
+ return Status::InternalError(
+ "unexpected Segment writer path for {}: expected {}
vertical flushes, "
+ "observed {}",
+ case_name, expected_vertical_segment_ids.size(),
vertical_segment_ids.size());
+ }
+ *output_collector = std::move(segment_collector);
+ return Status::OK();
+ }
+
void configure_partial_update_context(
RowsetWriterContext& context, const TabletSharedPtr& tablet,
const std::shared_ptr<PartialUpdateInfo>& partial_update_info,
@@ -3094,7 +3330,8 @@ protected:
RowsetWriterContext& context, const TabletSharedPtr& source_tablet,
const TabletSharedPtr& binlog_tablet,
const std::shared_ptr<PartialUpdateInfo>& partial_update_info =
nullptr,
- const std::vector<RowsetSharedPtr>& history = {}, bool need_before
= false) const {
+ const std::vector<RowsetSharedPtr>& history = {}, bool need_before
= false,
+ int64_t rows_per_segment = 3) const {
context.tablet_id = binlog_tablet->tablet_id();
context.tablet = binlog_tablet;
context.data_dir = binlog_tablet->data_dir();
@@ -3115,7 +3352,7 @@ protected:
options.source.source_write_type = DataWriteType::TYPE_DIRECT;
for (int64_t segment_id = 0; segment_id < 2; ++segment_id) {
auto lsn_ids = std::make_shared<std::vector<int64_t>>();
- for (int64_t row = 0; row < 3; ++row) {
+ for (int64_t row = 0; row < rows_per_segment; ++row) {
lsn_ids->push_back(1000 + segment_id * 100 + row);
}
context.insert_segment_allocated_lsns(segment_id,
std::move(lsn_ids));
@@ -3231,6 +3468,169 @@ TEST_F(SegmentFlusherTransformFormatTest,
}
}
+TEST_F(SegmentFlusherTransformFormatTest,
+ FlexiblePartialUpdateRowBinlogMaterializesFullBeforeAndAfterRows) {
+ const auto tablets = create_binlog_tablets(22008, true, true, true, true);
+ ASSERT_NE(tablets.source_tablet, nullptr);
+ ASSERT_NE(tablets.binlog_tablet, nullptr);
+
+ auto history_result =
+ write_mow_history(tablets.source_tablet,
tablets.source_tablet->tablet_schema(), 31012);
+ ASSERT_TRUE(history_result.has_value()) << history_result.error();
+ std::vector<RowsetSharedPtr> history {history_result.value()};
+
+ auto partial_update_info = std::make_shared<PartialUpdateInfo>();
+ ASSERT_TRUE(partial_update_info
+ ->init(tablets.source_tablet->tablet_id(), 1,
+ *tablets.source_tablet->tablet_schema(),
+ UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS,
+ PartialUpdateNewRowPolicyPB::APPEND, {}, false,
0, 0, "UTC", "")
+ .ok());
+ auto block_result =
+
create_flexible_row_binlog_update_block(tablets.source_tablet->tablet_schema());
+ ASSERT_TRUE(block_result.has_value()) << block_result.error();
+
+ constexpr std::string_view case_name = "mow_flexible_partial_row_binlog";
+ auto flush_status = flush_row_binlog_block(
+ case_name, tablets.binlog_tablet->tablet_schema(),
std::move(block_result).value(),
+ [this, tablets, partial_update_info, history](RowsetWriterContext&
context) {
+ configure_row_binlog_context(context, tablets.source_tablet,
tablets.binlog_tablet,
+ partial_update_info, history,
true, 9);
+ });
+ ASSERT_TRUE(flush_status.ok()) << flush_status;
+
+ const auto verify_status =
verify_flexible_row_binlog_segment(tablets.binlog_tablet, case_name);
+ ASSERT_TRUE(verify_status.ok()) << verify_status;
+}
+
+TEST_F(SegmentFlusherTransformFormatTest,
+ FlexiblePartialUpdateRowBinlogKeepsSequenceLoserAlignedWithBaseRowId) {
+ const auto tablets = create_binlog_tablets(22009, true, false, true, true);
+ ASSERT_NE(tablets.source_tablet, nullptr);
+ ASSERT_NE(tablets.binlog_tablet, nullptr);
+
+ auto history_result =
+ write_mow_history(tablets.source_tablet,
tablets.source_tablet->tablet_schema(), 31013);
+ ASSERT_TRUE(history_result.has_value()) << history_result.error();
+ std::vector<RowsetSharedPtr> history {history_result.value()};
+
+ auto partial_update_info = std::make_shared<PartialUpdateInfo>();
+ ASSERT_TRUE(partial_update_info
+ ->init(tablets.source_tablet->tablet_id(), 1,
+ *tablets.source_tablet->tablet_schema(),
+ UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS,
+ PartialUpdateNewRowPolicyPB::APPEND, {}, false,
0, 0, "UTC", "")
+ .ok());
+
+ const std::vector<FlexibleSequenceRow> rows {
+ {.key = 0, .v1 = 9000, .v2 = 9900, .sequence = 4999, .is_delete =
false},
+ {.key = 1, .v1 = 9001, .v2 = 9901, .sequence = 5000, .is_delete =
true},
+ {.key = 2, .v1 = 9002, .v2 = 9902, .sequence = 6002, .is_delete =
false},
+ };
+ auto base_block_result =
+
create_flexible_sequence_block(tablets.source_tablet->tablet_schema(), rows);
+ ASSERT_TRUE(base_block_result.has_value()) << base_block_result.error();
+ auto binlog_block_result =
+
create_flexible_sequence_block(tablets.source_tablet->tablet_schema(), rows);
+ ASSERT_TRUE(binlog_block_result.has_value()) <<
binlog_block_result.error();
+
+ auto mow_context = make_mow_context(tablets.source_tablet->tablet_id(),
history);
+ constexpr std::string_view base_case_name =
"mow_flexible_sequence_alignment_base";
+ std::shared_ptr<TestSegmentCollector> base_collector;
+ auto base_flush_status = flush_blocks_without_golden(
+ base_case_name, tablets.source_tablet->tablet_schema(),
+ {std::move(base_block_result).value()}, false, true,
+ [this, tablets, partial_update_info, history,
+ mow_context](RowsetWriterContext& context) {
+ configure_partial_update_context(context,
tablets.source_tablet,
+ partial_update_info, history);
+ context.mow_context = mow_context;
+ },
+ &base_collector);
+ ASSERT_TRUE(base_flush_status.ok()) << base_flush_status;
+ ASSERT_NE(base_collector, nullptr);
+ ASSERT_EQ(base_collector->segment_ids, (std::vector<uint32_t> {0}));
+ ASSERT_EQ(base_collector->segment_statistics.size(), 1);
+ ASSERT_EQ(base_collector->segment_statistics[0].row_num, 3);
+
+ RowsetId writing_rowset_id;
+ writing_rowset_id.init(10002);
+ EXPECT_TRUE(mow_context->delete_bitmap->contains(
+ {writing_rowset_id, 0, DeleteBitmap::TEMP_VERSION_COMMON}, 0));
+ EXPECT_TRUE(mow_context->delete_bitmap->contains(
+ {writing_rowset_id, 0, DeleteBitmap::TEMP_VERSION_COMMON}, 1));
+ EXPECT_FALSE(mow_context->delete_bitmap->contains(
+ {writing_rowset_id, 0, DeleteBitmap::TEMP_VERSION_COMMON}, 2));
+
+ RowsetWriterContext base_read_context;
+ base_read_context.tablet_schema = tablets.source_tablet->tablet_schema();
+ base_read_context.tablet_id = tablets.source_tablet->tablet_id();
+ base_read_context.rowset_id = writing_rowset_id;
+ auto base_segment =
+ read_logical_segment(fmt::format("{}/{}/segment_0.dat", kTestDir,
base_case_name), 0,
+ tablets.source_tablet->tablet_schema(),
base_read_context, true);
+ ASSERT_TRUE(base_segment.has_value()) << base_segment.error();
+ ASSERT_EQ(base_segment->block.rows(), 3);
+ EXPECT_TRUE(verify_segment_field(base_segment->block, "k1", 0,
+ Field::create_field<TYPE_INT>(Int32(0)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(base_segment->block, "k1", 1,
+ Field::create_field<TYPE_INT>(Int32(1)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(base_segment->block, "k1", 2,
+ Field::create_field<TYPE_INT>(Int32(2)))
+ .ok());
+
+ constexpr std::string_view binlog_case_name =
"mow_flexible_sequence_alignment_binlog";
+ std::shared_ptr<TestSegmentCollector> binlog_collector;
+ auto binlog_flush_status = flush_blocks_without_golden(
+ binlog_case_name, tablets.binlog_tablet->tablet_schema(),
+ {std::move(binlog_block_result).value()}, false, false,
+ [this, tablets, partial_update_info, history,
+ mow_context](RowsetWriterContext& context) {
+ configure_row_binlog_context(context, tablets.source_tablet,
tablets.binlog_tablet,
+ partial_update_info, history,
false, 3);
+
context.write_binlog_opt().write_binlog_config().source.mow_context =
mow_context;
+ },
+ &binlog_collector);
+ ASSERT_TRUE(binlog_flush_status.ok()) << binlog_flush_status;
+ ASSERT_NE(binlog_collector, nullptr);
+ ASSERT_EQ(binlog_collector->segment_ids, (std::vector<uint32_t> {0}));
+ ASSERT_EQ(binlog_collector->segment_statistics.size(), 1);
+ ASSERT_EQ(binlog_collector->segment_statistics[0].row_num, 3);
+
+ auto binlog_segment = read_row_binlog_segment(tablets.binlog_tablet,
binlog_case_name, 0);
+ ASSERT_TRUE(binlog_segment.has_value()) << binlog_segment.error();
+ ASSERT_EQ(binlog_segment->rows(), 3);
+ EXPECT_TRUE(
+ verify_segment_field(*binlog_segment, "k1", 0,
Field::create_field<TYPE_INT>(Int32(0)))
+ .ok());
+ EXPECT_TRUE(
+ verify_segment_field(*binlog_segment, "k1", 1,
Field::create_field<TYPE_INT>(Int32(1)))
+ .ok());
+ EXPECT_TRUE(
+ verify_segment_field(*binlog_segment, "k1", 2,
Field::create_field<TYPE_INT>(Int32(2)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(*binlog_segment, BINLOG_LSN_COL, 0,
+
Field::create_field<TYPE_BIGINT>(Int64(1000)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(*binlog_segment, BINLOG_OP_COL, 0,
+
Field::create_field<TYPE_BIGINT>(Int64(ROW_BINLOG_UPDATE)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(*binlog_segment, BINLOG_LSN_COL, 1,
+
Field::create_field<TYPE_BIGINT>(Int64(1001)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(*binlog_segment, BINLOG_OP_COL, 1,
+
Field::create_field<TYPE_BIGINT>(Int64(ROW_BINLOG_DELETE)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(*binlog_segment, BINLOG_LSN_COL, 2,
+
Field::create_field<TYPE_BIGINT>(Int64(1002)))
+ .ok());
+ EXPECT_TRUE(verify_segment_field(*binlog_segment, BINLOG_OP_COL, 2,
+
Field::create_field<TYPE_BIGINT>(Int64(ROW_BINLOG_APPEND)))
+ .ok());
+}
+
TEST_F(SegmentFlusherFormatTest, VariantLogicalComparisonPreservesScalarTypes)
{
VariantMap boolean_object;
boolean_object.try_emplace(
diff --git a/be/test/storage/transform/row_binlog_derive_test.cpp
b/be/test/storage/transform/row_binlog_derive_test.cpp
index 4ea6bb7ab17..07fc267d211 100644
--- a/be/test/storage/transform/row_binlog_derive_test.cpp
+++ b/be/test/storage/transform/row_binlog_derive_test.cpp
@@ -27,8 +27,10 @@
// existing one, DELETE on a delete sign), AFTER rebuilt for a fixed
// partial update, the BEFORE image mirroring history, the
// zero-value-column no-op, and the sequence-column source path
-// Rejects* flexible partial update, a fixed partial update of the wrong
-// width, a missing source schema, and a negative segment id
+// Flexible partial update materialization through the MoW derive stage,
+// including aligned LSN and operation output
+// Rejects* a fixed partial update of the wrong width, a missing source
+// schema, and a negative segment id
//
// Oracle values are ported from the deleted RowBinlogSegmentWriter.
@@ -39,6 +41,7 @@
#include <vector>
#include "common/config.h"
+#include "core/column/column_complex.h"
#include "core/column/column_nullable.h"
#include "core/column/column_vector.h"
#include "storage/binlog.h"
@@ -1393,16 +1396,25 @@ TEST_F(RowBinlogDeriveTest,
MowSeqSourceSeqLosesStillReadsHistory) {
}
// ===========================================================================
-// §5.4 Reject / negative paths.
+// §5.4 Flexible support and reject / negative paths.
// ===========================================================================
-// N4 (B9): binlog<row> does not support flexible partial update.
-TEST_F(RowBinlogDeriveTest, MowRejectsFlexiblePartialUpdate) {
- auto binlog_tablet =
- create_binlog_tablets(/*tablet_id=*/8501, TKeysType::DUP_KEYS,
/*mow=*/false).binlog;
- ASSERT_TRUE(binlog_tablet != nullptr);
- auto binlog_schema = binlog_tablet->tablet_schema();
- auto source_schema = create_binlog_pu_source_schema();
+// A flexible partial update reaches the same MoW derive stage as a fixed
+// partial update, but materializes its full AFTER row from the skip bitmap.
+TEST_F(RowBinlogDeriveTest, MowMaterializesFlexiblePartialUpdate) {
+ TabletSchemaPB source_pb;
+ create_flexible_mow_schema()->to_schema_pb(&source_pb);
+ source_pb.mutable_column(source_pb.delete_sign_idx())->set_visible(false);
+
source_pb.mutable_column(source_pb.skip_bitmap_col_idx())->set_visible(false);
+ auto source_schema = std::make_shared<TabletSchema>();
+ source_schema->init_from_pb(source_pb);
+
+ TabletSchemaPB binlog_pb;
+ create_binlog_seq_schema()->to_schema_pb(&binlog_pb);
+ binlog_pb.mutable_column(0)->set_name("k");
+ binlog_pb.mutable_column(1)->set_name("v");
+ auto binlog_schema = std::make_shared<TabletSchema>();
+ binlog_schema->init_from_pb(binlog_pb);
auto probe_tablet = make_tablet(source_schema, /*tablet_id=*/8502);
auto mow = make_mow_context(100, {});
@@ -1425,7 +1437,7 @@ TEST_F(RowBinlogDeriveTest,
MowRejectsFlexiblePartialUpdate) {
cfg.source.is_transient_rowset_writer = false;
cfg.source.mow_context = mow;
cfg.write_before = false;
- register_segment_lsns(rwc, 2);
+ register_segment_lsns(rwc, 1);
auto chain = build_transform_chain(rwc);
EXPECT_EQ(chain.stage_names(), (std::vector<std::string_view>
{"MowRowBinlogDerive"}));
@@ -1435,19 +1447,25 @@ TEST_F(RowBinlogDeriveTest,
MowRejectsFlexiblePartialUpdate) {
ctx.mow_context = mow;
ctx.partial_update_info = pui;
- Block block = source_schema->create_storage_block({0, 1});
+ Block block = source_schema->create_storage_block();
{
auto guard = block.mutate_columns_scoped();
auto& cols = guard.mutable_columns();
int32_t k = 1, v1 = 10;
+ int8_t delete_sign = 0;
cols[0]->insert_data(reinterpret_cast<const char*>(&k),
sizeof(int32_t));
cols[1]->insert_data(reinterpret_cast<const char*>(&v1),
sizeof(int32_t));
+ cols[2]->insert_data(reinterpret_cast<const char*>(&delete_sign),
sizeof(int8_t));
+ assert_cast<ColumnBitmap*>(cols[3].get())->insert_value(BitmapValue
{});
}
- auto st = chain.apply(ctx, &block);
- EXPECT_FALSE(st.ok());
- EXPECT_NE(st.to_string().find("does not support flexible partial update"),
std::string::npos)
- << st;
+ ASSERT_TRUE(chain.apply(ctx, &block).ok());
+ ASSERT_EQ(block.columns(), binlog_schema->num_columns());
+ ASSERT_EQ(block.rows(), 1);
+ EXPECT_EQ(read_int(block, 0, 0), 1);
+ EXPECT_EQ(read_int(block, 1, 0), 10);
+ EXPECT_EQ(read_lsn(block, binlog_schema->binlog_lsn_col_idx(), 0), 1000);
+ EXPECT_EQ(read_op(block, binlog_schema->binlog_op_col_idx(), 0),
ROW_BINLOG_APPEND);
}
// N5/N6 (B10): the fixed-PU MoW derive rejects a block that is too narrow
diff --git
a/regression-test/data/row_binlog_p0/test_row_binlog_flexible_partial_update.out
b/regression-test/data/row_binlog_p0/test_row_binlog_flexible_partial_update.out
new file mode 100644
index 00000000000..faffe70fcb9
--- /dev/null
+++
b/regression-test/data/row_binlog_p0/test_row_binlog_flexible_partial_update.out
@@ -0,0 +1,69 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !flexible_rejected_sequence_visible --
+5 50 e 105 500
+6 60 f 106 500
+
+-- !flexible_rejected_sequence_binlog --
+5 50 e 105 500 0
+6 60 f 106 500 0
+
+-- !flexible_delete_then_insert_visible --
+7 0 default \N 160
+
+-- !flexible_delete_then_insert_binlog --
+0 7 70 g 107 100 \N \N \N \N
+1 7 0 default \N 160 70 g 107 100
+
+-- !flexible_visible --
+1 11 aa 110
+2 20 \N 120
+3 30 default 130
+5 50 e 500
+6 60 f 500
+7 0 default 160
+
+-- !flexible_raw_binlog --
+0 1 10 a 100 \N \N \N
+0 2 20 b 100 \N \N \N
+0 4 40 d 100 \N \N \N
+0 5 50 e 500 \N \N \N
+0 6 60 f 500 \N \N \N
+0 7 70 g 100 \N \N \N
+1 1 11 aa 110 10 a 100
+1 2 20 \N 120 20 b 100
+0 3 30 default 130 \N \N \N
+2 4 40 d 140 40 d 100
+1 7 0 default 160 70 g 100
+
+-- !flexible_detail --
+1 10 a 100 0
+2 20 b 100 0
+4 40 d 100 0
+5 50 e 500 0
+6 60 f 500 0
+7 70 g 100 0
+1 10 a 100 2
+1 11 aa 110 3
+2 20 b 100 2
+2 20 \N 120 3
+3 30 default 130 0
+4 40 d 100 1
+7 70 g 100 2
+7 0 default 160 3
+
+-- !flexible_min_delta --
+1 11 aa 110 0
+2 20 \N 120 0
+3 30 default 130 0
+5 50 e 500 0
+6 60 f 500 0
+7 0 default 160 0
+
+-- !flexible_append_only --
+1 10 a 100 0
+2 20 b 100 0
+4 40 d 100 0
+5 50 e 500 0
+6 60 f 500 0
+7 70 g 100 0
+3 30 default 130 0
diff --git
a/regression-test/data/row_binlog_p0/test_row_binlog_flexible_partial_update_no_sequence.out
b/regression-test/data/row_binlog_p0/test_row_binlog_flexible_partial_update_no_sequence.out
new file mode 100644
index 00000000000..c2d4d89d3bd
--- /dev/null
+++
b/regression-test/data/row_binlog_p0/test_row_binlog_flexible_partial_update_no_sequence.out
@@ -0,0 +1,44 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !no_sequence_visible --
+1 11 aa 101
+2 20 \N 102
+3 30 default \N
+5 0 default \N
+
+-- !no_sequence_raw_binlog --
+0 1 10 a 101 \N \N \N
+0 2 20 b 102 \N \N \N
+0 4 40 d 104 \N \N \N
+0 5 50 e 105 \N \N \N
+1 1 11 aa 101 10 a 101
+1 2 20 \N 102 20 b 102
+0 3 30 default \N \N \N \N
+2 4 40 d 104 40 d 104
+1 5 0 default \N 50 e 105
+
+-- !no_sequence_detail --
+1 10 a 101 0
+2 20 b 102 0
+4 40 d 104 0
+5 50 e 105 0
+1 10 a 101 2
+1 11 aa 101 3
+2 20 b 102 2
+2 20 \N 102 3
+3 30 default \N 0
+4 40 d 104 1
+5 50 e 105 2
+5 0 default \N 3
+
+-- !no_sequence_min_delta --
+1 11 aa 101 0
+2 20 \N 102 0
+3 30 default \N 0
+5 0 default \N 0
+
+-- !no_sequence_append_only --
+1 10 a 101 0
+2 20 b 102 0
+4 40 d 104 0
+5 50 e 105 0
+3 30 default \N 0
diff --git
a/regression-test/suites/row_binlog_p0/test_row_binlog_flexible_partial_update.groovy
b/regression-test/suites/row_binlog_p0/test_row_binlog_flexible_partial_update.groovy
new file mode 100644
index 00000000000..81a693be9d0
--- /dev/null
+++
b/regression-test/suites/row_binlog_p0/test_row_binlog_flexible_partial_update.groovy
@@ -0,0 +1,176 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_row_binlog_flexible_partial_update") {
+ def tableName = "test_row_binlog_flexible_partial_update"
+ sql "DROP TABLE IF EXISTS ${tableName} FORCE"
+ try {
+ sql """
+ CREATE TABLE ${tableName} (
+ id INT,
+ v1 INT NOT NULL DEFAULT "0",
+ v2 VARCHAR(16) NULL DEFAULT "default",
+ v3 INT NULL,
+ seq INT
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "light_schema_change" = "true",
+ "enable_unique_key_skip_bitmap_column" = "true",
+ "function_column.sequence_col" = "seq",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql """
+ INSERT INTO ${tableName} VALUES
+ (1, 10, 'a', 101, 100),
+ (2, 20, 'b', 102, 100),
+ (4, 40, 'd', 104, 100),
+ (5, 50, 'e', 105, 500),
+ (6, 60, 'f', 106, 500),
+ (7, 70, 'g', 107, 100)
+ """
+
+ def flexibleRows = """
+ {"id":1,"v1":11,"seq":110}
+ {"id":1,"v2":"aa"}
+ {"id":2,"v2":null,"seq":120}
+ {"id":3,"v1":30,"seq":130}
+ {"id":4,"__DORIS_DELETE_SIGN__":1,"seq":140}
+ """.stripIndent().trim()
+ streamLoad {
+ table tableName
+ set "format", "json"
+ set "read_json_by_line", "true"
+ set "strict_mode", "false"
+ set "unique_key_update_mode", "UPDATE_FLEXIBLE_COLUMNS"
+ inputStream new ByteArrayInputStream(flexibleRows.getBytes())
+ time 20000
+ }
+
+ // A single lower-sequence update loses to the visible row. It must
neither change the
+ // base table nor produce a visible Row Binlog change.
+ streamLoad {
+ table tableName
+ set "format", "json"
+ set "read_json_by_line", "true"
+ set "strict_mode", "false"
+ set "unique_key_update_mode", "UPDATE_FLEXIBLE_COLUMNS"
+ inputStream new ByteArrayInputStream(
+ '{"id":5,"v1":5000,"seq":400}'.getBytes())
+ time 20000
+ }
+
+ // The same rule applies to a lower-sequence delete.
+ streamLoad {
+ table tableName
+ set "format", "json"
+ set "read_json_by_line", "true"
+ set "strict_mode", "false"
+ set "unique_key_update_mode", "UPDATE_FLEXIBLE_COLUMNS"
+ inputStream new ByteArrayInputStream(
+ '{"id":6,"__DORIS_DELETE_SIGN__":1,"seq":400}'.getBytes())
+ time 20000
+ }
+
+ // DELETE followed by INSERT for an existing key in one load keeps the
old row as BEFORE,
+ // but omitted AFTER columns use default/null instead of resurrecting
historical values.
+ def deleteThenInsertRows = """
+ {"id":7,"__DORIS_DELETE_SIGN__":1,"seq":150}
+ {"id":7,"seq":160}
+ """.stripIndent().trim()
+ streamLoad {
+ table tableName
+ set "format", "json"
+ set "read_json_by_line", "true"
+ set "strict_mode", "false"
+ set "unique_key_update_mode", "UPDATE_FLEXIBLE_COLUMNS"
+ inputStream new
ByteArrayInputStream(deleteThenInsertRows.getBytes())
+ time 20000
+ }
+ sql "sync"
+
+ order_qt_flexible_rejected_sequence_visible """
+ SELECT id, v1, v2, v3, seq
+ FROM ${tableName}
+ WHERE id IN (5, 6)
+ ORDER BY id
+ """
+
+ qt_flexible_rejected_sequence_binlog """
+ SELECT id, v1, v2, v3, seq, __DORIS_BINLOG_OP__
+ FROM binlog("table" = "${tableName}")
+ WHERE id IN (5, 6)
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
+ order_qt_flexible_delete_then_insert_visible """
+ SELECT id, v1, v2, v3, seq
+ FROM ${tableName}
+ WHERE id = 7
+ """
+
+ qt_flexible_delete_then_insert_binlog """
+ SELECT __DORIS_BINLOG_OP__, id, v1, v2, v3, seq,
+ __BEFORE__v1__, __BEFORE__v2__, __BEFORE__v3__,
__BEFORE__seq__
+ FROM binlog("table" = "${tableName}")
+ WHERE id = 7
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
+ order_qt_flexible_visible """
+ SELECT id, v1, v2, seq
+ FROM ${tableName}
+ ORDER BY id
+ """
+
+ qt_flexible_raw_binlog """
+ SELECT __DORIS_BINLOG_OP__, id, v1, v2, seq,
+ __BEFORE__v1__, __BEFORE__v2__, __BEFORE__seq__
+ FROM binlog("table" = "${tableName}")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
+ if (!isCloudMode()) {
+ qt_flexible_detail """
+ SELECT id, v1, v2, seq, __DORIS_BINLOG_OP__
+ FROM ${tableName}@incr("incrementType" = "DETAIL")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__,
__DORIS_BINLOG_OP__
+ """
+
+ order_qt_flexible_min_delta """
+ SELECT id, v1, v2, seq, __DORIS_BINLOG_OP__
+ FROM ${tableName}@incr("incrementType" = "MIN_DELTA")
+ ORDER BY id, __DORIS_BINLOG_OP__
+ """
+
+ qt_flexible_append_only """
+ SELECT id, v1, v2, seq, __DORIS_BINLOG_OP__
+ FROM ${tableName}@incr("incrementType" = "APPEND_ONLY")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+ }
+ } finally {
+ sql "DROP TABLE IF EXISTS ${tableName} FORCE"
+ }
+}
diff --git
a/regression-test/suites/row_binlog_p0/test_row_binlog_flexible_partial_update_no_sequence.groovy
b/regression-test/suites/row_binlog_p0/test_row_binlog_flexible_partial_update_no_sequence.groovy
new file mode 100644
index 00000000000..c6bf42eecd3
--- /dev/null
+++
b/regression-test/suites/row_binlog_p0/test_row_binlog_flexible_partial_update_no_sequence.groovy
@@ -0,0 +1,101 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_row_binlog_flexible_partial_update_no_sequence", "nonConcurrent") {
+ sql "DROP TABLE IF EXISTS
test_row_binlog_flexible_partial_update_no_sequence FORCE"
+
+ sql """
+ CREATE TABLE test_row_binlog_flexible_partial_update_no_sequence (
+ id INT,
+ v1 INT NOT NULL DEFAULT "0",
+ v2 VARCHAR(16) NULL DEFAULT "default",
+ v3 INT NULL
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "light_schema_change" = "true",
+ "enable_unique_key_skip_bitmap_column" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql """
+ INSERT INTO test_row_binlog_flexible_partial_update_no_sequence VALUES
+ (1, 10, 'a', 101),
+ (2, 20, 'b', 102),
+ (4, 40, 'd', 104),
+ (5, 50, 'e', 105)
+ """
+
+ // Cover duplicate-key aggregation, history fill, explicit NULL, a new key,
+ // DELETE, and DELETE -> INSERT without relying on a sequence column.
+ def flexibleRows = """
+ {"id":1,"v1":11}
+ {"id":1,"v2":"aa"}
+ {"id":2,"v2":null}
+ {"id":3,"v1":30}
+ {"id":4,"__DORIS_DELETE_SIGN__":1}
+ {"id":5,"__DORIS_DELETE_SIGN__":1}
+ {"id":5}
+ """.stripIndent().trim()
+ streamLoad {
+ table "test_row_binlog_flexible_partial_update_no_sequence"
+ set "format", "json"
+ set "read_json_by_line", "true"
+ set "strict_mode", "false"
+ set "unique_key_update_mode", "UPDATE_FLEXIBLE_COLUMNS"
+ inputStream new ByteArrayInputStream(flexibleRows.getBytes())
+ time 20000
+ }
+ sql "sync"
+
+ order_qt_no_sequence_visible """
+ SELECT id, v1, v2, v3
+ FROM test_row_binlog_flexible_partial_update_no_sequence
+ ORDER BY id
+ """
+
+ qt_no_sequence_raw_binlog """
+ SELECT __DORIS_BINLOG_OP__, id, v1, v2, v3,
+ __BEFORE__v1__, __BEFORE__v2__, __BEFORE__v3__
+ FROM binlog("table" =
"test_row_binlog_flexible_partial_update_no_sequence")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
+ qt_no_sequence_detail """
+ SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
+ FROM
test_row_binlog_flexible_partial_update_no_sequence@incr("incrementType" =
"DETAIL")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__,
__DORIS_BINLOG_OP__
+ """
+
+ order_qt_no_sequence_min_delta """
+ SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
+ FROM
test_row_binlog_flexible_partial_update_no_sequence@incr("incrementType" =
"MIN_DELTA")
+ ORDER BY id, __DORIS_BINLOG_OP__
+ """
+
+ qt_no_sequence_append_only """
+ SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
+ FROM
test_row_binlog_flexible_partial_update_no_sequence@incr("incrementType" =
"APPEND_ONLY")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
