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 2023/01/29 08:13:37 UTC
[iotdb] branch geely_1.0.1 updated (1e59f59b5b -> a5efa3fa01)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a change to branch geely_1.0.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
discard 1e59f59b5b [IOTDB-5338] WAL buffer flush threshold optimaztion (#8832)
discard 625ea5da01 [IOTDB-5401] Reduce the estimated size of memory retained after calling next in SeriesAggregationScanOperator
discard 0c13eca957 add error log when target partition list is emtpy
add 573097af3f [To rc/1.0.1] [IOTDB-5266] Refine the code of cross selector and fix the bug when using FileTImeIndex (#8899)
new 7169d47fe6 add error log when target partition list is emtpy
new 2c8391bda9 [IOTDB-5401] Reduce the estimated size of memory retained after calling next in SeriesAggregationScanOperator
new a5efa3fa01 [IOTDB-5338] WAL buffer flush threshold optimaztion (#8832)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (1e59f59b5b)
\
N -- N -- N refs/heads/geely_1.0.1 (a5efa3fa01)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 3 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:
.../db/engine/compaction/CompactionScheduler.java | 8 +-
.../compaction/cross/ICrossSpaceSelector.java | 4 +-
.../cross/rewrite/CrossCompactionTaskResource.java | 119 +++
.../rewrite/CrossSpaceCompactionCandidate.java | 259 ++++++
.../rewrite/CrossSpaceCompactionResource.java | 74 --
.../RewriteCrossSpaceCompactionSelector.java | 340 ++------
.../compaction/task/ICompactionSelector.java | 4 +-
.../db/engine/storagegroup/TsFileResource.java | 19 +
.../storagegroup/timeindex/DeviceTimeIndex.java | 4 +
.../apache/iotdb/db/exception/MergeException.java | 2 +-
.../CrossSpaceCompactionWithFastPerformerTest.java | 21 +-
...eCompactionWithFastPerformerValidationTest.java | 937 +++++++++++----------
...sSpaceCompactionWithReadPointPerformerTest.java | 21 +-
...actionWithReadPointPerformerValidationTest.java | 906 ++++++++++----------
.../engine/compaction/cross/MergeUpgradeTest.java | 4 +-
.../cross/RewriteCompactionFileSelectorTest.java | 121 +--
16 files changed, 1508 insertions(+), 1335 deletions(-)
create mode 100644 server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossCompactionTaskResource.java
create mode 100644 server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionCandidate.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/CrossSpaceCompactionResource.java
[iotdb] 01/03: add error log when target partition list is emtpy
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch geely_1.0.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7169d47fe6dcfec51c00e9d3fac9f82089551eae
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Fri Jan 13 21:00:57 2023 +0800
add error log when target partition list is emtpy
---
.../java/org/apache/iotdb/commons/partition/DataPartition.java | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
index ffdc345e1d..73d2a36780 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
@@ -23,6 +23,9 @@ import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.utils.PathUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -33,6 +36,7 @@ import java.util.stream.Collectors;
// TODO: Remove this class
public class DataPartition extends Partition {
+ private static final Logger logger = LoggerFactory.getLogger(DataPartition.class);
public static final TRegionReplicaSet NOT_ASSIGNED = new TRegionReplicaSet();
// Map<StorageGroup, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionMessage>>>>
private Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
@@ -99,6 +103,11 @@ public class DataPartition extends Partition {
for (TTimePartitionSlot timePartitionSlot : timePartitionSlotList) {
List<TRegionReplicaSet> targetRegionList = slotReplicaSetMap.get(timePartitionSlot);
if (targetRegionList == null || targetRegionList.size() == 0) {
+ logger.error(
+ "met error when calculating partition for writing. device: {}, target timePartitionSlotList: {}, dataPartitionMap: {}",
+ deviceName,
+ timePartitionSlotList,
+ dataPartitionMap);
throw new RuntimeException(
String.format(
"targetRegionList is empty. device: %s, timeSlot: %s",
[iotdb] 03/03: [IOTDB-5338] WAL buffer flush threshold optimaztion (#8832)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch geely_1.0.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit a5efa3fa01c3e056d8d26fa20f123f2a2e4455ca
Author: Alan Choo <43...@users.noreply.github.com>
AuthorDate: Sun Jan 29 16:09:39 2023 +0800
[IOTDB-5338] WAL buffer flush threshold optimaztion (#8832)
---
.../org/apache/iotdb/db/wal/buffer/WALBuffer.java | 58 +++++++++++++---------
1 file changed, 35 insertions(+), 23 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
index c1ef5932e4..db2d627ef5 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
@@ -58,6 +58,7 @@ public class WALBuffer extends AbstractWALBuffer {
private static final Logger logger = LoggerFactory.getLogger(WALBuffer.class);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final int HALF_WAL_BUFFER_SIZE = config.getWalBufferSize() / 2;
+ private static final double FSYNC_BUFFER_RATIO = 0.95;
private static final int QUEUE_CAPACITY = config.getWalBufferQueueCapacity();
/** whether close method is called */
@@ -142,9 +143,9 @@ public class WALBuffer extends AbstractWALBuffer {
/** This task serializes WALEntry to workingBuffer and will call fsync at last. */
private class SerializeTask implements Runnable {
- private final ByteBufferView byteBufferVew = new ByteBufferView();
+ private final ByteBufferView byteBufferView = new ByteBufferView();
private final SerializeInfo info = new SerializeInfo();
- private int batchSize = 0;
+ private int totalSize = 0;
@Override
public void run() {
@@ -172,20 +173,21 @@ public class WALBuffer extends AbstractWALBuffer {
Thread.currentThread().interrupt();
}
- // for better fsync performance, sleep a while to enlarge write batch
- long fsyncDelay = config.getFsyncWalDelayInMs();
- if (fsyncDelay > 0) {
+ // try to get more WALEntries with blocking interface to enlarge write batch
+ while (totalSize < HALF_WAL_BUFFER_SIZE * FSYNC_BUFFER_RATIO) {
+ WALEntry walEntry = null;
try {
- Thread.sleep(fsyncDelay);
+ // for better fsync performance, wait a while to enlarge write batch
+ walEntry = walEntries.poll(config.getFsyncWalDelayInMs(), TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
- logger.warn("Interrupted when sleeping a while to enlarge wal write batch.");
+ logger.warn(
+ "Interrupted when waiting for taking WALEntry from blocking queue to serialize.");
Thread.currentThread().interrupt();
}
- }
- // try to get more WALEntries with non-blocking interface to enlarge write batch
- while (walEntries.peek() != null && batchSize < QUEUE_CAPACITY) {
- WALEntry walEntry = walEntries.poll();
+ if (walEntry == null) {
+ break;
+ }
boolean returnFlag = handleWALEntry(walEntry);
if (returnFlag) {
return;
@@ -193,7 +195,7 @@ public class WALBuffer extends AbstractWALBuffer {
}
// call fsync at last and set fsyncListeners
- if (batchSize > 0) {
+ if (totalSize > 0) {
fsyncWorkingBuffer(currentSearchIndex, currentFileStatus, info);
}
}
@@ -209,7 +211,6 @@ public class WALBuffer extends AbstractWALBuffer {
boolean success = handleInfoEntry(walEntry);
if (success) {
- ++batchSize;
info.fsyncListeners.add(walEntry.getWalFlushListener());
}
return false;
@@ -221,11 +222,10 @@ public class WALBuffer extends AbstractWALBuffer {
* @return true if serialization is successful.
*/
private boolean handleInfoEntry(WALEntry walEntry) {
- int size = byteBufferVew.position();
+ int size = byteBufferView.position();
try {
- walEntry.serialize(byteBufferVew);
- size = byteBufferVew.position() - size;
- logger.debug("wal entry size is: {}", size);
+ walEntry.serialize(byteBufferView);
+ size = byteBufferView.position() - size;
} catch (Exception e) {
logger.error(
"Fail to serialize WALEntry to wal node-{}'s buffer, discard it.", identifier, e);
@@ -245,6 +245,7 @@ public class WALBuffer extends AbstractWALBuffer {
currentFileStatus = WALFileStatus.CONTAINS_SEARCH_INDEX;
}
}
+ totalSize += size;
info.metaData.add(size, searchIndex);
return true;
}
@@ -256,16 +257,20 @@ public class WALBuffer extends AbstractWALBuffer {
private boolean handleSignalEntry(WALSignalEntry walSignalEntry) {
switch (walSignalEntry.getType()) {
case ROLL_WAL_LOG_WRITER_SIGNAL:
- logger.debug("Handle roll log writer signal for wal node-{}.", identifier);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Handle roll log writer signal for wal node-{}.", identifier);
+ }
info.rollWALFileWriterListener = walSignalEntry.getWalFlushListener();
fsyncWorkingBuffer(currentSearchIndex, currentFileStatus, info);
return true;
case CLOSE_SIGNAL:
- logger.debug(
- "Handle close signal for wal node-{}, there are {} entries left.",
- identifier,
- walEntries.size());
- boolean dataExists = batchSize > 0;
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Handle close signal for wal node-{}, there are {} entries left.",
+ identifier,
+ walEntries.size());
+ }
+ boolean dataExists = totalSize > 0;
if (dataExists) {
fsyncWorkingBuffer(currentSearchIndex, currentFileStatus, info);
}
@@ -420,6 +425,13 @@ public class WALBuffer extends AbstractWALBuffer {
public void run() {
currentWALFileWriter.updateFileStatus(fileStatus);
+ if (logger.isDebugEnabled()) {
+ double usedRatio = (double) syncingBuffer.position() / syncingBuffer.capacity();
+ logger.debug(
+ "Sync wal buffer, forceFlag: {}, buffer used: {} / {} = {}%",
+ forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
+ }
+
// flush buffer to os
try {
currentWALFileWriter.write(syncingBuffer, info.metaData);
[iotdb] 02/03: [IOTDB-5401] Reduce the estimated size of memory retained after calling next in SeriesAggregationScanOperator
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch geely_1.0.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 2c8391bda9191ea26958d353604af541fc0f2d1d
Author: liuminghui233 <36...@users.noreply.github.com>
AuthorDate: Thu Jan 12 11:19:58 2023 +0800
[IOTDB-5401] Reduce the estimated size of memory retained after calling next in SeriesAggregationScanOperator
---
.../AbstractSeriesAggregationScanOperator.java | 8 ++---
.../mpp/execution/operator/OperatorMemoryTest.java | 35 ++++++++++------------
2 files changed, 19 insertions(+), 24 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java
index 5f285de538..3566e1d834 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/source/AbstractSeriesAggregationScanOperator.java
@@ -64,7 +64,7 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
protected boolean finished = false;
- private final long maxRetainedSize;
+ private final long cachedRawDataSize;
private final long maxReturnSize;
public AbstractSeriesAggregationScanOperator(
@@ -92,14 +92,14 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
}
this.resultTsBlockBuilder = new TsBlockBuilder(dataTypes);
- this.maxRetainedSize =
+ this.cachedRawDataSize =
(1L + subSensorSize) * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
this.maxReturnSize = maxReturnSize;
}
@Override
public long calculateMaxPeekMemory() {
- return maxRetainedSize + maxReturnSize;
+ return cachedRawDataSize + maxReturnSize;
}
@Override
@@ -109,7 +109,7 @@ public abstract class AbstractSeriesAggregationScanOperator extends AbstractData
@Override
public long calculateRetainedSizeAfterCallingNext() {
- return maxRetainedSize;
+ return isGroupByQuery ? cachedRawDataSize : 0;
}
@Override
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
index 8bc9f944a4..447a91c955 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
@@ -1149,16 +1149,13 @@ public class OperatorMemoryTest {
TimeColumn.SIZE_IN_BYTES_PER_POSITION
+ 512 * Byte.BYTES
+ LongColumn.SIZE_IN_BYTES_PER_POSITION;
- long expectedMaxRetainSize =
- 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
+ long cachedRawDataSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
assertEquals(
- expectedMaxReturnSize + expectedMaxRetainSize,
+ expectedMaxReturnSize + cachedRawDataSize,
seriesAggregationScanOperator1.calculateMaxPeekMemory());
assertEquals(expectedMaxReturnSize, seriesAggregationScanOperator1.calculateMaxReturnSize());
- assertEquals(
- expectedMaxRetainSize,
- seriesAggregationScanOperator1.calculateRetainedSizeAfterCallingNext());
+ assertEquals(0, seriesAggregationScanOperator1.calculateRetainedSizeAfterCallingNext());
// case2: without group by, step is PARTIAL
List<AggregationDescriptor> aggregationDescriptors2 =
@@ -1184,15 +1181,13 @@ public class OperatorMemoryTest {
TimeColumn.SIZE_IN_BYTES_PER_POSITION
+ 512 * Byte.BYTES
+ 2 * LongColumn.SIZE_IN_BYTES_PER_POSITION;
- expectedMaxRetainSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
+ cachedRawDataSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
assertEquals(
- expectedMaxReturnSize + expectedMaxRetainSize,
+ expectedMaxReturnSize + cachedRawDataSize,
seriesAggregationScanOperator2.calculateMaxPeekMemory());
assertEquals(expectedMaxReturnSize, seriesAggregationScanOperator2.calculateMaxReturnSize());
- assertEquals(
- expectedMaxRetainSize,
- seriesAggregationScanOperator2.calculateRetainedSizeAfterCallingNext());
+ assertEquals(0, seriesAggregationScanOperator2.calculateRetainedSizeAfterCallingNext());
long maxTsBlockLineNumber =
TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();
@@ -1229,14 +1224,14 @@ public class OperatorMemoryTest {
* (TimeColumn.SIZE_IN_BYTES_PER_POSITION
+ 512 * Byte.BYTES
+ LongColumn.SIZE_IN_BYTES_PER_POSITION);
- expectedMaxRetainSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
+ cachedRawDataSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
assertEquals(
- expectedMaxReturnSize + expectedMaxRetainSize,
+ expectedMaxReturnSize + cachedRawDataSize,
seriesAggregationScanOperator3.calculateMaxPeekMemory());
assertEquals(expectedMaxReturnSize, seriesAggregationScanOperator3.calculateMaxReturnSize());
assertEquals(
- expectedMaxRetainSize,
+ cachedRawDataSize,
seriesAggregationScanOperator3.calculateRetainedSizeAfterCallingNext());
// case4: with group by, total window num > 1000
@@ -1267,14 +1262,14 @@ public class OperatorMemoryTest {
* (TimeColumn.SIZE_IN_BYTES_PER_POSITION
+ 512 * Byte.BYTES
+ LongColumn.SIZE_IN_BYTES_PER_POSITION));
- expectedMaxRetainSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
+ cachedRawDataSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
assertEquals(
- expectedMaxReturnSize + expectedMaxRetainSize,
+ expectedMaxReturnSize + cachedRawDataSize,
seriesAggregationScanOperator4.calculateMaxPeekMemory());
assertEquals(expectedMaxReturnSize, seriesAggregationScanOperator4.calculateMaxReturnSize());
assertEquals(
- expectedMaxRetainSize,
+ cachedRawDataSize,
seriesAggregationScanOperator4.calculateRetainedSizeAfterCallingNext());
// case5: over DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES
@@ -1303,14 +1298,14 @@ public class OperatorMemoryTest {
typeProvider);
expectedMaxReturnSize = DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES;
- expectedMaxRetainSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
+ cachedRawDataSize = 2L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
assertEquals(
- expectedMaxReturnSize + expectedMaxRetainSize,
+ expectedMaxReturnSize + cachedRawDataSize,
seriesAggregationScanOperator5.calculateMaxPeekMemory());
assertEquals(expectedMaxReturnSize, seriesAggregationScanOperator5.calculateMaxReturnSize());
assertEquals(
- expectedMaxRetainSize,
+ cachedRawDataSize,
seriesAggregationScanOperator5.calculateRetainedSizeAfterCallingNext());
} catch (IllegalPathException e) {
e.printStackTrace();