Skip to content

Introduce delay analyzer to get watermark between sequence data and unsequece data based on ICDE 2022 #16886

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

Open
Pengzna wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Introduce delay analyzer to get watermark between sequence data and unsequece data based on ICDE 2022 #16886

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0260b78
introduce delay analyzer based on ICDE 2022
Pengzna Dec 9, 2025
b027a87
adopt to data-region
Pengzna Dec 9, 2025
d909cea
fix ut
Pengzna Dec 9, 2025
90a3278
fix
Pengzna Dec 9, 2025
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
33 changes: 33 additions & 0 deletions ...-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,7 @@
import org.apache.iotdb.db.storageengine.rescon.memory.TimePartitionManager;
import org.apache.iotdb.db.storageengine.rescon.memory.TsFileResourceManager;
import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager;
import org.apache.iotdb.db.tools.DelayAnalyzer;
import org.apache.iotdb.db.tools.settle.TsFileAndModSettleTool;
import org.apache.iotdb.db.utils.CommonUtils;
import org.apache.iotdb.db.utils.DateTimeUtils;
Expand Down Expand Up @@ -369,6 +370,9 @@ public class DataRegion implements IDataRegionForQuery {
private ILoadDiskSelector ordinaryLoadDiskSelector;
private ILoadDiskSelector pipeAndIoTV2LoadDiskSelector;

/** Delay analyzer for tracking data arrival delays and calculating safe watermarks */
private final DelayAnalyzer delayAnalyzer;

Comment on lines +373 to +375
Copy link
Contributor

@jt2594838 jt2594838 Dec 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add a switch for this to avoid affecting performance of existing scenarios.

/**
* Construct a database processor.
*
Expand All @@ -387,6 +391,7 @@ public DataRegion(
this.dataRegionId = new DataRegionId(Integer.parseInt(dataRegionIdString));
this.databaseName = databaseName;
this.fileFlushPolicy = fileFlushPolicy;
this.delayAnalyzer = new DelayAnalyzer();
acquireDirectBufferMemory();

dataRegionSysDir = SystemFileFactory.INSTANCE.getFile(systemDir, dataRegionIdString);
Expand Down Expand Up @@ -453,6 +458,7 @@ public DataRegion(String databaseName, String dataRegionIdString) {
partitionMaxFileVersions.put(0L, 0L);
upgradeModFileThreadPool = null;
this.metrics = new DataRegionMetrics(this);
this.delayAnalyzer = new DelayAnalyzer();

initDiskSelector();
}
Expand Down Expand Up @@ -1151,6 +1157,10 @@ public void insert(InsertRowNode insertRowNode) throws WriteProcessException {
if (deleted) {
return;
}
long arrivalTime = System.currentTimeMillis();
long generationTime = insertRowNode.getTime();
delayAnalyzer.update(generationTime, arrivalTime);

// init map
long timePartitionId = TimePartitionUtils.getTimePartitionId(insertRowNode.getTime());
initFlushTimeMap(timePartitionId);
Expand Down Expand Up @@ -1315,6 +1325,11 @@ public void insertTablet(InsertTabletNode insertTabletNode)
"Won't insert tablet {}, because region is deleted", insertTabletNode.getSearchIndex());
return;
}
long arrivalTime = System.currentTimeMillis();
long[] times = insertTabletNode.getTimes();
for (long generationTime : times) {
delayAnalyzer.update(generationTime, arrivalTime);
}
TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
long[] infoForMetrics = new long[5];
Expand Down Expand Up @@ -4105,9 +4120,11 @@ public void insert(InsertRowsOfOneDeviceNode insertRowsOfOneDeviceNode)
return;
}
long ttl = getTTL(insertRowsOfOneDeviceNode);
long arrivalTime = System.currentTimeMillis();
Map<TsFileProcessor, InsertRowsNode> tsFileProcessorMap = new HashMap<>();
for (int i = 0; i < insertRowsOfOneDeviceNode.getInsertRowNodeList().size(); i++) {
InsertRowNode insertRowNode = insertRowsOfOneDeviceNode.getInsertRowNodeList().get(i);
delayAnalyzer.update(insertRowNode.getTime(), arrivalTime);
if (!CommonUtils.isAlive(insertRowNode.getTime(), ttl)) {
// we do not need to write these part of data, as they can not be queried
// or the sub-plan has already been executed, we are retrying other sub-plans
Expand Down Expand Up @@ -4223,8 +4240,10 @@ public void insert(InsertRowsNode insertRowsNode)
}
boolean[] areSequence = new boolean[insertRowsNode.getInsertRowNodeList().size()];
long[] timePartitionIds = new long[insertRowsNode.getInsertRowNodeList().size()];
long arrivalTime = System.currentTimeMillis();
for (int i = 0; i < insertRowsNode.getInsertRowNodeList().size(); i++) {
InsertRowNode insertRowNode = insertRowsNode.getInsertRowNodeList().get(i);
delayAnalyzer.update(insertRowNode.getTime(), arrivalTime);
long ttl = getTTL(insertRowNode);
if (!CommonUtils.isAlive(insertRowNode.getTime(), ttl)) {
insertRowsNode
Expand Down Expand Up @@ -4314,8 +4333,13 @@ public void insertTablets(InsertMultiTabletsNode insertMultiTabletsNode)
// infoForMetrics[2]: ScheduleWalTimeCost
// infoForMetrics[3]: ScheduleMemTableTimeCost
// infoForMetrics[4]: InsertedPointsNumber
long arrivalTime = System.currentTimeMillis();
for (int i = 0; i < insertMultiTabletsNode.getInsertTabletNodeList().size(); i++) {
InsertTabletNode insertTabletNode = insertMultiTabletsNode.getInsertTabletNodeList().get(i);
long[] times = insertTabletNode.getTimes();
for (long generationTime : times) {
delayAnalyzer.update(generationTime, arrivalTime);
}
TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
boolean noFailure = false;
Expand Down Expand Up @@ -4619,6 +4643,15 @@ public TsFileManager getTsFileManager() {
return tsFileManager;
}

/**
* Get the delay analyzer instance for this data region
*
* @return DelayAnalyzer instance for tracking data arrival delays and calculating safe watermarks
*/
public DelayAnalyzer getDelayAnalyzer() {
return delayAnalyzer;
}

private long getTTL(InsertNode insertNode) {
if (insertNode.getTableName() == null) {
return DataNodeTTLCache.getInstance().getTTLForTree(insertNode.getTargetPath().getNodes());
Expand Down
Loading
Loading