This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 532766764ea [Fix](Export) Fix the NPE exception when cancel an export
job #29913
532766764ea is described below
commit 532766764ea1b802cb620988cb89b4ce3bc47699
Author: Tiewei Fang <[email protected]>
AuthorDate: Fri Jan 12 16:30:58 2024 +0800
[Fix](Export) Fix the NPE exception when cancel an export job #29913
---
.../main/java/org/apache/doris/load/ExportJob.java | 27 ++++++++++++++--------
.../org/apache/doris/task/ExportExportingTask.java | 3 ++-
2 files changed, 20 insertions(+), 10 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
index 8521034f82e..0585311a1ab 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
@@ -163,9 +163,7 @@ public class ExportJob implements Writable {
private Integer parallelNum;
- public Map<String, Long> getPartitionToVersion() {
- return partitionToVersion;
- }
+ private Collection<Partition> partitionList = new ArrayList<Partition>();
private Map<String, Long> partitionToVersion = Maps.newHashMap();
@@ -314,7 +312,6 @@ public class ExportJob implements Writable {
List<Long> tabletIdList = Lists.newArrayList();
table.readLock();
try {
- Collection<Partition> partitions = new ArrayList<Partition>();
// get partitions
// user specifies partitions, already checked in ExportStmt
if (this.partitionNames != null) {
@@ -323,19 +320,18 @@ public class ExportJob implements Writable {
+ " of partitions allowed by a export job");
}
for (String partName : this.partitionNames) {
- partitions.add(table.getPartition(partName));
+ partitionList.add(table.getPartition(partName));
}
} else {
if (table.getPartitions().size() >
Config.maximum_number_of_export_partitions) {
throw new UserException("The partitions number of this
export job is larger than the maximum number"
+ " of partitions allowed by a export job");
}
- partitions = table.getPartitions();
+ partitionList = table.getPartitions();
}
// get tablets
- for (Partition partition : partitions) {
- partitionToVersion.put(partition.getName(),
partition.getVisibleVersion());
+ for (Partition partition : partitionList) {
for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE)) {
tabletIdList.addAll(index.getTabletIdsInOrder());
}
@@ -525,6 +521,15 @@ public class ExportJob implements Writable {
this.outfileInfo = outfileInfo;
}
+ public synchronized Map<String, Long> getPartitionToVersion() {
+ if (partitionToVersion.isEmpty()) {
+ // get version of partitions
+ for (Partition partition : partitionList) {
+ partitionToVersion.put(partition.getName(),
partition.getVisibleVersion());
+ }
+ }
+ return partitionToVersion;
+ }
public synchronized Thread getDoExportingThread() {
return doExportingThread;
@@ -582,7 +587,11 @@ public class ExportJob implements Writable {
// maybe user cancel this job
if (task != null && state == JobState.EXPORTING && stmtExecutorList !=
null) {
for (int idx = 0; idx < stmtExecutorList.size(); ++idx) {
- stmtExecutorList.get(idx).cancel();
+ // because a exporting task may be cancelled due to a load
operation,
+ // then it's StmtExecutor is null
+ if (stmtExecutorList.get(idx) != null) {
+ stmtExecutorList.get(idx).cancel();
+ }
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/task/ExportExportingTask.java
b/fe/fe-core/src/main/java/org/apache/doris/task/ExportExportingTask.java
index 142cfddd707..d5a18f19d56 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/ExportExportingTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/ExportExportingTask.java
@@ -130,6 +130,7 @@ public class ExportExportingTask extends MasterTask {
job.getTableName().getDb());
OlapTable table =
db.getOlapTableOrAnalysisException(job.getTableName().getTbl());
table.readLock();
+ Map<String, Long> partitionToVersion =
job.getPartitionToVersion();
try {
SelectStmt selectStmt = selectStmtList.get(idx);
List<Long> tabletIds =
selectStmt.getTableRefs().get(0).getSampleTabletIds();
@@ -138,7 +139,7 @@ public class ExportExportingTask extends MasterTask {
tabletId);
Partition partition =
table.getPartition(tabletMeta.getPartitionId());
long nowVersion = partition.getVisibleVersion();
- long oldVersion =
job.getPartitionToVersion().get(partition.getName());
+ long oldVersion =
partitionToVersion.get(partition.getName());
if (nowVersion != oldVersion) {
LOG.warn("Tablet {} has changed version, old
version = {}, now version = {}",
tabletId, oldVersion, nowVersion);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
