This is an automated email from the ASF dual-hosted git repository.
hello-stephen 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 977eaa7f4a8 [fix](row binlog): op should be insert but update when
insert after delete (#66046)
977eaa7f4a8 is described below
commit 977eaa7f4a8e85d8037c3be4b424ce870eb4ce01
Author: Userwhite <[email protected]>
AuthorDate: Fri Jul 31 11:53:44 2026 +0800
[fix](row binlog): op should be insert but update when insert after delete
(#66046)
### What problem does this PR solve?
Issue Number: close https://github.com/apache/doris/issues/65808
The main reason for the OP error is that the data has already been
deleted, but in Doris, deletion may only set the delete sign to 1, so
the data can still be queried when retrieving historical data.
Therefore, we need to proactively detect this situation.
---------
Co-authored-by: Gavin Chou <[email protected]>
---
be/src/storage/partial_update_info.cpp | 36 +++++++++--
be/src/storage/partial_update_info.h | 6 +-
.../storage/segment/historical_row_retriever.cpp | 43 +++++++++++--
be/src/storage/segment/historical_row_retriever.h | 8 +++
.../storage/segment/row_binlog_segment_writer.cpp | 21 ++++---
.../segment/historical_row_retriever_test.cpp | 67 ++++++++++++++++++++
.../row_binlog_p0/test_binlog_changes_syntax.out | 16 ++---
.../data/row_binlog_p0/test_row_binlog_basic.out | 13 +++-
.../test_binlog_changes_syntax.groovy | 14 ++---
.../row_binlog_p0/test_row_binlog_basic.groovy | 62 +++++++++++++++++++
.../table_stream_p0/test_min_delta_stream.groovy | 72 +++++++++++++++++++---
11 files changed, 314 insertions(+), 44 deletions(-)
diff --git a/be/src/storage/partial_update_info.cpp
b/be/src/storage/partial_update_info.cpp
index b41df738472..1bea6e05cb7 100644
--- a/be/src/storage/partial_update_info.cpp
+++ b/be/src/storage/partial_update_info.cpp
@@ -384,12 +384,36 @@ Status FixedReadPlan::read_columns_by_plan(
return Status::OK();
}
+Status FixedReadPlan::fill_old_delete_signs(const Block& old_value_block,
+ const std::map<uint32_t,
uint32_t>& read_index,
+ size_t num_rows,
+ std::vector<signed char>*
old_delete_signs) const {
+ if (old_delete_signs == nullptr) {
+ return Status::OK();
+ }
+ const auto* old_delete_sign_column_data =
+ BaseTablet::get_delete_sign_column_data(old_value_block);
+ if (old_delete_sign_column_data == nullptr) {
+ return Status::InternalError("old delete signs column not found,
block: {}",
+ old_value_block.dump_structure());
+ }
+ old_delete_signs->assign(num_rows, 0);
+ for (size_t idx = 0; idx < num_rows; ++idx) {
+ auto it = read_index.find(cast_set<uint32_t>(idx));
+ if (it != read_index.end()) {
+ (*old_delete_signs)[idx] = old_delete_sign_column_data[it->second];
+ }
+ }
+ return Status::OK();
+}
+
Status FixedReadPlan::fill_missing_columns(
const segment_v2::HistoricalRowRetrieverContext& historical_context,
const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
const TabletSchema& tablet_schema, Block& full_block,
const std::vector<bool>& use_default_or_null_flag, bool
has_default_or_nullable,
- uint32_t segment_start_pos, const Block* block) const {
+ uint32_t segment_start_pos, const Block* block,
+ std::vector<signed char>* old_delete_signs) const {
auto mutable_full_columns_guard = full_block.mutate_columns_scoped();
auto& mutable_full_columns = mutable_full_columns_guard.mutable_columns();
// create old value columns
@@ -413,11 +437,14 @@ Status FixedReadPlan::fill_missing_columns(
RETURN_IF_ERROR(read_columns_by_plan(tablet_schema, missing_cids,
rsid_to_rowset,
old_value_block, &read_index, true,
nullptr));
- const auto* old_delete_signs =
BaseTablet::get_delete_sign_column_data(old_value_block);
- if (old_delete_signs == nullptr) {
+ const auto* old_delete_sign_column_data =
+ BaseTablet::get_delete_sign_column_data(old_value_block);
+ if (old_delete_sign_column_data == nullptr) {
return Status::InternalError("old delete signs column not found,
block: {}",
old_value_block.dump_structure());
}
+ RETURN_IF_ERROR(fill_old_delete_signs(old_value_block, read_index,
+ use_default_or_null_flag.size(),
old_delete_signs));
// build default value columns
auto default_value_block = old_value_block.clone_empty();
RETURN_IF_ERROR(BaseTablet::generate_default_value_block(tablet_schema,
missing_cids,
@@ -439,8 +466,7 @@ Status FixedReadPlan::fill_missing_columns(
bool should_use_default = use_default_or_null_flag[idx];
if (!should_use_default) {
- bool old_row_delete_sign =
- (old_delete_signs != nullptr &&
old_delete_signs[pos_in_old_block] != 0);
+ bool old_row_delete_sign =
old_delete_sign_column_data[pos_in_old_block] != 0;
if (old_row_delete_sign) {
if (!tablet_schema.has_sequence_col()) {
should_use_default = true;
diff --git a/be/src/storage/partial_update_info.h
b/be/src/storage/partial_update_info.h
index 0192f35b9db..2066bf732e7 100644
--- a/be/src/storage/partial_update_info.h
+++ b/be/src/storage/partial_update_info.h
@@ -135,7 +135,11 @@ public:
const TabletSchema& tablet_schema, Block&
full_block,
const std::vector<bool>&
use_default_or_null_flag,
bool has_default_or_nullable, uint32_t
segment_start_pos,
- const Block* block) const;
+ const Block* block,
+ std::vector<signed char>* old_delete_signs =
nullptr) const;
+ Status fill_old_delete_signs(const Block& old_value_block,
+ const std::map<uint32_t, uint32_t>&
read_index, size_t num_rows,
+ std::vector<signed char>* old_delete_signs)
const;
private:
std::map<RowsetId, std::map<uint32_t /* segment_id */,
std::vector<RidAndPos>>> plan;
diff --git a/be/src/storage/segment/historical_row_retriever.cpp
b/be/src/storage/segment/historical_row_retriever.cpp
index b56daef47cb..4108365deed 100644
--- a/be/src/storage/segment/historical_row_retriever.cpp
+++ b/be/src/storage/segment/historical_row_retriever.cpp
@@ -86,8 +86,6 @@ Status
PrimaryKeyModelRowRetriever::retrieve_historical_row(const Int8* delete_s
auto* tablet = static_cast<Tablet*>(_context.tablet.get());
auto& tablet_schema = _context.tablet_schema;
- DCHECK(_context.partial_update_info);
-
std::vector<RowsetSharedPtr> specified_rowsets;
{
std::shared_lock rlock(_context.tablet->get_header_lock());
@@ -169,7 +167,7 @@ Status
PrimaryKeyModelRowRetriever::build_after_block(Block* block, size_t row_p
}
return _rssid_to_rid.fill_missing_columns(
_context, _rsid_to_rowset, *_context.tablet_schema, *block,
_use_default_or_null_flag,
- _has_default_or_nullable, cast_set<uint32_t>(row_pos), block);
+ _has_default_or_nullable, cast_set<uint32_t>(row_pos), block,
&_old_delete_signs);
}
Status PrimaryKeyModelRowRetriever::build_before_block(Block* before_block,
@@ -193,8 +191,9 @@ Status
PrimaryKeyModelRowRetriever::build_before_block(Block* before_block,
// key: logical row index in current batch; value: index in old_value_block
std::map<uint32_t, uint32_t> read_index;
RETURN_IF_ERROR(_rssid_to_rid.read_columns_by_plan(*tablet_schema,
value_cids, _rsid_to_rowset,
- old_value_block,
&read_index, false,
+ old_value_block,
&read_index, true,
nullptr));
+ RETURN_IF_ERROR(_fill_old_delete_signs(old_value_block, read_index,
num_rows));
{
auto mutable_before_columns_guard =
before_block->mutate_columns_scoped();
@@ -202,8 +201,8 @@ Status
PrimaryKeyModelRowRetriever::build_before_block(Block* before_block,
// Fill each row in before_block.
for (uint32_t idx = 0; idx < num_rows; ++idx) {
auto it = read_index.find(idx);
- if (it == read_index.end()) {
- // No historical row, fill BEFORE with NULL.
+ if (it == read_index.end() || _old_delete_signs[idx] != 0) {
+ // No live historical row, fill BEFORE with NULL.
for (size_t i = 0; i < value_cids.size(); ++i) {
auto* nullable_column =
assert_cast<ColumnNullable*>(mutable_before_columns[i].get());
@@ -223,6 +222,38 @@ Status
PrimaryKeyModelRowRetriever::build_before_block(Block* before_block,
return Status::OK();
}
+Status PrimaryKeyModelRowRetriever::revise_operators_by_old_delete_sign(size_t
num_rows) {
+ if (_operators.empty() || _rssid_to_rid.empty()) {
+ return Status::OK();
+ }
+ DCHECK_EQ(_operators.size(), num_rows);
+
+ if (_old_delete_signs.empty()) {
+ // If no BEFORE/missing column was read, read only old delete signs
here.
+ Block old_delete_sign_block;
+ std::map<uint32_t, uint32_t> read_index;
+ RETURN_IF_ERROR(_rssid_to_rid.read_columns_by_plan(
+ *_context.tablet_schema, std::vector<uint32_t> {},
_rsid_to_rowset,
+ old_delete_sign_block, &read_index, true, nullptr));
+ RETURN_IF_ERROR(_fill_old_delete_signs(old_delete_sign_block,
read_index, num_rows));
+ }
+ DCHECK_EQ(_old_delete_signs.size(), num_rows);
+
+ for (size_t idx = 0; idx < num_rows; ++idx) {
+ if (_operators[idx] == ROW_BINLOG_UPDATE && _old_delete_signs[idx] !=
0) {
+ _operators[idx] = ROW_BINLOG_APPEND;
+ }
+ }
+ return Status::OK();
+}
+
+Status PrimaryKeyModelRowRetriever::_fill_old_delete_signs(
+ const Block& old_value_block, const std::map<uint32_t, uint32_t>&
read_index,
+ size_t num_rows) {
+ return _rssid_to_rid.fill_old_delete_signs(old_value_block, read_index,
num_rows,
+ &_old_delete_signs);
+}
+
std::string PrimaryKeyModelRowRetriever::_full_encode_keys(
const std::vector<IOlapColumnDataAccessor*>& key_columns, size_t pos,
bool null_first) {
return _full_encode_keys(_key_coders, key_columns, pos, null_first);
diff --git a/be/src/storage/segment/historical_row_retriever.h
b/be/src/storage/segment/historical_row_retriever.h
index 5949896a476..56a5b7664e7 100644
--- a/be/src/storage/segment/historical_row_retriever.h
+++ b/be/src/storage/segment/historical_row_retriever.h
@@ -82,6 +82,8 @@ public:
Status build_before_block(Block* before_block, const
std::vector<uint32_t>& value_cids,
size_t /*row_pos*/, size_t num_rows) override;
+ Status revise_operators_by_old_delete_sign(size_t num_rows);
+
void clear() override {
_key_columns.clear();
_seq_column = nullptr;
@@ -90,11 +92,15 @@ public:
_rssid_to_rid.clear();
_rsid_to_rowset.clear();
_operators.clear();
+ _old_delete_signs.clear();
}
std::vector<int64_t>& get_operators() override { return _operators; };
private:
+ Status _fill_old_delete_signs(const Block& old_value_block,
+ const std::map<uint32_t, uint32_t>&
read_index, size_t num_rows);
+
void _maybe_invalid_row_cache(const std::string& key);
// used for unique-key with merge on write and segment min_max key
@@ -128,6 +134,8 @@ private:
// cache operator for fill_binlog_columns
std::vector<int64_t> _operators;
+ // Aligned by row position in the current append batch.
+ std::vector<signed char> _old_delete_signs;
};
} // namespace segment_v2
diff --git a/be/src/storage/segment/row_binlog_segment_writer.cpp
b/be/src/storage/segment/row_binlog_segment_writer.cpp
index 6a5737a613f..e0430c98f48 100644
--- a/be/src/storage/segment/row_binlog_segment_writer.cpp
+++ b/be/src/storage/segment/row_binlog_segment_writer.cpp
@@ -194,8 +194,9 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
// We read historical rows only when we really need them:
// 1. partial update: build the full AFTER row.
// 2. write_before: fill __BEFORE__* columns.
- // Otherwise we do not compare with old rows here, so row binlog op only
- // keeps the simple meaning: append for non-delete rows, delete for delete
rows.
+ // 3. partial update / write_before need the old delete sign to identify
INSERT -> DELETE
+ // -> INSERT correctly; otherwise the last INSERT may be treated as
UPDATE because lookup
+ // hits the tombstone row.
if (is_partial_update || _write_before) {
auto* pk_retriever =
dynamic_cast<PrimaryKeyModelRowRetriever*>(_historical_data_writer.get());
@@ -263,15 +264,22 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
}
}
+ if (_write_before) {
+ RETURN_IF_ERROR(_fill_before_columns(num_rows));
+ }
+
+ if (is_partial_update || _write_before) {
+ auto* pk_retriever =
+
dynamic_cast<PrimaryKeyModelRowRetriever*>(_historical_data_writer.get());
+ DCHECK(pk_retriever != nullptr);
+
RETURN_IF_ERROR(pk_retriever->revise_operators_by_old_delete_sign(num_rows));
+ }
+
RETURN_IF_ERROR(_fill_binlog_columns(num_rows, operators));
// row-binlog key don't need seq column
RETURN_IF_ERROR(build_key_index(_converted_key_columns, nullptr,
num_rows));
- if (_write_before) {
- RETURN_IF_ERROR(_fill_before_columns(num_rows));
- }
-
_num_rows_written += num_rows;
// need to clean olap_data_convertor that be used when fill binlog columns
and build key index
_olap_data_convertor->clear_source_content();
@@ -334,7 +342,6 @@ Status RowBinlogSegmentWriter::_fill_binlog_columns(size_t
num_rows,
assert_cast<ColumnInt64*>(lsn_col_ptr)->insert_value(_lsn_ids->at(i));
}
- // wrong op only happens when partial-update, it will be fixed by
delete bitmap when publish
const FieldType op_col_type =
_tablet_schema->column(binlog_cids[2]).type();
IColumn* op_col_ptr = binlog_prefix_columns[2].get();
auto* op_nullable_column =
check_and_get_column<ColumnNullable>(op_col_ptr);
diff --git a/be/test/storage/segment/historical_row_retriever_test.cpp
b/be/test/storage/segment/historical_row_retriever_test.cpp
new file mode 100644
index 00000000000..05933e5e31b
--- /dev/null
+++ b/be/test/storage/segment/historical_row_retriever_test.cpp
@@ -0,0 +1,67 @@
+// 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.
+
+#include "storage/segment/historical_row_retriever.h"
+
+#include <gtest/gtest.h>
+
+#include <map>
+#include <memory>
+#include <vector>
+
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "storage/binlog.h"
+#include "storage/olap_common.h"
+#include "storage/utils.h"
+
+namespace doris::segment_v2 {
+
+TEST(HistoricalRowRetrieverTest, ReviseOperatorWithOldDeleteSign) {
+ auto delete_sign_column = ColumnInt8::create();
+ delete_sign_column->insert_value(1);
+ delete_sign_column->insert_value(0);
+ delete_sign_column->insert_value(1);
+
+ Block old_value_block;
+ old_value_block.insert(
+ {std::move(delete_sign_column), std::make_shared<DataTypeInt8>(),
DELETE_SIGN});
+
+ std::map<uint32_t, uint32_t> read_index {
+ {0, 0},
+ {1, 1},
+ {3, 2},
+ };
+
+ PrimaryKeyModelRowRetriever retriever;
+ ASSERT_TRUE(retriever._fill_old_delete_signs(old_value_block, read_index,
4).ok());
+ ASSERT_EQ((std::vector<signed char> {1, 0, 0, 1}),
retriever._old_delete_signs);
+
+ retriever._operators = {ROW_BINLOG_UPDATE, ROW_BINLOG_UPDATE,
ROW_BINLOG_UPDATE,
+ ROW_BINLOG_DELETE};
+ RowsetId rowset_id;
+ rowset_id.init(1);
+ retriever._rssid_to_rid.prepare_to_read(RowLocation(rowset_id, 0, 0), 0);
+ ASSERT_TRUE(retriever.revise_operators_by_old_delete_sign(4).ok());
+
+ EXPECT_EQ(ROW_BINLOG_APPEND, retriever._operators[0]);
+ EXPECT_EQ(ROW_BINLOG_UPDATE, retriever._operators[1]);
+ EXPECT_EQ(ROW_BINLOG_UPDATE, retriever._operators[2]);
+ EXPECT_EQ(ROW_BINLOG_DELETE, retriever._operators[3]);
+}
+
+} // namespace doris::segment_v2
diff --git a/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
b/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
index f2fa93f1918..5f056221642 100644
--- a/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
+++ b/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
@@ -65,6 +65,7 @@
-- !mow_append_range --
2 20 b 0
+2 21 b1 0
4 40 d 0
5 50 e 0
@@ -82,10 +83,9 @@
1 12 a2 2
1 12 a2 3
1 13 a3 3
-2 \N \N 2
2 20 b 0
2 20 b 1
-2 21 b1 3
+2 21 b1 0
3 30 c 1
4 40 d 0
5 50 e 0
@@ -98,10 +98,9 @@
1 12 2
1 12 3
1 13 3
-2 \N 2
2 20 0
2 20 1
-2 21 3
+2 21 0
3 30 1
4 40 0
5 50 0
@@ -116,10 +115,9 @@
1 12 2
1 12 3
1 13 3
-2 \N 2
2 20 0
2 20 1
-2 21 3
+2 21 0
3 30 0
3 30 1
4 40 0
@@ -134,10 +132,9 @@
1 12 2
1 12 3
1 13 3
-2 \N 2
2 20 0
2 20 1
-2 21 3
+2 21 0
3 30 0
3 30 1
4 40 0
@@ -167,10 +164,9 @@
1 12 2
1 12 3
1 13 3
-2 \N 2
2 20 0
2 20 1
-2 21 3
+2 21 0
3 30 0
3 30 1
4 40 0
diff --git a/regression-test/data/row_binlog_p0/test_row_binlog_basic.out
b/regression-test/data/row_binlog_p0/test_row_binlog_basic.out
index b84e49b5040..1a251f75297 100644
--- a/regression-test/data/row_binlog_p0/test_row_binlog_basic.out
+++ b/regression-test/data/row_binlog_p0/test_row_binlog_basic.out
@@ -39,6 +39,16 @@
1 2 2 2 2000 201 2000 200
2 1 1 1 10 10 10 10
+-- !mow_reinsert_before_binlog --
+0 1 10 \N
+2 1 10 10
+0 1 10 \N
+
+-- !mow_reinsert_binlog --
+0 2 20 partial
+2 2 \N default
+0 2 20 default
+
-- !mow_seq_raw --
-- !mow_seq_binlog --
@@ -55,4 +65,5 @@
1 1 1 1 300 350 300 300
1 1 1 1 400 350 300 350
1 1 1 1 400 360 400 350
-2 1 1 1 400 350 400 350
\ No newline at end of file
+2 1 1 1 400 350 400 350
+
diff --git
a/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
b/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
index 80d98d13f70..e443a9012a4 100644
--- a/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
+++ b/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
@@ -255,8 +255,9 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
sql "INSERT INTO ${mowTable} VALUES (6, 60, 'f')"
sql "sync"
- // 2.1 APPEND_ONLY [mowT0, mowT1]: only brand-new keys with their first
- // INSERT. Updates / deletes / resurrections are filtered out.
+ // 2.1 APPEND_ONLY [mowT0, mowT1]: APPEND rows in the window.
+ // Updates / deletes are filtered out. key=2 delete-then-reinsert
+ // is kept because the previous version is a tombstone, not a live
row.
// Expected:
// key=2 first INSERT(20) is APPEND (key did not exist before).
// key=4 INSERT(40) is APPEND.
@@ -290,16 +291,15 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
// 2.3 DETAIL [mowT0, mowT1]: every raw binlog row in the window.
// Each "INSERT into UNIQUE KEY MoW that hits an existing key"
emits
// a UPDATE_BEFORE/UPDATE_AFTER pair instead of a plain INSERT.
After
- // a DELETE the key version still exists in row binlog, so a later
- // INSERT on the same key is also recorded as an UPDATE (BEFORE is
- // populated from the deleted snapshot, hence NULL value columns).
+ // a DELETE, a later INSERT on the same key is recorded as APPEND
+ // because the previous version is a tombstone, not a live row.
// Count breakdown:
// key=1: 3 updates -> 3 * 2 = 6 rows
- // key=2: INSERT(20) + DELETE(20) + UPDATE(NULL -> 21) = 4 rows
+ // key=2: INSERT(20) + DELETE(20) + INSERT(21) = 3 rows
// key=3: DELETE(30) = 1 row
// key=4: INSERT(40) = 1 row
// key=5: INSERT(50) + DELETE(50) = 2 rows
- // Total = 6 + 4 + 1 + 1 + 2 = 14.
+ // Total = 6 + 3 + 1 + 1 + 2 = 13.
order_qt_mow_detail_range """
SELECT id, v1, v2, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr('startTimestamp' = '${mowT0}',
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 d810420346b..219893fd526 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
@@ -23,6 +23,8 @@ suite("test_row_binlog_basic", "nonConcurrent") {
sql "DROP TABLE IF EXISTS test_dup_with_binlog FORCE"
sql "DROP TABLE IF EXISTS test_mow_with_binlog FORCE"
sql "DROP TABLE IF EXISTS test_mow_with_before_binlog FORCE"
+ sql "DROP TABLE IF EXISTS test_mow_reinsert_with_before_binlog FORCE"
+ sql "DROP TABLE IF EXISTS test_mow_reinsert_binlog FORCE"
sql "DROP TABLE IF EXISTS test_mow_seq_with_binlog FORCE"
sql "DROP TABLE IF EXISTS test_empty_rowset FORCE"
@@ -82,6 +84,38 @@ suite("test_row_binlog_basic", "nonConcurrent") {
)
"""
+ sql """
+ CREATE TABLE test_mow_reinsert_with_before_binlog (
+ k1 INT,
+ v1 INT
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql """
+ CREATE TABLE test_mow_reinsert_binlog (
+ k1 INT,
+ v1 INT,
+ v2 STRING DEFAULT "default"
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW"
+ )
+ """
+
sql """
CREATE TABLE test_mow_seq_with_binlog (
k1 INT,
@@ -228,6 +262,34 @@ suite("test_row_binlog_basic", "nonConcurrent") {
ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
+ sql "INSERT INTO test_mow_reinsert_with_before_binlog VALUES (1, 10)"
+ sql "DELETE FROM test_mow_reinsert_with_before_binlog WHERE k1 = 1"
+ sql "INSERT INTO test_mow_reinsert_with_before_binlog VALUES (1, 10)"
+
+ qt_mow_reinsert_before_binlog """
+ SELECT __DORIS_BINLOG_OP__ AS op,
+ k1,
+ v1,
+ __BEFORE__v1__
+ FROM binlog("table" = "test_mow_reinsert_with_before_binlog")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
+ sql "INSERT INTO test_mow_reinsert_binlog VALUES (2, 20, 'partial')"
+ sql "DELETE FROM test_mow_reinsert_binlog WHERE k1 = 2"
+ sql "SET enable_unique_key_partial_update = true"
+ sql "INSERT INTO test_mow_reinsert_binlog(k1, v1) VALUES (2, 20)"
+ sql "SET enable_unique_key_partial_update = false"
+
+ qt_mow_reinsert_binlog """
+ SELECT __DORIS_BINLOG_OP__ AS op,
+ k1,
+ v1,
+ v2
+ FROM binlog("table" = "test_mow_reinsert_binlog")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+
sql """
INSERT INTO test_mow_seq_with_binlog(k1, k2, k3, v1, v2,
__DORIS_SEQUENCE_COL__)
VALUES (1, 1, 1, 100, '100', 2)
diff --git
a/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy
b/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy
index bd85855cd79..425a4895178 100644
--- a/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy
+++ b/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy
@@ -29,6 +29,9 @@ suite("test_min_delta_stream", "nonConcurrent") {
def ukStream = "md_uk_stream"
def ukSkipBase = "md_uk_skip_base"
def ukSkipStream = "md_uk_skip_stream"
+ def ukReinsertBase = "md_uk_reinsert_base"
+ def ukReinsertStream = "md_uk_reinsert_stream"
+ def ukReinsertTarget = "md_uk_reinsert_target"
def ukMultiBase = "md_uk_multi_base"
def ukMultiStream = "md_uk_multi_stream"
def ukDeleteBase = "md_uk_delete_base"
@@ -57,6 +60,9 @@ suite("test_min_delta_stream", "nonConcurrent") {
sql "DROP TABLE IF EXISTS ${ukBase}"
sql "DROP STREAM IF EXISTS ${ukSkipStream}"
sql "DROP TABLE IF EXISTS ${ukSkipBase}"
+ sql "DROP STREAM IF EXISTS ${ukReinsertStream}"
+ sql "DROP TABLE IF EXISTS ${ukReinsertBase}"
+ sql "DROP TABLE IF EXISTS ${ukReinsertTarget}"
sql "DROP STREAM IF EXISTS ${ukMultiStream}"
sql "DROP TABLE IF EXISTS ${ukMultiBase}"
sql "DROP STREAM IF EXISTS ${ukDeleteStream}"
@@ -187,7 +193,59 @@ suite("test_min_delta_stream", "nonConcurrent") {
def ukSkipRows = sql "SELECT id, v1, __DORIS_STREAM_CHANGE_TYPE_COL__
FROM ${ukSkipStream}"
assertEquals(0, ukSkipRows.size())
- // 3) UNIQUE KEY + MIN_DELTA: verify multiple UPDATEs on the same key
keep only first/last as BEFORE/AFTER.
+ // 3) UNIQUE KEY + MIN_DELTA: after INSERT+DELETE is consumed,
reinserting the same key is APPEND.
+ sql """
+ CREATE TABLE ${ukReinsertBase} (
+ id BIGINT,
+ v1 INT
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """
+ CREATE STREAM ${ukReinsertStream}
+ ON TABLE ${ukReinsertBase}
+ PROPERTIES (
+ "type" = "min_delta",
+ "show_initial_rows" = "false"
+ )
+ """
+ sql """
+ CREATE TABLE ${ukReinsertTarget} (
+ id BIGINT,
+ v1 INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql "INSERT INTO ${ukReinsertBase} VALUES (20, 200)"
+ sql "DELETE FROM ${ukReinsertBase} WHERE id = 20"
+ sql "sync"
+ sql "INSERT INTO ${ukReinsertTarget} SELECT id, v1 FROM
${ukReinsertStream}"
+ assertEquals(0, sql("SELECT id, v1 FROM ${ukReinsertStream}").size())
+
+ sql "INSERT INTO ${ukReinsertBase} VALUES (20, 201)"
+ sql "sync"
+
+ def ukReinsertRows = sql """
+ SELECT id, v1, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM ${ukReinsertStream}
+ ORDER BY id, v1, __DORIS_STREAM_CHANGE_TYPE_COL__
+ """
+ assertEquals(1, ukReinsertRows.size())
+ assertEquals("20", ukReinsertRows[0][0].toString())
+ assertEquals("201", ukReinsertRows[0][1].toString())
+ assertEquals("APPEND", ukReinsertRows[0][2].toString())
+
+ // 4) UNIQUE KEY + MIN_DELTA: verify multiple UPDATEs on the same key
keep only first/last as BEFORE/AFTER.
sql """
CREATE TABLE ${ukMultiBase} (
id BIGINT,
@@ -748,7 +806,7 @@ suite("test_min_delta_stream", "nonConcurrent") {
FROM ${incrBase}@incr('startTimestamp' = '${startTimestamp}',
"endTimestamp" = "${endTimestamp}",
"incrementType" = "DETAIL")
- ORDER BY __DORIS_BINLOG_LSN__
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
assertEquals(6, detailWithRangeRows.size())
assertEquals("2", detailWithRangeRows[0][0].toString())
@@ -774,14 +832,14 @@ suite("test_min_delta_stream", "nonConcurrent") {
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${incrBase}@incr('startTimestamp' = '${startTimestamp}',
"incrementType" = "DETAIL")
- ORDER BY __DORIS_BINLOG_LSN__
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
assertEquals(detailWithRangeRows, detailWithStartRows)
def minDeltaDefaultRows = sql """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${incrBase}@incr("incrementType" = "MIN_DELTA")
- ORDER BY __DORIS_BINLOG_LSN__
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
assertEquals(2, minDeltaDefaultRows.size())
assertEquals("2", minDeltaDefaultRows[0][0].toString())
@@ -794,7 +852,7 @@ suite("test_min_delta_stream", "nonConcurrent") {
def emptyIncrRows = sql """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${incrBase}@incr()
- ORDER BY __DORIS_BINLOG_LSN__
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
assertEquals(minDeltaDefaultRows, emptyIncrRows)
@@ -825,7 +883,7 @@ suite("test_min_delta_stream", "nonConcurrent") {
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${incrDupBase}@incr('startTimestamp' = '${dupStartTimestamp}',
"incrementType" = "DETAIL")
- ORDER BY __DORIS_BINLOG_LSN__
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
assertEquals(1, dupDetailRows.size())
assertEquals("3", dupDetailRows[0][0].toString())
@@ -870,7 +928,7 @@ suite("test_min_delta_stream", "nonConcurrent") {
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${incrMowNoHistoryBase}@incr('startTimestamp' =
'${mowNoHistoryStartTimestamp}',
"incrementType" = "DETAIL")
- ORDER BY __DORIS_BINLOG_LSN__
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
assertEquals(1, mowNoHistoryDetailRows.size())
assertEquals("1", mowNoHistoryDetailRows[0][0].toString())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
