Skip to content

Commit e857c5e

Browse files
LchangliangLchangliang
committed
[Schema Change] support fast add/drop column (apache#49)
* [feature](schema-change) support fast schema change. coauthor: yixiutt * [schema change] Using columns desc from fe to read data. coauthor: Lchangliang * [feature](schema change) schema change optimize for add/drop columns. 1.add uniqueId field for class column. 2.schema change for add/drop columns directly update schema meta Co-authored-by: yixiutt <yixiu@selectdb.com> Co-authored-by: SWJTU-ZhangLei <1091517373@qq.com> [Feature](schema change) fix write and add regression test (apache#69) Co-authored-by: yixiutt <yixiu@selectdb.com> [schema change] be ssupport that delete use newest schema add delete regression test fix regression case (apache#107) tmp [feature](schema change) light schema change exclude rollup and agg/uniq/dup key type. [feature](schema change) fe olapTable maxUniqueId write in disk. [feature](schema change) add rpc iface for sc add column. [feature](schema change) add columnsDesc to TPushReq for ligtht sc. resolve the deadlock when schema change (apache#124) fix columns from fe don't has bitmap_index flag (apache#134) add update/delete case construct MATERIALIZED schema from origin schema when insert fix not vectorized compaction coredump use segment cache choose newest schema by schema version when compaction (apache#182) [bugfix](schema change) fix ligth schema change problem. [feature](schema change) light schema change add alter job. (#1) fix be ut [bug] (schema change) unique drop key column should not light schema change [feature](schema change) add schema change regression-test. fix regression test [bugfix](schema change) fix multi alter clauses for light schema change. (#2) [bugfix](schema change) fix multi clauses calculate column unique id (#3) modify PushTask process (apache#217) [Bugfix](schema change) fix jobId replay cause bdbje exception. [bug](schema change) fix max col unique id repeatitive. (apache#232) [optimize](schema change) modify pendingMaxColUniqueId generate rule. fix compaction error * fix be ut * fix snapshot load core fix unique_id error (apache#278) [refact](fe) remove redundant code for light schema change. (#4) [refact](fe) remove redundant code for light schema change. (#4) format fe core format be core fix be ut modify fe meta version fix rebase error flush schema into rowset_meta in old table [refactor](schema change) refact fe light schema change. (#5) delete the change of schemahash and support get max version schema
1 parent 3229730 commit e857c5e

File tree

99 files changed

+6751
-2257
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

99 files changed

+6751
-2257
lines changed

‎be/src/exec/olap_scanner.cpp

+20-8
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,9 @@
2323
#include "exprs/expr_context.h"
2424
#include "gen_cpp/PaloInternalService_types.h"
2525
#include "olap/decimal12.h"
26+
#include "olap/field.h"
2627
#include "olap/storage_engine.h"
28+
#include "olap/tablet_schema.h"
2729
#include "olap/uint24.h"
2830
#include "olap_scan_node.h"
2931
#include "olap_utils.h"
@@ -86,6 +88,14 @@ Status OlapScanner::prepare(
8688
LOG(WARNING) << ss.str();
8789
return Status::InternalError(ss.str());
8890
}
91+
_tablet_schema = _tablet->tablet_schema();
92+
if (!_parent->_olap_scan_node.columns_desc.empty() &&
93+
_parent->_olap_scan_node.columns_desc[0].col_unique_id >= 0) {
94+
_tablet_schema.clear_columns();
95+
for (const auto& column_desc : _parent->_olap_scan_node.columns_desc) {
96+
_tablet_schema.append_column(TabletColumn(column_desc));
97+
}
98+
}
8999
{
90100
std::shared_lock rdlock(_tablet->get_header_lock());
91101
const RowsetSharedPtr rowset = _tablet->rowset_with_max_version();
@@ -168,6 +178,7 @@ Status OlapScanner::_init_tablet_reader_params(
168178
RETURN_IF_ERROR(_init_return_columns(!_tablet_reader_params.direct_mode));
169179

170180
_tablet_reader_params.tablet = _tablet;
181+
_tablet_reader_params.tablet_schema = &_tablet_schema;
171182
_tablet_reader_params.reader_type = READER_QUERY;
172183
_tablet_reader_params.aggregation = _aggregation;
173184
_tablet_reader_params.version = Version(0, _version);
@@ -208,7 +219,7 @@ Status OlapScanner::_init_tablet_reader_params(
208219
_tablet_reader_params.return_columns.push_back(i);
209220
}
210221
for (auto index : _return_columns) {
211-
if (_tablet->tablet_schema().column(index).is_key()) {
222+
if (_tablet_schema.column(index).is_key()) {
212223
continue;
213224
} else {
214225
_tablet_reader_params.return_columns.push_back(index);
@@ -217,13 +228,12 @@ Status OlapScanner::_init_tablet_reader_params(
217228
}
218229

219230
// use _tablet_reader_params.return_columns, because reader use this to merge sort
220-
Status res =
221-
_read_row_cursor.init(_tablet->tablet_schema(), _tablet_reader_params.return_columns);
231+
Status res = _read_row_cursor.init(_tablet_schema, _tablet_reader_params.return_columns);
222232
if (!res.ok()) {
223233
LOG(WARNING) << "fail to init row cursor.res = " << res;
224234
return Status::InternalError("failed to initialize storage read row cursor");
225235
}
226-
_read_row_cursor.allocate_memory_for_string_type(_tablet->tablet_schema());
236+
_read_row_cursor.allocate_memory_for_string_type(_tablet_schema);
227237

228238
// If a agg node is this scan node direct parent
229239
// we will not call agg object finalize method in scan node,
@@ -242,15 +252,17 @@ Status OlapScanner::_init_return_columns(bool need_seq_col) {
242252
if (!slot->is_materialized()) {
243253
continue;
244254
}
245-
int32_t index = _tablet->field_index(slot->col_name());
255+
int32_t index = slot->col_unique_id() >= 0
256+
? _tablet_schema.field_index(slot->col_unique_id())
257+
: _tablet_schema.field_index(slot->col_name());
246258
if (index < 0) {
247259
std::stringstream ss;
248260
ss << "field name is invalid. field=" << slot->col_name();
249261
LOG(WARNING) << ss.str();
250262
return Status::InternalError(ss.str());
251263
}
252264
_return_columns.push_back(index);
253-
if (slot->is_nullable() && !_tablet->tablet_schema().column(index).is_nullable())
265+
if (slot->is_nullable() && !_tablet_schema.column(index).is_nullable())
254266
_tablet_columns_convert_to_null_set.emplace(index);
255267
_query_slots.push_back(slot);
256268
}
@@ -259,13 +271,13 @@ Status OlapScanner::_init_return_columns(bool need_seq_col) {
259271
if (_tablet->tablet_schema().has_sequence_col() && need_seq_col) {
260272
bool has_replace_col = false;
261273
for (auto col : _return_columns) {
262-
if (_tablet->tablet_schema().column(col).aggregation() ==
274+
if (_tablet_schema.column(col).aggregation() ==
263275
FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE) {
264276
has_replace_col = true;
265277
break;
266278
}
267279
}
268-
if (auto sequence_col_idx = _tablet->tablet_schema().sequence_col_idx();
280+
if (auto sequence_col_idx = _tablet_schema.sequence_col_idx();
269281
has_replace_col && std::find(_return_columns.begin(), _return_columns.end(),
270282
sequence_col_idx) == _return_columns.end()) {
271283
_return_columns.push_back(sequence_col_idx);

‎be/src/exec/olap_scanner.h

+2
Original file line numberDiff line numberDiff line change
@@ -145,6 +145,8 @@ class OlapScanner {
145145
MonotonicStopWatch _watcher;
146146

147147
std::shared_ptr<MemTracker> _mem_tracker;
148+
149+
TabletSchema _tablet_schema;
148150
};
149151

150152
} // namespace doris

‎be/src/exec/tablet_info.cpp

+13
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@ void OlapTableIndexSchema::to_protobuf(POlapTableIndexSchema* pindex) const {
3131
for (auto slot : slots) {
3232
pindex->add_columns(slot->col_name());
3333
}
34+
for (auto column : columns) {
35+
column->to_schema_pb(pindex->add_columns_desc());
36+
}
3437
}
3538

3639
Status OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema) {
@@ -57,6 +60,11 @@ Status OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema) {
5760
}
5861
index->slots.emplace_back(it->second);
5962
}
63+
for (auto& pcolumn_desc : p_index.columns_desc()) {
64+
TabletColumn* tc = _obj_pool.add(new TabletColumn());
65+
tc->init_from_pb(pcolumn_desc);
66+
index->columns.emplace_back(tc);
67+
}
6068
_indexes.emplace_back(index);
6169
}
6270

@@ -90,6 +98,11 @@ Status OlapTableSchemaParam::init(const TOlapTableSchemaParam& tschema) {
9098
}
9199
index->slots.emplace_back(it->second);
92100
}
101+
for (auto& tcolumn_desc : t_index.columns_desc) {
102+
TabletColumn* tc = _obj_pool.add(new TabletColumn());
103+
tc->init_from_thrift(tcolumn_desc);
104+
index->columns.emplace_back(tc);
105+
}
93106
_indexes.emplace_back(index);
94107
}
95108

‎be/src/exec/tablet_info.h

+2
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
#include "common/status.h"
2828
#include "gen_cpp/Descriptors_types.h"
2929
#include "gen_cpp/descriptors.pb.h"
30+
#include "olap/tablet_schema.h"
3031
#include "runtime/descriptors.h"
3132
#include "runtime/raw_value.h"
3233
#include "runtime/tuple.h"
@@ -41,6 +42,7 @@ struct OlapTableIndexSchema {
4142
int64_t index_id;
4243
std::vector<SlotDescriptor*> slots;
4344
int32_t schema_hash;
45+
std::vector<TabletColumn*> columns;
4446

4547
void to_protobuf(POlapTableIndexSchema* pindex) const;
4648
};

‎be/src/olap/base_tablet.cpp

+11
Original file line numberDiff line numberDiff line change
@@ -72,4 +72,15 @@ void BaseTablet::_gen_tablet_path() {
7272
}
7373
}
7474

75+
bool BaseTablet::set_tablet_schema_into_rowset_meta() {
76+
bool flag = false;
77+
for (RowsetMetaSharedPtr rowset_meta : _tablet_meta->all_mutable_rs_metas()) {
78+
if (!rowset_meta->get_rowset_pb().has_tablet_schema()) {
79+
rowset_meta->set_tablet_schema(&_schema);
80+
flag = true;
81+
}
82+
}
83+
return flag;
84+
}
85+
7586
} /* namespace doris */

‎be/src/olap/base_tablet.h

+3-1
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,9 @@ class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
6666
}
6767

6868
// properties encapsulated in TabletSchema
69-
const TabletSchema& tablet_schema() const;
69+
virtual const TabletSchema& tablet_schema() const;
70+
71+
bool set_tablet_schema_into_rowset_meta();
7072

7173
protected:
7274
void _gen_tablet_path();

‎be/src/olap/collect_iterator.cpp

+4-4
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,9 @@ Status CollectIterator::add_child(RowsetReaderSharedPtr rs_reader) {
5555
// then merged with the base rowset.
5656
void CollectIterator::build_heap(const std::vector<RowsetReaderSharedPtr>& rs_readers) {
5757
DCHECK(rs_readers.size() == _children.size());
58-
_reverse = _reader->_tablet->tablet_schema().keys_type() == KeysType::UNIQUE_KEYS;
59-
SortType sort_type = _reader->_tablet->tablet_schema().sort_type();
60-
int sort_col_num = _reader->_tablet->tablet_schema().sort_col_num();
58+
_reverse = _reader->_tablet_schema->keys_type() == KeysType::UNIQUE_KEYS;
59+
SortType sort_type = _reader->_tablet_schema->sort_type();
60+
int sort_col_num = _reader->_tablet_schema->sort_col_num();
6161
if (_children.empty()) {
6262
_inner_iter.reset(nullptr);
6363
return;
@@ -200,7 +200,7 @@ CollectIterator::Level0Iterator::Level0Iterator(RowsetReaderSharedPtr rs_reader,
200200
CollectIterator::Level0Iterator::~Level0Iterator() = default;
201201

202202
Status CollectIterator::Level0Iterator::init() {
203-
RETURN_NOT_OK_LOG(_row_cursor.init(_reader->_tablet->tablet_schema(), _reader->_seek_columns),
203+
RETURN_NOT_OK_LOG(_row_cursor.init(*_reader->_tablet_schema, _reader->_seek_columns),
204204
"failed to init row cursor");
205205
return (this->*_refresh_current_row)();
206206
}

‎be/src/olap/compaction.cpp

+18-8
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,10 @@
1717

1818
#include "olap/compaction.h"
1919

20+
#include "common/status.h"
2021
#include "gutil/strings/substitute.h"
22+
#include "olap/rowset/rowset_meta.h"
23+
#include "olap/tablet.h"
2124
#include "util/time.h"
2225
#include "util/trace.h"
2326

@@ -141,8 +144,15 @@ Status Compaction::do_compaction_impl(int64_t permits) {
141144

142145
LOG(INFO) << "start " << merge_type << compaction_name() << ". tablet=" << _tablet->full_name()
143146
<< ", output_version=" << _output_version << ", permits: " << permits;
144-
145-
RETURN_NOT_OK(construct_output_rowset_writer());
147+
// get cur schema if rowset schema exist, rowset schema must be newer than tablet schema
148+
std::vector<RowsetMetaSharedPtr> rowset_metas;
149+
rowset_metas.resize(_input_rowsets.size());
150+
std::transform(_input_rowsets.begin(), _input_rowsets.end(), rowset_metas.begin(),
151+
[](const RowsetSharedPtr& rowset) { return rowset->rowset_meta(); });
152+
auto rowset_meta = Tablet::rowset_meta_with_max_schema_version(rowset_metas);
153+
const TabletSchema* cur_tablet_schema = rowset_meta->tablet_schema();
154+
155+
RETURN_NOT_OK(construct_output_rowset_writer(cur_tablet_schema));
146156
RETURN_NOT_OK(construct_input_rowset_readers());
147157
TRACE("prepare finished");
148158

@@ -152,11 +162,11 @@ Status Compaction::do_compaction_impl(int64_t permits) {
152162
Status res;
153163

154164
if (use_vectorized_compaction) {
155-
res = Merger::vmerge_rowsets(_tablet, compaction_type(), _input_rs_readers,
156-
_output_rs_writer.get(), &stats);
165+
res = Merger::vmerge_rowsets(_tablet, compaction_type(), cur_tablet_schema,
166+
_input_rs_readers, _output_rs_writer.get(), &stats);
157167
} else {
158-
res = Merger::merge_rowsets(_tablet, compaction_type(), _input_rs_readers,
159-
_output_rs_writer.get(), &stats);
168+
res = Merger::merge_rowsets(_tablet, compaction_type(), cur_tablet_schema,
169+
_input_rs_readers, _output_rs_writer.get(), &stats);
160170
}
161171

162172
if (!res.ok()) {
@@ -219,8 +229,8 @@ Status Compaction::do_compaction_impl(int64_t permits) {
219229
return Status::OK();
220230
}
221231

222-
Status Compaction::construct_output_rowset_writer() {
223-
return _tablet->create_rowset_writer(_output_version, VISIBLE, NONOVERLAPPING,
232+
Status Compaction::construct_output_rowset_writer(const TabletSchema* schema) {
233+
return _tablet->create_rowset_writer(_output_version, VISIBLE, NONOVERLAPPING, schema,
224234
_oldest_write_timestamp, _newest_write_timestamp,
225235
&_output_rs_writer);
226236
}

‎be/src/olap/compaction.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ class Compaction {
6767
Status modify_rowsets();
6868
void gc_output_rowset();
6969

70-
Status construct_output_rowset_writer();
70+
Status construct_output_rowset_writer(const TabletSchema* schema);
7171
Status construct_input_rowset_readers();
7272

7373
Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets);

0 commit comments

Comments
 (0)