You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2022/12/22 04:40:27 UTC
[iotdb] branch xingtanzjr/refine_cross_selection updated (dbf2cedd80 -> 226c470de2)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a change to branch xingtanzjr/refine_cross_selection
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from dbf2cedd80 spotless
new 5a8e43df3d fix the bug when checking target seq files
new 226c470de2 complete deviceTimeIndex build
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rewrite/CrossSpaceCompactionCandidate.java | 77 ++++++++++++++++------
.../db/engine/storagegroup/TsFileResource.java | 15 +++++
.../storagegroup/timeindex/DeviceTimeIndex.java | 4 ++
3 files changed, 76 insertions(+), 20 deletions(-)
[iotdb] 02/02: complete deviceTimeIndex build
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/refine_cross_selection
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 226c470de2893282db4553b70f6e5728af228d8b
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Thu Dec 22 12:40:11 2022 +0800
complete deviceTimeIndex build
---
.../rewrite/CrossSpaceCompactionCandidate.java | 59 +++++++++++++++++++---
.../db/engine/storagegroup/TsFileResource.java | 15 ++++++
.../storagegroup/timeindex/DeviceTimeIndex.java | 4 ++
3 files changed, 70 insertions(+), 8 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java
index f3ddb7c230..0965ead3f9 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java
@@ -21,9 +21,14 @@ package org.apache.iotdb.db.engine.compaction.cross.rewrite;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
+import org.apache.iotdb.db.engine.storagegroup.timeindex.DeviceTimeIndex;
+import org.apache.iotdb.db.engine.storagegroup.timeindex.ITimeIndex;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
/**
@@ -56,7 +61,7 @@ public class CrossSpaceCompactionCandidate {
this.nextUnseqFileIndex = 0;
}
- public boolean hasNextSplit() {
+ public boolean hasNextSplit() throws IOException {
if (nextUnseqFileIndex >= unseqFiles.size()) {
return false;
}
@@ -67,10 +72,15 @@ public class CrossSpaceCompactionCandidate {
return nextSplit;
}
- private boolean prepareNextSplit() {
+ private boolean prepareNextSplit() throws IOException {
TsFileResourceCandidate unseqFile = unseqFiles.get(nextUnseqFileIndex);
List<TsFileResourceCandidate> ret = new ArrayList<>();
+ // The startTime and endTime of each device are different in one TsFile. So we need to do the
+ // check
+ // one by one. And we cannot skip any device in the unseq file because it may lead to omission
+ // of
+ // target seq file
for (DeviceInfo unseqDeviceInfo : unseqFile.getDevices()) {
for (TsFileResourceCandidate seqFile : seqFiles) {
if (!seqFile.containsDevice(unseqDeviceInfo.deviceId)) {
@@ -167,6 +177,7 @@ public class CrossSpaceCompactionCandidate {
protected TsFileResource resource;
protected boolean selected;
protected boolean isValidCandidate;
+ private Map<String, DeviceInfo> deviceInfoMap;
protected TsFileResourceCandidate(TsFileResource tsFileResource) {
this.resource = tsFileResource;
@@ -176,20 +187,46 @@ public class CrossSpaceCompactionCandidate {
this.isValidCandidate = tsFileResource.isClosed() && tsFileResource.getTsFile().exists();
}
+ private void prepareDeviceInfos() throws IOException {
+ if (deviceInfoMap != null) {
+ return;
+ }
+ deviceInfoMap = new LinkedHashMap<>();
+ if (resource.getTimeIndexType() == ITimeIndex.FILE_TIME_INDEX_TYPE) {
+ DeviceTimeIndex timeIndex = resource.buildDeviceTimeIndex();
+ for (String deviceId : timeIndex.getDevices()) {
+ deviceInfoMap.put(
+ deviceId,
+ new DeviceInfo(
+ deviceId, timeIndex.getStartTime(deviceId), timeIndex.getEndTime(deviceId)));
+ }
+ } else {
+ for (String deviceId : resource.getDevices()) {
+ deviceInfoMap.put(
+ deviceId,
+ new DeviceInfo(
+ deviceId, resource.getStartTime(deviceId), resource.getEndTime(deviceId)));
+ }
+ }
+ }
+
protected void markAsSelected() {
this.selected = true;
}
- protected List<DeviceInfo> getDevices() {
- return null;
+ protected List<DeviceInfo> getDevices() throws IOException {
+ prepareDeviceInfos();
+ return new ArrayList<>(deviceInfoMap.values());
}
- protected DeviceInfo getDeviceInfoById(String deviceId) {
- return null;
+ protected DeviceInfo getDeviceInfoById(String deviceId) throws IOException {
+ prepareDeviceInfos();
+ return deviceInfoMap.get(deviceId);
}
- protected boolean containsDevice(String deviceId) {
- return false;
+ protected boolean containsDevice(String deviceId) throws IOException {
+ prepareDeviceInfos();
+ return deviceInfoMap.containsKey(deviceId);
}
}
@@ -197,5 +234,11 @@ public class CrossSpaceCompactionCandidate {
protected String deviceId;
protected long startTime;
protected long endTime;
+
+ public DeviceInfo(String deviceId, long startTime, long endTime) {
+ this.deviceId = deviceId;
+ this.startTime = startTime;
+ this.endTime = endTime;
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index a73ada0e33..2f9e194130 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -411,6 +411,21 @@ public class TsFileResource {
return timeIndex.getDevices(file.getPath(), this);
}
+ public DeviceTimeIndex buildDeviceTimeIndex() throws IOException {
+ readLock();
+ try (InputStream inputStream =
+ FSFactoryProducer.getFSFactory()
+ .getBufferedInputStream(file.getPath() + TsFileResource.RESOURCE_SUFFIX)) {
+ DeviceTimeIndex deviceTimeIndex = new DeviceTimeIndex();
+ return deviceTimeIndex.deserialize(inputStream);
+ } catch (Exception e) {
+ throw new IOException(
+ "Can't read file " + file.getPath() + TsFileResource.RESOURCE_SUFFIX + " from disk", e);
+ } finally {
+ readUnlock();
+ }
+ }
+
/**
* Whether this TsFileResource contains this device, if false, it must not contain this device, if
* true, it may or may not contain this device
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/timeindex/DeviceTimeIndex.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/timeindex/DeviceTimeIndex.java
index 3a745f532d..2d50171bcb 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/timeindex/DeviceTimeIndex.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/timeindex/DeviceTimeIndex.java
@@ -150,6 +150,10 @@ public class DeviceTimeIndex implements ITimeIndex {
endTimes = Arrays.copyOfRange(endTimes, 0, deviceToIndex.size());
}
+ public Set<String> getDevices() {
+ return deviceToIndex.keySet();
+ }
+
@Override
public Set<String> getDevices(String tsFilePath, TsFileResource tsFileResource) {
return deviceToIndex.keySet();
[iotdb] 01/02: fix the bug when checking target seq files
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/refine_cross_selection
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5a8e43df3d0d4de881dd6747a2638cc858866894
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Thu Dec 22 00:14:01 2022 +0800
fix the bug when checking target seq files
---
.../cross/rewrite/CrossSpaceCompactionCandidate.java | 18 ++++++------------
1 file changed, 6 insertions(+), 12 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java
index 4ff2759af4..f3ddb7c230 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java
@@ -70,12 +70,9 @@ public class CrossSpaceCompactionCandidate {
private boolean prepareNextSplit() {
TsFileResourceCandidate unseqFile = unseqFiles.get(nextUnseqFileIndex);
List<TsFileResourceCandidate> ret = new ArrayList<>();
- boolean lastTargetSeqFileSelected = false;
- for (TsFileResourceCandidate seqFile : seqFiles) {
- if (lastTargetSeqFileSelected) {
- break;
- }
- for (DeviceInfo unseqDeviceInfo : unseqFile.getDevices()) {
+
+ for (DeviceInfo unseqDeviceInfo : unseqFile.getDevices()) {
+ for (TsFileResourceCandidate seqFile : seqFiles) {
if (!seqFile.containsDevice(unseqDeviceInfo.deviceId)) {
continue;
}
@@ -92,24 +89,21 @@ public class CrossSpaceCompactionCandidate {
// avoid duplication selection
if (!seqFile.selected) {
ret.add(seqFile);
+ seqFile.markAsSelected();
}
- // prepare
- lastTargetSeqFileSelected = true;
+ // if this condition is satisfied, all subsequent seq files is unnecessary to check
break;
} else if (unseqDeviceInfo.startTime <= seqDeviceInfo.endTime) {
if (!seqFile.selected) {
ret.add(seqFile);
+ seqFile.markAsSelected();
}
- break;
}
}
}
// mark candidates in next split as selected even though it may not be added to the final
// TaskResource
unseqFile.markAsSelected();
- for (TsFileResourceCandidate fileCandidate : ret) {
- fileCandidate.markAsSelected();
- }
nextSplit = new CrossCompactionTaskResourceSplit(unseqFile, ret);
return true;
}