This is an automated email from the ASF dual-hosted git repository.
seawinde 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 df6a3242797 [fix](binlog) Support hidden key columns in row binlog
(#65076)
df6a3242797 is described below
commit df6a32427974ac04821253da30e062a0021ffd0f
Author: seawinde <[email protected]>
AuthorDate: Wed Jul 15 16:47:00 2026 +0800
[fix](binlog) Support hidden key columns in row binlog (#65076)
### What problem does this PR solve?
Issue Number: #65418
Related PR: #63110
Problem Summary:
Row binlog schema and the BE row-binlog writer previously treated normal
row-binlog columns as visible source columns only. This misses hidden
key columns, while hidden non-key internal columns such as
sequence/delete/version/skip-bitmap columns should still be excluded
from row binlog.
Root cause: In `OlapTable.generateTableRowBinlogSchema()` and
`SchemaChangeHandler.addColumnRowBinlog()`, FE generated and maintained
row-binlog schema from visible columns only. In `RowBinlogSegmentWriter`
/ `RowBinlogSourceDataWriter`, BE also used visible-column counts to map
source columns to row-binlog normal columns.
This PR makes row-binlog normal columns follow a simple source-schema
prefix contract: visible columns plus hidden key columns are written,
and trailing hidden non-key columns are skipped. FE schema generation
and add-column schema-change sync now use this contract. BE row-binlog
writing uses the same normal-column count for full writes, partial
update filtering, key column materialization, and BEFORE value columns.
| File | Change Description |
|------|-------------------|
| `OlapTable.java` | Generate row-binlog schema from full base schema,
include hidden key columns, skip hidden non-key columns, and reject
visible/key columns after hidden non-key columns. |
| `SchemaChangeHandler.java` | Keep hidden key columns when syncing ADD
COLUMN changes to row-binlog schema, while skipping hidden non-key
columns. |
| `row_binlog_segment_writer.*` | Use row-binlog normal column count
instead of visible column count, filter partial-update source cids to
normal columns, and collect all key columns in the normal prefix. |
| FE/BE/regression tests | Cover hidden key schema generation/writing
and hidden non-key exclusion. |
---
.../storage/segment/row_binlog_segment_writer.cpp | 144 ++++++++++++-------
be/src/storage/segment/row_binlog_segment_writer.h | 12 +-
be/test/olap/rowset/group_rowset_writer_test.cpp | 127 ++++++++++++++++-
.../segment/row_binlog_segment_writer_test.cpp | 157 +++++++++++++++++++++
.../apache/doris/alter/SchemaChangeHandler.java | 6 +-
.../java/org/apache/doris/catalog/OlapTable.java | 5 +-
.../doris/alter/SchemaChangeHandlerTest.java | 41 ++++++
.../catalog/OlapTableRowBinlogSchemaTest.java | 54 ++++++-
.../row_binlog_p0/test_row_binlog_basic.groovy | 10 +-
9 files changed, 490 insertions(+), 66 deletions(-)
diff --git a/be/src/storage/segment/row_binlog_segment_writer.cpp
b/be/src/storage/segment/row_binlog_segment_writer.cpp
index c1102b318bb..6a5737a613f 100644
--- a/be/src/storage/segment/row_binlog_segment_writer.cpp
+++ b/be/src/storage/segment/row_binlog_segment_writer.cpp
@@ -17,6 +17,9 @@
#include "storage/segment/row_binlog_segment_writer.h"
+#include <algorithm>
+#include <iterator>
+
#include "cloud/config.h"
#include "common/cast_set.h"
#include "core/block/column_with_type_and_name.h"
@@ -76,7 +79,7 @@ Status RowBinlogSegmentWriter::init() {
_binlog_op_col_id = static_cast<uint32_t>(op_col_id);
_normal_col_start_id = tso_col_id == 0 ? BINLOG_COLNUM : 0;
- uint32_t normal_col_num =
cast_set<uint32_t>(source_schema->num_visible_columns());
+ uint32_t normal_col_num =
cast_set<uint32_t>(_source_data_writer->normal_column_count());
_before_col_start_id = _normal_col_start_id + normal_col_num;
if (!_write_before && _tablet_schema->num_columns() > normal_col_num +
BINLOG_COLNUM) {
@@ -117,6 +120,7 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
if (UNLIKELY(source_schema == nullptr)) {
return Status::InternalError("binlog<row> writer missing
source_tablet_schema");
}
+ const size_t normal_column_count =
_source_data_writer->normal_column_count();
bool is_partial_update = _binlog_opts.source.partial_update_info &&
_binlog_opts.source.partial_update_info->is_partial_update() &&
@@ -125,6 +129,7 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
std::vector<uint32_t> partial_cids =
is_partial_update ?
_binlog_opts.source.partial_update_info->update_cids
: std::vector<uint32_t>();
+ std::vector<uint32_t> row_binlog_partial_cids = partial_cids;
if (is_partial_update) {
if (block->columns() <= source_schema->num_key_columns() ||
block->columns() >= source_schema->num_columns()) {
@@ -135,11 +140,12 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
_tablet_schema->num_columns()));
}
- // binlog don't need invisible column
- auto erase_invisible_col = std::remove_if(
- partial_cids.begin(), partial_cids.end(),
- [&](uint32_t cid) { return cid >=
source_schema->num_visible_columns(); });
- partial_cids.erase(erase_invisible_col, partial_cids.end());
+ // Partial update lists source cids. Row-binlog writes visible columns
and hidden key
+ // columns, so hidden non-key source columns are skipped.
+ auto erase_non_normal_col = std::remove_if(
+ row_binlog_partial_cids.begin(), row_binlog_partial_cids.end(),
+ [&](uint32_t cid) { return
!_source_data_writer->is_normal_column(cid); });
+ row_binlog_partial_cids.erase(erase_non_normal_col,
row_binlog_partial_cids.end());
}
// get delete_sign_column from source block if has
@@ -181,9 +187,9 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
row_pos,
num_rows));
}
- size_t max_normal_col_id = _normal_col_start_id +
source_schema->num_visible_columns();
- RETURN_IF_ERROR(_source_data_writer->fill_normal_columns(_column_writers,
_normal_col_start_id,
-
max_normal_col_id, partial_cids));
+ size_t max_normal_col_id = _normal_col_start_id + normal_column_count;
+ RETURN_IF_ERROR(_source_data_writer->fill_normal_columns(
+ _column_writers, _normal_col_start_id, max_normal_col_id,
row_binlog_partial_cids));
// We read historical rows only when we really need them:
// 1. partial update: build the full AFTER row.
@@ -203,9 +209,14 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
if (is_partial_update) {
std::vector<uint32_t> row_binlog_missing_column_ids;
- _source_data_writer->filter_source_ids(
- _binlog_opts.source.partial_update_info->missing_cids,
- row_binlog_missing_column_ids);
+ row_binlog_missing_column_ids.reserve(
+ _binlog_opts.source.partial_update_info->missing_cids.size());
+ // Missing cids are source cids too. Only normal columns have AFTER
writers.
+ for (uint32_t cid :
_binlog_opts.source.partial_update_info->missing_cids) {
+ if (_source_data_writer->is_normal_column(cid)) {
+ row_binlog_missing_column_ids.emplace_back(cid);
+ }
+ }
// build AFTER block (fill missing columns in full_block)
RETURN_IF_ERROR(_historical_data_writer->build_after_block(&full_block,
row_pos, num_rows));
@@ -215,7 +226,8 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
RETURN_IF_ERROR(after_convertor->set_source_content_with_specifid_columns(
&full_block, row_pos, num_rows,
row_binlog_missing_column_ids));
for (auto cid : row_binlog_missing_column_ids) {
- auto converted_cid = _normal_col_start_id + cid;
+ auto converted_cid =
+ _normal_col_start_id +
_source_data_writer->normal_column_ordinal(cid);
auto converted_result = after_convertor->convert_column_data(cid);
if (!converted_result.first.ok()) {
return converted_result.first;
@@ -230,9 +242,9 @@ Status RowBinlogSegmentWriter::append_block(const Block*
block, size_t row_pos,
DCHECK(!_tablet_schema->has_sequence_col());
// _converted_key_columns must be resized before fill binlog columns
_converted_key_columns.resize(_tablet_schema->num_key_columns());
- for (size_t i = _normal_col_start_id; i <
_tablet_schema->num_key_columns(); i++) {
- _converted_key_columns[i] = _source_data_writer->get_converted_column(
- cast_set<uint32_t>(i - _normal_col_start_id));
+ const auto& source_key_columns = _source_data_writer->source_key_columns();
+ for (size_t i = 0; i < source_key_columns.size(); ++i) {
+ _converted_key_columns[_normal_col_start_id + i] =
source_key_columns[i];
}
std::vector<int64_t> no_operators = std::vector<int64_t> {};
@@ -374,7 +386,14 @@ Status RowBinlogSegmentWriter::_fill_before_columns(size_t
num_rows) {
if (UNLIKELY(source_schema == nullptr)) {
return Status::InternalError("row binlog writer missing
source_tablet_schema");
}
- size_t value_column_num = source_schema->num_visible_value_columns();
+ std::vector<uint32_t> value_cids;
+ for (uint32_t cid = 0; cid < source_schema->num_columns(); ++cid) {
+ const auto& column = source_schema->column(cid);
+ if (column.visible() && !column.is_key()) {
+ value_cids.emplace_back(cid);
+ }
+ }
+ size_t value_column_num = value_cids.size();
if (value_column_num == 0) {
// No BEFORE columns in row binlog schema.
return Status::OK();
@@ -401,13 +420,6 @@ Status RowBinlogSegmentWriter::_fill_before_columns(size_t
num_rows) {
} else {
DCHECK(_historical_data_writer != nullptr);
- std::vector<uint32_t> value_cids;
- uint32_t value_start =
cast_set<uint32_t>(source_schema->num_key_columns());
- uint32_t value_end =
cast_set<uint32_t>(source_schema->num_visible_columns());
- for (uint32_t cid = value_start; cid < value_end; ++cid) {
- value_cids.emplace_back(cid);
- }
-
DCHECK_EQ(before_cids.size(), value_cids.size());
RETURN_IF_ERROR(_historical_data_writer->build_before_block(&before_block,
value_cids, 0,
num_rows));
@@ -429,18 +441,35 @@ Status
RowBinlogSegmentWriter::_fill_before_columns(size_t num_rows) {
return Status::OK();
}
+bool RowBinlogSourceDataWriter::is_normal_column(uint32_t source_cid) const {
+ return std::find(_normal_column_ids.begin(), _normal_column_ids.end(),
source_cid) !=
+ _normal_column_ids.end();
+}
+
+size_t RowBinlogSourceDataWriter::normal_column_ordinal(uint32_t source_cid)
const {
+ auto it = std::find(_normal_column_ids.begin(), _normal_column_ids.end(),
source_cid);
+ DCHECK(it != _normal_column_ids.end()) << source_cid;
+ return cast_set<size_t>(std::distance(_normal_column_ids.begin(), it));
+}
+
Status RowBinlogSourceDataWriter::init() {
_olap_data_convertor = std::make_unique<OlapBlockDataConvertor>();
- // _normal_column_ids: the columns which we need to write into binlog from
source block
if (UNLIKELY(_opt.source.tablet_schema == nullptr)) {
return Status::InternalError("row binlog writer missing
source_tablet_schema");
}
- for (uint32_t i = 0; i < _opt.source.tablet_schema->num_visible_columns();
i++) {
- _normal_column_ids.emplace_back(i);
+
+ const auto& source_schema = _opt.source.tablet_schema;
+ _normal_column_ids.clear();
+ for (uint32_t cid = 0; cid < source_schema->num_columns(); ++cid) {
+ const auto& column = source_schema->column(cid);
+ if (!column.visible() && !column.is_key()) {
+ continue;
+ }
+ _normal_column_ids.emplace_back(cid);
}
- _olap_data_convertor->reserve(_opt.source.tablet_schema->num_columns());
- for (size_t cid = 0; cid < _opt.source.tablet_schema->num_columns();
cid++) {
-
_olap_data_convertor->add_column_data_convertor(_opt.source.tablet_schema->column(cid));
+ _olap_data_convertor->reserve(source_schema->num_columns());
+ for (size_t cid = 0; cid < source_schema->num_columns(); cid++) {
+
_olap_data_convertor->add_column_data_convertor(source_schema->column(cid));
}
return Status::OK();
}
@@ -448,31 +477,43 @@ Status RowBinlogSourceDataWriter::init() {
Status RowBinlogSourceDataWriter::prepare_by_source_block(
const Block* block, size_t row_pos, size_t num_rows,
std::vector<uint32_t>& partial_source_cids, Block* full_block) {
- _converted_columns.resize(_normal_column_ids.size());
+ TabletSchemaSPtr tablet_schema = _opt.source.tablet_schema;
+ // OlapBlockDataConvertor and historical lookup are indexed by source cid,
so keep
+ // converted columns sparse over the source schema instead of row-binlog
normal order.
+ _converted_columns.assign(tablet_schema->num_columns(), nullptr);
// LOG(INFO) << block->dump_data(0, num_rows);
// convert column data from engine format to storage layer format
size_t col_pos_in_block = 0;
- TabletSchemaSPtr tablet_schema = _opt.source.tablet_schema;
- const auto& including_cids =
- partial_source_cids.empty() ? _normal_column_ids :
partial_source_cids;
- for (auto& cid : including_cids) {
- const ColumnWithTypeAndName& col =
block->get_by_position(col_pos_in_block++);
+ const bool is_partial_update = !partial_source_cids.empty();
+ const size_t column_count =
+ is_partial_update ? partial_source_cids.size() :
_normal_column_ids.size();
+ for (size_t ordinal = 0; ordinal < column_count; ++ordinal) {
+ const uint32_t source_cid =
+ is_partial_update ? partial_source_cids[ordinal] :
_normal_column_ids[ordinal];
+ DCHECK_LT(source_cid, tablet_schema->num_columns());
+ const ColumnWithTypeAndName& col =
+ block->get_by_position(is_partial_update ? col_pos_in_block++
: source_cid);
RETURN_IF_ERROR(_olap_data_convertor->set_source_content_with_specifid_column(
- col, row_pos, num_rows, cid));
+ col, row_pos, num_rows, source_cid));
// olap data convertor alway start from id = 0
- auto converted_result = _olap_data_convertor->convert_column_data(cid);
+ auto converted_result =
_olap_data_convertor->convert_column_data(source_cid);
if (!converted_result.first.ok()) {
return converted_result.first;
}
- _converted_columns[cid] = converted_result.second;
+ _converted_columns[source_cid] = converted_result.second;
- if (cid < tablet_schema->num_key_columns()) {
- _key_columns.push_back(converted_result.second);
+ full_block->replace_by_position(source_cid, col.column);
+ }
+ for (uint32_t cid : _normal_column_ids) {
+ if (!tablet_schema->column(cid).is_key()) {
+ continue;
}
- full_block->replace_by_position(cid, col.column);
+ DCHECK_LT(cid, _converted_columns.size());
+ DCHECK(_converted_columns[cid] != nullptr);
+ _key_columns.push_back(_converted_columns[cid]);
}
_num_rows = num_rows;
@@ -497,15 +538,20 @@ Status RowBinlogSourceDataWriter::fill_normal_columns(
std::vector<uint32_t>& partial_source_cids) {
DCHECK_EQ(end - start, _normal_column_ids.size());
- const auto& including_cids =
- partial_source_cids.empty() ? _normal_column_ids :
partial_source_cids;
- for (size_t cid : including_cids) {
- DCHECK(column_writers[start + cid]->get_column()->type() ==
+ const bool is_partial_update = !partial_source_cids.empty();
+ const size_t column_count =
+ is_partial_update ? partial_source_cids.size() :
_normal_column_ids.size();
+ for (size_t ordinal = 0; ordinal < column_count; ++ordinal) {
+ uint32_t cid =
+ is_partial_update ? partial_source_cids[ordinal] :
_normal_column_ids[ordinal];
+ size_t normal_ordinal = is_partial_update ? normal_column_ordinal(cid)
: ordinal;
+ DCHECK(column_writers[start + normal_ordinal]->get_column()->type() ==
_opt.source.tablet_schema->columns()[cid]->type())
<< cid;
- RETURN_IF_ERROR(column_writers[start +
cid]->append(_converted_columns[cid]->get_nullmap(),
-
_converted_columns[cid]->get_data(),
- _num_rows));
+ DCHECK(_converted_columns[cid] != nullptr);
+ RETURN_IF_ERROR(column_writers[start + normal_ordinal]->append(
+ _converted_columns[cid]->get_nullmap(),
_converted_columns[cid]->get_data(),
+ _num_rows));
}
return Status::OK();
diff --git a/be/src/storage/segment/row_binlog_segment_writer.h
b/be/src/storage/segment/row_binlog_segment_writer.h
index 3438bff6b7b..69023d8a3a7 100644
--- a/be/src/storage/segment/row_binlog_segment_writer.h
+++ b/be/src/storage/segment/row_binlog_segment_writer.h
@@ -44,8 +44,6 @@ public:
void clear();
- IOlapColumnDataAccessor* get_converted_column(uint32_t cid) { return
_converted_columns[cid]; }
-
bool need_before() const { return _opt.write_before; }
const std::vector<IOlapColumnDataAccessor*>& source_key_columns() const {
return _key_columns; }
@@ -53,11 +51,11 @@ public:
std::unique_ptr<OlapBlockDataConvertor>& olap_data_convertor() { return
_olap_data_convertor; }
- void filter_source_ids(std::vector<uint32_t>& full_cids,
std::vector<uint32_t>& res_cids) {
- res_cids.reserve(full_cids.size());
- std::set_intersection(_normal_column_ids.begin(),
_normal_column_ids.end(),
- full_cids.begin(), full_cids.end(),
std::back_inserter(res_cids));
- }
+ size_t normal_column_count() const { return _normal_column_ids.size(); }
+
+ bool is_normal_column(uint32_t source_cid) const;
+
+ size_t normal_column_ordinal(uint32_t source_cid) const;
private:
const SegmentWriteBinlogOptions& _opt;
diff --git a/be/test/olap/rowset/group_rowset_writer_test.cpp
b/be/test/olap/rowset/group_rowset_writer_test.cpp
index 4ebd2db4ca3..159a77a6480 100644
--- a/be/test/olap/rowset/group_rowset_writer_test.cpp
+++ b/be/test/olap/rowset/group_rowset_writer_test.cpp
@@ -23,6 +23,7 @@
#include <chrono>
#include <memory>
+#include <set>
#include <string>
#include <thread>
#include <vector>
@@ -36,6 +37,9 @@
#include "runtime/runtime_profile.h"
#include "storage/binlog.h"
#include "storage/olap_define.h"
+#include "storage/partial_update_info.h"
+#include "storage/rowset/rowset_reader.h"
+#include "storage/rowset/rowset_reader_context.h"
#include "storage/storage_engine.h"
#include "storage/tablet/tablet.h"
#include "storage/tablet/tablet_manager.h"
@@ -72,10 +76,24 @@ protected:
ExecEnv::GetInstance()->set_storage_engine(std::move(engine));
_request = testutil::create_tablet_request(
- 10010, 270068390, 10001, 1, TKeysType::UNIQUE_KEYS,
- {{"k1", TPrimitiveType::INT, true}, {"v1",
TPrimitiveType::INT, false}});
+ 10010, 270068390, 10001, 2, TKeysType::UNIQUE_KEYS,
+ {{"k1", TPrimitiveType::INT, true},
+ {"__DORIS_TEST_HIDDEN_KEY__", TPrimitiveType::BIGINT, true},
+ {"__DORIS_TEST_HIDDEN_VALUE__", TPrimitiveType::INT, false},
+ {"v1", TPrimitiveType::INT, false},
+ {"v2", TPrimitiveType::BIGINT, false},
+ {std::string(DELETE_SIGN), TPrimitiveType::TINYINT, false}});
+ _request.tablet_schema.columns[1].__set_visible(false);
+ _request.tablet_schema.columns[2].__set_visible(false);
+ _request.tablet_schema.columns[5].__set_visible(false);
+ _request.tablet_schema.columns[2].__set_is_allow_null(true);
+ _request.tablet_schema.columns[4].__set_is_allow_null(true);
_request.__set_enable_unique_key_merge_on_write(true);
testutil::enable_row_binlog(&_request);
+
_request.row_binlog_schema.columns.erase(_request.row_binlog_schema.columns.begin()
+ 5);
+
_request.row_binlog_schema.columns.erase(_request.row_binlog_schema.columns.begin()
+ 2);
+ _request.row_binlog_schema.columns[5].__set_is_allow_null(true);
+ _request.row_binlog_schema.columns[6].__set_is_allow_null(true);
auto profile =
std::make_unique<RuntimeProfile>("GroupRowsetWriterTest");
ASSERT_TRUE(engine_ptr->create_tablet(_request, profile.get()).ok());
_tablet = engine_ptr->tablet_manager()->get_tablet(_request.tablet_id);
@@ -101,13 +119,31 @@ protected:
auto& columns = columns_guard.mutable_columns();
for (int i = 0; i < num_rows; ++i) {
columns[0]->insert(Field::create_field<PrimitiveType::TYPE_INT>(start_key + i));
- columns[1]->insert(
+
columns[1]->insert(Field::create_field<PrimitiveType::TYPE_BIGINT>(
+ static_cast<int64_t>(1000 + start_key + i)));
+
columns[2]->insert(Field::create_field<PrimitiveType::TYPE_INT>(
+ static_cast<int32_t>(10000 + start_key + i)));
+ columns[3]->insert(
Field::create_field<PrimitiveType::TYPE_INT>((start_key + i) * 10));
+
columns[4]->insert(Field::create_field<PrimitiveType::TYPE_BIGINT>(
+ static_cast<int64_t>((start_key + i) * 100)));
+
columns[5]->insert(Field::create_field<PrimitiveType::TYPE_TINYINT>(0));
}
}
return block;
}
+ Block create_partial_update_block() const {
+ Block block = _tablet->tablet_schema()->create_block_by_cids({0, 1, 2,
3});
+ auto columns_guard = block.mutate_columns_scoped();
+ auto& columns = columns_guard.mutable_columns();
+ columns[0]->insert(Field::create_field<PrimitiveType::TYPE_INT>(1));
+
columns[1]->insert(Field::create_field<PrimitiveType::TYPE_BIGINT>(1001));
+ columns[2]->insert(Field::create_field<PrimitiveType::TYPE_INT>(200));
+ columns[3]->insert(Field::create_field<PrimitiveType::TYPE_INT>(20));
+ return block;
+ }
+
Status create_group_rowset_writer(std::unique_ptr<GroupRowsetWriter>*
group_writer,
RowsetId* data_rowset_id, size_t
num_rows) {
RowsetWriterContext data_context;
@@ -225,4 +261,89 @@ TEST_F(GroupRowsetWriterTest, success) {
EXPECT_TRUE(file_exists(second_segment_path));
}
+TEST_F(GroupRowsetWriterTest, partialUpdateSkipsHiddenNonKeyColumns) {
+ ASSERT_EQ(1,
_tablet->row_binlog_tablet_schema()->field_index("__DORIS_TEST_HIDDEN_KEY__"));
+ ASSERT_EQ(-1,
_tablet->row_binlog_tablet_schema()->field_index("__DORIS_TEST_HIDDEN_VALUE__"));
+ ASSERT_EQ(-1,
_tablet->row_binlog_tablet_schema()->field_index(DELETE_SIGN));
+
+ auto partial_update_info = std::make_shared<PartialUpdateInfo>();
+ ASSERT_TRUE(
+ partial_update_info
+ ->init(_tablet->tablet_id(), 1, *_tablet->tablet_schema(),
+ UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS,
+ PartialUpdateNewRowPolicyPB::APPEND,
+ {"k1", "__DORIS_TEST_HIDDEN_KEY__",
"__DORIS_TEST_HIDDEN_VALUE__", "v1"},
+ false, 0, 0, "", "")
+ .ok());
+ EXPECT_EQ((std::vector<uint32_t> {0, 1, 2, 3}),
partial_update_info->update_cids);
+ EXPECT_EQ((std::vector<uint32_t> {4, 5}),
partial_update_info->missing_cids);
+
+ RowsetWriterContext row_binlog_context;
+ row_binlog_context.tablet = _tablet;
+ row_binlog_context.tablet_schema = _tablet->row_binlog_tablet_schema();
+ row_binlog_context.rowset_state = PREPARED;
+ row_binlog_context.segments_overlap = NONOVERLAPPING;
+ row_binlog_context.max_rows_per_segment = 1024;
+ row_binlog_context.write_type = DataWriteType::TYPE_DIRECT;
+ row_binlog_context.partial_update_info = partial_update_info;
+ row_binlog_context.mow_context =
+ std::make_shared<MowContext>(1, 1,
std::make_shared<RowsetIdUnorderedSet>(),
+ std::vector<RowsetSharedPtr> {},
nullptr);
+ row_binlog_context.write_binlog_opt().enable = true;
+ auto& binlog_options =
row_binlog_context.write_binlog_opt().write_binlog_config();
+ binlog_options.source.tablet_schema = _tablet->tablet_schema();
+ binlog_options.source.partial_update_info = partial_update_info;
+ binlog_options.source.mow_context = row_binlog_context.mow_context;
+ binlog_options.source.source_write_type = DataWriteType::TYPE_DIRECT;
+
+ auto lsn_buffer = AutoIncIDBuffer::create_shared(1, 1,
kBinlogLsnAutoIncId);
+ lsn_buffer->append_range_for_test(1000, 1);
+ std::shared_ptr<std::vector<int64_t>> lsn_ids;
+ ASSERT_TRUE(allocate_binlog_lsn(lsn_buffer, 1, &lsn_ids).ok());
+ binlog_options.insert_seg_lsn(0, lsn_ids);
+
+ auto row_binlog_writer_res =
_tablet->create_rowset_writer(row_binlog_context, false);
+ ASSERT_TRUE(row_binlog_writer_res.has_value());
+ auto row_binlog_writer = std::move(row_binlog_writer_res.value());
+
+ Block block = create_partial_update_block();
+ ASSERT_TRUE(row_binlog_writer->flush_single_block(&block).ok());
+
+ RowsetSharedPtr row_binlog_rowset;
+ ASSERT_TRUE(row_binlog_writer->build(row_binlog_rowset).ok());
+ ASSERT_EQ(1, row_binlog_rowset->num_segments());
+
+ const auto& row_binlog_schema = _tablet->row_binlog_tablet_schema();
+ ASSERT_EQ(7, row_binlog_schema->num_columns());
+ std::vector<uint32_t> return_columns {0, 1, 2, 3, 4, 5, 6};
+ RowsetReaderContext reader_context;
+ reader_context.tablet_schema = row_binlog_schema;
+ reader_context.need_ordered_result = false;
+ reader_context.return_columns = &return_columns;
+
+ RowsetReaderSharedPtr rowset_reader;
+ ASSERT_TRUE(row_binlog_rowset->create_reader(&rowset_reader).ok());
+ ASSERT_TRUE(rowset_reader->init(&reader_context).ok());
+
+ Block output_block = row_binlog_schema->create_block();
+ auto status = rowset_reader->next_batch(&output_block);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(1, output_block.rows());
+ ASSERT_EQ(return_columns.size(), output_block.columns());
+
+ EXPECT_EQ(1, (*output_block.get_by_position(0).column)[0].get<TYPE_INT>());
+ EXPECT_EQ(1001,
(*output_block.get_by_position(1).column)[0].get<TYPE_BIGINT>());
+ EXPECT_EQ(20,
(*output_block.get_by_position(2).column)[0].get<TYPE_INT>());
+ EXPECT_TRUE(output_block.get_by_position(3).column->is_null_at(0));
+ EXPECT_EQ(static_cast<int128_t>(1000),
+
(*output_block.get_by_position(4).column)[0].get<TYPE_LARGEINT>());
+ EXPECT_FALSE(output_block.get_by_position(5).column->is_null_at(0));
+ EXPECT_EQ(ROW_BINLOG_APPEND,
(*output_block.get_by_position(5).column)[0].get<TYPE_BIGINT>());
+ EXPECT_TRUE(output_block.get_by_position(6).column->is_null_at(0));
+
+ Block eof_block = row_binlog_schema->create_block();
+ status = rowset_reader->next_batch(&eof_block);
+ EXPECT_TRUE(status.is<ErrorCode::END_OF_FILE>()) << status;
+}
+
} // namespace doris
diff --git a/be/test/storage/segment/row_binlog_segment_writer_test.cpp
b/be/test/storage/segment/row_binlog_segment_writer_test.cpp
new file mode 100644
index 00000000000..0c0c4b9972f
--- /dev/null
+++ b/be/test/storage/segment/row_binlog_segment_writer_test.cpp
@@ -0,0 +1,157 @@
+// 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/row_binlog_segment_writer.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "core/field.h"
+#include "core/value/decimalv2_value.h"
+#include "storage/binlog.h"
+#include "storage/iterator/olap_data_convertor.h"
+#include "storage/tablet/tablet_schema.h"
+
+namespace doris::segment_v2 {
+
+namespace {
+
+TabletColumn create_column(int32_t unique_id, const std::string& name,
FieldType type, bool is_key,
+ bool visible) {
+ TabletColumn column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
type, !is_key);
+ column.set_unique_id(unique_id);
+ column.set_name(name);
+ column.set_is_key(is_key);
+ column.set_length(type == FieldType::OLAP_FIELD_TYPE_LARGEINT ? 16 : 4);
+ column.set_index_length(column.length());
+ column._visible = visible;
+ return column;
+}
+
+TabletSchemaSPtr create_source_schema() {
+ auto schema = std::make_shared<TabletSchema>();
+ schema->append_column(create_column(0, "k1",
FieldType::OLAP_FIELD_TYPE_INT, true, true));
+ schema->append_column(create_column(1, "__DORIS_TEST_HIDDEN_KEY__",
+ FieldType::OLAP_FIELD_TYPE_LARGEINT,
true, false));
+ schema->append_column(create_column(2, "v1",
FieldType::OLAP_FIELD_TYPE_INT, false, true));
+ schema->append_column(create_column(3, "__DORIS_TEST_HIDDEN_VALUE__",
+ FieldType::OLAP_FIELD_TYPE_INT, false,
false));
+ schema->_keys_type = UNIQUE_KEYS;
+ return schema;
+}
+
+Block create_source_block(const TabletSchemaSPtr& schema) {
+ Block block = schema->create_block();
+ auto columns_guard = block.mutate_columns_scoped();
+ auto& columns = columns_guard.mutable_columns();
+ for (int i = 0; i < 2; ++i) {
+ columns[0]->insert(Field::create_field<TYPE_INT>(10 + i));
+
columns[1]->insert(Field::create_field<TYPE_LARGEINT>(static_cast<int128_t>(1000
+ i)));
+ columns[2]->insert(Field::create_field<TYPE_INT>(100 + i));
+ columns[3]->insert(Field::create_field<TYPE_INT>(10000 + i));
+ }
+ return block;
+}
+
+} // namespace
+
+TEST(RowBinlogSegmentWriterTest, collectHiddenKeyInSourceDataWriter) {
+ auto source_schema = create_source_schema();
+ SegmentWriteBinlogOptions options;
+ options.source.tablet_schema = source_schema;
+
+ RowBinlogSourceDataWriter writer(options);
+ ASSERT_TRUE(writer.init().ok());
+ EXPECT_EQ(3, writer.normal_column_count());
+
+ Block block = create_source_block(source_schema);
+ Block full_block = source_schema->create_block();
+ std::vector<uint32_t> partial_source_cids;
+ ASSERT_TRUE(
+ writer.prepare_by_source_block(&block, 0, 2, partial_source_cids,
&full_block).ok());
+
+ const auto& key_columns = writer.source_key_columns();
+ ASSERT_EQ(2, key_columns.size());
+
+ ASSERT_NE(nullptr, key_columns[0]->get_data_at(1));
+ EXPECT_EQ(11, *reinterpret_cast<const
int32_t*>(key_columns[0]->get_data_at(1)));
+
+ ASSERT_NE(nullptr, key_columns[1]->get_data_at(1));
+ EXPECT_EQ(static_cast<int128_t>(1001),
+ *reinterpret_cast<const
int128_t*>(key_columns[1]->get_data_at(1)));
+}
+
+TEST(RowBinlogSegmentWriterTest, skipHiddenNonKeyBeforeVisibleColumn) {
+ auto source_schema = std::make_shared<TabletSchema>();
+ source_schema->append_column(
+ create_column(0, "k1", FieldType::OLAP_FIELD_TYPE_INT, true,
true));
+ source_schema->append_column(create_column(1,
"__DORIS_TEST_HIDDEN_VALUE__",
+ FieldType::OLAP_FIELD_TYPE_INT,
false, false));
+ source_schema->append_column(
+ create_column(2, "v1", FieldType::OLAP_FIELD_TYPE_INT, false,
true));
+ source_schema->_keys_type = UNIQUE_KEYS;
+
+ SegmentWriteBinlogOptions options;
+ options.source.tablet_schema = source_schema;
+
+ RowBinlogSourceDataWriter writer(options);
+ ASSERT_TRUE(writer.init().ok());
+ EXPECT_EQ(2, writer.normal_column_count());
+ EXPECT_TRUE(writer.is_normal_column(0));
+ EXPECT_FALSE(writer.is_normal_column(1));
+ EXPECT_TRUE(writer.is_normal_column(2));
+ EXPECT_EQ(0, writer.normal_column_ordinal(0));
+ EXPECT_EQ(1, writer.normal_column_ordinal(2));
+}
+
+TEST(RowBinlogSegmentWriterTest, skipMultipleHiddenNonKeyColumns) {
+ auto source_schema = std::make_shared<TabletSchema>();
+ source_schema->append_column(
+ create_column(0, "k1", FieldType::OLAP_FIELD_TYPE_INT, true,
true));
+ source_schema->append_column(create_column(1,
"__DORIS_TEST_HIDDEN_VALUE_1__",
+ FieldType::OLAP_FIELD_TYPE_INT,
false, false));
+ source_schema->append_column(
+ create_column(2, "v1", FieldType::OLAP_FIELD_TYPE_INT, false,
true));
+ source_schema->append_column(create_column(3,
"__DORIS_TEST_HIDDEN_VALUE_2__",
+ FieldType::OLAP_FIELD_TYPE_INT,
false, false));
+ source_schema->append_column(
+ create_column(4, "v2", FieldType::OLAP_FIELD_TYPE_INT, false,
true));
+ source_schema->append_column(create_column(5,
"__DORIS_TEST_HIDDEN_VALUE_3__",
+ FieldType::OLAP_FIELD_TYPE_INT,
false, false));
+ source_schema->_keys_type = UNIQUE_KEYS;
+
+ SegmentWriteBinlogOptions options;
+ options.source.tablet_schema = source_schema;
+
+ RowBinlogSourceDataWriter writer(options);
+ ASSERT_TRUE(writer.init().ok());
+ EXPECT_EQ(3, writer.normal_column_count());
+ EXPECT_TRUE(writer.is_normal_column(0));
+ EXPECT_FALSE(writer.is_normal_column(1));
+ EXPECT_TRUE(writer.is_normal_column(2));
+ EXPECT_FALSE(writer.is_normal_column(3));
+ EXPECT_TRUE(writer.is_normal_column(4));
+ EXPECT_FALSE(writer.is_normal_column(5));
+ EXPECT_EQ(0, writer.normal_column_ordinal(0));
+ EXPECT_EQ(1, writer.normal_column_ordinal(2));
+ EXPECT_EQ(2, writer.normal_column_ordinal(4));
+}
+
+} // namespace doris::segment_v2
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
index f818fcecd12..233cead1bdc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
@@ -322,9 +322,9 @@ public class SchemaChangeHandler extends AlterHandler {
private void addColumnRowBinlog(List<Column> rowBinlogSchema, Column
newColumn, ColumnPosition columnPos,
Set<String> newColNameSet, boolean
needHistoricalValue,
IntSupplier columnUniqueIdSupplier) throws
DdlException {
- if (!newColumn.isVisible()) {
- // row binlog schema is generated from visible columns only, so
schema change must not
- // sync hidden system columns such as
sequence/delete/version/skip-bitmap columns.
+ if (!newColumn.isVisible() && !newColumn.isKey()) {
+ // Row-binlog writes visible columns plus hidden key columns. Skip
hidden non-key
+ // system columns such as sequence/delete/version/skip-bitmap
columns.
return;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
index 0174f379c4c..bcd9e76e7c4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
@@ -2433,7 +2433,10 @@ public class OlapTable extends Table implements
MTMVRelatedTableIf, GsonPostProc
boolean needHistoricalValue =
getBinlogConfig().getNeedHistoricalValue();
List<Column> beforeColumns = new ArrayList<>();
- for (Column column : getBaseSchema(false)) {
+ for (Column column : getBaseSchema(true)) {
+ if (!column.isVisible() && !column.isKey()) {
+ continue;
+ }
Preconditions.checkState(!column.getType().isVariantType(),
"binlog<Row> does not support VARIANT column: " +
column.getName());
Preconditions.checkState(!column.isAutoInc(),
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
index 040b43aaf0a..8294a566fda 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
@@ -24,6 +24,7 @@ import org.apache.doris.catalog.Index;
import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.MaterializedIndexMeta;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.info.ColumnPosition;
import org.apache.doris.catalog.info.IndexType;
@@ -37,6 +38,7 @@ import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.utframe.TestWithFeService;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.logging.log4j.LogManager;
@@ -46,8 +48,11 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.IntSupplier;
import java.util.stream.Collectors;
public class SchemaChangeHandlerTest extends TestWithFeService {
@@ -888,6 +893,42 @@ public class SchemaChangeHandlerTest extends
TestWithFeService {
}
+ @Test
+ public void testRowBinlogSchemaChangeKeepsHiddenKeyColumn() throws
Exception {
+ SchemaChangeHandler schemaChangeHandler = new SchemaChangeHandler();
+ List<Column> rowBinlogSchema = Lists.newArrayList();
+ Column key = new Column("k1", PrimitiveType.INT);
+ key.setIsKey(true);
+ rowBinlogSchema.add(Column.generateRowBinlogKeyColumn(key));
+ rowBinlogSchema.add(new Column(Column.BINLOG_LSN_COL,
PrimitiveType.LARGEINT));
+ rowBinlogSchema.add(new Column(Column.BINLOG_OPERATION_COL,
PrimitiveType.BIGINT));
+ rowBinlogSchema.add(new Column(Column.BINLOG_TIMESTAMP_COL,
PrimitiveType.BIGINT));
+
+ Column hiddenKey = new Column("__DORIS_TEST_HIDDEN_KEY__",
PrimitiveType.BIGINT);
+ hiddenKey.setIsKey(true);
+ hiddenKey.setIsVisible(false);
+ AtomicInteger uniqueId = new AtomicInteger(10);
+ IntSupplier uniqueIdSupplier = uniqueId::getAndIncrement;
+ Method addColumnRowBinlog =
SchemaChangeHandler.class.getDeclaredMethod(
+ "addColumnRowBinlog", List.class, Column.class,
ColumnPosition.class,
+ java.util.Set.class, boolean.class, IntSupplier.class);
+ addColumnRowBinlog.setAccessible(true);
+ addColumnRowBinlog.invoke(schemaChangeHandler, rowBinlogSchema,
hiddenKey, null,
+ Sets.newHashSet(hiddenKey.getName()), false, uniqueIdSupplier);
+
+ List<String> columnNames =
rowBinlogSchema.stream().map(Column::getName).collect(Collectors.toList());
+ Assertions.assertEquals(1,
columnNames.indexOf("__DORIS_TEST_HIDDEN_KEY__"));
+ Assertions.assertEquals(2, columnNames.indexOf(Column.BINLOG_LSN_COL));
+
+ Column hiddenValue = new Column("__DORIS_TEST_HIDDEN_VALUE__",
PrimitiveType.INT);
+ hiddenValue.setIsKey(false);
+ hiddenValue.setIsVisible(false);
+ addColumnRowBinlog.invoke(schemaChangeHandler, rowBinlogSchema,
hiddenValue, null,
+ Sets.newHashSet(hiddenValue.getName()), false,
uniqueIdSupplier);
+ columnNames =
rowBinlogSchema.stream().map(Column::getName).collect(Collectors.toList());
+
Assertions.assertFalse(columnNames.contains("__DORIS_TEST_HIDDEN_VALUE__"));
+ }
+
@Test
public void testAggAddOrDropInvertedIndex() throws Exception {
LOG.info("dbName: {}", Env.getCurrentInternalCatalog().getDbNames());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
index 42a3bc8e078..b1fe029106a 100755
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
@@ -30,12 +30,15 @@ import java.util.stream.Collectors;
public class OlapTableRowBinlogSchemaTest {
private static OlapTable newTestTable(BinlogConfig binlogConfig) {
- long baseIndexId = 1L;
Column key = new Column("k1", PrimitiveType.INT);
key.setIsKey(true);
Column value = new Column("v1", Type.INT, false, null, false, "7", "");
value.setIsKey(false);
- List<Column> baseSchema = Lists.newArrayList(key, value);
+ return newTestTable(binlogConfig, Lists.newArrayList(key, value));
+ }
+
+ private static OlapTable newTestTable(BinlogConfig binlogConfig,
List<Column> baseSchema) {
+ long baseIndexId = 1L;
// Construct a minimal olap table for row binlog schema generation.
OlapTable table = new OlapTable(1L, "tbl", baseSchema,
KeysType.PRIMARY_KEYS, null, null);
@@ -113,4 +116,51 @@ public class OlapTableRowBinlogSchemaTest {
Assertions.assertFalse(table.needRowBinlog());
Assertions.assertTrue(table.getBaseIndexMeta().getRowBinlogIndexId()
<= 0);
}
+
+ @Test
+ public void testRowBinlogSchemaIncludesHiddenKeyColumns() {
+ Column key = new Column("k1", PrimitiveType.INT);
+ key.setIsKey(true);
+ Column hiddenKey = new Column("__DORIS_TEST_HIDDEN_KEY__",
PrimitiveType.BIGINT);
+ hiddenKey.setIsKey(true);
+ hiddenKey.setIsVisible(false);
+ Column value = new Column("v1", PrimitiveType.INT);
+ value.setIsKey(false);
+ Column hiddenValue = new Column("__DORIS_TEST_HIDDEN_VALUE__",
PrimitiveType.INT);
+ hiddenValue.setIsKey(false);
+ hiddenValue.setIsVisible(false);
+
+ OlapTable table =
newTestTable(BinlogTestUtils.newTestRowBinlogConfig(true, true),
+ Lists.newArrayList(key, hiddenKey, value, hiddenValue));
+
+ List<String> columnNames =
table.getRowBinlogMeta().getSchema(true).stream().map(Column::getName)
+ .collect(Collectors.toList());
+ Assertions.assertEquals("k1", columnNames.get(0));
+ Assertions.assertEquals("__DORIS_TEST_HIDDEN_KEY__",
columnNames.get(1));
+ Assertions.assertEquals("v1", columnNames.get(2));
+
Assertions.assertFalse(columnNames.contains("__DORIS_TEST_HIDDEN_VALUE__"));
+
Assertions.assertTrue(columnNames.contains(Column.generateBeforeColName("v1")));
+
Assertions.assertFalse(columnNames.contains(Column.generateBeforeColName("__DORIS_TEST_HIDDEN_KEY__")));
+ Assertions.assertEquals(4, columnNames.indexOf(Column.BINLOG_LSN_COL));
+ }
+
+ @Test
+ public void
testRowBinlogSchemaSkipsHiddenNonKeyColumnBeforeVisibleColumn() {
+ Column key = new Column("k1", PrimitiveType.INT);
+ key.setIsKey(true);
+ Column hiddenValue = new Column("__DORIS_TEST_HIDDEN_VALUE__",
PrimitiveType.INT);
+ hiddenValue.setIsKey(false);
+ hiddenValue.setIsVisible(false);
+ Column value = new Column("v1", PrimitiveType.INT);
+ value.setIsKey(false);
+
+ OlapTable table =
newTestTable(BinlogTestUtils.newTestRowBinlogConfig(true, false),
+ Lists.newArrayList(key, hiddenValue, value));
+ List<String> columnNames =
table.getRowBinlogMeta().getSchema(true).stream().map(Column::getName)
+ .collect(Collectors.toList());
+ Assertions.assertEquals("k1", columnNames.get(0));
+ Assertions.assertEquals("v1", columnNames.get(1));
+
Assertions.assertFalse(columnNames.contains("__DORIS_TEST_HIDDEN_VALUE__"));
+ Assertions.assertEquals(2, columnNames.indexOf(Column.BINLOG_LSN_COL));
+ }
}
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 c1e40c33ba8..d810420346b 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
@@ -293,6 +293,14 @@ suite("test_row_binlog_basic", "nonConcurrent") {
FROM binlog("table" = "test_mow_seq_with_binlog")
ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
"""
+
+ test {
+ sql """
+ SELECT __DORIS_SEQUENCE_COL__
+ FROM binlog("table" = "test_mow_seq_with_binlog")
+ """
+ exception "Unknown column"
+ }
sql "SET skip_delete_bitmap = false"
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
