[feature](load) collect loaded rows on table level after txn published (

apache#24346) As title. Stream load 20 lines ``` 2023-09-14 11:40:04,186 DEBUG (PUBLISH_VERSION|23) [DatabaseTransactionMgr.updateCatalogAfterVisible():1769] table id to loaded rows:{51016=20} ``` ``` mysql> select count(*) from dup_tbl_basic; +----------+ | count(*) | +----------+ | 20 | +----------+ 1 row in set (0.05 sec) ```
dataroaring · Sep 19, 2023 · b092bda · b092bda
1 parent 80bcb43
commit b092bda
Show file tree

Hide file tree

Showing 9 changed files with 83 additions and 9 deletions.
diff --git a/be/src/agent/task_worker_pool.cpp b/be/src/agent/task_worker_pool.cpp
@@ -1529,14 +1529,16 @@ void PublishVersionTaskPool::_publish_version_worker_thread_callback() {
  std::map<TTabletId, TVersion> succ_tablets;
  // partition_id, tablet_id, publish_version
  std::vector<std::tuple<int64_t, int64_t, int64_t>> discontinuous_version_tablets;
+ std::map<TTabletId, int64_t> tablet_id_to_num_delta_rows;
  uint32_t retry_time = 0;
  Status status;
  bool is_task_timeout = false;
  while (retry_time < PUBLISH_VERSION_MAX_RETRY) {
  succ_tablets.clear();
  error_tablet_ids.clear();
  EnginePublishVersionTask engine_task(publish_version_req, &error_tablet_ids,
- &succ_tablets, &discontinuous_version_tablets);
+ &succ_tablets, &discontinuous_version_tablets,
+ &tablet_id_to_num_delta_rows);
  status = StorageEngine::instance()->execute_task(&engine_task);
  if (status.ok()) {
  break;
@@ -1621,7 +1623,7 @@ void PublishVersionTaskPool::_publish_version_worker_thread_callback() {
  finish_task_request.__set_succ_tablets(succ_tablets);
  finish_task_request.__set_error_tablet_ids(
  std::vector<TTabletId>(error_tablet_ids.begin(), error_tablet_ids.end()));
-
+ finish_task_request.__set_tablet_id_to_delta_num_rows(tablet_id_to_num_delta_rows);
  _finish_task(finish_task_request);
  _remove_task_info(agent_task_req.task_type, agent_task_req.signature);
  }

diff --git a/be/src/olap/task/engine_publish_version_task.cpp b/be/src/olap/task/engine_publish_version_task.cpp
@@ -71,11 +71,13 @@ void TabletPublishStatistics::record_in_bvar() {
 EnginePublishVersionTask::EnginePublishVersionTask(
  const TPublishVersionRequest& publish_version_req, std::set<TTabletId>* error_tablet_ids,
  std::map<TTabletId, TVersion>* succ_tablets,
- std::vector<std::tuple<int64_t, int64_t, int64_t>>* discontinuous_version_tablets)
+ std::vector<std::tuple<int64_t, int64_t, int64_t>>* discontinuous_version_tablets,
+ std::map<TTabletId, int64_t>* tablet_id_to_num_delta_rows)
  : _publish_version_req(publish_version_req),
  _error_tablet_ids(error_tablet_ids),
  _succ_tablets(succ_tablets),
- _discontinuous_version_tablets(discontinuous_version_tablets) {}
+ _discontinuous_version_tablets(discontinuous_version_tablets),
+ _tablet_id_to_num_delta_rows(tablet_id_to_num_delta_rows) {}
 
 void EnginePublishVersionTask::add_error_tablet_id(int64_t tablet_id) {
  std::lock_guard<std::mutex> lck(_tablet_ids_mutex);
@@ -186,6 +188,9 @@ Status EnginePublishVersionTask::finish() {
  continue;
  }
  }
+ auto rowset_meta_ptr = rowset->rowset_meta();
+ _tablet_id_to_num_delta_rows->insert(
+ {rowset_meta_ptr->tablet_id(), rowset_meta_ptr->num_rows()});
  auto tablet_publish_txn_ptr = std::make_shared<TabletPublishTxnTask>(
  this, tablet, rowset, partition_id, transaction_id, version, tablet_info);
  auto submit_st = token->submit_func([=]() { tablet_publish_txn_ptr->handle(); });

diff --git a/be/src/olap/task/engine_publish_version_task.h b/be/src/olap/task/engine_publish_version_task.h
@@ -87,10 +87,11 @@ class EnginePublishVersionTask : public EngineTask {
  EnginePublishVersionTask(
  const TPublishVersionRequest& publish_version_req,
  std::set<TTabletId>* error_tablet_ids, std::map<TTabletId, TVersion>* succ_tablets,
- std::vector<std::tuple<int64_t, int64_t, int64_t>>* discontinous_version_tablets);
- ~EnginePublishVersionTask() {}
+ std::vector<std::tuple<int64_t, int64_t, int64_t>>* discontinous_version_tablets,
+ std::map<TTabletId, int64_t>* tablet_id_to_num_delta_rows);
+ ~EnginePublishVersionTask() override = default;
 
- virtual Status finish() override;
+ Status finish() override;
 
  void add_error_tablet_id(int64_t tablet_id);
 
@@ -102,6 +103,7 @@ class EnginePublishVersionTask : public EngineTask {
  std::set<TTabletId>* _error_tablet_ids;
  std::map<TTabletId, TVersion>* _succ_tablets;
  std::vector<std::tuple<int64_t, int64_t, int64_t>>* _discontinuous_version_tablets;
+ std::map<TTabletId, int64_t>* _tablet_id_to_num_delta_rows;
 };
 
 class AsyncTabletPublishTask {

diff --git a/fe/fe-core/src/main/java/org/apache/doris/master/MasterImpl.java b/fe/fe-core/src/main/java/org/apache/doris/master/MasterImpl.java
@@ -491,6 +491,9 @@ private void finishPublishVersion(AgentTask task, TFinishTaskRequest request) {
  // not remove the task from queue and be will retry
  return;
  }
+ if (request.isSetTabletIdToDeltaNumRows()) {
+ publishVersionTask.setTabletIdToDeltaNumRows(request.getTabletIdToDeltaNumRows());
+ }
  AgentTaskQueue.removeTask(publishVersionTask.getBackendId(),
  publishVersionTask.getTaskType(),
  publishVersionTask.getSignature());

diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/PublishVersionTask.java b/fe/fe-core/src/main/java/org/apache/doris/task/PublishVersionTask.java
@@ -21,6 +21,7 @@
 import org.apache.doris.thrift.TPublishVersionRequest;
 import org.apache.doris.thrift.TTaskType;
 
+import com.google.common.collect.Maps;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -38,6 +39,11 @@ public class PublishVersionTask extends AgentTask {
  // tabletId => version, current version = 0
  private Map<Long, Long> succTablets;
 
+ /**
+ * To collect loaded rows for each tablet from each BE
+ */
+ private final Map<Long, Long> tabletIdToDeltaNumRows = Maps.newHashMap();
+
  public PublishVersionTask(long backendId, long transactionId, long dbId,
  List<TPartitionVersionInfo> partitionVersionInfos, long createTime) {
  super(null, backendId, TTaskType.PUBLISH_VERSION, dbId, -1L, -1L, -1L, -1L, transactionId, createTime);
@@ -81,4 +87,12 @@ public synchronized void addErrorTablets(List<Long> errorTablets) {
  }
  this.errorTablets.addAll(errorTablets);
  }
+
+ public void setTabletIdToDeltaNumRows(Map<Long, Long> tabletIdToDeltaNumRows) {
+ this.tabletIdToDeltaNumRows.putAll(tabletIdToDeltaNumRows);
+ }
+
+ public Map<Long, Long> getTabletIdToDeltaNumRows() {
+ return tabletIdToDeltaNumRows;
+ }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
@@ -51,6 +51,7 @@
 import org.apache.doris.persist.BatchRemoveTransactionsOperationV2;
 import org.apache.doris.persist.EditLog;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.statistics.AnalysisManager;
 import org.apache.doris.task.AgentBatchTask;
 import org.apache.doris.task.AgentTaskExecutor;
 import org.apache.doris.task.ClearTransactionTask;
@@ -1787,6 +1788,9 @@ private boolean updateCatalogAfterVisible(TransactionState transactionState, Dat
  }
  }
  }
+ AnalysisManager analysisManager = Env.getCurrentEnv().getAnalysisManager();
+ LOG.debug("table id to loaded rows:{}", transactionState.getTableIdToNumDeltaRows());
+ transactionState.getTableIdToNumDeltaRows().forEach(analysisManager::updateUpdatedRows);
  return true;
  }
 

diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/PublishVersionDaemon.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/PublishVersionDaemon.java
@@ -18,6 +18,8 @@
 package org.apache.doris.transaction;
 
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TabletInvertedIndex;
+import org.apache.doris.catalog.TabletMeta;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.util.MasterDaemon;
 import org.apache.doris.metric.MetricRepo;
@@ -29,13 +31,17 @@
 import org.apache.doris.thrift.TPartitionVersionInfo;
 import org.apache.doris.thrift.TTaskType;
 
+import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.stream.Stream;
 
 public class PublishVersionDaemon extends MasterDaemon {
 
@@ -121,12 +127,39 @@ private void publishVersion() {
  AgentTaskExecutor.submit(batchTask);
  }
 
+ TabletInvertedIndex tabletInvertedIndex = Env.getCurrentEnv().getTabletInvertedIndex();
+ Set<Long> tabletIdFilter = Sets.newHashSet();
+ Map<Long, Long> tableIdToNumDeltaRows = Maps.newHashMap();
  // try to finish the transaction, if failed just retry in next loop
  for (TransactionState transactionState : readyTransactionStates) {
- boolean hasBackendAliveAndUnfinishTask = transactionState.getPublishVersionTasks().values().stream()
+ Stream<PublishVersionTask> publishVersionTaskStream = transactionState
+ .getPublishVersionTasks()
+ .values()
+ .stream()
+ .peek(task -> {
+ if (task.isFinished() && CollectionUtils.isEmpty(task.getErrorTablets())) {
+ Map<Long, Long> tabletIdToDeltaNumRows =
+ task.getTabletIdToDeltaNumRows();
+ tabletIdToDeltaNumRows.forEach((tabletId, numRows) -> {
+ if (!tabletIdFilter.add(tabletId)) {
+ // means the delta num rows for this tablet id has been collected
+ return;
+ }
+ TabletMeta tabletMeta = tabletInvertedIndex.getTabletMeta(tabletId);
+ if (tabletMeta == null) {
+ // for delete, drop, schema change etc. here may be a null value
+ return;
+ }
+ long tableId = tabletMeta.getTableId();
+ tableIdToNumDeltaRows.computeIfPresent(tableId, (tblId, orgNum) -> orgNum + numRows);
+ tableIdToNumDeltaRows.putIfAbsent(tableId, numRows);
+ });
+ }
+ });
+ boolean hasBackendAliveAndUnfinishedTask = publishVersionTaskStream
  .anyMatch(task -> !task.isFinished() && infoService.checkBackendAlive(task.getBackendId()));
 
- boolean shouldFinishTxn = !hasBackendAliveAndUnfinishTask || transactionState.isPublishTimeout();
+ boolean shouldFinishTxn = !hasBackendAliveAndUnfinishedTask || transactionState.isPublishTimeout();
  if (shouldFinishTxn) {
  try {
  // one transaction exception should not affect other transaction

diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionState.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionState.java
@@ -253,6 +253,8 @@ public String toString() {
  // tbl id -> (index ids)
  private Map<Long, Set<Long>> loadedTblIndexes = Maps.newHashMap();
 
+ private Map<Long, Long> tableIdToNumDeltaRows = Maps.newHashMap();
+
  private String errorLogUrl = null;
 
  // record some error msgs during the transaction operation.
@@ -701,6 +703,14 @@ public void readFields(DataInput in) throws IOException {
  }
  }
 
+ public Map<Long, Long> getTableIdToNumDeltaRows() {
+ return tableIdToNumDeltaRows;
+ }
+
+ public void setTableIdToNumDeltaRows(Map<Long, Long> tableIdToNumDeltaRows) {
+ this.tableIdToNumDeltaRows.putAll(tableIdToNumDeltaRows);
+ }
+
  public void setErrorMsg(String errMsg) {
  this.errMsg = errMsg;
  }

diff --git a/gensrc/thrift/MasterService.thrift b/gensrc/thrift/MasterService.thrift
@@ -66,6 +66,7 @@ struct TFinishTaskRequest {
  15: optional i64 copy_size
  16: optional i64 copy_time_ms
  17: optional map<Types.TTabletId, Types.TVersion> succ_tablets
+ 18: optional map<i64, i64> tablet_id_to_delta_num_rows
 }
 
 struct TTablet {