This is an automated email from the ASF dual-hosted git repository.
zclllyybb 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 bd16e5d68fd [fix](fe) Guard auto partition result against concurrent
drops (#65282)
bd16e5d68fd is described below
commit bd16e5d68fd280343ccf6da61b2d47a9d3c2615d
Author: zclllyybb <[email protected]>
AuthorDate: Thu Jul 23 16:45:17 2026 +0800
[fix](fe) Guard auto partition result against concurrent drops (#65282)
Problem Summary: Auto partition creation can race with dynamic partition
retention. After new partitions are added, the response builder reads
the table partition map and partition info without holding the table
metadata lock. If a retention drop removes one of those partitions in
the same window, the response builder can observe a missing partition or
partition item and hit a null pointer while constructing partition and
tablet metadata. This change snapshots the required partition metadata
under the table read lock, returns a normal error when the partition was
already dropped before the snapshot, and keeps tablet
location/cache/backend resolution work outside the table lock.
---
.../apache/doris/service/FrontendServiceImpl.java | 438 ++++++++++++---------
.../doris/service/FrontendServiceImplTest.java | 49 +++
2 files changed, 303 insertions(+), 184 deletions(-)
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 b737ed2bc94..6a5695cebb0 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
@@ -40,6 +40,7 @@ import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.PartitionInfo;
+import org.apache.doris.catalog.PartitionItem;
import org.apache.doris.catalog.PartitionType;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Table;
@@ -4702,26 +4703,6 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
}
}
- // check partition's number limit. because partitions in
addPartitionClauseMap may be duplicated with existing
- // partitions, which would lead to false positive. so we should check
the partition number AFTER adding new
- // partitions using its ACTUAL NUMBER, rather than the sum of existing
and requested partitions.
- int partitionNum = olapTable.getPartitionNum();
- int autoPartitionLimit = Config.max_auto_partition_num;
- if (partitionNum > autoPartitionLimit) {
- String errorMessage = String.format(
- "partition numbers %d exceeded limit of variable
max_auto_partition_num %d",
- partitionNum, autoPartitionLimit);
- LOG.warn(errorMessage);
- errorStatus.setErrorMsgs(Lists.newArrayList(errorMessage));
- result.setStatus(errorStatus);
- LOG.warn("send create partition error status: {}", result);
- return result;
- } else if (partitionNum > autoPartitionLimit * 8 / 10) {
- LOG.warn("Table {}.{} auto partition count {} is approaching limit
{} (>80%)."
- + " Consider increasing max_auto_partition_num.",
- db.getFullName(), olapTable.getName(), partitionNum,
autoPartitionLimit);
- }
-
// build partition & tablets
List<TTabletLocation> tablets = new ArrayList<>();
List<TTabletLocation> slaveTablets = new ArrayList<>();
@@ -4733,41 +4714,59 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
&& request.isEnableAdaptiveRandomBucket();
boolean loadToSingleTablet = request.isSetLoadToSingleTablet() &&
request.isLoadToSingleTablet();
final boolean hasBeEndpoint = request.isSetBeEndpoint();
- // Lazy: resolved on the first CloudTablet that needs it (skipped on
cache-hit).
- String cachedClusterId = null;
- for (String partitionName : addPartitionClauseMap.keySet()) {
- Partition partition = table.getPartition(partitionName);
- // For thread safety, we preserve the tablet distribution
information of each partition
- // before calling getOrSetAutoPartitionInfo, but not check the
partition first
- List<TTabletLocation> partitionTablets = new ArrayList<>();
- List<TTabletLocation> partitionSlaveTablets = new ArrayList<>();
- TOlapTablePartition tPartition = new TOlapTablePartition();
- tPartition.setId(partition.getId());
- int partColNum = partitionInfo.getPartitionColumns().size();
+ List<PartitionResultSnapshot> partitionSnapshots = new ArrayList<>();
+
+ olapTable.readLock();
+ try {
+ // check partition's number limit. because partitions in
addPartitionClauseMap may be duplicated with
+ // existing partitions, which would lead to false positive. so we
should check the partition number AFTER
+ // adding new partitions using its ACTUAL NUMBER, rather than the
sum of existing and requested partitions.
+ int partitionNum = olapTable.getPartitionNum();
+ int autoPartitionLimit = Config.max_auto_partition_num;
+ if (partitionNum > autoPartitionLimit) {
+ String errorMessage = String.format(
+ "partition numbers %d exceeded limit of variable
max_auto_partition_num %d",
+ partitionNum, autoPartitionLimit);
+ LOG.warn(errorMessage);
+ errorStatus.setErrorMsgs(Lists.newArrayList(errorMessage));
+ result.setStatus(errorStatus);
+ LOG.warn("send create partition error status: {}", result);
+ return result;
+ } else if (partitionNum > autoPartitionLimit * 8 / 10) {
+ LOG.warn("Table {}.{} auto partition count {} is approaching
limit {} (>80%)."
+ + " Consider increasing max_auto_partition_num.",
+ db.getFullName(), olapTable.getName(), partitionNum,
autoPartitionLimit);
+ }
+
try {
- OlapTableSink.setPartitionKeys(tPartition,
partitionInfo.getItem(partition.getId()), partColNum);
+
partitionSnapshots.addAll(snapshotPartitionResultsByName(olapTable,
addPartitionClauseMap.keySet(),
+ loadToSingleTablet, enableAdaptiveRandomBucket, "auto
partition"));
} catch (UserException ex) {
errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
result.setStatus(errorStatus);
LOG.warn("send create partition error status: {}", result);
return result;
}
- for (MaterializedIndex index :
partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
- tPartition.addToIndexes(new
TOlapTableIndexTablets(index.getId(), Lists.newArrayList(
-
index.getTablets().stream().map(Tablet::getId).collect(Collectors.toList()))));
- tPartition.setNumBuckets(index.getTablets().size());
- }
-
tPartition.setIsMutable(olapTable.getPartitionInfo().getIsMutable(partition.getId()));
- boolean randomDistribution =
- partition.getDistributionInfo().getType() ==
DistributionInfo.DistributionInfoType.RANDOM;
- boolean cacheLoadTabletIdx =
- (loadToSingleTablet || enableAdaptiveRandomBucket) &&
randomDistribution;
+ } finally {
+ olapTable.readUnlock();
+ }
+
+ // Lazy: resolved on the first CloudTablet that needs it (skipped on
cache-hit).
+ String cachedClusterId = null;
+ for (PartitionResultSnapshot partitionSnapshot : partitionSnapshots) {
+ Partition partition = partitionSnapshot.partition;
+ long partitionId = partitionSnapshot.partitionId;
+ TOlapTablePartition tPartition = partitionSnapshot.tPartition;
+ boolean cacheLoadTabletIdx = partitionSnapshot.cacheLoadTabletIdx;
partitions.add(tPartition);
- // tablet
+ // For thread safety, we preserve the tablet distribution
information of each partition
+ // before calling getOrSetAutoPartitionInfo, but not check the
partition first
+ List<TTabletLocation> partitionTablets = new ArrayList<>();
+ List<TTabletLocation> partitionSlaveTablets = new ArrayList<>();
AtomicLong cachedLoadTabletIdx = new AtomicLong(-1);
if (needUseCache
&&
Env.getCurrentGlobalTransactionMgr().getAutoPartitionCacheMgr()
- .getAutoPartitionInfo(txnId, partition.getId(),
partitionTablets,
+ .getAutoPartitionInfo(txnId, partitionId,
partitionTablets,
partitionSlaveTablets,
cachedLoadTabletIdx)) {
if (cacheLoadTabletIdx) {
tPartition.setLoadTabletIdx(cachedLoadTabletIdx.get());
@@ -4791,52 +4790,49 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
return result;
}
}
- int quorum =
olapTable.getPartitionInfo().getReplicaAllocation(partition.getId()).getTotalReplicaNum()
/ 2
- + 1;
- for (MaterializedIndex index :
partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
- for (Tablet tablet : index.getTablets()) {
- // we should ensure the replica backend is alive
- // otherwise, there will be a 'unknown node id, id=xxx'
error for stream load
- // BE id -> path hash
- Multimap<Long, Long> bePathsMap;
- try {
- if (tablet instanceof CloudTablet) {
- CloudTablet cloudTablet = (CloudTablet) tablet;
- if (hasBeEndpoint) {
- bePathsMap =
cloudTablet.getNormalReplicaBackendPathMap(request.be_endpoint);
- } else {
- if (cachedClusterId == null) {
- cachedClusterId =
((CloudSystemInfoService) Env.getCurrentSystemInfo())
- .getCurrentClusterId();
- }
- bePathsMap =
cloudTablet.getNormalReplicaBackendPathMapByClusterId(cachedClusterId);
- }
+ int quorum = partitionSnapshot.quorum;
+ for (Tablet tablet : partitionSnapshot.tablets) {
+ // we should ensure the replica backend is alive
+ // otherwise, there will be a 'unknown node id, id=xxx' error
for stream load
+ // BE id -> path hash
+ Multimap<Long, Long> bePathsMap;
+ try {
+ if (tablet instanceof CloudTablet) {
+ CloudTablet cloudTablet = (CloudTablet) tablet;
+ if (hasBeEndpoint) {
+ bePathsMap =
cloudTablet.getNormalReplicaBackendPathMap(request.be_endpoint);
} else {
- bePathsMap =
tablet.getNormalReplicaBackendPathMap();
+ if (cachedClusterId == null) {
+ cachedClusterId = ((CloudSystemInfoService)
Env.getCurrentSystemInfo())
+ .getCurrentClusterId();
+ }
+ bePathsMap =
cloudTablet.getNormalReplicaBackendPathMapByClusterId(cachedClusterId);
}
- } catch (UserException ex) {
-
errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
- result.setStatus(errorStatus);
- LOG.warn("send create partition error status: {}",
result);
- return result;
- }
- if (bePathsMap.keySet().size() < quorum) {
- LOG.warn("auto go quorum exception");
- }
- if (request.isSetWriteSingleReplica() &&
request.isWriteSingleReplica()) {
- Long[] nodes = bePathsMap.keySet().toArray(new
Long[0]);
- Random random = new SecureRandom();
- Long masterNode = nodes[random.nextInt(nodes.length)];
- Multimap<Long, Long> slaveBePathsMap = bePathsMap;
- slaveBePathsMap.removeAll(masterNode);
- partitionTablets.add(new
TTabletLocation(tablet.getId(),
-
Lists.newArrayList(Sets.newHashSet(masterNode))));
- partitionSlaveTablets.add(new
TTabletLocation(tablet.getId(),
- Lists.newArrayList(slaveBePathsMap.keySet())));
} else {
- partitionTablets.add(new
TTabletLocation(tablet.getId(),
- Lists.newArrayList(bePathsMap.keySet())));
+ bePathsMap = tablet.getNormalReplicaBackendPathMap();
}
+ } catch (UserException ex) {
+
errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
+ result.setStatus(errorStatus);
+ LOG.warn("send create partition error status: {}", result);
+ return result;
+ }
+ if (bePathsMap.keySet().size() < quorum) {
+ LOG.warn("auto go quorum exception");
+ }
+ if (request.isSetWriteSingleReplica() &&
request.isWriteSingleReplica()) {
+ Long[] nodes = bePathsMap.keySet().toArray(new Long[0]);
+ Random random = new SecureRandom();
+ Long masterNode = nodes[random.nextInt(nodes.length)];
+ Multimap<Long, Long> slaveBePathsMap = bePathsMap;
+ slaveBePathsMap.removeAll(masterNode);
+ partitionTablets.add(new TTabletLocation(tablet.getId(),
+ Lists.newArrayList(Sets.newHashSet(masterNode))));
+ partitionSlaveTablets.add(new
TTabletLocation(tablet.getId(),
+ Lists.newArrayList(slaveBePathsMap.keySet())));
+ } else {
+ partitionTablets.add(new TTabletLocation(tablet.getId(),
+ Lists.newArrayList(bePathsMap.keySet())));
}
}
@@ -4864,7 +4860,7 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
if (needUseCache) {
long loadTabletIdx = cacheLoadTabletIdx ?
tPartition.getLoadTabletIdx() : -1;
long cachedTabletIdx =
Env.getCurrentGlobalTransactionMgr().getAutoPartitionCacheMgr()
- .getOrSetAutoPartitionInfo(txnId, partition.getId(),
partitionTablets,
+ .getOrSetAutoPartitionInfo(txnId, partitionId,
partitionTablets,
partitionSlaveTablets, loadTabletIdx);
if (cacheLoadTabletIdx) {
tPartition.setLoadTabletIdx(cachedTabletIdx);
@@ -4987,6 +4983,14 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
}
}
+ Backend requestBackend = request.isSetBeEndpoint() ?
resolveBeEndpoint(request.getBeEndpoint()) : null;
+ long adaptiveBucketBeId = requestBackend != null ?
requestBackend.getId() : -1L;
+ TUniqueId queryId = request.isSetQueryId() ? request.getQueryId() :
null;
+ boolean enableAdaptiveRandomBucket =
request.isSetEnableAdaptiveRandomBucket()
+ && request.isEnableAdaptiveRandomBucket();
+ boolean loadToSingleTablet = request.isSetLoadToSingleTablet() &&
request.isLoadToSingleTablet();
+ final boolean replaceHasBeEndpoint = request.isSetBeEndpoint();
+
InsertOverwriteManager overwriteManager =
Env.getCurrentEnv().getInsertOverwriteManager();
ReentrantLock taskLock = overwriteManager.getLock(taskGroupId);
if (taskLock == null) {
@@ -5001,6 +5005,7 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
ArrayList<Long> pendingPartitionIds = new ArrayList<>(); // pending:
[1 2]
ArrayList<Long> newPartitionIds = new ArrayList<>(); // requested temp
partition ids. for [7 8]
boolean needReplace = false;
+ List<PartitionResultSnapshot> partitionSnapshots = new ArrayList<>();
try {
taskLock.lock();
// double check lock. maybe taskLock is not null, but has been
removed from the Map. means the task failed.
@@ -5048,29 +5053,37 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
}
}
}
- } catch (DdlException | RuntimeException ex) {
- errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
- result.setStatus(errorStatus);
- LOG.warn("send create partition error status: {}", result);
- return result;
- } finally {
- taskLock.unlock();
- }
- // result: [1 2 5 6], make it [7 8 5 6]
- int idx = 0;
- if (needReplace) {
- for (int i = 0; i < reqPartitionIds.size(); i++) {
- if (reqPartitionIds.get(i).equals(resultPartitionIds.get(i))) {
- resultPartitionIds.set(i, newPartitionIds.get(idx++));
+ // result: [1 2 5 6], make it [7 8 5 6]
+ int idx = 0;
+ if (needReplace) {
+ for (int i = 0; i < reqPartitionIds.size(); i++) {
+ if
(reqPartitionIds.get(i).equals(resultPartitionIds.get(i))) {
+ resultPartitionIds.set(i, newPartitionIds.get(idx++));
+ }
}
}
- }
- if (idx != newPartitionIds.size()) {
- errorStatus.addToErrorMsgs("changed partition number " + idx + "
is not correct");
+ if (idx != newPartitionIds.size()) {
+ errorStatus.addToErrorMsgs("changed partition number " + idx +
" is not correct");
+ result.setStatus(errorStatus);
+ LOG.warn("send create partition error status: {}", result);
+ return result;
+ }
+
+ olapTable.readLock();
+ try {
+
partitionSnapshots.addAll(snapshotPartitionResultsById(olapTable,
resultPartitionIds,
+ loadToSingleTablet, enableAdaptiveRandomBucket,
"replace partition"));
+ } finally {
+ olapTable.readUnlock();
+ }
+ } catch (UserException | RuntimeException ex) {
+ errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
result.setStatus(errorStatus);
LOG.warn("send create partition error status: {}", result);
return result;
+ } finally {
+ taskLock.unlock();
}
if (LOG.isDebugEnabled()) {
@@ -5087,51 +5100,23 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
List<TOlapTablePartition> partitions = new ArrayList<>();
List<TTabletLocation> tablets = new ArrayList<>();
List<TTabletLocation> slaveTablets = new ArrayList<>();
- PartitionInfo partitionInfo = olapTable.getPartitionInfo();
- Backend requestBackend = request.isSetBeEndpoint() ?
resolveBeEndpoint(request.getBeEndpoint()) : null;
- long adaptiveBucketBeId = requestBackend != null ?
requestBackend.getId() : -1L;
- TUniqueId queryId = request.isSetQueryId() ? request.getQueryId() :
null;
- boolean enableAdaptiveRandomBucket =
request.isSetEnableAdaptiveRandomBucket()
- && request.isEnableAdaptiveRandomBucket();
- boolean loadToSingleTablet = request.isSetLoadToSingleTablet() &&
request.isLoadToSingleTablet();
- final boolean replaceHasBeEndpoint = request.isSetBeEndpoint();
// Lazy: resolved on the first CloudTablet that needs it.
String replaceCachedClusterId = null;
- for (long partitionId : resultPartitionIds) {
- Partition partition = olapTable.getPartition(partitionId);
+ for (PartitionResultSnapshot partitionSnapshot : partitionSnapshots) {
+ Partition partition = partitionSnapshot.partition;
+ long partitionId = partitionSnapshot.partitionId;
+ TOlapTablePartition tPartition = partitionSnapshot.tPartition;
+ boolean cacheLoadTabletIdx = partitionSnapshot.cacheLoadTabletIdx;
+ partitions.add(tPartition);
// For thread safety, we preserve the tablet distribution
information of each partition
// before calling getOrSetAutoPartitionInfo, but not check the
partition first
List<TTabletLocation> partitionTablets = new ArrayList<>();
List<TTabletLocation> partitionSlaveTablets = new ArrayList<>();
- TOlapTablePartition tPartition = new TOlapTablePartition();
- tPartition.setId(partition.getId());
-
- // set partition keys
- int partColNum = partitionInfo.getPartitionColumns().size();
- try {
- OlapTableSink.setPartitionKeys(tPartition,
partitionInfo.getItem(partition.getId()), partColNum);
- } catch (UserException ex) {
- errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
- result.setStatus(errorStatus);
- LOG.warn("send replace partition error status: {}", result);
- return result;
- }
- for (MaterializedIndex index :
partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
- tPartition.addToIndexes(new
TOlapTableIndexTablets(index.getId(), Lists.newArrayList(
-
index.getTablets().stream().map(Tablet::getId).collect(Collectors.toList()))));
- tPartition.setNumBuckets(index.getTablets().size());
- }
-
tPartition.setIsMutable(olapTable.getPartitionInfo().getIsMutable(partition.getId()));
- boolean randomDistribution =
- partition.getDistributionInfo().getType() ==
DistributionInfo.DistributionInfoType.RANDOM;
- boolean cacheLoadTabletIdx =
- (loadToSingleTablet || enableAdaptiveRandomBucket) &&
randomDistribution;
- partitions.add(tPartition);
// tablet
AtomicLong cachedLoadTabletIdx = new AtomicLong(-1);
if (needUseCache && txnId != 0
&&
Env.getCurrentGlobalTransactionMgr().getAutoPartitionCacheMgr()
- .getAutoPartitionInfo(txnId, partition.getId(),
partitionTablets,
+ .getAutoPartitionInfo(txnId, partitionId,
partitionTablets,
partitionSlaveTablets,
cachedLoadTabletIdx)) {
if (cacheLoadTabletIdx) {
tPartition.setLoadTabletIdx(cachedLoadTabletIdx.get());
@@ -5155,53 +5140,50 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
return result;
}
}
- int quorum =
olapTable.getPartitionInfo().getReplicaAllocation(partition.getId()).getTotalReplicaNum()
/ 2
- + 1;
- for (MaterializedIndex index :
partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
- for (Tablet tablet : index.getTablets()) {
- // we should ensure the replica backend is alive
- // otherwise, there will be a 'unknown node id, id=xxx'
error for stream load
- // BE id -> path hash
- Multimap<Long, Long> bePathsMap;
- try {
- if (tablet instanceof CloudTablet) {
- CloudTablet cloudTablet = (CloudTablet) tablet;
- if (replaceHasBeEndpoint) {
- bePathsMap =
cloudTablet.getNormalReplicaBackendPathMap(request.be_endpoint);
- } else {
- if (replaceCachedClusterId == null) {
- replaceCachedClusterId =
((CloudSystemInfoService) Env.getCurrentSystemInfo())
- .getCurrentClusterId();
- }
- bePathsMap = cloudTablet
-
.getNormalReplicaBackendPathMapByClusterId(replaceCachedClusterId);
- }
+ int quorum = partitionSnapshot.quorum;
+ for (Tablet tablet : partitionSnapshot.tablets) {
+ // we should ensure the replica backend is alive
+ // otherwise, there will be a 'unknown node id, id=xxx' error
for stream load
+ // BE id -> path hash
+ Multimap<Long, Long> bePathsMap;
+ try {
+ if (tablet instanceof CloudTablet) {
+ CloudTablet cloudTablet = (CloudTablet) tablet;
+ if (replaceHasBeEndpoint) {
+ bePathsMap =
cloudTablet.getNormalReplicaBackendPathMap(request.be_endpoint);
} else {
- bePathsMap =
tablet.getNormalReplicaBackendPathMap();
+ if (replaceCachedClusterId == null) {
+ replaceCachedClusterId =
((CloudSystemInfoService) Env.getCurrentSystemInfo())
+ .getCurrentClusterId();
+ }
+ bePathsMap = cloudTablet
+
.getNormalReplicaBackendPathMapByClusterId(replaceCachedClusterId);
}
- } catch (UserException ex) {
-
errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
- result.setStatus(errorStatus);
- LOG.warn("send replace partition error status: {}",
result);
- return result;
- }
- if (bePathsMap.keySet().size() < quorum) {
- LOG.warn("auto go quorum exception");
- }
- if (request.isSetWriteSingleReplica() &&
request.isWriteSingleReplica()) {
- Long[] nodes = bePathsMap.keySet().toArray(new
Long[0]);
- Random random = new SecureRandom();
- Long masterNode = nodes[random.nextInt(nodes.length)];
- Multimap<Long, Long> slaveBePathsMap = bePathsMap;
- slaveBePathsMap.removeAll(masterNode);
- partitionTablets.add(new
TTabletLocation(tablet.getId(),
-
Lists.newArrayList(Sets.newHashSet(masterNode))));
- partitionSlaveTablets.add(new
TTabletLocation(tablet.getId(),
- Lists.newArrayList(slaveBePathsMap.keySet())));
} else {
- partitionTablets.add(new
TTabletLocation(tablet.getId(),
- Lists.newArrayList(bePathsMap.keySet())));
+ bePathsMap = tablet.getNormalReplicaBackendPathMap();
}
+ } catch (UserException ex) {
+
errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage()));
+ result.setStatus(errorStatus);
+ LOG.warn("send replace partition error status: {}",
result);
+ return result;
+ }
+ if (bePathsMap.keySet().size() < quorum) {
+ LOG.warn("auto go quorum exception");
+ }
+ if (request.isSetWriteSingleReplica() &&
request.isWriteSingleReplica()) {
+ Long[] nodes = bePathsMap.keySet().toArray(new Long[0]);
+ Random random = new SecureRandom();
+ Long masterNode = nodes[random.nextInt(nodes.length)];
+ Multimap<Long, Long> slaveBePathsMap = bePathsMap;
+ slaveBePathsMap.removeAll(masterNode);
+ partitionTablets.add(new TTabletLocation(tablet.getId(),
+ Lists.newArrayList(Sets.newHashSet(masterNode))));
+ partitionSlaveTablets.add(new
TTabletLocation(tablet.getId(),
+ Lists.newArrayList(slaveBePathsMap.keySet())));
+ } else {
+ partitionTablets.add(new TTabletLocation(tablet.getId(),
+ Lists.newArrayList(bePathsMap.keySet())));
}
}
@@ -5233,14 +5215,14 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
if (needUseCache) {
long loadTabletIdx = cacheLoadTabletIdx ?
tPartition.getLoadTabletIdx() : -1;
long cachedTabletIdx =
Env.getCurrentGlobalTransactionMgr().getAutoPartitionCacheMgr()
- .getOrSetAutoPartitionInfo(txnId, partition.getId(),
partitionTablets,
+ .getOrSetAutoPartitionInfo(txnId, partitionId,
partitionTablets,
partitionSlaveTablets, loadTabletIdx);
if (cacheLoadTabletIdx) {
tPartition.setLoadTabletIdx(cachedTabletIdx);
}
if (LOG.isDebugEnabled()) {
LOG.debug("Cache auto partition info, txnId: {},
partitionId: {}, "
- + "tablets: {}, slaveTablets: {}", txnId,
partition.getId(),
+ + "tablets: {}, slaveTablets: {}", txnId,
partitionId,
partitionTablets.size(),
partitionSlaveTablets.size());
}
}
@@ -5272,6 +5254,94 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
return result;
}
+ private static final class PartitionResultSnapshot {
+ private final Partition partition;
+ private final long partitionId;
+ private final TOlapTablePartition tPartition;
+ private final List<Tablet> tablets;
+ private final int quorum;
+ private final boolean cacheLoadTabletIdx;
+
+ private PartitionResultSnapshot(Partition partition, long partitionId,
+ TOlapTablePartition tPartition, List<Tablet> tablets, int
quorum, boolean cacheLoadTabletIdx) {
+ this.partition = partition;
+ this.partitionId = partitionId;
+ this.tPartition = tPartition;
+ this.tablets = tablets;
+ this.quorum = quorum;
+ this.cacheLoadTabletIdx = cacheLoadTabletIdx;
+ }
+ }
+
+ private static List<PartitionResultSnapshot>
snapshotPartitionResultsByName(OlapTable olapTable,
+ Collection<String> partitionNames, boolean loadToSingleTablet,
boolean enableAdaptiveRandomBucket,
+ String resultName) throws UserException {
+ PartitionInfo partitionInfo = olapTable.getPartitionInfo();
+ int partColNum = partitionInfo.getPartitionColumns().size();
+ List<PartitionResultSnapshot> partitionSnapshots = new ArrayList<>();
+ for (String partitionName : partitionNames) {
+ Partition partition = olapTable.getPartition(partitionName);
+ if (partition == null) {
+ throw new UserException(String.format(
+ "partition %s was dropped concurrently while building
%s result, please retry",
+ partitionName, resultName));
+ }
+ partitionSnapshots.add(snapshotPartitionResult(partitionInfo,
partColNum, partition, partitionName,
+ loadToSingleTablet, enableAdaptiveRandomBucket,
resultName));
+ }
+ return partitionSnapshots;
+ }
+
+ private static List<PartitionResultSnapshot>
snapshotPartitionResultsById(OlapTable olapTable,
+ Collection<Long> partitionIds, boolean loadToSingleTablet, boolean
enableAdaptiveRandomBucket,
+ String resultName) throws UserException {
+ PartitionInfo partitionInfo = olapTable.getPartitionInfo();
+ int partColNum = partitionInfo.getPartitionColumns().size();
+ List<PartitionResultSnapshot> partitionSnapshots = new ArrayList<>();
+ for (long partitionId : partitionIds) {
+ Partition partition = olapTable.getPartition(partitionId);
+ if (partition == null) {
+ throw new UserException(String.format(
+ "partition %d was dropped concurrently while building
%s result, please retry",
+ partitionId, resultName));
+ }
+ partitionSnapshots.add(snapshotPartitionResult(partitionInfo,
partColNum, partition,
+ String.valueOf(partitionId), loadToSingleTablet,
enableAdaptiveRandomBucket, resultName));
+ }
+ return partitionSnapshots;
+ }
+
+ private static PartitionResultSnapshot
snapshotPartitionResult(PartitionInfo partitionInfo, int partColNum,
+ Partition partition, String partitionLabel, boolean
loadToSingleTablet,
+ boolean enableAdaptiveRandomBucket, String resultName) throws
UserException {
+ long partitionId = partition.getId();
+ PartitionItem partitionItem = partitionInfo.getItem(partitionId);
+ if (partitionItem == null) {
+ throw new UserException(String.format(
+ "partition item of %s was dropped concurrently while
building %s result, please retry",
+ partitionLabel, resultName));
+ }
+
+ TOlapTablePartition tPartition = new TOlapTablePartition();
+ tPartition.setId(partitionId);
+ OlapTableSink.setPartitionKeys(tPartition, partitionItem, partColNum);
+ List<Tablet> partitionTabletSnapshot = new ArrayList<>();
+ for (MaterializedIndex index :
partition.getMaterializedIndices(MaterializedIndex.IndexExtState.ALL)) {
+ List<Tablet> indexTablets = new ArrayList<>(index.getTablets());
+ tPartition.addToIndexes(new TOlapTableIndexTablets(index.getId(),
Lists.newArrayList(
+
indexTablets.stream().map(Tablet::getId).collect(Collectors.toList()))));
+ tPartition.setNumBuckets(indexTablets.size());
+ partitionTabletSnapshot.addAll(indexTablets);
+ }
+ tPartition.setIsMutable(partitionInfo.getIsMutable(partitionId));
+ boolean randomDistribution =
+ partition.getDistributionInfo().getType() ==
DistributionInfo.DistributionInfoType.RANDOM;
+ boolean cacheLoadTabletIdx = (loadToSingleTablet ||
enableAdaptiveRandomBucket) && randomDistribution;
+ int quorum =
partitionInfo.getReplicaAllocation(partitionId).getTotalReplicaNum() / 2 + 1;
+ return new PartitionResultSnapshot(partition, partitionId, tPartition,
partitionTabletSnapshot, quorum,
+ cacheLoadTabletIdx);
+ }
+
public TGetMetaResult getMeta(TGetMetaRequest request) throws TException {
String clientAddr = getClientAddrAsString();
if (LOG.isDebugEnabled()) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
index c017cd71bc2..87cb3f8e2c3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
@@ -88,6 +88,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
public class FrontendServiceImplTest {
@@ -372,6 +373,54 @@ public class FrontendServiceImplTest {
Assert.assertNotNull(p20230807);
}
+ @Test
+ public void
testCreatePartitionReturnsRetryErrorWhenResultPartitionIsMissing() throws
Exception {
+ String createOlapTblStmt = "CREATE TABLE
test.partition_dropped_before_result_snapshot(\n"
+ + " event_day DATETIME NOT NULL,\n"
+ + " site_id INT\n"
+ + ")\n"
+ + "DUPLICATE KEY(event_day, site_id)\n"
+ + "AUTO PARTITION BY RANGE (date_trunc(event_day, 'day')) ()\n"
+ + "DISTRIBUTED BY HASH(event_day) BUCKETS 1\n"
+ + "PROPERTIES(\"replication_num\" = \"1\");";
+ createTable(createOlapTblStmt);
+
+ Database db =
Env.getCurrentInternalCatalog().getDbOrAnalysisException("test");
+ OlapTable table = (OlapTable)
db.getTableOrAnalysisException("partition_dropped_before_result_snapshot");
+ OlapTable spyTable = Mockito.spy(table);
+ String partitionName = "p20230808000000";
+ AtomicBoolean hideResultPartition = new AtomicBoolean(true);
+ Mockito.doAnswer(invocation -> {
+ Partition partition = (Partition) invocation.callRealMethod();
+ // Model the lookup result after a concurrent retention drop
without timing-dependent test threads.
+ if (partition != null && hideResultPartition.compareAndSet(true,
false)) {
+ return null;
+ }
+ return partition;
+ }).when(spyTable).getPartition(partitionName);
+
+ db.unregisterTable(table.getName());
+ db.registerTable(spyTable);
+ try {
+ TNullableStringLiteral start = new TNullableStringLiteral();
+ start.setValue("2023-08-08 00:00:00");
+ TCreatePartitionRequest request = new TCreatePartitionRequest();
+ request.setDbId(db.getId());
+ request.setTableId(spyTable.getId());
+
request.setPartitionValues(Collections.singletonList(Collections.singletonList(start)));
+
+ TCreatePartitionResult result = new
FrontendServiceImpl(exeEnv).createPartition(request);
+
+ Assert.assertEquals(TStatusCode.RUNTIME_ERROR,
result.getStatus().getStatusCode());
+ Assert.assertTrue(result.getStatus().getErrorMsgs().get(0)
+ .contains("was dropped concurrently while building auto
partition result, please retry"));
+ Assert.assertFalse(hideResultPartition.get());
+ } finally {
+ db.unregisterTable(spyTable.getName());
+ db.registerTable(table);
+ }
+ }
+
@Test
public void testCreatePartitionList() throws Exception {
String createOlapTblStmt = new String("CREATE TABLE
test.partition_list(\n"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
