This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1e3b0d3 [Rollup] Change table's state right after all rollup jobs are
done (#2904)
1e3b0d3 is described below
commit 1e3b0d31ea937ff119ad0929247ca4d02665c5a4
Author: Mingyu Chen <[email protected]>
AuthorDate: Fri Feb 14 21:28:51 2020 +0800
[Rollup] Change table's state right after all rollup jobs are done (#2904)
In the current implementation, the state of the table will be set until the
next round of job scheduling. So there may be tens of seconds between job
completion and table state changes to NORMAL.
And also, I made the synchronized range smaller by replacing the
synchronized methods with synchronized blocks, which may solve the problem
described in #2903
---
.../java/org/apache/doris/alter/AlterHandler.java | 3 +-
.../java/org/apache/doris/alter/AlterJobV2.java | 6 +-
.../doris/alter/MaterializedViewHandler.java | 141 ++++++++++++++-------
.../java/org/apache/doris/alter/RollupJobV2.java | 2 +-
.../org/apache/doris/analysis/GroupingInfo.java | 7 +-
.../org/apache/doris/journal/JournalEntity.java | 2 +-
.../java/org/apache/doris/persist/EditLog.java | 4 +-
.../org/apache/doris/persist/OperationType.java | 2 +-
.../org/apache/doris/alter/BatchRollupJobTest.java | 101 +++++++++++++++
.../org/apache/doris/utframe/UtFrameUtils.java | 66 ++++++++++
10 files changed, 277 insertions(+), 57 deletions(-)
diff --git a/fe/src/main/java/org/apache/doris/alter/AlterHandler.java
b/fe/src/main/java/org/apache/doris/alter/AlterHandler.java
index f874fab..9e630c2 100644
--- a/fe/src/main/java/org/apache/doris/alter/AlterHandler.java
+++ b/fe/src/main/java/org/apache/doris/alter/AlterHandler.java
@@ -318,11 +318,12 @@ public abstract class AlterHandler extends MasterDaemon {
@Override
protected void runAfterCatalogReady() {
+ long curTime = System.currentTimeMillis();
// clean history job
Iterator<AlterJob> iter = finishedOrCancelledAlterJobs.iterator();
while (iter.hasNext()) {
AlterJob historyJob = iter.next();
- if ((System.currentTimeMillis() - historyJob.getCreateTimeMs()) /
1000 > Config.history_job_keep_max_second) {
+ if ((curTime - historyJob.getCreateTimeMs()) / 1000 >
Config.history_job_keep_max_second) {
iter.remove();
LOG.info("remove history {} job[{}]. created at {}",
historyJob.getType(),
historyJob.getTableId(),
TimeUtils.longToTimeString(historyJob.getCreateTimeMs()));
diff --git a/fe/src/main/java/org/apache/doris/alter/AlterJobV2.java
b/fe/src/main/java/org/apache/doris/alter/AlterJobV2.java
index 709c040..df64eb5 100644
--- a/fe/src/main/java/org/apache/doris/alter/AlterJobV2.java
+++ b/fe/src/main/java/org/apache/doris/alter/AlterJobV2.java
@@ -107,7 +107,7 @@ public abstract class AlterJobV2 implements Writable {
return tableName;
}
- private boolean isTimeout() {
+ public boolean isTimeout() {
return System.currentTimeMillis() - createTimeMs > timeoutMs;
}
@@ -115,6 +115,10 @@ public abstract class AlterJobV2 implements Writable {
return jobState.isFinalState();
}
+ public long getFinishedTimeMs() {
+ return finishedTimeMs;
+ }
+
/*
* The keyword 'synchronized' only protects 2 methods:
* run() and cancel()
diff --git
a/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java
b/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java
index 1d8c809..e4eaabb 100644
--- a/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java
+++ b/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java
@@ -18,11 +18,11 @@
package org.apache.doris.alter;
import org.apache.doris.alter.AlterJob.JobState;
-import org.apache.doris.analysis.CreateMaterializedViewStmt;
import org.apache.doris.analysis.AddRollupClause;
import org.apache.doris.analysis.AlterClause;
import org.apache.doris.analysis.CancelAlterTableStmt;
import org.apache.doris.analysis.CancelStmt;
+import org.apache.doris.analysis.CreateMaterializedViewStmt;
import org.apache.doris.analysis.DropRollupClause;
import org.apache.doris.analysis.MVColumnItem;
import org.apache.doris.catalog.AggregateType;
@@ -54,13 +54,13 @@ import org.apache.doris.persist.DropInfo;
import org.apache.doris.persist.EditLog;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TStorageFormat;
+import org.apache.doris.thrift.TStorageMedium;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
-import org.apache.doris.thrift.TStorageMedium;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -90,58 +90,69 @@ public class MaterializedViewHandler extends AlterHandler {
super("materialized view");
}
-
// for batch submit rollup job, tableId -> jobId
// keep table's not final state job size. The job size determine's table's
state, = 0 means table is normal, otherwise is rollup
private Map<Long, Set<Long>> tableNotFinalStateJobMap = new
ConcurrentHashMap<>();
// keep table's running job,used for concurrency limit
+ // table id -> set of running job ids
private Map<Long, Set<Long>> tableRunningJobMap = new
ConcurrentHashMap<>();
- public synchronized void addAlterJobV2(AlterJobV2 alterJob) {
+ public void addAlterJobV2(AlterJobV2 alterJob) {
super.addAlterJobV2(alterJob);
addAlterJobV2ToTableNotFinalStateJobMap(alterJob);
}
- protected synchronized void batchAddAlterJobV2(List<AlterJobV2>
alterJobV2List) {
+ protected void batchAddAlterJobV2(List<AlterJobV2> alterJobV2List) {
for (AlterJobV2 alterJobV2 : alterJobV2List) {
addAlterJobV2(alterJobV2);
}
}
- private void addAlterJobV2ToTableNotFinalStateJobMap(AlterJobV2
alterJobV2) {
+ // return true iff job is actually added this time
+ private boolean addAlterJobV2ToTableNotFinalStateJobMap(AlterJobV2
alterJobV2) {
if (alterJobV2.isDone()) {
LOG.warn("try to add a final job({}) to a unfinal set",
alterJobV2.getJobId());
- return;
+ return false;
}
+
Long tableId = alterJobV2.getTableId();
Long jobId = alterJobV2.getJobId();
- Set<Long> tableNotFinalStateJobIdSet =
tableNotFinalStateJobMap.get(tableId);
- if (tableNotFinalStateJobIdSet == null) {
- tableNotFinalStateJobIdSet = new HashSet<>();
- tableNotFinalStateJobMap.put(tableId, tableNotFinalStateJobIdSet);
+
+ synchronized (tableNotFinalStateJobMap) {
+ Set<Long> tableNotFinalStateJobIdSet =
tableNotFinalStateJobMap.get(tableId);
+ if (tableNotFinalStateJobIdSet == null) {
+ tableNotFinalStateJobIdSet = new HashSet<>();
+ tableNotFinalStateJobMap.put(tableId,
tableNotFinalStateJobIdSet);
+ }
+ return tableNotFinalStateJobIdSet.add(jobId);
}
- tableNotFinalStateJobIdSet.add(jobId);
}
/**
*
* @param alterJobV2
- * @return true current table doesn't have not not final rollup
job,table'state is normal
- * false table status is rollup
+ * @return true iif we really removed a job from tableNotFinalStateJobMap,
+ * and there is no running job of this table
+ * false otherwise.
*/
private boolean removeAlterJobV2FromTableNotFinalStateJobMap(AlterJobV2
alterJobV2) {
Long tableId = alterJobV2.getTableId();
Long jobId = alterJobV2.getJobId();
- Set<Long> tableNotFinalStateJobIdset =
tableNotFinalStateJobMap.get(tableId);
- if (tableNotFinalStateJobIdset == null) {
- return true;
- }
- tableNotFinalStateJobIdset.remove(jobId);
- if (tableNotFinalStateJobIdset.size() == 0) {
- tableNotFinalStateJobMap.remove(tableId);
- return true;
+
+ synchronized (tableNotFinalStateJobMap) {
+ Set<Long> tableNotFinalStateJobIdset =
tableNotFinalStateJobMap.get(tableId);
+ if (tableNotFinalStateJobIdset == null) {
+ // This could happen when this job is already removed before.
+ // return false, so that we will not set table's to NORMAL
again.
+ return false;
+ }
+ tableNotFinalStateJobIdset.remove(jobId);
+ if (tableNotFinalStateJobIdset.size() == 0) {
+ tableNotFinalStateJobMap.remove(tableId);
+ return true;
+ }
+ return false;
}
- return false;
}
/**
@@ -260,12 +271,17 @@ public class MaterializedViewHandler extends AlterHandler
{
throw e;
}
+ // set table' state to ROLLUP before adding rollup jobs.
+ // so that when the AlterHandler thread run the jobs, it will see the
expected table's state.
+ // ATTN: This order is not mandatory, because database lock will
protect us,
+ // but this order is more reasonable
+ olapTable.setState(OlapTableState.ROLLUP);
+
// 2 batch submit rollup job
List<AlterJobV2> rollupJobV2List = new
ArrayList<>(rollupNameJobMap.values());
batchAddAlterJobV2(rollupJobV2List);
- BatchAlterJobPersistInfo batchAlterJobV2 = new
BatchAlterJobPersistInfo(rollupJobV2List);
- olapTable.setState(OlapTableState.ROLLUP);
+ BatchAlterJobPersistInfo batchAlterJobV2 = new
BatchAlterJobPersistInfo(rollupJobV2List);
Catalog.getCurrentCatalog().getEditLog().logBatchAlterJob(batchAlterJobV2);
LOG.info("finished to create materialized view job: {}", logJobIdSet);
}
@@ -720,16 +736,18 @@ public class MaterializedViewHandler extends AlterHandler
{
runAlterJobV2();
}
- private synchronized Map<Long, AlterJobV2> getAlterJobsCopy () {
+ private Map<Long, AlterJobV2> getAlterJobsCopy() {
return new HashMap<>(alterJobsV2);
}
private void removeJobFromRunningQueue(RollupJobV2 rollupJobV2) {
- Set<Long> runningJobIdSet =
tableRunningJobMap.get(rollupJobV2.getTableId());
- if (runningJobIdSet != null) {
- runningJobIdSet.remove(rollupJobV2.getJobId());
- if (runningJobIdSet.size() == 0) {
- tableRunningJobMap.remove(rollupJobV2.getTableId());
+ synchronized (tableRunningJobMap) {
+ Set<Long> runningJobIdSet =
tableRunningJobMap.get(rollupJobV2.getTableId());
+ if (runningJobIdSet != null) {
+ runningJobIdSet.remove(rollupJobV2.getJobId());
+ if (runningJobIdSet.size() == 0) {
+ tableRunningJobMap.remove(rollupJobV2.getTableId());
+ }
}
}
}
@@ -749,14 +767,14 @@ public class MaterializedViewHandler extends AlterHandler
{
}
// replay the alter job v2
+ @Override
public void replayAlterJobV2(AlterJobV2 alterJob) {
super.replayAlterJobV2(alterJob);
if (!alterJob.isDone()) {
addAlterJobV2ToTableNotFinalStateJobMap(alterJob);
changeTableStatus(alterJob.getDbId(), alterJob.getTableId(),
OlapTableState.ROLLUP);
} else {
- boolean tableIsNormal =
removeAlterJobV2FromTableNotFinalStateJobMap(alterJob);
- if (tableIsNormal) {
+ if (removeAlterJobV2FromTableNotFinalStateJobMap(alterJob)) {
changeTableStatus(alterJob.getDbId(), alterJob.getTableId(),
OlapTableState.NORMAL);
}
}
@@ -766,18 +784,43 @@ public class MaterializedViewHandler extends AlterHandler
{
* create tablet and alter tablet in be is thread safe,so we can run
rollup job for one table concurrently
*/
private void runAlterJobWithConcurrencyLimit(RollupJobV2 rollupJobV2) {
- Set<Long> tableRunningJobSet =
tableRunningJobMap.get(rollupJobV2.getTableId());
- if (tableRunningJobSet == null) {
- tableRunningJobSet = new HashSet<>();
- tableRunningJobMap.put(rollupJobV2.getTableId(),
tableRunningJobSet);
+ if (rollupJobV2.isDone()) {
+ return;
}
- // current job is already in running
- if (tableRunningJobSet.contains(rollupJobV2.getJobId())) {
+ if (rollupJobV2.isTimeout()) {
+ // in run(), the timeout job will be cancelled.
rollupJobV2.run();
- } else if (tableRunningJobSet.size() <
Config.max_running_rollup_job_num_per_table) {
- // add current job to running queue
- tableRunningJobSet.add(rollupJobV2.getJobId());
+ return;
+ }
+
+ // check if rollup job can be run within limitation.
+ long tblId = rollupJobV2.getTableId();
+ long jobId = rollupJobV2.getJobId();
+ boolean shouldJobRun = false;
+ synchronized (tableRunningJobMap) {
+ Set<Long> tableRunningJobSet = tableRunningJobMap.get(tblId);
+ if (tableRunningJobSet == null) {
+ tableRunningJobSet = new HashSet<>();
+ tableRunningJobMap.put(tblId, tableRunningJobSet);
+ }
+
+ // current job is already in running
+ if (tableRunningJobSet.contains(jobId)) {
+ shouldJobRun = true;
+ } else if (tableRunningJobSet.size() <
Config.max_running_rollup_job_num_per_table) {
+ // add current job to running queue
+ tableRunningJobSet.add(jobId);
+ shouldJobRun = true;
+ } else {
+ LOG.debug("number of running alter job {} in table {} exceed
limit {}. job {} is suspended",
+ tableRunningJobSet.size(), rollupJobV2.getTableId(),
+ Config.max_running_rollup_job_num_per_table,
rollupJobV2.getJobId());
+ shouldJobRun = false;
+ }
+ }
+
+ if (shouldJobRun) {
rollupJobV2.run();
}
}
@@ -787,17 +830,21 @@ public class MaterializedViewHandler extends AlterHandler
{
while (iterator.hasNext()) {
Map.Entry<Long, AlterJobV2> entry = iterator.next();
RollupJobV2 alterJob = (RollupJobV2)entry.getValue();
+ // run alter job
+ runAlterJobWithConcurrencyLimit(alterJob);
+ // the following check should be right after job's running, so
that the table's state
+ // can be changed to NORMAL immediately after the last alter job
of the table is done.
+ //
+ // ATTN(cmy): there is still a short gap between "job finish" and
"table become normal",
+ // so if user send next alter job right after the "job finish",
+ // it may encounter "table's state not NORMAL" error.
if (alterJob.isDone()) {
removeJobFromRunningQueue(alterJob);
-
- boolean tableIsNormal =
removeAlterJobV2FromTableNotFinalStateJobMap(alterJob);
- if (tableIsNormal) {
+ if (removeAlterJobV2FromTableNotFinalStateJobMap(alterJob)) {
changeTableStatus(alterJob.getDbId(),
alterJob.getTableId(), OlapTableState.NORMAL);
}
continue;
}
- // run alter job
- runAlterJobWithConcurrencyLimit(alterJob);
}
}
diff --git a/fe/src/main/java/org/apache/doris/alter/RollupJobV2.java
b/fe/src/main/java/org/apache/doris/alter/RollupJobV2.java
index 8c78135..d7ea4e2 100644
--- a/fe/src/main/java/org/apache/doris/alter/RollupJobV2.java
+++ b/fe/src/main/java/org/apache/doris/alter/RollupJobV2.java
@@ -42,10 +42,10 @@ import org.apache.doris.task.AgentTaskExecutor;
import org.apache.doris.task.AgentTaskQueue;
import org.apache.doris.task.AlterReplicaTask;
import org.apache.doris.task.CreateReplicaTask;
+import org.apache.doris.thrift.TStorageFormat;
import org.apache.doris.thrift.TStorageMedium;
import org.apache.doris.thrift.TStorageType;
import org.apache.doris.thrift.TTaskType;
-import org.apache.doris.thrift.TStorageFormat;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
diff --git a/fe/src/main/java/org/apache/doris/analysis/GroupingInfo.java
b/fe/src/main/java/org/apache/doris/analysis/GroupingInfo.java
index 391d2e6..e1e7961 100644
--- a/fe/src/main/java/org/apache/doris/analysis/GroupingInfo.java
+++ b/fe/src/main/java/org/apache/doris/analysis/GroupingInfo.java
@@ -32,6 +32,8 @@ import java.util.Set;
import java.util.stream.Collectors;
public class GroupingInfo {
+ public static final String COL_GROUPING_ID = "GROUPING_ID";
+
private VirtualSlotRef groupingIDSlot;
private TupleDescriptor virtualTuple;
private Set<VirtualSlotRef> groupingSlots;
@@ -43,8 +45,7 @@ public class GroupingInfo {
this.groupingType = groupingType;
groupingSlots = new LinkedHashSet<>();
virtualTuple =
analyzer.getDescTbl().createTupleDescriptor("VIRTUAL_TUPLE");
- String colName = "GROUPING_ID";
- groupingIDSlot = new VirtualSlotRef(colName, Type.BIGINT,
virtualTuple, new ArrayList<>());
+ groupingIDSlot = new VirtualSlotRef(COL_GROUPING_ID, Type.BIGINT,
virtualTuple, new ArrayList<>());
groupingIDSlot.analyze(analyzer);
groupingSlots.add(groupingIDSlot);
}
@@ -138,7 +139,7 @@ public class GroupingInfo {
for (BitSet bitSet : groupingIdList) {
long l = 0L;
// for all column, using for group by
- if ("GROUPING_ID".equals(slot.getColumnName())) {
+ if (slot.getColumnName().equalsIgnoreCase(COL_GROUPING_ID)) {
BitSet newBitSet = new BitSet();
for (int i = 0; i < bitSetAll.length(); ++i) {
newBitSet.set(i, bitSet.get(bitSetAll.length() - i -
1));
diff --git a/fe/src/main/java/org/apache/doris/journal/JournalEntity.java
b/fe/src/main/java/org/apache/doris/journal/JournalEntity.java
index 93f94c1..f0d57f1 100644
--- a/fe/src/main/java/org/apache/doris/journal/JournalEntity.java
+++ b/fe/src/main/java/org/apache/doris/journal/JournalEntity.java
@@ -500,7 +500,7 @@ public class JournalEntity implements Writable {
isRead = true;
break;
}
- case OperationType.OP_BATCH_ALTER_JOB_V2: {
+ case OperationType.OP_BATCH_ADD_ROLLUP: {
data = BatchAlterJobPersistInfo.read(in);
isRead = true;
break;
diff --git a/fe/src/main/java/org/apache/doris/persist/EditLog.java
b/fe/src/main/java/org/apache/doris/persist/EditLog.java
index 37fde77..03ba620 100644
--- a/fe/src/main/java/org/apache/doris/persist/EditLog.java
+++ b/fe/src/main/java/org/apache/doris/persist/EditLog.java
@@ -701,7 +701,7 @@ public class EditLog {
}
break;
}
- case OperationType.OP_BATCH_ALTER_JOB_V2: {
+ case OperationType.OP_BATCH_ADD_ROLLUP: {
BatchAlterJobPersistInfo batchAlterJobV2 =
(BatchAlterJobPersistInfo)journal.getData();
for (AlterJobV2 alterJobV2 :
batchAlterJobV2.getAlterJobV2List()) {
catalog.getRollupHandler().replayAlterJobV2(alterJobV2);
@@ -1222,7 +1222,7 @@ public class EditLog {
}
public void logBatchAlterJob(BatchAlterJobPersistInfo batchAlterJobV2) {
- logEdit(OperationType.OP_BATCH_ALTER_JOB_V2, batchAlterJobV2);
+ logEdit(OperationType.OP_BATCH_ADD_ROLLUP, batchAlterJobV2);
}
public void logModifyDistributionType(TableInfo tableInfo) {
diff --git a/fe/src/main/java/org/apache/doris/persist/OperationType.java
b/fe/src/main/java/org/apache/doris/persist/OperationType.java
index 315a94e..b6f37d2 100644
--- a/fe/src/main/java/org/apache/doris/persist/OperationType.java
+++ b/fe/src/main/java/org/apache/doris/persist/OperationType.java
@@ -57,7 +57,7 @@ public class OperationType {
public static final short OP_RENAME_ROLLUP = 120;
public static final short OP_ALTER_JOB_V2 = 121;
public static final short OP_MODIFY_DISTRIBUTION_TYPE = 122;
- public static final short OP_BATCH_ALTER_JOB_V2 = 123;
+ public static final short OP_BATCH_ADD_ROLLUP = 123;
public static final short OP_BATCH_DROP_ROLLUP = 124;
// 30~39 130~139 230~239 ...
diff --git a/fe/src/test/java/org/apache/doris/alter/BatchRollupJobTest.java
b/fe/src/test/java/org/apache/doris/alter/BatchRollupJobTest.java
new file mode 100644
index 0000000..1c92db4
--- /dev/null
+++ b/fe/src/test/java/org/apache/doris/alter/BatchRollupJobTest.java
@@ -0,0 +1,101 @@
+// 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.
+
+package org.apache.doris.alter;
+
+import org.apache.doris.analysis.AlterTableStmt;
+import org.apache.doris.analysis.CreateDbStmt;
+import org.apache.doris.analysis.CreateTableStmt;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.OlapTable.OlapTableState;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.utframe.UtFrameUtils;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Map;
+import java.util.UUID;
+
+public class BatchRollupJobTest {
+
+ private static String runningDir = "fe/mocked/BatchRollupJobTest/" +
UUID.randomUUID().toString() + "/";
+
+ private static ConnectContext ctx = UtFrameUtils.createDefaultCtx();
+
+ @BeforeClass
+ public static void setup() throws Exception {
+ UtFrameUtils.createMinDorisCluster(runningDir);
+ }
+
+ @Test
+ public void test() throws Exception {
+ System.out.println("xxx");
+ // create database db1
+ String createDbStmtStr = "create database db1;";
+ CreateDbStmt createDbStmt = (CreateDbStmt)
UtFrameUtils.parseAndAnalyzeStmt(createDbStmtStr, ctx);
+ Catalog.getCurrentCatalog().createDb(createDbStmt);
+ System.out.println(Catalog.getCurrentCatalog().getDbNames());
+ // create table tbl1
+ String createTblStmtStr1 = "create table db1.tbl1(k1 int, k2 int, k3
int) distributed by hash(k1) buckets 3 properties('replication_num' = '1');";
+ CreateTableStmt createTableStmt = (CreateTableStmt)
UtFrameUtils.parseAndAnalyzeStmt(createTblStmtStr1, ctx);
+ Catalog.getCurrentCatalog().createTable(createTableStmt);
+
+ // batch add 3 rollups
+ String stmtStr = "alter table db1.tbl1 add rollup r1(k1) duplicate
key(k1), r2(k1, k2) duplicate key(k1), r3(k2) duplicate key(k2);";
+ AlterTableStmt alterTableStmt = (AlterTableStmt)
UtFrameUtils.parseAndAnalyzeStmt(stmtStr, ctx);
+
Catalog.getCurrentCatalog().getAlterInstance().processAlterTable(alterTableStmt);
+
+ Map<Long, AlterJobV2> alterJobs =
Catalog.getCurrentCatalog().getRollupHandler().getAlterJobsV2();
+ Assert.assertEquals(3, alterJobs.size());
+
+ Database db = Catalog.getCurrentCatalog().getDb("default_cluster:db1");
+ Assert.assertNotNull(db);
+ OlapTable tbl = (OlapTable) db.getTable("tbl1");
+ Assert.assertNotNull(tbl);
+
+ int finishedNum = 0;
+ for (AlterJobV2 alterJobV2 : alterJobs.values()) {
+ if (alterJobV2.getType() != AlterJobV2.JobType.ROLLUP) {
+ continue;
+ }
+ while (!alterJobV2.getJobState().isFinalState()) {
+ System.out.println(
+ "rollup job " + alterJobV2.getJobId() + " is running.
state: " + alterJobV2.getJobState());
+ Thread.sleep(5000);
+ }
+ System.out.println("rollup job " + alterJobV2.getJobId() + " is
done. state: " + alterJobV2.getJobState());
+ Assert.assertEquals(AlterJobV2.JobState.FINISHED,
alterJobV2.getJobState());
+ ++finishedNum;
+ if (finishedNum == 3) {
+ Thread.sleep(100);
+ Assert.assertEquals(OlapTableState.NORMAL, tbl.getState());
+ } else {
+ Assert.assertEquals(OlapTableState.ROLLUP, tbl.getState());
+ }
+ }
+
+ for (Partition partition : tbl.getPartitions()) {
+ Assert.assertEquals(4,
partition.getMaterializedIndices(IndexExtState.VISIBLE).size());
+ }
+ }
+}
diff --git a/fe/src/test/java/org/apache/doris/utframe/UtFrameUtils.java
b/fe/src/test/java/org/apache/doris/utframe/UtFrameUtils.java
index d6811a2..325afc9 100644
--- a/fe/src/test/java/org/apache/doris/utframe/UtFrameUtils.java
+++ b/fe/src/test/java/org/apache/doris/utframe/UtFrameUtils.java
@@ -23,11 +23,28 @@ import org.apache.doris.analysis.SqlScanner;
import org.apache.doris.analysis.StatementBase;
import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Catalog;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.Pair;
import org.apache.doris.mysql.privilege.PaloAuth;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TNetworkAddress;
+import
org.apache.doris.utframe.MockedBackendFactory.DefaultBeThriftServiceImpl;
+import
org.apache.doris.utframe.MockedBackendFactory.DefaultHeartbeatServiceImpl;
+import
org.apache.doris.utframe.MockedBackendFactory.DefaultPBackendServiceImpl;
+import org.apache.doris.utframe.MockedFrontend.EnvVarNotSetException;
+import org.apache.doris.utframe.MockedFrontend.FeStartException;
+import org.apache.doris.utframe.MockedFrontend.NotInitException;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import java.io.IOException;
import java.io.StringReader;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
public class UtFrameUtils {
@@ -54,4 +71,53 @@ public class UtFrameUtils {
statementBase.analyze(analyzer);
return statementBase;
}
+
+ public static void createMinDorisCluster(String runningDir) throws
EnvVarNotSetException, IOException,
+ FeStartException, NotInitException, DdlException,
InterruptedException {
+ // get DORIS_HOME
+ final String dorisHome = System.getenv("DORIS_HOME");
+ if (Strings.isNullOrEmpty(dorisHome)) {
+ throw new EnvVarNotSetException("env DORIS_HOME is not set");
+ }
+
+ Random r = new Random(System.currentTimeMillis());
+ int basePort = 20000 + r.nextInt(9000);
+ int fe_http_port = basePort + 1;
+ int fe_rpc_port = basePort + 2;
+ int fe_query_port = basePort + 3;
+ int fe_edit_log_port = basePort + 4;
+
+ int be_heartbeat_port = basePort + 5;
+ int be_thrift_port = basePort + 6;
+ int be_brpc_port = basePort + 7;
+ int be_http_port = basePort + 8;
+
+ // start fe in "DORIS_HOME/fe/mocked/"
+ MockedFrontend frontend = MockedFrontend.getInstance();
+ Map<String, String> feConfMap = Maps.newHashMap();
+ // set additional fe config
+ feConfMap.put("http_port", String.valueOf(fe_http_port));
+ feConfMap.put("rpc_port", String.valueOf(fe_rpc_port));
+ feConfMap.put("query_port", String.valueOf(fe_query_port));
+ feConfMap.put("edit_log_port", String.valueOf(fe_edit_log_port));
+ feConfMap.put("tablet_create_timeout_second", "10");
+ frontend.init(dorisHome + "/" + runningDir, feConfMap);
+ frontend.start(new String[0]);
+
+ // start be
+ MockedBackend backend = MockedBackendFactory.createBackend("127.0.0.1",
+ be_heartbeat_port, be_thrift_port, be_brpc_port, be_http_port,
+ new DefaultHeartbeatServiceImpl(be_thrift_port, be_http_port,
be_brpc_port),
+ new DefaultBeThriftServiceImpl(), new
DefaultPBackendServiceImpl());
+ backend.setFeAddress(new TNetworkAddress("127.0.0.1",
frontend.getRpcPort()));
+ backend.start();
+
+ // add be
+ List<Pair<String, Integer>> bes = Lists.newArrayList();
+ bes.add(Pair.create(backend.getHost(), backend.getHeartbeatPort()));
+ Catalog.getCurrentSystemInfo().addBackends(bes, false,
"default_cluster");
+
+ // sleep to wait first heartbeat
+ Thread.sleep(6000);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
