You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by da...@apache.org on 2024/02/28 12:47:46 UTC
(doris) branch master updated: [Fix](cloud) Fix auto partition cannot load data(#31491)
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 26a32099519 [Fix](cloud) Fix auto partition cannot load data(#31491)
26a32099519 is described below
commit 26a32099519d6e32f52c8f1408180be4859425b2
Author: zclllyybb <zh...@selectdb.com>
AuthorDate: Wed Feb 28 20:47:39 2024 +0800
[Fix](cloud) Fix auto partition cannot load data(#31491)
---
be/src/service/backend_options.cpp | 8 +++++--
be/src/service/backend_options.h | 1 +
be/src/vec/sink/vrow_distribution.cpp | 8 ++++++-
be/src/vec/sink/writer/vtablet_writer.cpp | 1 +
.../main/java/org/apache/doris/catalog/Tablet.java | 25 ++++++++++++++++++----
.../apache/doris/cloud/catalog/CloudReplica.java | 25 ++++++++++++++--------
.../apache/doris/cloud/catalog/CloudTablet.java | 16 ++++++++++----
.../doris/cloud/system/CloudSystemInfoService.java | 9 ++++++++
.../apache/doris/service/FrontendServiceImpl.java | 8 ++++++-
gensrc/thrift/FrontendService.thrift | 2 ++
10 files changed, 82 insertions(+), 21 deletions(-)
diff --git a/be/src/service/backend_options.cpp b/be/src/service/backend_options.cpp
index a8c48fd710e..6eeee2f2881 100644
--- a/be/src/service/backend_options.cpp
+++ b/be/src/service/backend_options.cpp
@@ -67,6 +67,10 @@ const std::string& BackendOptions::get_localhost() {
return _s_localhost;
}
+std::string BackendOptions::get_be_endpoint() {
+ return _s_localhost + ":" + std::to_string(config::heartbeat_service_port);
+}
+
TBackend BackendOptions::get_local_backend() {
_backend.__set_host(_s_localhost);
_backend.__set_be_port(config::be_port);
@@ -99,7 +103,7 @@ const char* BackendOptions::get_service_bind_address_without_bracket() {
bool BackendOptions::analyze_priority_cidrs(const std::string& priority_networks,
std::vector<CIDR>* cidrs) {
- if (priority_networks == "") {
+ if (priority_networks.empty()) {
return true;
}
LOG(INFO) << "priority cidrs: " << priority_networks;
@@ -119,7 +123,7 @@ bool BackendOptions::analyze_priority_cidrs(const std::string& priority_networks
bool BackendOptions::analyze_localhost(std::string& localhost, bool& bind_ipv6,
std::vector<CIDR>* cidrs, std::vector<InetAddress>* hosts) {
- std::vector<InetAddress>::iterator addr_it = hosts->begin();
+ auto addr_it = hosts->begin();
if (!cidrs->empty()) {
for (; addr_it != hosts->end(); ++addr_it) {
VLOG_CRITICAL << "check ip=" << addr_it->get_host_address();
diff --git a/be/src/service/backend_options.h b/be/src/service/backend_options.h
index 8f504ba2ea7..c757f31d321 100644
--- a/be/src/service/backend_options.h
+++ b/be/src/service/backend_options.h
@@ -33,6 +33,7 @@ class BackendOptions {
public:
static bool init();
static const std::string& get_localhost();
+ static std::string get_be_endpoint();
static TBackend get_local_backend();
static void set_backend_id(int64_t backend_id);
static void set_localhost(const std::string& host);
diff --git a/be/src/vec/sink/vrow_distribution.cpp b/be/src/vec/sink/vrow_distribution.cpp
index 9ddee3b8f0b..358f8d3783e 100644
--- a/be/src/vec/sink/vrow_distribution.cpp
+++ b/be/src/vec/sink/vrow_distribution.cpp
@@ -23,10 +23,12 @@
#include <memory>
+#include "common/logging.h"
#include "common/status.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
+#include "service/backend_options.h"
#include "util/doris_metrics.h"
#include "util/thrift_rpc_helper.h"
#include "vec/columns/column_const.h"
@@ -60,10 +62,12 @@ Status VRowDistribution::_save_missing_values(std::vector<std::vector<std::strin
}
// to avoid too large mem use
- if (_batching_rows > _batch_size) {
+ if (_batching_block->rows() > _batch_size) {
_deal_batched = true;
}
+ VLOG_NOTICE << "pushed some batching lines, now numbers = " << _batching_rows;
+
return Status::OK();
}
@@ -77,10 +81,12 @@ Status VRowDistribution::automatic_create_partition() {
SCOPED_TIMER(_add_partition_request_timer);
TCreatePartitionRequest request;
TCreatePartitionResult result;
+ string be_endpoint = BackendOptions::get_be_endpoint();
request.__set_txn_id(_txn_id);
request.__set_db_id(_vpartition->db_id());
request.__set_table_id(_vpartition->table_id());
request.__set_partitionValues(_partitions_need_create);
+ request.__set_be_endpoint(be_endpoint);
VLOG(1) << "automatic partition rpc begin request " << request;
TNetworkAddress master_addr = ExecEnv::GetInstance()->master_info()->network_address;
diff --git a/be/src/vec/sink/writer/vtablet_writer.cpp b/be/src/vec/sink/writer/vtablet_writer.cpp
index 2d808b32af9..3bd0ef94592 100644
--- a/be/src/vec/sink/writer/vtablet_writer.cpp
+++ b/be/src/vec/sink/writer/vtablet_writer.cpp
@@ -446,6 +446,7 @@ void VNodeChannel::open() {
}
void VNodeChannel::incremental_open() {
+ VLOG_DEBUG << "incremental opening node channel" << _node_id;
_open_internal(true);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
index ccf4a20eda9..6c305587806 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
@@ -20,6 +20,7 @@ package org.apache.doris.catalog;
import org.apache.doris.catalog.Replica.ReplicaState;
import org.apache.doris.clone.TabletSchedCtx;
import org.apache.doris.clone.TabletSchedCtx.Priority;
+import org.apache.doris.cloud.catalog.CloudReplica;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.FeMetaVersion;
@@ -222,13 +223,17 @@ public class Tablet extends MetaObject implements Writable {
}
}
- // return map of (BE id -> path hash) of normal replicas
- // for load plan.
- public Multimap<Long, Long> getNormalReplicaBackendPathMap() throws UserException {
+ @FunctionalInterface
+ interface BackendIdGetter {
+ long get(Replica rep, String be);
+ }
+
+ private Multimap<Long, Long> getNormalReplicaBackendPathMapImpl(String beEndpoint, BackendIdGetter idGetter)
+ throws UserException {
Multimap<Long, Long> map = HashMultimap.create();
SystemInfoService infoService = Env.getCurrentSystemInfo();
for (Replica replica : replicas) {
- long backendId = replica.getBackendId();
+ long backendId = idGetter.get(replica, beEndpoint);
if (!infoService.checkBackendAlive(backendId)) {
continue;
}
@@ -246,6 +251,18 @@ public class Tablet extends MetaObject implements Writable {
return map;
}
+ // return map of (BE id -> path hash) of normal replicas
+ // for load plan.
+ public Multimap<Long, Long> getNormalReplicaBackendPathMap() throws UserException {
+ return getNormalReplicaBackendPathMapImpl(null, (rep, be) -> rep.getBackendId());
+ }
+
+ // for cloud mode without ConnectContext. use BE IP to find replica
+ protected Multimap<Long, Long> getNormalReplicaBackendPathMapCloud(String beEndpoint) throws UserException {
+ return getNormalReplicaBackendPathMapImpl(beEndpoint,
+ (rep, be) -> ((CloudReplica) rep).getBackendId(be));
+ }
+
// for query
public List<Replica> getQueryableReplicas(long visibleVersion, boolean allowFailedVersion) {
List<Replica> allQueryableReplica = Lists.newArrayListWithCapacity(replicas.size());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
index dbd340ee17a..e49ff6cafa8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
@@ -110,6 +110,11 @@ public class CloudReplica extends Replica {
return pickedBeId;
}
+ public long getBackendId(String beEndpoint) {
+ String cluster = ((CloudSystemInfoService) Env.getCurrentSystemInfo()).getClusterIdByBeAddr(beEndpoint);
+ return getBackendIdImpl(cluster);
+ }
+
@Override
public long getBackendId() {
String cluster = null;
@@ -139,20 +144,22 @@ public class CloudReplica extends Replica {
}
return -1;
}
+ return getBackendIdImpl(cluster);
+ }
+ private long getBackendIdImpl(String cluster) {
// check default cluster valid.
- if (!Strings.isNullOrEmpty(cluster)) {
- boolean exist = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
- .getCloudClusterNames().contains(cluster);
- if (!exist) {
- //can't use this default cluster, plz change another
- LOG.warn("cluster: {} is not existed", cluster);
- return -1;
- }
- } else {
+ if (Strings.isNullOrEmpty(cluster)) {
LOG.warn("failed to get available be, clusterName: {}", cluster);
return -1;
}
+ boolean exist = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
+ .getCloudClusterNames().contains(cluster);
+ if (!exist) {
+ // can't use this default cluster, plz change another
+ LOG.warn("cluster: {} is not existed", cluster);
+ return -1;
+ }
// if cluster is SUSPENDED, wait
try {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java
index 566c06b234f..28384bfdcf3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java
@@ -45,10 +45,7 @@ public class CloudTablet extends Tablet {
return null;
}
- @Override
- public Multimap<Long, Long> getNormalReplicaBackendPathMap() throws UserException {
- Multimap<Long, Long> pathMap = super.getNormalReplicaBackendPathMap();
-
+ private Multimap<Long, Long> backendPathMapReprocess(Multimap<Long, Long> pathMap) throws UserException {
if (pathMap.containsKey(-1L)) {
pathMap.removeAll(-1L);
if (pathMap.isEmpty()) {
@@ -60,6 +57,17 @@ public class CloudTablet extends Tablet {
return pathMap;
}
+ @Override
+ public Multimap<Long, Long> getNormalReplicaBackendPathMap() throws UserException {
+ Multimap<Long, Long> pathMap = super.getNormalReplicaBackendPathMap();
+ return backendPathMapReprocess(pathMap);
+ }
+
+ public Multimap<Long, Long> getNormalReplicaBackendPathMapCloud(String beEndpoint) throws UserException {
+ Multimap<Long, Long> pathMap = super.getNormalReplicaBackendPathMapCloud(beEndpoint);
+ return backendPathMapReprocess(pathMap);
+ }
+
@Override
protected boolean isLatestReplicaAndDeleteOld(Replica newReplica) {
boolean delete = false;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
index c6c9c3f1064..ae43da4b244 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
@@ -322,6 +322,15 @@ public class CloudSystemInfoService extends SystemInfoService {
return clusterIdToBackend.getOrDefault(clusterId, new ArrayList<>());
}
+ public String getClusterIdByBeAddr(String beEndpoint) {
+ for (Map.Entry<String, List<Backend>> idBe : clusterIdToBackend.entrySet()) {
+ if (idBe.getValue().stream().anyMatch(be -> be.getAddress().equals(beEndpoint))) {
+ return getClusterNameByClusterId(idBe.getKey());
+ }
+ }
+ return null;
+ }
+
public List<String> getCloudClusterIds() {
return new ArrayList<>(clusterIdToBackend.keySet());
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 73dc65db522..de70930f7a7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -48,6 +48,7 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.TabletMeta;
+import org.apache.doris.cloud.catalog.CloudTablet;
import org.apache.doris.cloud.planner.CloudStreamLoadPlanner;
import org.apache.doris.cluster.ClusterNamespace;
import org.apache.doris.common.AnalysisException;
@@ -3397,7 +3398,12 @@ public class FrontendServiceImpl implements FrontendService.Iface {
// BE id -> path hash
Multimap<Long, Long> bePathsMap;
try {
- bePathsMap = tablet.getNormalReplicaBackendPathMap();
+ if (Config.isCloudMode() && request.isSetBeEndpoint()) {
+ bePathsMap = ((CloudTablet) tablet)
+ .getNormalReplicaBackendPathMapCloud(request.be_endpoint);
+ } else {
+ bePathsMap = tablet.getNormalReplicaBackendPathMap();
+ }
} catch (UserException ex) {
errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
result.setStatus(errorStatus);
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 89ad46345d1..4b67bd1ec9d 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -1258,6 +1258,8 @@ struct TCreatePartitionRequest {
3: optional i64 table_id
// for each partition column's partition values. [missing_rows, partition_keys]->Left bound(for range) or Point(for list)
4: optional list<list<Exprs.TStringLiteral>> partitionValues
+ // be_endpoint = <ip>:<heartbeat_port> to distinguish a particular BE
+ 5: optional string be_endpoint
}
struct TCreatePartitionResult {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org