You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/05/20 09:46:04 UTC
(doris) branch branch-4.1 updated: branch-4.1: [fix](streaming-job) recompute derived fields after replay and ALTER #62936 (#63261)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 53829fc1e80 branch-4.1: [fix](streaming-job) recompute derived fields after replay and ALTER #62936 (#63261)
53829fc1e80 is described below
commit 53829fc1e8019343db823e6f5b0769ded997a774
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Wed May 20 17:45:56 2026 +0800
branch-4.1: [fix](streaming-job) recompute derived fields after replay and ALTER #62936 (#63261)
Cherry-picked from #62936
Co-authored-by: wudi <wu...@selectdb.com>
---
.../insert/streaming/StreamingInsertJob.java | 20 +++++++-
.../StreamingInsertJobCheckDataQualityTest.java | 54 ++++++++++++++++++++++
2 files changed, 72 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index 5fe72415c0e..7e6c2c3830b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -297,21 +297,35 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
try {
this.jobProperties = new StreamingJobProperties(properties);
jobProperties.validate();
- this.sampleWindowMs = jobProperties.getMaxIntervalSecond() * 10 * 1000;
resolveCloudCluster();
// build time definition
JobExecutionConfiguration execConfig = getJobConfig();
TimerDefinition timerDefinition = new TimerDefinition();
- timerDefinition.setInterval(jobProperties.getMaxIntervalSecond());
timerDefinition.setIntervalUnit(IntervalUnit.SECOND);
timerDefinition.setStartTimeMs(execConfig.getTimerDefinition().getStartTimeMs());
execConfig.setTimerDefinition(timerDefinition);
+ recomputeDerivedFields();
} catch (AnalysisException ae) {
log.warn("parse streaming insert job failed, props: {}", properties, ae);
throw new RuntimeException(ae.getMessage());
}
}
+ private void recomputeDerivedFields() {
+ if (jobProperties == null) {
+ return;
+ }
+ long maxIntervalSec = jobProperties.getMaxIntervalSecond();
+ this.sampleWindowMs = maxIntervalSec * 10 * 1000;
+ JobExecutionConfiguration execConfig = getJobConfig();
+ if (execConfig != null && execConfig.getTimerDefinition() != null) {
+ execConfig.getTimerDefinition().setInterval(maxIntervalSec);
+ }
+ this.sampleStartTime = System.currentTimeMillis();
+ this.sampleWindowScannedRows = 0L;
+ this.sampleWindowFilteredRows = 0L;
+ }
+
private void resolveCloudCluster() throws AnalysisException {
String requested = validateComputeGroupProperty(properties);
if (requested != null) {
@@ -956,6 +970,7 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
}
this.properties.putAll(inputProperties);
this.jobProperties = new StreamingJobProperties(this.properties);
+ recomputeDerivedFields();
}
private void resetCloudProgress(Offset offset) throws JobException {
@@ -1335,6 +1350,7 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
if (jobProperties == null && properties != null) {
jobProperties = new StreamingJobProperties(properties);
}
+ recomputeDerivedFields();
if (null == getSucceedTaskCount()) {
setSucceedTaskCount(new AtomicLong(0));
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java
index a0ca82075d4..06c5e1a5c75 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java
@@ -18,6 +18,8 @@
package org.apache.doris.job.extensions.insert.streaming;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.base.JobExecutionConfiguration;
+import org.apache.doris.job.base.TimerDefinition;
import org.apache.doris.job.cdc.request.CommitOffsetRequest;
import org.apache.doris.job.common.JobStatus;
import org.apache.doris.job.exception.JobException;
@@ -144,6 +146,58 @@ public class StreamingInsertJobCheckDataQualityTest {
Assert.assertTrue((long) Deencapsulation.getField(job, "sampleStartTime") > oldStartTime);
}
+ private static StreamingInsertJob jobWithProperties(String maxIntervalSec) {
+ StreamingInsertJob job = Deencapsulation.newInstance(StreamingInsertJob.class);
+ Map<String, String> props = new HashMap<>();
+ props.put(StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY, maxIntervalSec);
+ Deencapsulation.setField(job, "properties", props);
+ Deencapsulation.setField(job, "jobProperties", new StreamingJobProperties(props));
+
+ JobExecutionConfiguration cfg = new JobExecutionConfiguration();
+ cfg.setTimerDefinition(new TimerDefinition());
+ Deencapsulation.setField(job, "jobConfig", cfg);
+ return job;
+ }
+
+ @Test
+ public void testRecomputeDerivedFieldsRebuildsSampleWindowMs() throws Exception {
+ StreamingInsertJob job = jobWithProperties("10");
+ Deencapsulation.setField(job, "sampleWindowMs", 0L);
+ Deencapsulation.setField(job, "sampleStartTime", 0L);
+
+ Deencapsulation.invoke(job, "recomputeDerivedFields");
+
+ Assert.assertEquals(100_000L, (long) Deencapsulation.getField(job, "sampleWindowMs"));
+ Assert.assertTrue((long) Deencapsulation.getField(job, "sampleStartTime") > 0L);
+ }
+
+ @Test
+ public void testModifyPropertiesInternalRefreshesDerivedFields() throws Exception {
+ StreamingInsertJob job = jobWithProperties("10");
+ JobExecutionConfiguration cfg = (JobExecutionConfiguration) Deencapsulation.getField(job, "jobConfig");
+ cfg.getTimerDefinition().setInterval(10L);
+ Deencapsulation.setField(job, "sampleWindowMs", 100_000L);
+
+ Map<String, String> alter = new HashMap<>();
+ alter.put(StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY, "30");
+ Deencapsulation.invoke(job, "modifyPropertiesInternal", alter);
+
+ Assert.assertEquals(300_000L, (long) Deencapsulation.getField(job, "sampleWindowMs"));
+ Assert.assertEquals(30L, cfg.getTimerDefinition().getInterval().longValue());
+ }
+
+ @Test
+ public void testRecomputeDerivedFieldsResetsSampleCounters() throws Exception {
+ StreamingInsertJob job = jobWithProperties("10");
+ Deencapsulation.setField(job, "sampleWindowScannedRows", 1000L);
+ Deencapsulation.setField(job, "sampleWindowFilteredRows", 100L);
+
+ Deencapsulation.invoke(job, "recomputeDerivedFields");
+
+ Assert.assertEquals(0L, (long) Deencapsulation.getField(job, "sampleWindowScannedRows"));
+ Assert.assertEquals(0L, (long) Deencapsulation.getField(job, "sampleWindowFilteredRows"));
+ }
+
@Test
public void testMissingMaxFilterRatioIsNoop() throws Exception {
StreamingInsertJob job = newJob(0.10, 60_000L);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org