[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
Lchangliang · Jul 10, 2022 · dd711a1 · dd711a1
1 parent 9e9d6a4
commit dd711a1
Show file tree

Hide file tree

Showing 99 changed files with 6,751 additions and 2,257 deletions.
diff --git a/be/src/exec/olap_scanner.cpp b/be/src/exec/olap_scanner.cpp
@@ -23,7 +23,9 @@
 #include "exprs/expr_context.h"
 #include "gen_cpp/PaloInternalService_types.h"
 #include "olap/decimal12.h"
+#include "olap/field.h"
 #include "olap/storage_engine.h"
+#include "olap/tablet_schema.h"
 #include "olap/uint24.h"
 #include "olap_scan_node.h"
 #include "olap_utils.h"
@@ -86,6 +88,14 @@ Status OlapScanner::prepare(
  LOG(WARNING) << ss.str();
  return Status::InternalError(ss.str());
  }
+ _tablet_schema = _tablet->tablet_schema();
+ if (!_parent->_olap_scan_node.columns_desc.empty() &&
+ _parent->_olap_scan_node.columns_desc[0].col_unique_id >= 0) {
+ _tablet_schema.clear_columns();
+ for (const auto& column_desc : _parent->_olap_scan_node.columns_desc) {
+ _tablet_schema.append_column(TabletColumn(column_desc));
+ }
+ }
  {
  std::shared_lock rdlock(_tablet->get_header_lock());
  const RowsetSharedPtr rowset = _tablet->rowset_with_max_version();
@@ -170,6 +180,7 @@ Status OlapScanner::_init_tablet_reader_params(
  RETURN_IF_ERROR(_init_return_columns(!_tablet_reader_params.direct_mode));
 
  _tablet_reader_params.tablet = _tablet;
+ _tablet_reader_params.tablet_schema = &_tablet_schema;
  _tablet_reader_params.reader_type = READER_QUERY;
  _tablet_reader_params.aggregation = _aggregation;
  _tablet_reader_params.version = Version(0, _version);
@@ -210,7 +221,7 @@ Status OlapScanner::_init_tablet_reader_params(
  _tablet_reader_params.return_columns.push_back(i);
  }
  for (auto index : _return_columns) {
- if (_tablet->tablet_schema().column(index).is_key()) {
+ if (_tablet_schema.column(index).is_key()) {
  continue;
  } else {
  _tablet_reader_params.return_columns.push_back(index);
@@ -219,13 +230,12 @@ Status OlapScanner::_init_tablet_reader_params(
  }
 
  // use _tablet_reader_params.return_columns, because reader use this to merge sort
- Status res =
- _read_row_cursor.init(_tablet->tablet_schema(), _tablet_reader_params.return_columns);
+ Status res = _read_row_cursor.init(_tablet_schema, _tablet_reader_params.return_columns);
  if (!res.ok()) {
  LOG(WARNING) << "fail to init row cursor.res = " << res;
  return Status::InternalError("failed to initialize storage read row cursor");
  }
- _read_row_cursor.allocate_memory_for_string_type(_tablet->tablet_schema());
+ _read_row_cursor.allocate_memory_for_string_type(_tablet_schema);
 
  // If a agg node is this scan node direct parent
  // we will not call agg object finalize method in scan node,
@@ -244,15 +254,17 @@ Status OlapScanner::_init_return_columns(bool need_seq_col) {
  if (!slot->is_materialized()) {
  continue;
  }
- int32_t index = _tablet->field_index(slot->col_name());
+ int32_t index = slot->col_unique_id() >= 0
+ ? _tablet_schema.field_index(slot->col_unique_id())
+ : _tablet_schema.field_index(slot->col_name());
  if (index < 0) {
  std::stringstream ss;
  ss << "field name is invalid. field=" << slot->col_name();
  LOG(WARNING) << ss.str();
  return Status::InternalError(ss.str());
  }
  _return_columns.push_back(index);
- if (slot->is_nullable() && !_tablet->tablet_schema().column(index).is_nullable())
+ if (slot->is_nullable() && !_tablet_schema.column(index).is_nullable())
  _tablet_columns_convert_to_null_set.emplace(index);
  _query_slots.push_back(slot);
  }
@@ -261,13 +273,13 @@ Status OlapScanner::_init_return_columns(bool need_seq_col) {
  if (_tablet->tablet_schema().has_sequence_col() && need_seq_col) {
  bool has_replace_col = false;
  for (auto col : _return_columns) {
- if (_tablet->tablet_schema().column(col).aggregation() ==
+ if (_tablet_schema.column(col).aggregation() ==
  FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE) {
  has_replace_col = true;
  break;
  }
  }
- if (auto sequence_col_idx = _tablet->tablet_schema().sequence_col_idx();
+ if (auto sequence_col_idx = _tablet_schema.sequence_col_idx();
  has_replace_col && std::find(_return_columns.begin(), _return_columns.end(),
  sequence_col_idx) == _return_columns.end()) {
  _return_columns.push_back(sequence_col_idx);

diff --git a/be/src/exec/olap_scanner.h b/be/src/exec/olap_scanner.h
@@ -145,6 +145,8 @@ class OlapScanner {
  MonotonicStopWatch _watcher;
 
  std::shared_ptr<MemTracker> _mem_tracker;
+
+ TabletSchema _tablet_schema;
 };
 
 } // namespace doris
diff --git a/be/src/exec/tablet_info.cpp b/be/src/exec/tablet_info.cpp
@@ -31,6 +31,9 @@ void OlapTableIndexSchema::to_protobuf(POlapTableIndexSchema* pindex) const {
  for (auto slot : slots) {
  pindex->add_columns(slot->col_name());
  }
+ for (auto column : columns) {
+ column->to_schema_pb(pindex->add_columns_desc());
+ }
 }
 
 Status OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema) {
@@ -57,6 +60,11 @@ Status OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema) {
  }
  index->slots.emplace_back(it->second);
  }
+ for (auto& pcolumn_desc : p_index.columns_desc()) {
+ TabletColumn* tc = _obj_pool.add(new TabletColumn());
+ tc->init_from_pb(pcolumn_desc);
+ index->columns.emplace_back(tc);
+ }
  _indexes.emplace_back(index);
  }
 
@@ -90,6 +98,11 @@ Status OlapTableSchemaParam::init(const TOlapTableSchemaParam& tschema) {
  }
  index->slots.emplace_back(it->second);
  }
+ for (auto& tcolumn_desc : t_index.columns_desc) {
+ TabletColumn* tc = _obj_pool.add(new TabletColumn());
+ tc->init_from_thrift(tcolumn_desc);
+ index->columns.emplace_back(tc);
+ }
  _indexes.emplace_back(index);
  }
 

diff --git a/be/src/exec/tablet_info.h b/be/src/exec/tablet_info.h
@@ -27,6 +27,7 @@
 #include "common/status.h"
 #include "gen_cpp/Descriptors_types.h"
 #include "gen_cpp/descriptors.pb.h"
+#include "olap/tablet_schema.h"
 #include "runtime/descriptors.h"
 #include "runtime/raw_value.h"
 #include "runtime/tuple.h"
@@ -41,6 +42,7 @@ struct OlapTableIndexSchema {
  int64_t index_id;
  std::vector<SlotDescriptor*> slots;
  int32_t schema_hash;
+ std::vector<TabletColumn*> columns;
 
  void to_protobuf(POlapTableIndexSchema* pindex) const;
 };

diff --git a/be/src/olap/base_tablet.cpp b/be/src/olap/base_tablet.cpp
@@ -72,4 +72,15 @@ void BaseTablet::_gen_tablet_path() {
  }
 }
 
+bool BaseTablet::set_tablet_schema_into_rowset_meta() {
+ bool flag = false;
+ for (RowsetMetaSharedPtr rowset_meta : _tablet_meta->all_mutable_rs_metas()) {
+ if (!rowset_meta->get_rowset_pb().has_tablet_schema()) {
+ rowset_meta->set_tablet_schema(&_schema);
+ flag = true;
+ }
+ }
+ return flag;
+}
+
 } /* namespace doris */
diff --git a/be/src/olap/base_tablet.h b/be/src/olap/base_tablet.h
@@ -66,7 +66,9 @@ class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
  }
 
  // properties encapsulated in TabletSchema
- const TabletSchema& tablet_schema() const;
+ virtual const TabletSchema& tablet_schema() const;
+
+ bool set_tablet_schema_into_rowset_meta();
 
 protected:
  void _gen_tablet_path();

diff --git a/be/src/olap/collect_iterator.cpp b/be/src/olap/collect_iterator.cpp
@@ -55,9 +55,9 @@ Status CollectIterator::add_child(RowsetReaderSharedPtr rs_reader) {
 // then merged with the base rowset.
 void CollectIterator::build_heap(const std::vector<RowsetReaderSharedPtr>& rs_readers) {
  DCHECK(rs_readers.size() == _children.size());
- _reverse = _reader->_tablet->tablet_schema().keys_type() == KeysType::UNIQUE_KEYS;
- SortType sort_type = _reader->_tablet->tablet_schema().sort_type();
- int sort_col_num = _reader->_tablet->tablet_schema().sort_col_num();
+ _reverse = _reader->_tablet_schema->keys_type() == KeysType::UNIQUE_KEYS;
+ SortType sort_type = _reader->_tablet_schema->sort_type();
+ int sort_col_num = _reader->_tablet_schema->sort_col_num();
  if (_children.empty()) {
  _inner_iter.reset(nullptr);
  return;
@@ -200,7 +200,7 @@ CollectIterator::Level0Iterator::Level0Iterator(RowsetReaderSharedPtr rs_reader,
 CollectIterator::Level0Iterator::~Level0Iterator() = default;
 
 Status CollectIterator::Level0Iterator::init() {
- RETURN_NOT_OK_LOG(_row_cursor.init(_reader->_tablet->tablet_schema(), _reader->_seek_columns),
+ RETURN_NOT_OK_LOG(_row_cursor.init(*_reader->_tablet_schema, _reader->_seek_columns),
  "failed to init row cursor");
  return (this->*_refresh_current_row)();
 }

diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
@@ -17,7 +17,10 @@
 
 #include "olap/compaction.h"
 
+#include "common/status.h"
 #include "gutil/strings/substitute.h"
+#include "olap/rowset/rowset_meta.h"
+#include "olap/tablet.h"
 #include "util/time.h"
 #include "util/trace.h"
 
@@ -141,8 +144,15 @@ Status Compaction::do_compaction_impl(int64_t permits) {
 
  LOG(INFO) << "start " << merge_type << compaction_name() << ". tablet=" << _tablet->full_name()
  << ", output_version=" << _output_version << ", permits: " << permits;
-
- RETURN_NOT_OK(construct_output_rowset_writer());
+ // get cur schema if rowset schema exist, rowset schema must be newer than tablet schema
+ std::vector<RowsetMetaSharedPtr> rowset_metas;
+ rowset_metas.resize(_input_rowsets.size());
+ std::transform(_input_rowsets.begin(), _input_rowsets.end(), rowset_metas.begin(),
+ [](const RowsetSharedPtr& rowset) { return rowset->rowset_meta(); });
+ auto rowset_meta = Tablet::rowset_meta_with_max_schema_version(rowset_metas);
+ const TabletSchema* cur_tablet_schema = rowset_meta->tablet_schema();
+
+ RETURN_NOT_OK(construct_output_rowset_writer(cur_tablet_schema));
  RETURN_NOT_OK(construct_input_rowset_readers());
  TRACE("prepare finished");
 
@@ -152,11 +162,11 @@ Status Compaction::do_compaction_impl(int64_t permits) {
  Status res;
 
  if (use_vectorized_compaction) {
- res = Merger::vmerge_rowsets(_tablet, compaction_type(), _input_rs_readers,
- _output_rs_writer.get(), &stats);
+ res = Merger::vmerge_rowsets(_tablet, compaction_type(), cur_tablet_schema,
+ _input_rs_readers, _output_rs_writer.get(), &stats);
  } else {
- res = Merger::merge_rowsets(_tablet, compaction_type(), _input_rs_readers,
- _output_rs_writer.get(), &stats);
+ res = Merger::merge_rowsets(_tablet, compaction_type(), cur_tablet_schema,
+ _input_rs_readers, _output_rs_writer.get(), &stats);
  }
 
  if (!res.ok()) {
@@ -219,8 +229,8 @@ Status Compaction::do_compaction_impl(int64_t permits) {
  return Status::OK();
 }
 
-Status Compaction::construct_output_rowset_writer() {
- return _tablet->create_rowset_writer(_output_version, VISIBLE, NONOVERLAPPING,
+Status Compaction::construct_output_rowset_writer(const TabletSchema* schema) {
+ return _tablet->create_rowset_writer(_output_version, VISIBLE, NONOVERLAPPING, schema,
  _oldest_write_timestamp, _newest_write_timestamp,
  &_output_rs_writer);
 }

diff --git a/be/src/olap/compaction.h b/be/src/olap/compaction.h
@@ -67,7 +67,7 @@ class Compaction {
  Status modify_rowsets();
  void gc_output_rowset();
 
- Status construct_output_rowset_writer();
+ Status construct_output_rowset_writer(const TabletSchema* schema);
  Status construct_input_rowset_readers();
 
  Status check_version_continuity(const std::vector<RowsetSharedPtr>& rowsets);