You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2021/08/07 13:32:36 UTC
[incubator-doris] branch master updated: [Alter] Support doing
compaction for tablets under alter operation (#6365)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new c6aa37f [Alter] Support doing compaction for tablets under alter operation (#6365)
c6aa37f is described below
commit c6aa37f5efc931f740d9a56b2e2f4ab96480ad06
Author: Mingyu Chen <mo...@gmail.com>
AuthorDate: Sat Aug 7 21:32:26 2021 +0800
[Alter] Support doing compaction for tablets under alter operation (#6365)
The problem I want to solve is described in #6355.
This CL mainly changes:
1. Support compacting tablets under alter operations
On BE side, the compaction logic will select tablets which state is "TABLET_NOTREADY" to do cumulative compaction.
2. Remove "alter_task" field in tablet's meta on BE side.
"alter_task" field is never used long time ago
3. Support doing delete operation when table is doing alter operation.
Previously, when a table is doing alter operation, execution of delete will return error: Table's state is not NORMAL.
But now, delete can be executed successfully only if the condition column is not under schema change.
And delete condition will be applied to all materialized indexes.
---
be/src/olap/cumulative_compaction_policy.cpp | 147 ++++++++++++-------
be/src/olap/cumulative_compaction_policy.h | 6 +-
be/src/olap/delta_writer.cpp | 26 ----
be/src/olap/olap_snapshot_converter.cpp | 14 --
be/src/olap/olap_snapshot_converter.h | 5 -
be/src/olap/push_handler.cpp | 54 -------
be/src/olap/schema_change.cpp | 57 --------
be/src/olap/schema_change.h | 8 --
be/src/olap/snapshot_manager.cpp | 3 -
be/src/olap/tablet.cpp | 67 ++++-----
be/src/olap/tablet.h | 10 +-
be/src/olap/tablet_manager.cpp | 160 +++++----------------
be/src/olap/tablet_manager.h | 3 +-
be/src/olap/tablet_meta.cpp | 86 -----------
be/src/olap/tablet_meta.h | 39 -----
be/src/olap/task/engine_storage_migration_task.cpp | 8 --
.../java/org/apache/doris/analysis/InsertStmt.java | 12 +-
.../main/java/org/apache/doris/catalog/Column.java | 17 ++-
.../org/apache/doris/catalog/PrimitiveType.java | 4 -
.../java/org/apache/doris/load/DeleteHandler.java | 44 +++---
20 files changed, 207 insertions(+), 563 deletions(-)
diff --git a/be/src/olap/cumulative_compaction_policy.cpp b/be/src/olap/cumulative_compaction_policy.cpp
index 529a132..d1a712c 100644
--- a/be/src/olap/cumulative_compaction_policy.cpp
+++ b/be/src/olap/cumulative_compaction_policy.cpp
@@ -68,40 +68,53 @@ void SizeBasedCumulativeCompactionPolicy::calculate_cumulative_point(
// calculate promotion size
auto base_rowset_meta = existing_rss.begin();
- // check base rowset first version must be zero
- CHECK((*base_rowset_meta)->start_version() == 0);
- int64_t promotion_size = 0;
- _calc_promotion_size(*base_rowset_meta, &promotion_size);
+ if (tablet->tablet_state() == TABLET_RUNNING) {
+ // check base rowset first version must be zero
+ // for tablet which state is not TABLET_RUNNING, there may not have base version.
+ CHECK((*base_rowset_meta)->start_version() == 0);
- int64_t prev_version = -1;
- for (const RowsetMetaSharedPtr& rs : existing_rss) {
- if (rs->version().first > prev_version + 1) {
- // There is a hole, do not continue
- break;
- }
+ int64_t promotion_size = 0;
+ _calc_promotion_size(*base_rowset_meta, &promotion_size);
- bool is_delete = tablet->version_for_delete_predicate(rs->version());
+ int64_t prev_version = -1;
+ for (const RowsetMetaSharedPtr& rs : existing_rss) {
+ if (rs->version().first > prev_version + 1) {
+ // There is a hole, do not continue
+ break;
+ }
- // break the loop if segments in this rowset is overlapping.
- if (!is_delete && rs->is_segments_overlapping()) {
- *ret_cumulative_point = rs->version().first;
- break;
- }
+ bool is_delete = tablet->version_for_delete_predicate(rs->version());
- // check the rowset is whether less than promotion size
- if (!is_delete && rs->version().first != 0 && rs->total_disk_size() < promotion_size) {
- *ret_cumulative_point = rs->version().first;
- break;
- }
+ // break the loop if segments in this rowset is overlapping.
+ if (!is_delete && rs->is_segments_overlapping()) {
+ *ret_cumulative_point = rs->version().first;
+ break;
+ }
- // include one situation: When the segment is not deleted, and is singleton delta, and is NONOVERLAPPING, ret_cumulative_point increase
- prev_version = rs->version().second;
- *ret_cumulative_point = prev_version + 1;
- }
- VLOG_NOTICE << "cumulative compaction size_based policy, calculate cumulative point value = "
+ // check the rowset is whether less than promotion size
+ if (!is_delete && rs->version().first != 0 && rs->total_disk_size() < promotion_size) {
+ *ret_cumulative_point = rs->version().first;
+ break;
+ }
+
+ // include one situation: When the segment is not deleted, and is singleton delta, and is NONOVERLAPPING, ret_cumulative_point increase
+ prev_version = rs->version().second;
+ *ret_cumulative_point = prev_version + 1;
+ }
+ VLOG_NOTICE << "cumulative compaction size_based policy, calculate cumulative point value = "
<< *ret_cumulative_point << ", calc promotion size value = " << promotion_size
<< " tablet = " << tablet->full_name();
+ } else if (tablet->tablet_state() == TABLET_NOTREADY) {
+ // tablet under alter process
+ // we choose version next to the base version as cumulative point
+ for (const RowsetMetaSharedPtr& rs : existing_rss) {
+ if (rs->version().first > 0) {
+ *ret_cumulative_point = rs->version().first;
+ break;
+ }
+ }
+ }
}
void SizeBasedCumulativeCompactionPolicy::_calc_promotion_size(RowsetMetaSharedPtr base_rowset_meta,
@@ -126,11 +139,15 @@ void SizeBasedCumulativeCompactionPolicy::_refresh_tablet_size_based_promotion_s
void SizeBasedCumulativeCompactionPolicy::update_cumulative_point(
Tablet* tablet, const std::vector<RowsetSharedPtr>& input_rowsets,
RowsetSharedPtr output_rowset, Version& last_delete_version) {
+ if (tablet->tablet_state() != TABLET_RUNNING) {
+ // if tablet under alter process, do not update cumulative point
+ return;
+ }
// if rowsets have delete version, move to the last directly
if (last_delete_version.first != -1) {
tablet->set_cumulative_layer_point(output_rowset->end_version() + 1);
} else {
- // if rowsets have not delete version, check output_rowset total disk size
+ // if rowsets have no delete version, check output_rowset total disk size
// satisfies promotion size.
size_t total_size = output_rowset->rowset_meta()->total_disk_size();
if (total_size >= _tablet_size_based_promotion_size) {
@@ -139,7 +156,7 @@ void SizeBasedCumulativeCompactionPolicy::update_cumulative_point(
}
}
-void SizeBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(
+void SizeBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(TabletState state,
const std::vector<RowsetMetaSharedPtr>& all_metas, int64_t current_cumulative_point,
uint32_t* score) {
bool base_rowset_exist = false;
@@ -151,12 +168,18 @@ void SizeBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(
// check the base rowset and collect the rowsets of cumulative part
auto rs_meta_iter = all_metas.begin();
+ RowsetMetaSharedPtr first_meta;
+ int64_t first_version = INT64_MAX;
for (; rs_meta_iter != all_metas.end(); rs_meta_iter++) {
auto rs_meta = *rs_meta_iter;
+ if (rs_meta->start_version() < first_version) {
+ first_version = rs_meta->start_version();
+ first_meta = rs_meta;
+ }
// check base rowset
if (rs_meta->start_version() == 0) {
base_rowset_exist = true;
- _calc_promotion_size(rs_meta, &promotion_size);
+ // _calc_promotion_size(rs_meta, &promotion_size);
}
if (rs_meta->end_version() < point) {
// all_rs_metas() is not sorted, so we use _continue_ other than _break_ here.
@@ -169,9 +192,19 @@ void SizeBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(
}
}
- // If base version does not exist, it may be that tablet is doing alter table.
- // Do not select it and set *score = 0
- if (!base_rowset_exist) {
+ if (first_meta == nullptr) {
+ *score = 0;
+ return;
+ }
+
+ // Use "first"(not base) version to calc promotion size
+ // because some tablet do not have base version(under alter operation)
+ _calc_promotion_size(first_meta, &promotion_size);
+
+ // If base version does not exist, but its state is RUNNING.
+ // It is abnormal, do not select it and set *score = 0
+ if (!base_rowset_exist && state == TABLET_RUNNING) {
+ LOG(WARNING) << "tablet state is running but have no base version";
*score = 0;
return;
}
@@ -185,6 +218,9 @@ void SizeBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(
std::sort(rowset_to_compact.begin(), rowset_to_compact.end(), RowsetMeta::comparator);
// calculate the rowsets to do cumulative compaction
+ // eg: size of rowset_to_compact are:
+ // 128, 16, 16, 16
+ // we will choose [16,16,16] to compact.
for (auto& rs_meta : rowset_to_compact) {
int current_level = _level_size(rs_meta->total_disk_size());
int remain_level = _level_size(total_size - rs_meta->total_disk_size());
@@ -360,27 +396,17 @@ int NumBasedCumulativeCompactionPolicy::pick_input_rowsets(
return transient_size;
}
-void NumBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(
+void NumBasedCumulativeCompactionPolicy::calc_cumulative_compaction_score(TabletState state,
const std::vector<RowsetMetaSharedPtr>& all_rowsets, const int64_t current_cumulative_point,
uint32_t* score) {
- bool base_rowset_exist = false;
const int64_t point = current_cumulative_point;
for (auto& rs_meta : all_rowsets) {
- if (rs_meta->start_version() == 0) {
- base_rowset_exist = true;
- }
if (rs_meta->start_version() < point) {
// all_rs_metas() is not sorted, so we use _continue_ other than _break_ here.
continue;
}
*score += rs_meta->get_compaction_score();
}
-
- // If base version does not exist, it may be that tablet is doing alter table.
- // Do not select it and set *score = 0
- if (!base_rowset_exist) {
- *score = 0;
- }
}
void NumBasedCumulativeCompactionPolicy::calculate_cumulative_point(
@@ -404,20 +430,31 @@ void NumBasedCumulativeCompactionPolicy::calculate_cumulative_point(
return a->version().first < b->version().first;
});
- int64_t prev_version = -1;
- for (const RowsetMetaSharedPtr& rs : existing_rss) {
- if (rs->version().first > prev_version + 1) {
- // There is a hole, do not continue
- break;
+ if (tablet->tablet_state() == TABLET_RUNNING) {
+ int64_t prev_version = -1;
+ for (const RowsetMetaSharedPtr& rs : existing_rss) {
+ if (rs->version().first > prev_version + 1) {
+ // There is a hole, do not continue
+ break;
+ }
+ // break the loop if segments in this rowset is overlapping, or is a singleton.
+ if (rs->is_segments_overlapping() || rs->is_singleton_delta()) {
+ *ret_cumulative_point = rs->version().first;
+ break;
+ }
+
+ prev_version = rs->version().second;
+ *ret_cumulative_point = prev_version + 1;
}
- // break the loop if segments in this rowset is overlapping, or is a singleton.
- if (rs->is_segments_overlapping() || rs->is_singleton_delta()) {
- *ret_cumulative_point = rs->version().first;
- break;
+ } else if (tablet->tablet_state() == TABLET_NOTREADY) {
+ // tablet under alter process
+ // we choose version next to the base version as cumulative point
+ for (const RowsetMetaSharedPtr& rs : existing_rss) {
+ if (rs->version().first > 0) {
+ *ret_cumulative_point = rs->version().first;
+ break;
+ }
}
-
- prev_version = rs->version().second;
- *ret_cumulative_point = prev_version + 1;
}
}
diff --git a/be/src/olap/cumulative_compaction_policy.h b/be/src/olap/cumulative_compaction_policy.h
index 038b136..48c0233 100644
--- a/be/src/olap/cumulative_compaction_policy.h
+++ b/be/src/olap/cumulative_compaction_policy.h
@@ -64,7 +64,7 @@ public:
/// param all_rowsets, all rowsets in tablet.
/// param current_cumulative_point, current cumulative point value.
/// return score, the result score after calculate.
- virtual void calc_cumulative_compaction_score(
+ virtual void calc_cumulative_compaction_score(TabletState state,
const std::vector<RowsetMetaSharedPtr>& all_rowsets, int64_t current_cumulative_point,
uint32_t* score) = 0;
@@ -156,7 +156,7 @@ public:
/// Num based cumulative compaction policy implements calc cumulative compaction score function.
/// Its main policy is calculating the accumulative compaction score after current cumulative_point in tablet.
- void calc_cumulative_compaction_score(const std::vector<RowsetMetaSharedPtr>& all_rowsets,
+ void calc_cumulative_compaction_score(TabletState state, const std::vector<RowsetMetaSharedPtr>& all_rowsets,
int64_t current_cumulative_point,
uint32_t* score) override;
@@ -210,7 +210,7 @@ public:
/// Num based cumulative compaction policy implements calc cumulative compaction score function.
/// Its main policy is calculating the accumulative compaction score after current cumulative_point in tablet.
- void calc_cumulative_compaction_score(const std::vector<RowsetMetaSharedPtr>& all_rowsets,
+ void calc_cumulative_compaction_score(TabletState state, const std::vector<RowsetMetaSharedPtr>& all_rowsets,
int64_t current_cumulative_point,
uint32_t* score) override;
diff --git a/be/src/olap/delta_writer.cpp b/be/src/olap/delta_writer.cpp
index e51f4f8..defd345 100644
--- a/be/src/olap/delta_writer.cpp
+++ b/be/src/olap/delta_writer.cpp
@@ -119,32 +119,6 @@ OLAPStatus DeltaWriter::init() {
MutexLock push_lock(_tablet->get_push_lock());
RETURN_NOT_OK(_storage_engine->txn_manager()->prepare_txn(_req.partition_id, _tablet,
_req.txn_id, _req.load_id));
- if (_req.need_gen_rollup) {
- AlterTabletTaskSharedPtr alter_task = _tablet->alter_task();
- if (alter_task != nullptr && alter_task->alter_state() != ALTER_FAILED) {
- TTabletId new_tablet_id = alter_task->related_tablet_id();
- TSchemaHash new_schema_hash = alter_task->related_schema_hash();
- LOG(INFO) << "load with schema change. "
- << "old_tablet_id=" << _tablet->tablet_id() << ", "
- << ", old_schema_hash=" << _tablet->schema_hash() << ", "
- << ", new_tablet_id=" << new_tablet_id << ", "
- << ", new_schema_hash=" << new_schema_hash << ", "
- << ", transaction_id=" << _req.txn_id;
- _new_tablet = tablet_mgr->get_tablet(new_tablet_id, new_schema_hash);
- if (_new_tablet == nullptr) {
- LOG(WARNING) << "find alter task, but could not find new tablet. "
- << "new_tablet_id=" << new_tablet_id
- << ", new_schema_hash=" << new_schema_hash;
- return OLAP_ERR_TABLE_NOT_FOUND;
- }
- ReadLock new_migration_rlock(_new_tablet->get_migration_lock_ptr(), TRY_LOCK);
- if (!new_migration_rlock.own_lock()) {
- return OLAP_ERR_RWLOCK_ERROR;
- }
- RETURN_NOT_OK(_storage_engine->txn_manager()->prepare_txn(
- _req.partition_id, _new_tablet, _req.txn_id, _req.load_id));
- }
- }
}
RowsetWriterContext writer_context;
diff --git a/be/src/olap/olap_snapshot_converter.cpp b/be/src/olap/olap_snapshot_converter.cpp
index 4f8c2d1..334af43 100644
--- a/be/src/olap/olap_snapshot_converter.cpp
+++ b/be/src/olap/olap_snapshot_converter.cpp
@@ -255,20 +255,6 @@ OLAPStatus OlapSnapshotConverter::to_column_msg(const ColumnPB& column_pb,
return OLAP_SUCCESS;
}
-OLAPStatus OlapSnapshotConverter::to_alter_tablet_pb(
- const SchemaChangeStatusMessage& schema_change_msg, AlterTabletPB* alter_tablet_pb) {
- alter_tablet_pb->set_related_tablet_id(schema_change_msg.related_tablet_id());
- alter_tablet_pb->set_related_schema_hash(schema_change_msg.related_schema_hash());
- alter_tablet_pb->set_alter_type(
- static_cast<AlterTabletType>(schema_change_msg.schema_change_type()));
- if (schema_change_msg.versions_to_changed().size() == 0) {
- alter_tablet_pb->set_alter_state(AlterTabletState::ALTER_FINISHED);
- } else {
- alter_tablet_pb->set_alter_state(AlterTabletState::ALTER_FAILED);
- }
- return OLAP_SUCCESS;
-}
-
OLAPStatus OlapSnapshotConverter::save(const string& file_path,
const OLAPHeaderMessage& olap_header) {
DCHECK(!file_path.empty());
diff --git a/be/src/olap/olap_snapshot_converter.h b/be/src/olap/olap_snapshot_converter.h
index 46bb71e..05a8233 100644
--- a/be/src/olap/olap_snapshot_converter.h
+++ b/be/src/olap/olap_snapshot_converter.h
@@ -54,11 +54,6 @@ public:
OLAPStatus to_column_msg(const ColumnPB& column_pb, ColumnMessage* column_msg);
- // only convert schema change msg to alter tablet pb, not the other side because snapshot does not need
- // schema change status while restart and upgrade need schema change status
- OLAPStatus to_alter_tablet_pb(const SchemaChangeStatusMessage& schema_change_msg,
- AlterTabletPB* alter_tablet_pb);
-
OLAPStatus save(const string& file_path, const OLAPHeaderMessage& olap_header);
private:
diff --git a/be/src/olap/push_handler.cpp b/be/src/olap/push_handler.cpp
index 962acb0..8afb0bf 100644
--- a/be/src/olap/push_handler.cpp
+++ b/be/src/olap/push_handler.cpp
@@ -96,60 +96,6 @@ OLAPStatus PushHandler::_do_streaming_ingestion(TabletSharedPtr tablet, const TP
load_id.set_lo(0);
RETURN_NOT_OK(StorageEngine::instance()->txn_manager()->prepare_txn(
request.partition_id, tablet, request.transaction_id, load_id));
-
- // prepare txn will be always successful
- // if current tablet is under schema change, origin tablet is successful and
- // new tablet is not successful, it maybe a fatal error because new tablet has
- // not load successfully
-
- // only when fe sends schema_change true, should consider to push related
- // tablet
- if (_request.is_schema_changing) {
- VLOG_NOTICE << "push req specify schema changing is true. "
- << "tablet=" << tablet->full_name()
- << ", transaction_id=" << request.transaction_id;
- AlterTabletTaskSharedPtr alter_task = tablet->alter_task();
- if (alter_task != nullptr && alter_task->alter_state() != ALTER_FAILED) {
- TTabletId related_tablet_id = alter_task->related_tablet_id();
- TSchemaHash related_schema_hash = alter_task->related_schema_hash();
- LOG(INFO) << "find schema_change status when realtime push. "
- << "tablet=" << tablet->full_name()
- << ", related_tablet_id=" << related_tablet_id
- << ", related_schema_hash=" << related_schema_hash
- << ", transaction_id=" << request.transaction_id;
- TabletSharedPtr related_tablet =
- StorageEngine::instance()->tablet_manager()->get_tablet(related_tablet_id,
- related_schema_hash);
-
- // if related tablet not exists, only push current tablet
- if (related_tablet == nullptr) {
- LOG(WARNING) << "find alter task but not find related tablet, "
- << "related_tablet_id=" << related_tablet_id
- << ", related_schema_hash=" << related_schema_hash;
- tablet->release_push_lock();
- return OLAP_ERR_TABLE_NOT_FOUND;
- // if current tablet is new tablet, only push current tablet
- } else if (tablet->creation_time() > related_tablet->creation_time()) {
- LOG(INFO) << "current tablet is new, only push current tablet. "
- << "tablet=" << tablet->full_name()
- << " related_tablet=" << related_tablet->full_name();
- } else {
- ReadLock new_migration_rlock(related_tablet->get_migration_lock_ptr(), TRY_LOCK);
- if (!new_migration_rlock.own_lock()) {
- return OLAP_ERR_RWLOCK_ERROR;
- }
- PUniqueId load_id;
- load_id.set_hi(0);
- load_id.set_lo(0);
- RETURN_NOT_OK(StorageEngine::instance()->txn_manager()->prepare_txn(
- request.partition_id, related_tablet, request.transaction_id, load_id));
- // prepare txn will always be successful
- tablet_vars->push_back(TabletVars());
- TabletVars& new_item = tablet_vars->back();
- new_item.tablet = related_tablet;
- }
- }
- }
tablet->release_push_lock();
if (tablet_vars->size() == 1) {
diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp
index 3cc5a37..b5d5382 100644
--- a/be/src/olap/schema_change.cpp
+++ b/be/src/olap/schema_change.cpp
@@ -1774,63 +1774,6 @@ OLAPStatus SchemaChangeHandler::_get_versions_to_be_changed(
return OLAP_SUCCESS;
}
-OLAPStatus SchemaChangeHandler::_add_alter_task(
- AlterTabletType alter_tablet_type, TabletSharedPtr base_tablet, TabletSharedPtr new_tablet,
- const std::vector<Version>& versions_to_be_changed) {
- // check new tablet exists,
- // prevent to set base's status after new's dropping (clear base's status)
- if (StorageEngine::instance()->tablet_manager()->get_tablet(
- new_tablet->tablet_id(), new_tablet->schema_hash()) == nullptr) {
- LOG(WARNING) << "new_tablet does not exist. tablet=" << new_tablet->full_name();
- return OLAP_ERR_TABLE_NOT_FOUND;
- }
-
- // 1. 在新表和旧表中添加schema change标志
- base_tablet->delete_alter_task();
- base_tablet->add_alter_task(new_tablet->tablet_id(), new_tablet->schema_hash(),
- versions_to_be_changed, alter_tablet_type);
- base_tablet->save_meta();
- new_tablet->add_alter_task(base_tablet->tablet_id(), base_tablet->schema_hash(),
- std::vector<Version>(), // empty versions
- alter_tablet_type);
- new_tablet->save_meta();
- LOG(INFO) << "successfully add alter task to both base and new";
- return OLAP_SUCCESS;
-}
-
-OLAPStatus SchemaChangeHandler::_save_alter_state(AlterTabletState state,
- TabletSharedPtr base_tablet,
- TabletSharedPtr new_tablet) {
- WriteLock base_wlock(base_tablet->get_header_lock_ptr());
- WriteLock new_wlock(new_tablet->get_header_lock_ptr());
- AlterTabletTaskSharedPtr base_alter_task = base_tablet->alter_task();
- if (base_alter_task == nullptr) {
- LOG(INFO) << "could not find alter task info from base tablet " << base_tablet->full_name();
- return OLAP_ERR_ALTER_STATUS_ERR;
- }
- OLAPStatus res = base_tablet->set_alter_state(state);
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "failed to set alter state to " << state
- << " tablet=" << base_tablet->full_name() << " res=" << res;
- return res;
- }
- base_tablet->save_meta();
- AlterTabletTaskSharedPtr new_alter_task = new_tablet->alter_task();
- if (new_alter_task == nullptr) {
- LOG(INFO) << "could not find alter task info from new tablet " << new_tablet->full_name();
- return OLAP_ERR_ALTER_STATUS_ERR;
- }
- res = new_tablet->set_alter_state(state);
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "failed to set alter state to " << state << " tablet "
- << new_tablet->full_name() << " res" << res;
- return res;
- }
- new_tablet->save_meta();
-
- return OLAP_SUCCESS;
-}
-
OLAPStatus SchemaChangeHandler::_convert_historical_rowsets(const SchemaChangeParams& sc_params) {
LOG(INFO) << "begin to convert historical rowsets for new_tablet from base_tablet."
<< " base_tablet=" << sc_params.base_tablet->full_name()
diff --git a/be/src/olap/schema_change.h b/be/src/olap/schema_change.h
index ce5f107..73f784a 100644
--- a/be/src/olap/schema_change.h
+++ b/be/src/olap/schema_change.h
@@ -220,14 +220,6 @@ private:
std::unordered_map<std::string, AlterMaterializedViewParam> materialized_params_map;
};
- // add alter task to base_tablet and new_tablet.
- // add A->(B|C|...) relation chain to all of them.
- OLAPStatus _add_alter_task(AlterTabletType alter_tablet_type, TabletSharedPtr base_tablet,
- TabletSharedPtr new_tablet,
- const std::vector<Version>& versions_to_be_changed);
- OLAPStatus _save_alter_state(AlterTabletState state, TabletSharedPtr base_tablet,
- TabletSharedPtr new_tablet);
-
OLAPStatus _do_process_alter_tablet_v2(const TAlterTabletReqV2& request);
OLAPStatus _validate_alter_result(TabletSharedPtr new_tablet, const TAlterTabletReqV2& request);
diff --git a/be/src/olap/snapshot_manager.cpp b/be/src/olap/snapshot_manager.cpp
index 87ed672..812725f 100644
--- a/be/src/olap/snapshot_manager.cpp
+++ b/be/src/olap/snapshot_manager.cpp
@@ -441,9 +441,6 @@ OLAPStatus SnapshotManager::_create_snapshot_files(const TabletSharedPtr& ref_ta
break;
}
- // clear alter task info in snapshot files
- new_tablet_meta->delete_alter_task();
-
// The inc_rs_metas is deprecated since Doris version 0.13.
// Clear it for safety reason.
// Whether it is incremental or full snapshot, rowset information is stored in rs_meta.
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index e10e6f1..1746c5d 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -690,49 +690,31 @@ bool Tablet::version_for_load_deletion(const Version& version) {
return rowset->delete_flag();
}
-AlterTabletTaskSharedPtr Tablet::alter_task() {
- return _tablet_meta->alter_task();
-}
-
-void Tablet::add_alter_task(int64_t related_tablet_id, int32_t related_schema_hash,
- const std::vector<Version>& versions_to_alter,
- const AlterTabletType alter_type) {
- AlterTabletTask alter_task;
- alter_task.set_alter_state(ALTER_RUNNING);
- alter_task.set_related_tablet_id(related_tablet_id);
- alter_task.set_related_schema_hash(related_schema_hash);
- alter_task.set_alter_type(alter_type);
- _tablet_meta->add_alter_task(alter_task);
- LOG(INFO) << "successfully add alter task for tablet_id:" << this->tablet_id()
- << ", schema_hash:" << this->schema_hash() << ", related_tablet_id "
- << related_tablet_id << ", related_schema_hash " << related_schema_hash
- << ", alter_type " << alter_type;
-}
-
-void Tablet::delete_alter_task() {
- LOG(INFO) << "delete alter task from table. tablet=" << full_name();
- _tablet_meta->delete_alter_task();
-}
-
-OLAPStatus Tablet::set_alter_state(AlterTabletState state) {
- return _tablet_meta->set_alter_state(state);
-}
-
-bool Tablet::can_do_compaction() {
- // 如果table正在做schema change，则通过选路判断数据是否转换完成
- // 如果选路成功，则转换完成，可以进行compaction
- // 如果选路失败，则转换未完成，不能进行compaction
- ReadLock rdlock(&_meta_lock);
- const RowsetSharedPtr lastest_delta = rowset_with_max_version();
- if (lastest_delta == nullptr) {
+bool Tablet::can_do_compaction(size_t path_hash, CompactionType compaction_type) {
+ if (compaction_type == CompactionType::BASE_COMPACTION && tablet_state() != TABLET_RUNNING) {
+ // base compaction can only be done for tablet in TABLET_RUNNING state.
+ // but cumulative compaction can be done for TABLET_NOTREADY, such as tablet under alter process.
return false;
}
- Version test_version = Version(0, lastest_delta->end_version());
- if (OLAP_SUCCESS != capture_consistent_versions(test_version, nullptr)) {
+ if (data_dir()->path_hash() != path_hash || !is_used() || !init_succeeded()) {
return false;
}
+ if (tablet_state() == TABLET_RUNNING) {
+ // if tablet state is running, we need to check if it has consistent versions.
+ // tablet in other state such as TABLET_NOTREADY may not have complete versions.
+ ReadLock rdlock(&_meta_lock);
+ const RowsetSharedPtr lastest_delta = rowset_with_max_version();
+ if (lastest_delta == nullptr) {
+ return false;
+ }
+
+ Version test_version = Version(0, lastest_delta->end_version());
+ if (OLAP_SUCCESS != capture_consistent_versions(test_version, nullptr)) {
+ return false;
+ }
+ }
return true;
}
@@ -758,7 +740,7 @@ const uint32_t Tablet::_calc_cumulative_compaction_score(
}
#endif
uint32_t score = 0;
- _cumulative_compaction_policy->calc_cumulative_compaction_score(
+ _cumulative_compaction_policy->calc_cumulative_compaction_score(tablet_state(),
_tablet_meta->all_rs_metas(), cumulative_layer_point(), &score);
return score;
}
@@ -867,7 +849,6 @@ void Tablet::_max_continuous_version_from_beginning_unlocked(Version* version,
void Tablet::calculate_cumulative_point() {
WriteLock wrlock(&_meta_lock);
-
int64_t ret_cumulative_point;
_cumulative_compaction_policy->calculate_cumulative_point(
this, _tablet_meta->all_rs_metas(), _cumulative_point, &ret_cumulative_point);
@@ -1045,6 +1026,9 @@ TabletInfo Tablet::get_tablet_info() const {
void Tablet::pick_candidate_rowsets_to_cumulative_compaction(
int64_t skip_window_sec, std::vector<RowsetSharedPtr>* candidate_rowsets) {
+ if (_cumulative_point == K_INVALID_CUMULATIVE_POINT) {
+ return;
+ }
ReadLock rdlock(&_meta_lock);
_cumulative_compaction_policy->pick_candidate_rowsets(skip_window_sec, _rs_version_map,
_cumulative_point, candidate_rowsets);
@@ -1092,7 +1076,10 @@ void Tablet::get_compaction_status(std::string* json_result) {
_timestamped_version_tracker.get_stale_version_path_json_doc(path_arr);
}
rapidjson::Value cumulative_policy_type;
- std::string policy_type_str = _cumulative_compaction_policy->name();
+ std::string policy_type_str = "cumulative compaction policy not initializied";
+ if (_cumulative_compaction_policy.get() != nullptr) {
+ policy_type_str = _cumulative_compaction_policy->name();
+ }
cumulative_policy_type.SetString(policy_type_str.c_str(), policy_type_str.length(),
root.GetAllocator());
root.AddMember("cumulative policy type", cumulative_policy_type, root.GetAllocator());
diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
index 76649a3..437c136 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -134,14 +134,6 @@ public:
bool version_for_delete_predicate(const Version& version);
bool version_for_load_deletion(const Version& version);
- // message for alter task
- AlterTabletTaskSharedPtr alter_task();
- void add_alter_task(int64_t related_tablet_id, int32_t related_schema_hash,
- const std::vector<Version>& versions_to_alter,
- const AlterTabletType alter_type);
- void delete_alter_task();
- OLAPStatus set_alter_state(AlterTabletState state);
-
// meta lock
inline void obtain_header_rdlock() { _meta_lock.rdlock(); }
inline void obtain_header_wrlock() { _meta_lock.wrlock(); }
@@ -166,7 +158,7 @@ public:
inline RWMutex* get_migration_lock_ptr() { return &_migration_lock; }
// operation for compaction
- bool can_do_compaction();
+ bool can_do_compaction(size_t path_hash, CompactionType compaction_type);
uint32_t calc_compaction_score(
CompactionType compaction_type,
std::shared_ptr<CumulativeCompactionPolicy> cumulative_compaction_policy);
diff --git a/be/src/olap/tablet_manager.cpp b/be/src/olap/tablet_manager.cpp
index 8d3486d..a3bd45a 100644
--- a/be/src/olap/tablet_manager.cpp
+++ b/be/src/olap/tablet_manager.cpp
@@ -275,7 +275,7 @@ OLAPStatus TabletManager::create_tablet(const TCreateTabletReq& request,
// set alter type to schema-change. it is useless
TabletSharedPtr tablet = _internal_create_tablet_unlocked(
- AlterTabletType::SCHEMA_CHANGE, request, is_schema_change, base_tablet.get(), stores);
+ request, is_schema_change, base_tablet.get(), stores);
if (tablet == nullptr) {
LOG(WARNING) << "fail to create tablet. tablet_id=" << request.tablet_id;
DorisMetrics::instance()->create_tablet_requests_failed->increment(1);
@@ -289,7 +289,7 @@ OLAPStatus TabletManager::create_tablet(const TCreateTabletReq& request,
}
TabletSharedPtr TabletManager::_internal_create_tablet_unlocked(
- const AlterTabletType alter_type, const TCreateTabletReq& request,
+ const TCreateTabletReq& request,
const bool is_schema_change, const Tablet* base_tablet,
const std::vector<DataDir*>& data_dirs) {
// If in schema-change state, base_tablet must also be provided.
@@ -324,40 +324,33 @@ TabletSharedPtr TabletManager::_internal_create_tablet_unlocked(
LOG(WARNING) << "tablet init failed. tablet:" << tablet->full_name();
break;
}
- // TODO(lingbin): is it needed? because all type of create_tablet will be true.
- // 1. !is_schema_change: not in schema-change state;
- // 2. request.base_tablet_id > 0: in schema-change state;
- if (!is_schema_change || (request.__isset.base_tablet_id && request.base_tablet_id > 0)) {
- // Create init version if this is not a restore mode replica and request.version is set
- // bool in_restore_mode = request.__isset.in_restore_mode && request.in_restore_mode;
- // if (!in_restore_mode && request.__isset.version) {
- // create initial rowset before add it to storage engine could omit many locks
- res = _create_initial_rowset_unlocked(request, tablet.get());
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to create initial version for tablet. res=" << res;
- break;
- }
- TRACE("create initial rowset");
+
+ // Create init version if this is not a restore mode replica and request.version is set
+ // bool in_restore_mode = request.__isset.in_restore_mode && request.in_restore_mode;
+ // if (!in_restore_mode && request.__isset.version) {
+ // create initial rowset before add it to storage engine could omit many locks
+ res = _create_initial_rowset_unlocked(request, tablet.get());
+ if (res != OLAP_SUCCESS) {
+ LOG(WARNING) << "fail to create initial version for tablet. res=" << res;
+ break;
}
+ TRACE("create initial rowset");
+
if (is_schema_change) {
- if (request.__isset.base_tablet_id && request.base_tablet_id > 0) {
- LOG(INFO) << "request for alter-tablet v2, do not add alter task to tablet";
- // if this is a new alter tablet, has to set its state to not ready
- // because schema change handler depends on it to check whether history data
- // convert finished
- tablet->set_tablet_state(TabletState::TABLET_NOTREADY);
- } else {
- // add alter task to new tablet if it is a new tablet during schema change
- tablet->add_alter_task(base_tablet->tablet_id(), base_tablet->schema_hash(),
- std::vector<Version>(), alter_type);
- }
- // 有可能出现以下2种特殊情况：
- // 1. 因为操作系统时间跳变，导致新生成的表的creation_time小于旧表的creation_time时间
- // 2. 因为olap engine代码中统一以秒为单位，所以如果2个操作(比如create一个表,
- // 然后立即alter该表)之间的时间间隔小于1s，则alter得到的新表和旧表的creation_time会相同
- //
- // 当出现以上2种情况时，为了能够区分alter得到的新表和旧表，这里把新表的creation_time设置为
- // 旧表的creation_time加1
+ // if this is a new alter tablet, has to set its state to not ready
+ // because schema change handler depends on it to check whether history data
+ // convert finished
+ tablet->set_tablet_state(TabletState::TABLET_NOTREADY);
+ // The following two special situations may occur:
+ // 1. Because the operating system time jumps, the creation_time of the newly generated table
+ // is less than the creation_time of the old table
+ // 2. Because the unit of second is unified in the olap engine code,
+ // if two operations (such as creating a table, and then immediately altering the table)
+ // is less than 1s, then the creation_time of the new table and the old table obtained by alter will be the same
+ //
+ // When the above two situations occur, in order to be able to distinguish between the new tablet
+ // obtained by alter and the old tablet, the creation_time of the new tablet is set to
+ // the creation_time of the old tablet increased by 1
if (tablet->creation_time() <= base_tablet->creation_time()) {
LOG(WARNING) << "new tablet's create time is less than or equal to old tablet"
<< "new_tablet_create_time=" << tablet->creation_time()
@@ -488,75 +481,7 @@ OLAPStatus TabletManager::_drop_tablet_unlocked(TTabletId tablet_id, SchemaHash
return OLAP_SUCCESS;
}
- // Try to get schema change info, we can drop tablet directly if it is not
- // in schema-change state.
- AlterTabletTaskSharedPtr alter_task = to_drop_tablet->alter_task();
- if (alter_task == nullptr) {
- return _drop_tablet_directly_unlocked(tablet_id, schema_hash, keep_files);
- }
-
- AlterTabletState alter_state = alter_task->alter_state();
- TTabletId related_tablet_id = alter_task->related_tablet_id();
- TSchemaHash related_schema_hash = alter_task->related_schema_hash();
-
- TabletSharedPtr related_tablet = _get_tablet_unlocked(related_tablet_id, related_schema_hash);
- if (related_tablet == nullptr) {
- // TODO(lingbin): in what case, can this happen?
- LOG(WARNING) << "drop tablet directly when related tablet not found. "
- << " tablet_id=" << related_tablet_id
- << " schema_hash=" << related_schema_hash;
- return _drop_tablet_directly_unlocked(tablet_id, schema_hash, keep_files);
- }
-
- // Check whether the tablet we want to delete is in schema-change state
- bool is_schema_change_finished = (alter_state == ALTER_FINISHED || alter_state == ALTER_FAILED);
-
- // Check whether the tablet we want to delete is base-tablet
- bool is_dropping_base_tablet = false;
- if (to_drop_tablet->creation_time() < related_tablet->creation_time()) {
- is_dropping_base_tablet = true;
- }
-
- if (is_dropping_base_tablet && !is_schema_change_finished) {
- LOG(WARNING) << "fail to drop tablet. it is in schema-change state. tablet="
- << to_drop_tablet->full_name();
- return OLAP_ERR_PREVIOUS_SCHEMA_CHANGE_NOT_FINISHED;
- }
-
- // When the code gets here, there are two possibilities:
- // 1. The tablet currently being deleted is a base-tablet, and the corresponding
- // schema-change process has finished;
- // 2. The tablet we are currently trying to drop is not base-tablet(i.e. a tablet
- // generated from its base-tablet due to schema-change). For example, the current
- // request is triggered by cancel alter). In this scenario, the corresponding
- // schema-change task may still in process.
-
- // Drop specified tablet and clear schema-change info
- // NOTE: must first break the hard-link and then drop the tablet.
- // Otherwise, if first drop tablet, then break link. If BE restarts during execution,
- // after BE restarts, the tablet is no longer in metadata, but because the hard-link
- // is still there, the corresponding file may never be deleted from disk.
- related_tablet->obtain_header_wrlock();
- // should check the related tablet_id in alter task is current tablet to be dropped
- // For example: A related to B, BUT B related to C.
- // If drop A, should not clear B's alter task
- OLAPStatus res = OLAP_SUCCESS;
- AlterTabletTaskSharedPtr related_alter_task = related_tablet->alter_task();
- if (related_alter_task != nullptr && related_alter_task->related_tablet_id() == tablet_id &&
- related_alter_task->related_schema_hash() == schema_hash) {
- related_tablet->delete_alter_task();
- related_tablet->save_meta();
- }
- related_tablet->release_header_lock();
- res = _drop_tablet_directly_unlocked(tablet_id, schema_hash, keep_files);
- if (res != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to drop tablet which in schema change. tablet="
- << to_drop_tablet->full_name();
- return res;
- }
-
- LOG(INFO) << "finish to drop tablet. res=" << res;
- return res;
+ return _drop_tablet_directly_unlocked(tablet_id, schema_hash, keep_files);
}
OLAPStatus TabletManager::drop_tablets_on_error_root_path(
@@ -719,31 +644,12 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
ReadLock rlock(tablets_shard.lock.get());
for (const auto& tablet_map : tablets_shard.tablet_map) {
for (const TabletSharedPtr& tablet_ptr : tablet_map.second.table_arr) {
- auto search = tablet_submitted_compaction.find(tablet_ptr->tablet_id());
- if (search != tablet_submitted_compaction.end()) {
+ if (!tablet_ptr->can_do_compaction(data_dir->path_hash(), compaction_type)) {
continue;
}
- AlterTabletTaskSharedPtr cur_alter_task = tablet_ptr->alter_task();
- if (cur_alter_task != nullptr && cur_alter_task->alter_state() != ALTER_FINISHED &&
- cur_alter_task->alter_state() != ALTER_FAILED) {
- TabletSharedPtr related_tablet =
- _get_tablet_unlocked(cur_alter_task->related_tablet_id(),
- cur_alter_task->related_schema_hash());
- if (related_tablet != nullptr &&
- tablet_ptr->creation_time() > related_tablet->creation_time()) {
- // Current tablet is newly created during schema-change or rollup, skip it
- continue;
- }
- }
- // A not-ready tablet maybe a newly created tablet under schema-change, skip it
- if (tablet_ptr->tablet_state() == TABLET_NOTREADY) {
- continue;
- }
-
- if (tablet_ptr->data_dir()->path_hash() != data_dir->path_hash() ||
- !tablet_ptr->is_used() || !tablet_ptr->init_succeeded() ||
- !tablet_ptr->can_do_compaction()) {
+ auto search = tablet_submitted_compaction.find(tablet_ptr->tablet_id());
+ if (search != tablet_submitted_compaction.end()) {
continue;
}
@@ -753,7 +659,7 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
}
if (now_ms - last_failure_ms <=
config::min_compaction_failure_interval_sec * 1000) {
- VLOG_CRITICAL << "Too often to check compaction, skip it. "
+ VLOG_DEBUG << "Too often to check compaction, skip it. "
<< "compaction_type=" << compaction_type_str
<< ", last_failure_time_ms=" << last_failure_ms
<< ", tablet_id=" << tablet_ptr->tablet_id();
@@ -763,11 +669,13 @@ TabletSharedPtr TabletManager::find_best_tablet_to_compaction(
if (compaction_type == CompactionType::BASE_COMPACTION) {
MutexLock lock(tablet_ptr->get_base_lock(), TRY_LOCK);
if (!lock.own_lock()) {
+ LOG(INFO) << "can not get base lock: " << tablet_ptr->tablet_id();
continue;
}
} else {
MutexLock lock(tablet_ptr->get_cumulative_lock(), TRY_LOCK);
if (!lock.own_lock()) {
+ LOG(INFO) << "can not get cumu lock: " << tablet_ptr->tablet_id();
continue;
}
}
diff --git a/be/src/olap/tablet_manager.h b/be/src/olap/tablet_manager.h
index 5cdd2d2..db905c7 100644
--- a/be/src/olap/tablet_manager.h
+++ b/be/src/olap/tablet_manager.h
@@ -169,8 +169,7 @@ private:
TabletSharedPtr _get_tablet_unlocked(TTabletId tablet_id, SchemaHash schema_hash,
bool include_deleted, std::string* err);
- TabletSharedPtr _internal_create_tablet_unlocked(const AlterTabletType alter_type,
- const TCreateTabletReq& request,
+ TabletSharedPtr _internal_create_tablet_unlocked(const TCreateTabletReq& request,
const bool is_schema_change,
const Tablet* base_tablet,
const std::vector<DataDir*>& data_dirs);
diff --git a/be/src/olap/tablet_meta.cpp b/be/src/olap/tablet_meta.cpp
index ca942c5..a2a73a9 100644
--- a/be/src/olap/tablet_meta.cpp
+++ b/be/src/olap/tablet_meta.cpp
@@ -34,30 +34,6 @@ using std::vector;
namespace doris {
-void AlterTabletTask::init_from_pb(const AlterTabletPB& alter_task) {
- _alter_state = alter_task.alter_state();
- _related_tablet_id = alter_task.related_tablet_id();
- _related_schema_hash = alter_task.related_schema_hash();
- _alter_type = alter_task.alter_type();
-}
-
-void AlterTabletTask::to_alter_pb(AlterTabletPB* alter_task) {
- alter_task->set_alter_state(_alter_state);
- alter_task->set_related_tablet_id(_related_tablet_id);
- alter_task->set_related_schema_hash(_related_schema_hash);
- alter_task->set_alter_type(_alter_type);
-}
-
-OLAPStatus AlterTabletTask::set_alter_state(AlterTabletState alter_state) {
- if (_alter_state == ALTER_FAILED && alter_state != ALTER_FAILED) {
- return OLAP_ERR_ALTER_STATUS_ERR;
- } else if (_alter_state == ALTER_FINISHED && alter_state != ALTER_FINISHED) {
- return OLAP_ERR_ALTER_STATUS_ERR;
- }
- _alter_state = alter_state;
- return OLAP_SUCCESS;
-}
-
OLAPStatus TabletMeta::create(const TCreateTabletReq& request, const TabletUid& tablet_uid,
uint64_t shard_id, uint32_t next_unique_id,
const unordered_map<uint32_t, uint32_t>& col_ordinal_to_unique_id,
@@ -391,13 +367,6 @@ void TabletMeta::init_from_pb(const TabletMetaPB& tablet_meta_pb) {
_stale_rs_metas.push_back(std::move(rs_meta));
}
- // generate AlterTabletTask
- if (tablet_meta_pb.has_alter_task()) {
- AlterTabletTask* alter_tablet_task = new AlterTabletTask();
- alter_tablet_task->init_from_pb(tablet_meta_pb.alter_task());
- _alter_task.reset(alter_tablet_task);
- }
-
if (tablet_meta_pb.has_in_restore_mode()) {
_in_restore_mode = tablet_meta_pb.in_restore_mode();
}
@@ -442,9 +411,6 @@ void TabletMeta::to_meta_pb(TabletMetaPB* tablet_meta_pb) {
rs->to_rowset_pb(tablet_meta_pb->add_stale_rs_metas());
}
_schema.to_schema_pb(tablet_meta_pb->mutable_schema());
- if (_alter_task != nullptr) {
- _alter_task->to_alter_pb(tablet_meta_pb->mutable_alter_task());
- }
tablet_meta_pb->set_in_restore_mode(in_restore_mode());
@@ -548,9 +514,6 @@ void TabletMeta::modify_rs_metas(const std::vector<RowsetMetaSharedPtr>& to_add,
// is needed.
void TabletMeta::revise_rs_metas(std::vector<RowsetMetaSharedPtr>&& rs_metas) {
WriteLock wrlock(&_meta_lock);
- // delete alter task
- _alter_task.reset();
-
_rs_metas = std::move(rs_metas);
_stale_rs_metas.clear();
}
@@ -635,42 +598,6 @@ bool TabletMeta::version_for_delete_predicate(const Version& version) {
return false;
}
-// return value not reference
-// MVCC modification for alter task, upper application get a alter task mirror
-AlterTabletTaskSharedPtr TabletMeta::alter_task() {
- ReadLock rlock(&_meta_lock);
- return _alter_task;
-}
-
-void TabletMeta::add_alter_task(const AlterTabletTask& alter_task) {
- WriteLock wrlock(&_meta_lock);
- _alter_task.reset(new AlterTabletTask(alter_task));
-}
-
-void TabletMeta::delete_alter_task() {
- WriteLock wrlock(&_meta_lock);
- _alter_task.reset();
-}
-
-// if alter task is nullptr, return error?
-OLAPStatus TabletMeta::set_alter_state(AlterTabletState alter_state) {
- WriteLock wrlock(&_meta_lock);
- if (_alter_task == nullptr) {
- // alter state should be set to ALTER_PREPARED when starting to
- // alter tablet. In this scenario, _alter_task is null pointer.
- LOG(WARNING) << "original alter task is null, could not set state";
- return OLAP_ERR_ALTER_STATUS_ERR;
- } else {
- auto alter_tablet_task = new AlterTabletTask(*_alter_task);
- OLAPStatus reset_status = alter_tablet_task->set_alter_state(alter_state);
- if (reset_status != OLAP_SUCCESS) {
- return reset_status;
- }
- _alter_task.reset(alter_tablet_task);
- return OLAP_SUCCESS;
- }
-}
-
std::string TabletMeta::full_name() const {
std::stringstream ss;
ss << _tablet_id << "." << _schema_hash << "." << _tablet_uid.to_string();
@@ -686,18 +613,6 @@ OLAPStatus TabletMeta::set_partition_id(int64_t partition_id) {
return OLAP_SUCCESS;
}
-bool operator==(const AlterTabletTask& a, const AlterTabletTask& b) {
- if (a._alter_state != b._alter_state) return false;
- if (a._related_tablet_id != b._related_tablet_id) return false;
- if (a._related_schema_hash != b._related_schema_hash) return false;
- if (a._alter_type != b._alter_type) return false;
- return true;
-}
-
-bool operator!=(const AlterTabletTask& a, const AlterTabletTask& b) {
- return !(a == b);
-}
-
bool operator==(const TabletMeta& a, const TabletMeta& b) {
if (a._table_id != b._table_id) return false;
if (a._partition_id != b._partition_id) return false;
@@ -714,7 +629,6 @@ bool operator==(const TabletMeta& a, const TabletMeta& b) {
for (int i = 0; i < a._rs_metas.size(); ++i) {
if (a._rs_metas[i] != b._rs_metas[i]) return false;
}
- if (a._alter_task != b._alter_task) return false;
if (a._in_restore_mode != b._in_restore_mode) return false;
if (a._preferred_rowset_type != b._preferred_rowset_type) return false;
return true;
diff --git a/be/src/olap/tablet_meta.h b/be/src/olap/tablet_meta.h
index dbd594e..9f3ba46 100644
--- a/be/src/olap/tablet_meta.h
+++ b/be/src/olap/tablet_meta.h
@@ -68,40 +68,6 @@ class DataDir;
class TabletMeta;
using TabletMetaSharedPtr = std::shared_ptr<TabletMeta>;
-class AlterTabletTask {
-public:
- AlterTabletTask() {}
- void init_from_pb(const AlterTabletPB& alter_task);
- void to_alter_pb(AlterTabletPB* alter_task);
-
- inline const AlterTabletState& alter_state() const { return _alter_state; }
- OLAPStatus set_alter_state(AlterTabletState alter_state);
-
- inline int64_t related_tablet_id() const { return _related_tablet_id; }
- inline int32_t related_schema_hash() const { return _related_schema_hash; }
- inline void set_related_tablet_id(int64_t related_tablet_id) {
- _related_tablet_id = related_tablet_id;
- }
- inline void set_related_schema_hash(int32_t schema_hash) { _related_schema_hash = schema_hash; }
-
- inline const AlterTabletType& alter_type() const { return _alter_type; }
- inline void set_alter_type(AlterTabletType alter_type) { _alter_type = alter_type; }
-
- friend bool operator==(const AlterTabletTask& a, const AlterTabletTask& b);
- friend bool operator!=(const AlterTabletTask& a, const AlterTabletTask& b);
-
-private:
- AlterTabletState _alter_state = ALTER_PREPARED;
- int64_t _related_tablet_id = 0;
- int32_t _related_schema_hash = 0;
- AlterTabletType _alter_type = SCHEMA_CHANGE;
-};
-
-bool operator==(const AlterTabletTask& a, const AlterTabletTask& b);
-bool operator!=(const AlterTabletTask& a, const AlterTabletTask& b);
-
-typedef std::shared_ptr<AlterTabletTask> AlterTabletTaskSharedPtr;
-
// Class encapsulates meta of tablet.
// The concurrency control is handled in Tablet Class, not in this class.
class TabletMeta {
@@ -184,10 +150,6 @@ public:
void remove_delete_predicate_by_version(const Version& version);
DelPredicateArray delete_predicates() const;
bool version_for_delete_predicate(const Version& version);
- AlterTabletTaskSharedPtr alter_task();
- void add_alter_task(const AlterTabletTask& alter_task);
- void delete_alter_task();
- OLAPStatus set_alter_state(AlterTabletState alter_state);
std::string full_name() const;
@@ -231,7 +193,6 @@ private:
std::vector<RowsetMetaSharedPtr> _stale_rs_metas;
DelPredicateArray _del_pred_array;
- AlterTabletTaskSharedPtr _alter_task;
bool _in_restore_mode = false;
RowsetTypePB _preferred_rowset_type = ALPHA_ROWSET;
diff --git a/be/src/olap/task/engine_storage_migration_task.cpp b/be/src/olap/task/engine_storage_migration_task.cpp
index 67acb56..7894423 100644
--- a/be/src/olap/task/engine_storage_migration_task.cpp
+++ b/be/src/olap/task/engine_storage_migration_task.cpp
@@ -166,14 +166,6 @@ OLAPStatus EngineStorageMigrationTask::_migrate() {
res = OLAP_ERR_TABLE_NOT_FOUND;
break;
}
- AlterTabletTaskSharedPtr alter_task = _tablet->alter_task();
- if (alter_task != nullptr) {
- if (alter_task->alter_state() == ALTER_FINISHED) {
- new_tablet->set_alter_state(ALTER_FINISHED);
- } else {
- new_tablet->delete_alter_task();
- }
- }
} while (0);
_tablet->release_push_lock();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/InsertStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/InsertStmt.java
index 3a3cd91..7e31600 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/InsertStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/InsertStmt.java
@@ -615,10 +615,14 @@ public class InsertStmt extends DdlStmt {
if (entry == null) {
extentedRow.add(extentedRow.get(entry.first));
} else {
- ExprSubstitutionMap smap = new ExprSubstitutionMap();
- smap.getLhs().add(entry.second.getRefColumn());
- smap.getRhs().add(extentedRow.get(entry.first));
- extentedRow.add(Expr.substituteList(Lists.newArrayList(entry.second.getDefineExpr()), smap, analyzer, false).get(0));
+ if (entry.second == null) {
+ extentedRow.add(extentedRow.get(entry.first));
+ } else {
+ ExprSubstitutionMap smap = new ExprSubstitutionMap();
+ smap.getLhs().add(entry.second.getRefColumn());
+ smap.getRhs().add(extentedRow.get(entry.first));
+ extentedRow.add(Expr.substituteList(Lists.newArrayList(entry.second.getDefineExpr()), smap, analyzer, false).get(0));
+ }
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
index 48d85b6..8bc1b2f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
@@ -31,13 +31,13 @@ import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.thrift.TColumn;
import org.apache.doris.thrift.TColumnType;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.gson.annotations.SerializedName;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
@@ -437,6 +437,17 @@ public class Column implements Writable {
return colName;
}
+ public static String getShadowName(String colName) {
+ if (isShadowColumn(colName)) {
+ return colName;
+ }
+ return SchemaChangeHandler.SHADOW_NAME_PRFIX + colName;
+ }
+
+ public static boolean isShadowColumn(String colName) {
+ return colName.startsWith(SchemaChangeHandler.SHADOW_NAME_PRFIX);
+ }
+
public Expr getDefineExpr() {
return defineExpr;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/PrimitiveType.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/PrimitiveType.java
index 85572fd..761347a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/PrimitiveType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/PrimitiveType.java
@@ -481,11 +481,7 @@ public enum PrimitiveType {
compatibilityMatrix[TIME.ordinal()][TIME.ordinal()] = TIME;
}
- private static PrimitiveType[][] schemaChangeCompatibilityMatrix;
-
static {
- schemaChangeCompatibilityMatrix = new PrimitiveType[HLL.ordinal() + 1][HLL.ordinal() + 1];
-
// NULL_TYPE is compatible with any type and results in the non-null type.
compatibilityMatrix[NULL_TYPE.ordinal()][NULL_TYPE.ordinal()] = NULL_TYPE;
compatibilityMatrix[NULL_TYPE.ordinal()][BOOLEAN.ordinal()] = BOOLEAN;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
index 9aea5af..07016c4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
@@ -75,15 +75,15 @@ import org.apache.doris.transaction.TransactionState.TxnCoordinator;
import org.apache.doris.transaction.TransactionState.TxnSourceType;
import org.apache.doris.transaction.TransactionStatus;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.gson.annotations.SerializedName;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
@@ -146,7 +146,8 @@ public class DeleteHandler implements Writable {
OlapTable olapTable = (OlapTable) table;
if (olapTable.getState() != OlapTable.OlapTableState.NORMAL) {
- throw new DdlException("Table's state is not normal: " + tableName);
+ // table under alter operation can also do delete.
+ // just add a comment here to notice.
}
if (noPartitionSpecified) {
@@ -202,9 +203,10 @@ public class DeleteHandler implements Writable {
// task sent to be
AgentBatchTask batchTask = new AgentBatchTask();
// count total replica num
+ // Get ALL materialized indexes, because delete condition will be applied to all indexes
int totalReplicaNum = 0;
for (Partition partition : partitions) {
- for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.VISIBLE)) {
+ for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.ALL)) {
for (Tablet tablet : index.getTablets()) {
totalReplicaNum += tablet.getReplicas().size();
}
@@ -213,7 +215,7 @@ public class DeleteHandler implements Writable {
countDownLatch = new MarkedCountDownLatch<Long, Long>(totalReplicaNum);
for (Partition partition : partitions) {
- for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.VISIBLE)) {
+ for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.ALL)) {
long indexId = index.getId();
int schemaHash = olapTable.getSchemaHashByIndexId(indexId);
@@ -488,20 +490,13 @@ public class DeleteHandler implements Writable {
private void checkDeleteV2(OlapTable table, List<Partition> partitions, List<Predicate> conditions, List<String> deleteConditions)
throws DdlException {
- // check partition state
- for (Partition partition : partitions) {
- Partition.PartitionState state = partition.getState();
- if (state != Partition.PartitionState.NORMAL) {
- // ErrorReport.reportDdlException(ErrorCode.ERR_BAD_PARTITION_STATE, partition.getName(), state.name());
- throw new DdlException("Partition[" + partition.getName() + "]' state is not NORMAL: " + state.name());
- }
- }
-
// check condition column is key column and condition value
+ // Here we use "getFullSchema()" to get all columns including VISIBLE and SHADOW columns
Map<String, Column> nameToColumn = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
- for (Column column : table.getBaseSchema()) {
+ for (Column column : table.getFullSchema()) {
nameToColumn.put(column.getName(), column);
}
+
for (Predicate condition : conditions) {
SlotRef slotRef = getSlotRef(condition);
String columnName = slotRef.getColumnName();
@@ -509,6 +504,18 @@ public class DeleteHandler implements Writable {
ErrorReport.reportDdlException(ErrorCode.ERR_BAD_FIELD_ERROR, columnName, table.getName());
}
+ if (Column.isShadowColumn(columnName)) {
+ ErrorReport.reportDdlException(ErrorCode.ERR_COMMON_ERROR, "Can not apply delete condition to shadow column");
+ }
+
+ // Check if this column is under schema change, if yes, there will be a shadow column related to it.
+ // And we don't allow doing delete operation when a condition column is under schema change.
+ String shadowColName = Column.getShadowName(columnName);
+ if (nameToColumn.containsKey(shadowColName)) {
+ ErrorReport.reportDdlException(ErrorCode.ERR_COMMON_ERROR, "Column " + columnName + " is under" +
+ " schema change operation. Do not allow delete operation");
+ }
+
Column column = nameToColumn.get(columnName);
// Due to rounding errors, most floating-point numbers end up being slightly imprecise,
// it also means that numbers expected to be equal often differ slightly, so we do not allow compare with
@@ -570,7 +577,10 @@ public class DeleteHandler implements Writable {
// only need to check the first partition, because each partition has same materialized views
Map<Long, List<Column>> indexIdToSchema = table.getIndexIdToSchema();
Partition partition = partitions.get(0);
- for (MaterializedIndex index : partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE)) {
+ // Here we check ALL materialized views instead of just VISIBLE ones.
+ // For example, when a table is doing rollup or schema change. there will be some SHADOW indexes.
+ // And we also need to check these SHADOW indexes to see if the delete condition can be applied to them.
+ for (MaterializedIndex index : partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
if (table.getBaseIndexId() == index.getId()) {
continue;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org