Skip to content

[improvement](tablet scheduler) Adjust tablet sched priority to help load data succ #38528 #38884

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dataroaring merged 1 commit into from
Aug 5, 2024
Merged

[improvement](tablet scheduler) Adjust tablet sched priority to help load data succ #38528 #38884

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions fe/fe-common/src/main/java/org/apache/doris/common/Config.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1011,6 +1011,19 @@ public class Config extends ConfigBase {
@ConfField(mutable = true, masterOnly = true)
public static int tablet_further_repair_max_times = 5;

/**
* if tablet loaded txn failed recently, it will get higher priority to repair.
*/
@ConfField(mutable = true, masterOnly = true)
public static long tablet_recent_load_failed_second = 30 * 60;

/**
* base time for higher tablet scheduler task,
* set this config value bigger if want the high priority effect last longer.
*/
@ConfField(mutable = true, masterOnly = true)
public static long tablet_schedule_high_priority_second = 30 * 60;

/**
* the default slot number per path for hdd in tablet scheduler
* TODO(cmy): remove this config and dynamically adjust it by clone task statistic
Expand Down
16 changes: 9 additions & 7 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@
import org.apache.doris.catalog.Partition.PartitionState;
import org.apache.doris.catalog.Replica.ReplicaState;
import org.apache.doris.catalog.Tablet.TabletStatus;
import org.apache.doris.clone.TabletSchedCtx;
import org.apache.doris.clone.TabletScheduler;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
Expand Down Expand Up @@ -1841,11 +1840,11 @@ public boolean isStable(SystemInfoService infoService, TabletScheduler tabletSch
return false;
}

Pair<TabletStatus, TabletSchedCtx.Priority> statusPair = tablet.getHealthStatusWithPriority(
infoService, visibleVersion, replicaAlloc, aliveBeIds);
if (statusPair.first != TabletStatus.HEALTHY) {
TabletStatus status = tablet.getHealth(infoService, visibleVersion,
replicaAlloc, aliveBeIds).status;
if (status != TabletStatus.HEALTHY) {
LOG.info("table {} is not stable because tablet {} status is {}. replicas: {}",
id, tablet.getId(), statusPair.first, tablet.getReplicas());
id, tablet.getId(), status, tablet.getReplicas());
return false;
}
}
Expand Down Expand Up @@ -2482,6 +2481,10 @@ public boolean getEnableUniqueKeyMergeOnWrite() {
return tableProperty.getEnableUniqueKeyMergeOnWrite();
}

public boolean isUniqKeyMergeOnWrite() {
return getKeysType() == KeysType.UNIQUE_KEYS && getEnableUniqueKeyMergeOnWrite();
}

public boolean isDuplicateWithoutKey() {
return getKeysType() == KeysType.DUP_KEYS && getKeysNum() == 0;
}
Expand Down Expand Up @@ -2573,8 +2576,7 @@ public Set<Long> getPartitionKeys() {
}

public boolean isDupKeysOrMergeOnWrite() {
return keysType == KeysType.DUP_KEYS
|| (keysType == KeysType.UNIQUE_KEYS && getEnableUniqueKeyMergeOnWrite());
return keysType == KeysType.DUP_KEYS || isUniqKeyMergeOnWrite();
}

public void initAutoIncrementGenerator(long dbId) {
Expand Down
234 changes: 192 additions & 42 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
View file
Open in desktop

Large diffs are not rendered by default.

18 changes: 12 additions & 6 deletions fe/fe-core/src/main/java/org/apache/doris/clone/ColocateTableCheckerAndBalancer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.ReplicaAllocation;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.Tablet.TabletHealth;
import org.apache.doris.catalog.Tablet.TabletStatus;
import org.apache.doris.clone.TabletChecker.CheckerCounter;
import org.apache.doris.clone.TabletSchedCtx.Priority;
Expand Down Expand Up @@ -334,7 +335,7 @@ public void addGroup(GroupId groupId, ReplicaAllocation replicaAlloc, List<Set<L
* tablet to TabletScheduler.
* Otherwise, mark the group as stable
*/
protected void runAfterCatalogReady() {
public void runAfterCatalogReady() {
relocateAndBalanceGroups();
matchGroups();
}
Expand Down Expand Up @@ -512,6 +513,7 @@ private void matchGroups() {
if (olapTable == null || !colocateIndex.isColocateTable(olapTable.getId())) {
continue;
}
boolean isUniqKeyMergeOnWrite = olapTable.isUniqKeyMergeOnWrite();
olapTable.readLock();
try {
for (Partition partition : olapTable.getPartitions()) {
Expand All @@ -530,16 +532,20 @@ private void matchGroups() {
Preconditions.checkState(bucketsSeq.size() == replicationNum,
bucketsSeq.size() + " vs. " + replicationNum);
Tablet tablet = index.getTablet(tabletId);
TabletStatus st = tablet.getColocateHealthStatus(
TabletHealth tabletHealth = tablet.getColocateHealth(
visibleVersion, replicaAlloc, bucketsSeq);
if (st != TabletStatus.HEALTHY) {
if (tabletHealth.status != TabletStatus.HEALTHY) {
counter.unhealthyTabletNum++;
unstableReason = String.format("get unhealthy tablet %d in colocate table."
+ " status: %s", tablet.getId(), st);
+ " status: %s", tablet.getId(), tabletHealth.status);
if (LOG.isDebugEnabled()) {
LOG.debug(unstableReason);
}

if (tabletHealth.status == TabletStatus.UNRECOVERABLE) {
continue;
}

if (!tablet.readyToBeRepaired(infoService, Priority.NORMAL)) {
counter.tabletNotReady++;
continue;
Expand All @@ -550,9 +556,9 @@ private void matchGroups() {
db.getId(), tableId, partition.getId(), index.getId(), tablet.getId(),
replicaAlloc, System.currentTimeMillis());
// the tablet status will be set again when being scheduled
tabletCtx.setTabletStatus(st);
tabletCtx.setPriority(Priority.NORMAL);
tabletCtx.setTabletHealth(tabletHealth);
tabletCtx.setTabletOrderIdx(idx);
tabletCtx.setIsUniqKeyMergeOnWrite(isUniqKeyMergeOnWrite);

AddResult res = tabletScheduler.addTablet(tabletCtx, false /* not force */);
if (res == AddResult.LIMIT_EXCEED || res == AddResult.DISABLED) {
Expand Down
19 changes: 10 additions & 9 deletions fe/fe-core/src/main/java/org/apache/doris/clone/TabletChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.doris.catalog.Partition.PartitionState;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.Tablet.TabletHealth;
import org.apache.doris.catalog.Tablet.TabletStatus;
import org.apache.doris.clone.TabletScheduler.AddResult;
import org.apache.doris.common.Config;
Expand Down Expand Up @@ -197,7 +198,7 @@ private void removePrios(RepairTabletInfo repairTabletInfo) {
* If a tablet is not healthy, a TabletInfo will be created and sent to TabletScheduler for repairing.
*/
@Override
protected void runAfterCatalogReady() {
public void runAfterCatalogReady() {
int pendingNum = tabletScheduler.getPendingNum();
int runningNum = tabletScheduler.getRunningNum();
if (pendingNum > Config.max_scheduling_tablets
Expand Down Expand Up @@ -357,6 +358,7 @@ private LoopControlStatus handlePartitionTablet(Database db, OlapTable tbl, Part
return LoopControlStatus.CONTINUE;
}
boolean prioPartIsHealthy = true;
boolean isUniqKeyMergeOnWrite = tbl.isUniqKeyMergeOnWrite();
/*
* Tablet in SHADOW index can not be repaired of balanced
*/
Expand All @@ -369,26 +371,25 @@ private LoopControlStatus handlePartitionTablet(Database db, OlapTable tbl, Part
continue;
}

Pair<TabletStatus, TabletSchedCtx.Priority> statusWithPrio = tablet.getHealthStatusWithPriority(
infoService, partition.getVisibleVersion(),
TabletHealth tabletHealth = tablet.getHealth(infoService, partition.getVisibleVersion(),
tbl.getPartitionInfo().getReplicaAllocation(partition.getId()), aliveBeIds);

if (statusWithPrio.first == TabletStatus.HEALTHY) {
if (tabletHealth.status == TabletStatus.HEALTHY) {
// Only set last status check time when status is healthy.
tablet.setLastStatusCheckTime(startTime);
continue;
} else if (statusWithPrio.first == TabletStatus.UNRECOVERABLE) {
} else if (tabletHealth.status == TabletStatus.UNRECOVERABLE) {
// This tablet is not recoverable, do not set it into tablet scheduler
// all UNRECOVERABLE tablet can be seen from "show proc '/statistic'"
counter.unhealthyTabletNum++;
continue;
} else if (isInPrios) {
statusWithPrio.second = TabletSchedCtx.Priority.VERY_HIGH;
tabletHealth.priority = TabletSchedCtx.Priority.VERY_HIGH;
prioPartIsHealthy = false;
}

counter.unhealthyTabletNum++;
if (!tablet.readyToBeRepaired(infoService, statusWithPrio.second)) {
if (!tablet.readyToBeRepaired(infoService, tabletHealth.priority)) {
continue;
}

Expand All @@ -399,8 +400,8 @@ private LoopControlStatus handlePartitionTablet(Database db, OlapTable tbl, Part
tbl.getPartitionInfo().getReplicaAllocation(partition.getId()),
System.currentTimeMillis());
// the tablet status will be set again when being scheduled
tabletCtx.setTabletStatus(statusWithPrio.first);
tabletCtx.setPriority(statusWithPrio.second);
tabletCtx.setTabletHealth(tabletHealth);
tabletCtx.setIsUniqKeyMergeOnWrite(isUniqKeyMergeOnWrite);

AddResult res = tabletScheduler.addTablet(tabletCtx, false /* not force */);
if (res == AddResult.LIMIT_EXCEED || res == AddResult.DISABLED) {
Expand Down
Loading