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/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 02cb462eb25 [fix](load)(regression-test) fix broker load label
self-conflict on pending retry; deflake P2 suites (#66469)
02cb462eb25 is described below
commit 02cb462eb25fc848e3d7f445bdf359bacd3c0a70
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Aug 5 23:12:43 2026 +0800
[fix](load)(regression-test) fix broker load label self-conflict on pending
retry; deflake P2 suites (#66469)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #66468 (the branch-4.0 twin of this PR), #64945 / #65209
(existing master plugin wait behavior this builds on)
Problem Summary:
Forward-port of the fixes from a chronically red branch-4.0 P2
regression pipeline triage
(7 failures; no BE crash/OOM — all case-level). Every root cause below
also exists on master
(2 FE product bugs + 3 test/CI fixes). The branch-4.0 twin is #66468;
the #64822 backport
included there is not needed here.
1. **[load] Broker load pending-task retry self-conflicts on its own
label** (FE product
bug). `BrokerLoadPendingTask.executeTask()` runs `getAllFileStatus()`
then `beginTxn()`,
and `LoadTask.exec` retries the whole task on failure (retryTime=3),
with
`onTaskFinished()` inside the same try. Once the txn exists, any failure
makes every
retry hit `LabelAlreadyUsedException` against the job's OWN PREPARE txn,
burn all retries
within milliseconds and cancel the job with a misleading
`Label [...] has already been used, relate to txn [...], status
[PREPARE]` that masks the
real cause. **Root cause confirmed from the failing run's FE master
fe.log** (all within
57ms on one thread): `begin transaction: txn id 19118` succeeded; 2ms
later
`createLoadingTask` threw
`org.apache.doris.nereids.exceptions.AnalysisException`
`disk /mnt/... on backend ... exceed limit usage` (a CI disk was over
the watermark);
the three retries then all failed on the job's own label.
Fix: make `beginTxn()` retry-idempotent — reuse an already-assigned
`transactionId`, and
on `LabelAlreadyUsedException` adopt the label's txn iff it is ours
(`callbackId == job id`, status PREPARE). Foreign conflicts and lookup
failures still
rethrow the original exception. 3 Mockito unit tests.
2. **[load] Nereids planning failures must cancel the job with the real
cause** (FE product
bug, the trigger's escape route).
`org.apache.doris.nereids.exceptions.AnalysisException`
extends RuntimeException, so it bypassed `onPendingTaskFinished`'s
`catch (UserException)` (which cancels immediately with the real
message) and fell into
the generic retry path — that is exactly how the disk error above got
masked. Catch it
alongside UserException. The new test drives the real propagation path
(mocked
`NereidsLoadingTaskPlanner.plan` throws through the real
`LoadLoadingTask.init`/`createLoadingTask` chain) and asserts the job is
CANCELLED with
the real message and no loading tasks.
3. **plugin_compaction.groovy**: treat base compaction's `E-808`
(`BE_NO_SUITABLE_VERSION`,
the by-design "nothing to base-compact" result from
`BaseCompaction::pick_rowsets_to_compact`) as benign, exactly like
cumulative's
E-2000/E-2010. A lagging publish can legitimately make one replica's
cumulative trigger a
no-op (E-2000, already ignored), leaving its cumulative point behind so
the later base
trigger deterministically has nothing to merge on that replica —
observed killing
`test_base_compaction_no_value` 7/11 recent branch-4.0 P2 runs. Suites
still verify the
compaction effect via their own rowset/segment-count asserts.
4. **compaction_width_array_column**: with `BUCKETS 2` the fixture loads
a ~56GB 197-segment
overlapping rowset into a single tablet; compaction writes the full
output on the same
mount before deleting the input, which structurally ENOSPCs on 100GB CI
data disks
(11/11 recent branch-4.0 runs red; ASAN runs hit MEM_LIMIT for the same
oversize; the
same full disks also triggered the broker-load failure in item 1). Bump
to `BUCKETS 16`
(~7GB/tablet) and raise the load to 16G exec_mem_limit /
load_parallelism 1 (the load
phase was exceeding the default 8G with a ~12G peak).
5. **inverted_index_p2/test_show_data**: all four sub-suites
exact-compare physical index
sizes produced by different writer paths (inline-at-load vs ALTER+BUILD
INDEX vs
index-compaction merge vs rebuild). Settle-sequence analysis shows the
values are stable
per run yet differ by a fixed per-index-file overhead (~7KB per replica,
12-15% relative
on the httplogs fixture), so exact equality flakes chronically (6/11,
6/11, 3/11, 4/11
recent branch-4.0 runs across the four sub-suites). Replace the five
exact compares with
20% tolerance — a missing index still shows as a ~37% deficit and real
bloat as 2x+.
Pure-data (`no_index_size`) exact compares are kept.
6.
**cold_heat_separation_p2/table_modify_resouce_and_policy(+_by_hdfs)**:
both phases waited
only for `tablets[0]` to finish cooldown, then asserted
`remote_data_size > 0` for EVERY
replica row. Replicated cooldown is
leader-uploads/followers-follow-async, so a lagging
follower failed the assert with zero grace (7/11 recent branch-4.0
runs). Make the wait
cover all replica rows (local==0 && remote>0) with the same 100x10s
budget.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [x] Yes. (a) Broker load jobs whose pending task is retried after the
job's txn was
already begun now reuse (or adopt) that txn instead of being cancelled
with a
misleading "Label has already been used" error. (b) Nereids planning
failures now
cancel the load job immediately with the real cause instead of going
through the
generic retry path. No behavior change outside those failure paths.
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
Verification: `mvn test -pl fe-core -am -Dtest=BrokerLoadJobTest` (build
cache disabled)
green including the 4 new tests, 0 checkstyle violations; all changed
groovy files pass
offline `FileSystemCompiler` syntax check; test_show_data.groovy and
compaction_width_array_column.groovy end up byte-identical to their
fixed branch-4.0
(#66468) versions.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../apache/doris/load/loadv2/BrokerLoadJob.java | 59 ++++++--
.../doris/load/loadv2/BrokerLoadJobTest.java | 158 +++++++++++++++++++++
regression-test/plugins/plugin_compaction.groovy | 8 +-
.../table_modify_resouce_and_policy.groovy | 43 ++++--
.../table_modify_resouce_and_policy_by_hdfs.groovy | 43 ++++--
.../compaction_width_array_column.groovy | 7 +-
.../suites/compaction/ddl/column_witdh_array.sql | 5 +-
.../suites/inverted_index_p2/test_show_data.groovy | 37 ++++-
8 files changed, 321 insertions(+), 39 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
index d0ddcc5b1ae..c7b9415c68d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
@@ -62,6 +62,7 @@ import org.apache.doris.transaction.BeginTransactionException;
import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TransactionState.TxnCoordinator;
import org.apache.doris.transaction.TransactionState.TxnSourceType;
+import org.apache.doris.transaction.TransactionStatus;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
@@ -129,13 +130,52 @@ public class BrokerLoadJob extends BulkLoadJob {
public void beginTxn()
throws LabelAlreadyUsedException, BeginTransactionException,
AnalysisException, DuplicatedRequestException,
QuotaExceedException, MetaNotFoundException {
- transactionId = Env.getCurrentGlobalTransactionMgr()
- .beginTransaction(dbId,
Lists.newArrayList(fileGroupAggInfo.getAllTableIds()), label, null,
- new TxnCoordinator(TxnSourceType.FE, 0,
- FrontendOptions.getLocalHostAddress(),
- ExecuteEnv.getInstance().getStartupTime()),
- TransactionState.LoadJobSourceType.BATCH_LOAD_JOB, id,
- getTimeout());
+ if (transactionId > 0) {
+ // A previous attempt of the pending task already began our txn
and failed afterwards;
+ // the retried task must reuse it instead of failing
LabelAlreadyUsedException
+ // against the job's own txn.
+ LOG.info("broker load job {} reuses already begun txn {} on
pending task retry", id, transactionId);
+ return;
+ }
+ try {
+ transactionId = Env.getCurrentGlobalTransactionMgr()
+ .beginTransaction(dbId,
Lists.newArrayList(fileGroupAggInfo.getAllTableIds()), label, null,
+ new TxnCoordinator(TxnSourceType.FE, 0,
+ FrontendOptions.getLocalHostAddress(),
+ ExecuteEnv.getInstance().getStartupTime()),
+ TransactionState.LoadJobSourceType.BATCH_LOAD_JOB,
id,
+ getTimeout());
+ } catch (LabelAlreadyUsedException e) {
+ // The label may be occupied by our OWN txn: a previous attempt
registered it but threw
+ // before transactionId was assigned (e.g. edit log write
failure), and the pending
+ // task retry would otherwise burn all retries on this exception
and cancel the job
+ // with a misleading "Label has already been used".
+ Long ownTxnId = findSelfPreparedTxnByLabel();
+ if (ownTxnId == null) {
+ throw e;
+ }
+ LOG.info("broker load job {} adopts its own prepared txn {} for
label {} on pending task retry",
+ id, ownTxnId, label);
+ transactionId = ownTxnId;
+ }
+ }
+
+ private Long findSelfPreparedTxnByLabel() {
+ try {
+ Long txnId =
Env.getCurrentGlobalTransactionMgr().getTransactionIdByLabel(dbId, label,
+ Lists.newArrayList(TransactionStatus.PREPARE));
+ if (txnId == null) {
+ return null;
+ }
+ TransactionState existingTxn =
Env.getCurrentGlobalTransactionMgr().getTransactionState(dbId, txnId);
+ if (existingTxn != null && existingTxn.getCallbackId() == id
+ && existingTxn.getTransactionStatus() ==
TransactionStatus.PREPARE) {
+ return txnId;
+ }
+ } catch (Exception lookupException) {
+ LOG.warn("broker load job {} failed to look up txn by label {}",
id, label, lookupException);
+ }
+ return null;
}
@Override
@@ -203,7 +243,10 @@ public class BrokerLoadJob extends BulkLoadJob {
try {
Database db = getDb();
createLoadingTask(db, attachment);
- } catch (UserException e) {
+ } catch (UserException |
org.apache.doris.nereids.exceptions.AnalysisException e) {
+ // Nereids analysis errors extend RuntimeException, not
UserException; without this
+ // branch a deterministic planning failure (e.g. "disk ... exceed
limit usage") falls
+ // into the generic pending-task retry path and the real cause is
never reported.
LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
.add("database_id", dbId)
.add("error_msg", "Failed to divide job into loading
task.")
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
index 3764e0e286d..14a80c1933a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableProperty;
+import org.apache.doris.common.LabelAlreadyUsedException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.Status;
import org.apache.doris.common.jmockit.Deencapsulation;
@@ -31,6 +32,7 @@ import org.apache.doris.load.BrokerFileGroup;
import org.apache.doris.load.BrokerFileGroupAggInfo;
import org.apache.doris.load.BrokerFileGroupAggInfo.FileGroupAggKey;
import org.apache.doris.load.EtlStatus;
+import org.apache.doris.load.FailMsg;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.nereids.load.NereidsBrokerFileGroup;
import org.apache.doris.nereids.load.NereidsLoadingTaskPlanner;
@@ -42,6 +44,7 @@ import org.apache.doris.thrift.TBrokerFileStatus;
import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.transaction.GlobalTransactionMgrIface;
import org.apache.doris.transaction.TransactionState;
+import org.apache.doris.transaction.TransactionStatus;
import org.apache.doris.transaction.TxnStateCallbackFactory;
import com.google.common.collect.Lists;
@@ -400,4 +403,159 @@ public class BrokerLoadJobTest {
Assert.assertEquals(1, brokerLoadJob.getFinishTimestamp());
Assert.assertEquals(JobState.LOADING, brokerLoadJob.getState());
}
+
+ @Test
+ public void testBeginTxnReusesAlreadyBegunTxn() throws Exception {
+ // A retried pending task must not begin a second txn for the same job
(it would fail
+ // LabelAlreadyUsedException against the job's own txn).
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "transactionId", 12345L);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ brokerLoadJob.beginTxn();
+ }
+
+ Assert.assertEquals(12345L, (long)
Deencapsulation.getField(brokerLoadJob, "transactionId"));
+ Mockito.verifyNoInteractions(transactionMgr);
+ }
+
+ @Test
+ public void testBeginTxnAdoptsOwnPreparedTxn() throws Exception {
+ // First attempt registered the txn but threw before transactionId was
assigned
+ // (e.g. edit log write failure); the retry gets
LabelAlreadyUsedException for the job's
+ // OWN prepared txn and must adopt it instead of cancelling the job.
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TransactionState preparedTxn = Mockito.mock(TransactionState.class);
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "id", 1001L);
+ Deencapsulation.setField(brokerLoadJob, "dbId", 1L);
+ Deencapsulation.setField(brokerLoadJob, "label",
"label_self_conflict");
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ Mockito.when(transactionMgr.beginTransaction(Mockito.anyLong(),
Mockito.anyList(),
+ Mockito.anyString(), Mockito.any(), Mockito.any(),
Mockito.any(),
+ Mockito.anyLong(), Mockito.anyLong()))
+ .thenThrow(new
LabelAlreadyUsedException("label_self_conflict"));
+
Mockito.when(transactionMgr.getTransactionIdByLabel(Mockito.anyLong(),
Mockito.anyString(),
+ Mockito.anyList())).thenReturn(777L);
+ Mockito.when(transactionMgr.getTransactionState(Mockito.anyLong(),
Mockito.eq(777L)))
+ .thenReturn(preparedTxn);
+ Mockito.when(preparedTxn.getCallbackId()).thenReturn(1001L);
+
Mockito.when(preparedTxn.getTransactionStatus()).thenReturn(TransactionStatus.PREPARE);
+
+ brokerLoadJob.beginTxn();
+ }
+
+ Assert.assertEquals(777L, (long)
Deencapsulation.getField(brokerLoadJob, "transactionId"));
+ }
+
+ @Test
+ public void testBeginTxnRethrowsForeignLabelConflict() throws Exception {
+ // The label belongs to some other job's txn: the original exception
must propagate.
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TransactionState foreignTxn = Mockito.mock(TransactionState.class);
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "id", 1001L);
+ Deencapsulation.setField(brokerLoadJob, "dbId", 1L);
+ Deencapsulation.setField(brokerLoadJob, "label", "label_foreign");
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ Mockito.when(transactionMgr.beginTransaction(Mockito.anyLong(),
Mockito.anyList(),
+ Mockito.anyString(), Mockito.any(), Mockito.any(),
Mockito.any(),
+ Mockito.anyLong(), Mockito.anyLong()))
+ .thenThrow(new LabelAlreadyUsedException("label_foreign"));
+
Mockito.when(transactionMgr.getTransactionIdByLabel(Mockito.anyLong(),
Mockito.anyString(),
+ Mockito.anyList())).thenReturn(888L);
+ Mockito.when(transactionMgr.getTransactionState(Mockito.anyLong(),
Mockito.eq(888L)))
+ .thenReturn(foreignTxn);
+ Mockito.when(foreignTxn.getCallbackId()).thenReturn(9999L);
+
+ try {
+ brokerLoadJob.beginTxn();
+ Assert.fail("expected LabelAlreadyUsedException");
+ } catch (LabelAlreadyUsedException expected) {
+ // expected
+ }
+ }
+
+ Assert.assertEquals(0L, (long) Deencapsulation.getField(brokerLoadJob,
"transactionId"));
+ }
+
+ @Test
+ public void testPendingTaskOnFinishedWithNereidsPlanningError() throws
Exception {
+ // A Nereids planning error is a RuntimeException; it must cancel the
job with the real
+ // cause instead of escaping into the generic pending-task retry path
(which used to end
+ // in a misleading "Label has already been used" cancellation).
+ BrokerPendingTaskAttachment attachment =
Mockito.mock(BrokerPendingTaskAttachment.class);
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ Database database = Mockito.mock(Database.class);
+ BrokerFileGroupAggInfo fileGroupAggInfo =
Mockito.mock(BrokerFileGroupAggInfo.class);
+ BrokerFileGroup brokerFileGroup = Mockito.mock(BrokerFileGroup.class);
+ NereidsBrokerFileGroup nereidsBfg =
Mockito.mock(NereidsBrokerFileGroup.class);
+
Mockito.when(brokerFileGroup.toNereidsBrokerFileGroup()).thenReturn(nereidsBfg);
+ OlapTable olapTable = Mockito.mock(OlapTable.class);
+ GlobalTransactionMgrIface globalTxnMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ ProgressManager progressManager = Mockito.mock(ProgressManager.class);
+ ComputeGroupMgr computeGroupMgr = Mockito.mock(ComputeGroupMgr.class);
+ TableProperty tableProperty = Mockito.mock(TableProperty.class);
+
+ try (MockedStatic<Env> envMockedStatic = Mockito.mockStatic(Env.class);
+ MockedConstruction<NereidsLoadingTaskPlanner> ignored =
+
Mockito.mockConstruction(NereidsLoadingTaskPlanner.class, (mock, context) ->
+ Mockito.doThrow(new
org.apache.doris.nereids.exceptions.AnalysisException(
+ "disk /mnt/mock on backend 10001
exceed limit usage"))
+ .when(mock).plan(Mockito.any(),
Mockito.anyList(), Mockito.anyInt()))) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+
envMockedStatic.when(Env::getCurrentProgressManager).thenReturn(progressManager);
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(globalTxnMgr);
+
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "state", JobState.LOADING);
+ BrokerDesc brokerDesc = Mockito.mock(BrokerDesc.class);
+ Deencapsulation.setField(brokerLoadJob, "brokerDesc", brokerDesc);
+
+ Map<FileGroupAggKey, List<BrokerFileGroup>> aggKeyToFileGroups =
Maps.newHashMap();
+ FileGroupAggKey aggKey = new FileGroupAggKey(1L, null);
+ aggKeyToFileGroups.put(aggKey,
Lists.newArrayList(brokerFileGroup));
+ Deencapsulation.setField(brokerLoadJob, "fileGroupAggInfo",
fileGroupAggInfo);
+
+ Mockito.when(attachment.getTaskId()).thenReturn(1L);
+
Mockito.doReturn(database).when(catalog).getDbOrMetaException(Mockito.anyLong());
+ Mockito.doReturn(Lists.newArrayList()).when(database)
+ .getTablesOnIdOrderOrThrowException(Mockito.anyList());
+
Mockito.when(fileGroupAggInfo.getAggKeyToFileGroups()).thenReturn(aggKeyToFileGroups);
+
Mockito.when(fileGroupAggInfo.getAllTableIds()).thenReturn(Sets.newHashSet(1L));
+
Mockito.doReturn(olapTable).when(database).getTableNullable(Mockito.anyLong());
+ Mockito.when(olapTable.isTemporary()).thenReturn(false);
+
Mockito.when(olapTable.getTableProperty()).thenReturn(tableProperty);
+
Mockito.when(tableProperty.getUseSchemaLightChange()).thenReturn(false);
+ Mockito.when(olapTable.getIndexes()).thenReturn(null);
+ Mockito.when(attachment.getFileStatusByTable(aggKey)).thenReturn(
+ Collections.singletonList(Collections.singletonList(new
TBrokerFileStatus())));
+ Mockito.when(attachment.getFileNumByTable(aggKey)).thenReturn(1);
+ Mockito.when(env.getNextId()).thenReturn(1L);
+ Mockito.when(env.getComputeGroupMgr()).thenReturn(computeGroupMgr);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.when(computeGroupMgr.getAllBackendComputeGroup())
+ .thenReturn(new ComputeGroup("default", "default", null));
+ TxnStateCallbackFactory callbackFactory =
Mockito.mock(TxnStateCallbackFactory.class);
+
Mockito.when(globalTxnMgr.getCallbackFactory()).thenReturn(callbackFactory);
+
+ brokerLoadJob.onTaskFinished(attachment);
+
+ Assert.assertEquals(JobState.CANCELLED, brokerLoadJob.getState());
+ FailMsg failMsg = Deencapsulation.getField(brokerLoadJob,
"failMsg");
+ Assert.assertTrue(failMsg.getMsg().contains("exceed limit usage"));
+ Map<Long, LoadTask> idToTasks =
Deencapsulation.getField(brokerLoadJob, "idToTasks");
+ Assert.assertEquals(0, idToTasks.size());
+ }
+ }
}
diff --git a/regression-test/plugins/plugin_compaction.groovy
b/regression-test/plugins/plugin_compaction.groovy
index 67c16720204..d4d0966e504 100644
--- a/regression-test/plugins/plugin_compaction.groovy
+++ b/regression-test/plugins/plugin_compaction.groovy
@@ -114,8 +114,14 @@ Suite.metaClass.trigger_and_wait_compaction = { String
table_name, String compac
triggered_tablets.add(tablet) // compaction already in queue,
treat it as successfully triggered
} else if (!auto_compaction_disabled) {
// ignore the error if auto compaction enabled
- } else if (status_lower.contains("e-2000") ||
status_lower.contains("e-2010")) {
+ } else if (status_lower.contains("e-2000") ||
status_lower.contains("e-2010")
+ || status_lower.contains("e-808")) {
// ignore this tablet compaction.
+ // e-2000/e-2010: cumulative has no suitable version;
+ // e-808 (BE_NO_SUITABLE_VERSION): base compaction has nothing
to merge on this
+ // replica (e.g. only [0-1]+[2-y] with an empty [0-1]) — a
by-design no-op, the
+ // base analogue of e-2000. Replica layouts can legitimately
diverge here when a
+ // lagging publish made an earlier cumulative trigger a no-op
on one replica.
} else if (ignored_errors.any { error ->
status_lower.contains(error.toLowerCase()) }) {
// ignore this tablet compaction if the error is in the
ignored_errors list
} else {
diff --git
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
index ec8fcc8f255..db8a0fbe296 100644
---
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
+++
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
@@ -196,18 +196,28 @@ suite("table_modify_resouce") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
-
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
def try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
@@ -280,17 +290,28 @@ suite("table_modify_resouce") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
diff --git
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
index 574d9ef44fc..8082efb9774 100644
---
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
+++
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
@@ -192,18 +192,28 @@ suite("table_modify_resouce_by_hdfs") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
-
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
def try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
@@ -276,17 +286,28 @@ suite("table_modify_resouce_by_hdfs") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
diff --git
a/regression-test/suites/compaction/compaction_width_array_column.groovy
b/regression-test/suites/compaction/compaction_width_array_column.groovy
index d80e67c6cd8..225c412896b 100644
--- a/regression-test/suites/compaction/compaction_width_array_column.groovy
+++ b/regression-test/suites/compaction/compaction_width_array_column.groovy
@@ -33,14 +33,17 @@ suite('compaction_width_array_column', "p2") {
def s3BucketName = getS3BucketName()
def random = new Random();
+ // Loading wide array columns is memory-intensive: the default 8G
exec_mem_limit was exceeded
+ // (peak ~12G) and the load got cancelled with MEM_LIMIT_EXCEEDED. Raise
the limit to 16G and
+ // load with parallelism 1 to keep the per-load peak comfortably under the
limit.
def s3WithProperties = """WITH S3 (
|"AWS_ACCESS_KEY" = "${getS3AK()}",
|"AWS_SECRET_KEY" = "${getS3SK()}",
|"AWS_ENDPOINT" = "${getS3Endpoint()}",
|"AWS_REGION" = "${getS3Region()}")
|PROPERTIES(
- |"exec_mem_limit" = "8589934592",
- |"load_parallelism" = "3")""".stripMargin()
+ |"exec_mem_limit" = "17179869184",
+ |"load_parallelism" = "1")""".stripMargin()
// set fe configuration
sql "ADMIN SET FRONTEND CONFIG ('max_bytes_per_broker_scanner' =
'161061273600')"
diff --git a/regression-test/suites/compaction/ddl/column_witdh_array.sql
b/regression-test/suites/compaction/ddl/column_witdh_array.sql
index b51345a2084..c857221cdde 100644
--- a/regression-test/suites/compaction/ddl/column_witdh_array.sql
+++ b/regression-test/suites/compaction/ddl/column_witdh_array.sql
@@ -5,7 +5,10 @@ CREATE TABLE column_witdh_array (
k2 JSON NULL
) ENGINE=OLAP
DUPLICATE KEY(street, streetaddress)
-DISTRIBUTED BY HASH(street) BUCKETS 2
+-- 16 buckets keeps each tablet's single loaded rowset around 7GB; with 2
buckets it was ~56GB,
+-- and compaction (which writes the full output on the same mount before
deleting the input)
+-- structurally ENOSPCed on 100GB CI data disks.
+DISTRIBUTED BY HASH(street) BUCKETS 16
PROPERTIES (
"replication_num" = "1",
"disable_auto_compaction" = "true"
diff --git a/regression-test/suites/inverted_index_p2/test_show_data.groovy
b/regression-test/suites/inverted_index_p2/test_show_data.groovy
index 5d86c1e7c05..1638aa3606c 100644
--- a/regression-test/suites/inverted_index_p2/test_show_data.groovy
+++ b/regression-test/suites/inverted_index_p2/test_show_data.groovy
@@ -204,8 +204,14 @@ suite("test_show_data", "p2") {
create_httplogs_table_with_index.call(testTableWithIndex)
load_httplogs_data.call(testTableWithIndex,
'test_httplogs_load_with_index', 'true', 'json', 'documents-1000.json')
def another_with_index_size =
wait_for_show_data_finish(testTableWithIndex, 60000, 0)
+ assertTrue(another_with_index_size != "wait_timeout")
if (!isCloudMode()) {
- assertEquals(another_with_index_size, with_index_size)
+ // Inline-at-load vs BUILD INDEX writer paths differ by a fixed
per-index-file
+ // overhead (~7KB per replica observed); compare within 20%
tolerance, which still
+ // catches a missing index (~37% deficit) or gross bloat (2x+).
+ assertTrue(Math.abs(another_with_index_size - with_index_size)
+ <= 0.2 * Math.max(another_with_index_size,
with_index_size),
+ "index size mismatch beyond 20% tolerance:
inline_index=${another_with_index_size}, built_index=${with_index_size}")
}
} finally {
//try_sql("DROP TABLE IF EXISTS ${testTable}")
@@ -401,8 +407,12 @@ suite("test_show_data_for_bkd", "p2") {
create_httplogs_table_with_bkd_index.call(testTableWithBKDIndex)
load_httplogs_data.call(testTableWithBKDIndex,
'test_httplogs_load_with_bkd_index', 'true', 'json', 'documents-1000.json')
def another_with_index_size =
wait_for_show_data_finish(testTableWithBKDIndex, 60000, 0)
+ assertTrue(another_with_index_size != "wait_timeout")
if (!isCloudMode()) {
- assertEquals(another_with_index_size, with_index_size)
+ // Same rationale as test_show_data: writer-path dependent index
size, 20% tolerance.
+ assertTrue(Math.abs(another_with_index_size - with_index_size)
+ <= 0.2 * Math.max(another_with_index_size,
with_index_size),
+ "index size mismatch beyond 20% tolerance:
inline_index=${another_with_index_size}, built_index=${with_index_size}")
}
} finally {
//try_sql("DROP TABLE IF EXISTS ${testTable}")
@@ -604,8 +614,15 @@ suite("test_show_data_multi_add", "p2") {
create_httplogs_table_with_index.call(testTableWithIndex)
load_httplogs_data.call(testTableWithIndex,
'test_show_data_httplogs_multi_add_with_index', 'true', 'json',
'documents-1000.json')
def another_with_index_size =
wait_for_show_data_finish(testTableWithIndex, 60000, 0)
+ assertTrue(another_with_index_size != "wait_timeout")
if (!isCloudMode()) {
- assertEquals(another_with_index_size, with_index_size2)
+ // The inline-at-load and ALTER+BUILD INDEX writer paths produce
indexes whose on-disk
+ // size legitimately differs by a fixed per-index-file overhead
(~7KB per replica
+ // observed). Compare within 20% tolerance instead of exact
equality: a missing index
+ // still shows up as a ~37% deficit and gross bloat as 2x+.
+ assertTrue(Math.abs(another_with_index_size - with_index_size2)
+ <= 0.2 * Math.max(another_with_index_size,
with_index_size2),
+ "index size mismatch beyond 20% tolerance:
inline_index=${another_with_index_size}, built_index=${with_index_size2}")
}
} finally {
//try_sql("DROP TABLE IF EXISTS ${testTable}")
@@ -813,7 +830,14 @@ suite("test_show_data_with_compaction", "p2") {
assertTrue(another_with_index_size != "wait_timeout")
logger.info("with_index_size is {}, another_with_index_size is {}",
with_index_size, another_with_index_size)
- assertEquals(another_with_index_size, with_index_size)
+ // Index compaction merges per-segment index files; for identical data
the total
+ // on-disk size may differ from the non-compacted layout by a fixed
per-index-file
+ // overhead (merge vs rebuild writer paths, ~7KB per index file per
replica observed,
+ // 12-15% relative on this dataset). Compare within 20% tolerance
instead of exact
+ // equality to avoid flakiness, while still catching gross index bloat
or corruption.
+ assertTrue(Math.abs(with_index_size - another_with_index_size)
+ <= 0.2 * Math.max(with_index_size,
another_with_index_size),
+ "index size mismatch beyond 20% tolerance:
with_index=${with_index_size}, without_index=${another_with_index_size}")
set_be_config.call("inverted_index_compaction_enable", "true")
@@ -824,7 +848,10 @@ suite("test_show_data_with_compaction", "p2") {
def data_size_2 = create_table_run_compaction_and_wait(tableName)
logger.info("data_size_1 is {}, data_size_2 is {}", data_size_1,
data_size_2)
- assertEquals(data_size_1, data_size_2)
+ // Same rationale as above: compare index sizes within 20% tolerance,
not exact equality.
+ assertTrue(Math.abs(data_size_1 - data_size_2)
+ <= 0.2 * Math.max(data_size_1, data_size_2),
+ "index size mismatch beyond 20% tolerance:
data_size_1=${data_size_1}, data_size_2=${data_size_2}")
} finally {
// sql "DROP TABLE IF EXISTS ${tableWithIndexCompaction}"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
