This is an automated email from the ASF dual-hosted git repository.
dataroaring 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 10a968957bd [feature](cloud) Support delete statement in cloud mode
(#31323)
10a968957bd is described below
commit 10a968957bd7b2b9aa0dedf111150433a8b91dbe
Author: plat1ko <[email protected]>
AuthorDate: Fri Feb 23 19:43:56 2024 +0800
[feature](cloud) Support delete statement in cloud mode (#31323)
---
be/src/agent/agent_server.cpp | 4 +-
be/src/agent/task_worker_pool.cpp | 52 +++++++++++++
be/src/agent/task_worker_pool.h | 3 +
be/src/cloud/cloud_delete_task.cpp | 91 ++++++++++++++++++++++
be/src/cloud/cloud_delete_task.h | 32 ++++++++
be/src/olap/tablet_schema.cpp | 11 +++
be/src/olap/tablet_schema.h | 3 +
.../main/java/org/apache/doris/load/DeleteJob.java | 35 +++++----
8 files changed, 213 insertions(+), 18 deletions(-)
diff --git a/be/src/agent/agent_server.cpp b/be/src/agent/agent_server.cpp
index d2a6e605b88..7a462f4824c 100644
--- a/be/src/agent/agent_server.cpp
+++ b/be/src/agent/agent_server.cpp
@@ -189,7 +189,9 @@ void AgentServer::start_workers(StorageEngine& engine,
ExecEnv* exec_env) {
}
void AgentServer::cloud_start_workers(CloudStorageEngine& engine, ExecEnv*
exec_env) {
- // TODO(plat1ko): DELETE worker
+ _workers[TTaskType::PUSH] = std::make_unique<TaskWorkerPool>(
+ "PUSH", config::delete_worker_count,
+ [&engine](auto&& task) { cloud_push_callback(engine, task); });
// TODO(plat1ko): ALTER worker
// TODO(plat1ko): SUBMIT_TABLE_COMPACTION worker
// TODO(plat1ko): CALCULATE_DELETE_BITMAP worker
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 0c5ee012542..2b327f360fa 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -44,6 +44,7 @@
#include <vector>
#include "agent/utils.h"
+#include "cloud/cloud_delete_task.h"
#include "common/config.h"
#include "common/logging.h"
#include "common/status.h"
@@ -1446,6 +1447,57 @@ void push_callback(StorageEngine& engine, const
TAgentTaskRequest& req) {
remove_task_info(req.task_type, req.signature);
}
+void cloud_push_callback(CloudStorageEngine& engine, const TAgentTaskRequest&
req) {
+ const auto& push_req = req.push_req;
+
+ LOG(INFO) << "get push task. signature=" << req.signature
+ << " push_type=" << push_req.push_type;
+
+ // Return result to fe
+ TFinishTaskRequest finish_task_request;
+ finish_task_request.__set_backend(BackendOptions::get_local_backend());
+ finish_task_request.__set_task_type(req.task_type);
+ finish_task_request.__set_signature(req.signature);
+
+ // Only support DELETE in cloud mode now
+ if (push_req.push_type != TPushType::DELETE) {
+ finish_task_request.__set_task_status(
+ Status::NotSupported("push_type {} not is supported",
+ std::to_string(push_req.push_type))
+ .to_thrift());
+ return;
+ }
+
+ finish_task_request.__set_request_version(push_req.version);
+
+ DorisMetrics::instance()->delete_requests_total->increment(1);
+ auto st = CloudDeleteTask::execute(engine, req.push_req);
+ if (st.ok()) {
+ LOG_INFO("successfully execute push task")
+ .tag("signature", req.signature)
+ .tag("tablet_id", push_req.tablet_id)
+ .tag("push_type", push_req.push_type);
+ ++s_report_version;
+ auto& tablet_info =
finish_task_request.finish_tablet_infos.emplace_back();
+ // Just need tablet_id
+ tablet_info.tablet_id = push_req.tablet_id;
+ finish_task_request.__isset.finish_tablet_infos = true;
+ } else {
+ DorisMetrics::instance()->delete_requests_failed->increment(1);
+ LOG_WARNING("failed to execute push task")
+ .tag("signature", req.signature)
+ .tag("tablet_id", push_req.tablet_id)
+ .tag("push_type", push_req.push_type)
+ .error(st);
+ }
+
+ finish_task_request.__set_task_status(st.to_thrift());
+ finish_task_request.__set_report_version(s_report_version);
+
+ finish_task(finish_task_request);
+ remove_task_info(req.task_type, req.signature);
+}
+
PublishVersionWorkerPool::PublishVersionWorkerPool(StorageEngine& engine)
: TaskWorkerPool("PUBLISH_VERSION",
config::publish_version_worker_count,
[this](const TAgentTaskRequest& task) {
publish_version_callback(task); }),
diff --git a/be/src/agent/task_worker_pool.h b/be/src/agent/task_worker_pool.h
index 5d440377724..5f090539d4f 100644
--- a/be/src/agent/task_worker_pool.h
+++ b/be/src/agent/task_worker_pool.h
@@ -33,6 +33,7 @@ namespace doris {
class ExecEnv;
class StorageEngine;
+class CloudStorageEngine;
class Thread;
class ThreadPool;
class TMasterInfo;
@@ -158,6 +159,8 @@ void clear_transaction_task_callback(StorageEngine& engine,
const TAgentTaskRequ
void push_callback(StorageEngine& engine, const TAgentTaskRequest& req);
+void cloud_push_callback(CloudStorageEngine& engine, const TAgentTaskRequest&
req);
+
void update_tablet_meta_callback(StorageEngine& engine, const
TAgentTaskRequest& req);
void alter_tablet_callback(StorageEngine& engine, const TAgentTaskRequest&
req);
diff --git a/be/src/cloud/cloud_delete_task.cpp
b/be/src/cloud/cloud_delete_task.cpp
new file mode 100644
index 00000000000..2e3913533c3
--- /dev/null
+++ b/be/src/cloud/cloud_delete_task.cpp
@@ -0,0 +1,91 @@
+// 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 "cloud/cloud_delete_task.h"
+
+#include <gen_cpp/AgentService_types.h>
+#include <thrift/protocol/TDebugProtocol.h>
+
+#include "cloud/cloud_meta_mgr.h"
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet.h"
+#include "cloud/cloud_tablet_mgr.h"
+#include "common/logging.h"
+#include "olap/delete_handler.h"
+
+namespace doris {
+using namespace ErrorCode;
+
+Status CloudDeleteTask::execute(CloudStorageEngine& engine, const TPushReq&
request) {
+ VLOG_DEBUG << "begin to process delete data. request=" <<
ThriftDebugString(request);
+
+ if (!request.__isset.transaction_id) {
+ return Status::InvalidArgument("transaction_id is not set");
+ }
+
+ auto tablet = DORIS_TRY(engine.tablet_mgr().get_tablet(request.tablet_id));
+
+ using namespace std::chrono;
+ tablet->last_load_time_ms =
+
duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
+ // check if version number exceed limit
+ if (tablet->fetch_add_approximate_num_rowsets(0) >
config::max_tablet_version_num) {
+ LOG_WARNING("tablet exceeds max version num limit")
+ .tag("limit", config::max_tablet_version_num)
+ .tag("tablet_id", tablet->tablet_id());
+ return Status::Error<TOO_MANY_VERSION>("too many versions, versions={}
tablet={}",
+ config::max_tablet_version_num,
tablet->tablet_id());
+ }
+
+ // check delete condition if push for delete
+ DeletePredicatePB del_pred;
+ auto tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->update_tablet_columns(*tablet->tablet_schema(),
request.columns_desc);
+ RETURN_IF_ERROR(DeleteHandler::generate_delete_predicate(*tablet_schema,
+
request.delete_conditions, &del_pred));
+
+ PUniqueId load_id;
+ load_id.set_hi(0);
+ load_id.set_lo(0);
+ RowsetWriterContext context;
+ context.fs = engine.latest_fs();
+ context.txn_id = request.transaction_id;
+ context.load_id = load_id;
+ context.rowset_state = PREPARED;
+ context.segments_overlap = OVERLAP_UNKNOWN;
+ context.tablet_schema = tablet_schema;
+ // ATTN: `request.timeout` is always 0 in current version, so we MUST
ensure that the retention time of the
+ // recycler is much longer than the duration of the push task
+ context.txn_expiration = ::time(nullptr) + request.timeout;
+ auto rowset_writer = DORIS_TRY(tablet->create_rowset_writer(context,
false));
+
+ RowsetSharedPtr rowset;
+ RETURN_IF_ERROR(rowset_writer->build(rowset));
+ rowset->rowset_meta()->set_delete_predicate(std::move(del_pred));
+
+ auto st = engine.meta_mgr().commit_rowset(*rowset->rowset_meta(), true);
+
+ // Update tablet stats
+ tablet->fetch_add_approximate_num_rowsets(1);
+ tablet->fetch_add_approximate_cumu_num_rowsets(1);
+
+ // TODO(liaoxin): set_tablet_txn_info if enable_unique_key_merge_on_write
+
+ return Status::OK();
+}
+
+} // namespace doris
diff --git a/be/src/cloud/cloud_delete_task.h b/be/src/cloud/cloud_delete_task.h
new file mode 100644
index 00000000000..bdebcc8b6e1
--- /dev/null
+++ b/be/src/cloud/cloud_delete_task.h
@@ -0,0 +1,32 @@
+// 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.
+
+#pragma once
+
+#include "common/status.h"
+
+namespace doris {
+
+class CloudStorageEngine;
+class TPushReq;
+
+class CloudDeleteTask {
+public:
+ static Status execute(CloudStorageEngine& engine, const TPushReq& request);
+};
+
+} // namespace doris
diff --git a/be/src/olap/tablet_schema.cpp b/be/src/olap/tablet_schema.cpp
index 79806c4703e..e54cf99eac3 100644
--- a/be/src/olap/tablet_schema.cpp
+++ b/be/src/olap/tablet_schema.cpp
@@ -1231,6 +1231,17 @@ std::vector<const TabletIndex*>
TabletSchema::get_indexes_for_column(
return indexes_for_column;
}
+void TabletSchema::update_tablet_columns(const TabletSchema& tablet_schema,
+ const std::vector<TColumn>&
t_columns) {
+ copy_from(tablet_schema);
+ if (!t_columns.empty() && t_columns[0].col_unique_id >= 0) {
+ clear_columns();
+ for (const auto& column : t_columns) {
+ append_column(TabletColumn(column));
+ }
+ }
+}
+
bool TabletSchema::has_inverted_index(const TabletColumn& col) const {
// TODO use more efficient impl
int32_t col_unique_id = col.unique_id();
diff --git a/be/src/olap/tablet_schema.h b/be/src/olap/tablet_schema.h
index d4a1919101a..91193b998f9 100644
--- a/be/src/olap/tablet_schema.h
+++ b/be/src/olap/tablet_schema.h
@@ -401,6 +401,9 @@ public:
std::shared_ptr<TabletSchema> copy_without_extracted_columns();
+ void update_tablet_columns(const TabletSchema& tablet_schema,
+ const std::vector<TColumn>& t_columns);
+
private:
friend bool operator==(const TabletSchema& a, const TabletSchema& b);
friend bool operator!=(const TabletSchema& a, const TabletSchema& b);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
index e97640febc8..b1f27d2aadc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
@@ -95,8 +95,7 @@ public class DeleteJob extends AbstractTxnStateChangeCallback
implements DeleteJ
// jobId(listenerId). use in beginTransaction to callback function
private final long id;
- // transaction id.
- private long signature;
+ private long transactionId;
private final String label;
private final Set<Long> totalTablets;
private final Set<Long> quorumTablets;
@@ -120,7 +119,7 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
public DeleteJob(long id, long transactionId, String label,
Map<Long, Short> partitionReplicaNum, DeleteInfo
deleteInfo) {
this.id = id;
- this.signature = transactionId;
+ this.transactionId = transactionId;
this.label = label;
this.deleteInfo = deleteInfo;
totalTablets = Sets.newHashSet();
@@ -178,7 +177,7 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
LOG.info("check delete job quorum, transaction id: {}, total tablets:
{},"
+ " quorum tablets: {}, dropped tablets: {}",
- signature, totalTablets.size(), quorumTablets.size(),
dropCounter);
+ transactionId, totalTablets.size(), quorumTablets.size(),
dropCounter);
if (finishedTablets.containsAll(totalTablets)) {
this.state = DeleteState.FINISHED;
@@ -240,7 +239,7 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
}
public long getTransactionId() {
- return this.signature;
+ return this.transactionId;
}
public Collection<TabletDeleteInfo> getTabletDeleteInfo() {
@@ -284,7 +283,7 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
new TransactionState.TxnCoordinator(
TransactionState.TxnSourceType.FE,
FrontendOptions.getLocalHostAddress()),
TransactionState.LoadJobSourceType.FRONTEND, id,
Config.stream_load_default_timeout_second);
- this.signature = txnId;
+ this.transactionId = txnId;
Env.getCurrentGlobalTransactionMgr().getCallbackFactory().addCallback(this);
return txnId;
}
@@ -316,6 +315,9 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
countDownLatch.addMark(backendId, tabletId);
// create push task for each replica
+ // To ensure that upgrading to new signature
generating algo does not conflict with the old
+ // signature, adding 10 billion to `getNextId`. We are
confident that the old signature
+ // generated will not exceed this number.
PushTask pushTask = new PushTask(null,
replica.getBackendId(), targetDb.getId(),
targetTbl.getId(),
partition.getId(), indexId,
@@ -324,9 +326,8 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
-1, type, deleteConditions,
true, TPriority.NORMAL,
TTaskType.REALTIME_PUSH,
- signature,
- Env.getCurrentGlobalTransactionMgr()
-
.getTransactionIDGenerator().getNextTransactionId(),
+ transactionId,
+ Env.getCurrentEnv().getNextId() + 10000000000L,
columnsDesc);
pushTask.setIsSchemaChanging(false);
pushTask.setCountDownLatch(countDownLatch);
@@ -368,7 +369,7 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
switch (state) {
case UN_QUORUM:
LOG.warn("delete job timeout: transactionId {}, timeout {},
{}",
- signature, timeoutMs, errMsg);
+ transactionId, timeoutMs, errMsg);
throw new UserException(String.format("delete job timeout,
timeout(ms):%s, msg:%s", timeoutMs, errMsg));
case QUORUM_FINISHED:
case FINISHED:
@@ -382,7 +383,7 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
nowQuorumTimeMs = System.currentTimeMillis();
if (LOG.isDebugEnabled()) {
LOG.debug("wait for quorum finished delete job: {},
txn id: {}",
- id, signature);
+ id, transactionId);
}
}
break;
@@ -405,11 +406,11 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
}));
boolean visible = Env.getCurrentGlobalTransactionMgr()
.commitAndPublishTransaction(targetDb,
Lists.newArrayList(targetTbl),
- signature, tabletCommitInfos, getTimeoutMs());
+ transactionId, tabletCommitInfos, getTimeoutMs());
StringBuilder sb = new StringBuilder();
sb.append("{'label':'").append(label);
- sb.append("', 'txnId':'").append(signature)
+ sb.append("', 'txnId':'").append(transactionId)
.append("', 'status':'");
if (visible) {
sb.append(TransactionStatus.VISIBLE.name()).append("'");
@@ -428,17 +429,17 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
public void cancel(String reason) {
GlobalTransactionMgrIface globalTransactionMgr =
Env.getCurrentGlobalTransactionMgr();
try {
- globalTransactionMgr.abortTransaction(deleteInfo.getDbId(),
signature, reason);
+ globalTransactionMgr.abortTransaction(deleteInfo.getDbId(),
transactionId, reason);
} catch (Exception e) {
TransactionState state = globalTransactionMgr.getTransactionState(
- deleteInfo.getDbId(), signature);
+ deleteInfo.getDbId(), transactionId);
if (state == null) {
LOG.warn("cancel delete job failed because txn not found,
transactionId: {}",
- signature);
+ transactionId);
} else if (state.getTransactionStatus() ==
TransactionStatus.COMMITTED
|| state.getTransactionStatus() ==
TransactionStatus.VISIBLE) {
LOG.warn("cancel delete job failed because it has been
committed, transactionId: {}",
- signature);
+ transactionId);
} else {
LOG.warn("errors while abort transaction", e);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
