You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2019/06/13 12:48:14 UTC
[incubator-iotdb] branch feature_async_close_tsfile updated
(3b57abf -> 5ffd905)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
from 3b57abf add WAL and write .restore cost time log
new 21e4980 remove mergeMetadata in TsFileIoWriter
new 5ffd905 resolve conflict in overflow
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:
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 10 ++--
.../java/org/apache/iotdb/db/engine/Processor.java | 2 +-
.../engine/bufferwrite/BufferWriteProcessor.java | 63 +++++++++++-----------
.../bufferwrite/RestorableTsFileIOWriter.java | 32 +++++------
.../iotdb/db/engine/filenode/FileNodeManager.java | 22 ++++----
.../db/engine/filenode/FileNodeProcessor.java | 14 ++---
.../db/engine/filenode/FileNodeProcessorStore.java | 4 +-
.../db/engine/memcontrol/FlushPartialPolicy.java | 4 +-
.../db/engine/memcontrol/ForceFLushAllPolicy.java | 2 +-
.../db/engine/memtable/MemTableFlushTask.java | 18 +++----
.../iotdb/db/engine/overflow/io/OverflowIO.java | 3 +-
.../db/engine/overflow/io/OverflowProcessor.java | 22 ++++----
.../db/engine/overflow/io/OverflowResource.java | 2 +-
.../apache/iotdb/db/engine/pool/FlushManager.java | 8 +--
.../org/apache/iotdb/db/service/TSServiceImpl.java | 2 +-
.../iotdb/db/writelog/node/WriteLogNode.java | 4 +-
.../recover/ExclusiveLogRecoverPerformer.java | 2 +-
.../bufferwrite/BufferWriteProcessorNewTest.java | 8 +--
.../bufferwrite/BufferWriteProcessorTest.java | 12 ++---
.../bufferwrite/RestorableTsFileIOWriterTest.java | 11 ++--
.../memcontrol/BufferwriteFileSizeControlTest.java | 2 +-
.../memcontrol/BufferwriteMetaSizeControlTest.java | 2 +-
.../db/engine/memcontrol/IoTDBMemControlTest.java | 2 +-
.../memcontrol/OverflowFileSizeControlTest.java | 2 +-
.../memcontrol/OverflowMetaSizeControlTest.java | 2 +-
.../engine/overflow/io/OverflowProcessorTest.java | 8 +--
.../engine/overflow/io/OverflowResourceTest.java | 2 +-
.../apache/iotdb/db/integration/IOTDBFillIT.java | 8 +--
.../iotdb/db/integration/IOTDBGroupByIT.java | 8 +--
.../iotdb/db/integration/IoTDBAggregationIT.java | 6 +--
.../integration/IoTDBAggregationLargeDataIT.java | 2 +-
.../iotdb/db/integration/IoTDBDeletionIT.java | 2 +-
.../db/integration/IoTDBEngineTimeGeneratorIT.java | 2 +-
.../db/integration/IoTDBFloatPrecisionIT.java | 2 +-
.../db/integration/IoTDBFlushQueryMergeTest.java | 2 +-
.../iotdb/db/integration/IoTDBLargeDataIT.java | 4 +-
.../iotdb/db/integration/IoTDBMultiSeriesIT.java | 4 +-
.../db/integration/IoTDBSequenceDataQueryIT.java | 2 +-
.../iotdb/db/integration/IoTDBSeriesReaderIT.java | 4 +-
.../iotdb/db/integration/IoTDBVersionIT.java | 2 +-
.../iotdb/db/sync/sender/SingleClientSyncTest.java | 46 ++++++++--------
.../write/writer/NativeRestorableIOWriter.java | 2 +-
.../iotdb/tsfile/write/writer/TsFileIOWriter.java | 32 ++++-------
43 files changed, 184 insertions(+), 209 deletions(-)
[incubator-iotdb] 02/02: resolve conflict in overflow
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 5ffd9051e925caf362738ba8bee313d3c9fa5603
Merge: 21e4980 3b57abf
Author: qiaojialin <64...@qq.com>
AuthorDate: Thu Jun 13 20:47:57 2019 +0800
resolve conflict in overflow
.../iotdb/db/conf/directories/Directories.java | 5 +
.../engine/bufferwrite/BufferWriteProcessor.java | 6 +-
.../bufferwrite/RestorableTsFileIOWriter.java | 2 +
.../db/engine/overflow/io/OverflowMemtable.java | 9 +-
.../db/engine/overflow/io/OverflowProcessor.java | 25 +--
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 13 ++
.../physical/transfer/PhysicalPlanLogTransfer.java | 2 +-
.../org/apache/iotdb/db/writelog/io/LogWriter.java | 4 +
.../writelog/manager/MultiFileLogNodeManager.java | 36 +---
.../db/writelog/manager/WriteLogNodeManager.java | 5 +-
.../db/writelog/node/ExclusiveWriteLogNode.java | 181 +++++++++++----------
.../iotdb/db/writelog/node/WriteLogNode.java | 14 +-
.../engine/overflow/io/OverflowResourceTest.java | 8 +-
.../transfer/PhysicalPlanLogTransferTest.java | 29 ++--
.../org/apache/iotdb/db/tools/WalCheckerTest.java | 6 +-
.../iotdb/db/writelog/IoTDBLogFileSizeTest.java | 6 +-
.../apache/iotdb/db/writelog/PerformanceTest.java | 17 +-
.../org/apache/iotdb/db/writelog/RecoverTest.java | 12 +-
.../iotdb/db/writelog/WriteLogNodeManagerTest.java | 4 +-
.../apache/iotdb/db/writelog/WriteLogNodeTest.java | 18 +-
.../iotdb/db/writelog/io/LogWriterReaderTest.java | 5 +-
21 files changed, 192 insertions(+), 215 deletions(-)
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriter.java
index cb4007a,c327cae..6a492c9
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriter.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriter.java
@@@ -150,9 -150,10 +150,10 @@@ public class RestorableTsFileIOWriter e
}
}
- private void writeRestoreInfo() throws IOException {
+ public void writeRestoreInfo() throws IOException {
+ long start = System.currentTimeMillis();
long lastPosition = this.getPos();
- // TODO: no need to create a TsRowGroupBlockMetadata, flush RowGroupMetadata one by one is ok
+ // TODO: no need to create a TsRowGroupBlockMetadata, flushMetadata RowGroupMetadata one by one is ok
TsDeviceMetadata tsDeviceMetadata = new TsDeviceMetadata();
this.getAppendedRowGroupMetadata();
tsDeviceMetadata.setChunkGroupMetadataList(this.append);
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
index 2cf1294,173f615..5e3b777
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
@@@ -509,7 -509,7 +509,8 @@@ public class OverflowProcessor extends
filenodeFlushAction.act();
// write-ahead log
if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
- logNode.notifyEndFlush(null, walTaskId);
- logNode.notifyEndFlush(null, walTaskId, workResource.getInsertFile().getName());
++ //TODO
++// logNode.notifyEndFlush(null, walTaskId, workResource.getInsertFile().getName());
}
result = true;
} catch (IOException e) {
@@@ -570,14 -570,14 +571,15 @@@
throw new IOException(e);
}
long taskId = 0;
-- if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
-- try {
- taskId = logNode.notifyStartFlush();
- taskId = logNode.notifyStartFlush(workResource.getInsertFile().getName());
-- } catch (IOException e) {
-- LOGGER.error("Overflow processor {} encountered an error when notifying log node, {}",
-- getProcessorName(), e);
-- }
-- }
++// if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
++// try {
++ //TODO
++// taskId = logNode.notifyStartFlush(workResource.getInsertFile().getName());
++// } catch (IOException e) {
++// LOGGER.error("Overflow processor {} encountered an error when notifying log node, {}",
++// getProcessorName(), e);
++// }
++// }
final long walTaskId = taskId;
BasicMemController.getInstance().releaseUsage(this, memSize.get());
memSize.set(0);
diff --cc iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
index fa27a54,0ea4982..780baf6
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
@@@ -51,24 -49,18 +49,18 @@@ public interface WriteLogNode
*/
void forceSync() throws IOException;
- /*
- * Force OS to sync all written data to disk.
- */
- void force() throws IOException;
-
/**
- * When a FileNode attempts to start a flush, this method must be called to rename log file.
+ * When a FileNode attempts to start a flushMetadata, this method must be called to rename log file.
* @return the task id ( being used in the renamed log file)
*/
long notifyStartFlush() throws IOException;
/**
- * When the flush of a FlieNode ends, this method must be called to check if log file needs
+ * When the flushMetadata of a FlieNode ends, this method must be called to check if log file needs
* cleaning.
- * @param logPositions (deprecated)
- * @param taskId the task id that notifyStartFlush() returns.
+ * @param fileId
*/
- void notifyEndFlush(List<LogPosition> logPositions, long taskId);
+ void notifyEndFlush(long fileId);
/**
* return identifier of the log node.
diff --cc iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowResourceTest.java
index 24509f1,49457cf..3a4e013
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowResourceTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowResourceTest.java
@@@ -55,6 -57,10 +57,10 @@@ public class OverflowResourceTest
EnvironmentUtils.cleanDir(folderPath);
}
+ private void removeFlushedMemTable(IMemTable memTable, TsFileIOWriter overflowIOWriter) {
- overflowIOWriter.mergeChunkGroupMetaData();
++// overflowIOWriter.mergeChunkGroupMetaData();
+ }
+
@Test
public void testOverflowInsert() throws IOException {
OverflowTestUtils.produceInsertData(memtable);
[incubator-iotdb] 01/02: remove mergeMetadata in TsFileIoWriter
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 21e4980b92a0a97a005afbb95f7858d4c0b78908
Author: qiaojialin <64...@qq.com>
AuthorDate: Wed Jun 12 22:46:57 2019 +0800
remove mergeMetadata in TsFileIoWriter
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 10 ++--
.../java/org/apache/iotdb/db/engine/Processor.java | 2 +-
.../engine/bufferwrite/BufferWriteProcessor.java | 63 +++++++++++-----------
.../bufferwrite/RestorableTsFileIOWriter.java | 32 +++++------
.../iotdb/db/engine/filenode/FileNodeManager.java | 22 ++++----
.../db/engine/filenode/FileNodeProcessor.java | 14 ++---
.../db/engine/filenode/FileNodeProcessorStore.java | 4 +-
.../db/engine/memcontrol/FlushPartialPolicy.java | 4 +-
.../db/engine/memcontrol/ForceFLushAllPolicy.java | 2 +-
.../db/engine/memtable/MemTableFlushTask.java | 18 +++----
.../iotdb/db/engine/overflow/io/OverflowIO.java | 3 +-
.../db/engine/overflow/io/OverflowProcessor.java | 2 +-
.../db/engine/overflow/io/OverflowResource.java | 2 +-
.../apache/iotdb/db/engine/pool/FlushManager.java | 8 +--
.../org/apache/iotdb/db/service/TSServiceImpl.java | 2 +-
.../iotdb/db/writelog/node/WriteLogNode.java | 4 +-
.../recover/ExclusiveLogRecoverPerformer.java | 2 +-
.../bufferwrite/BufferWriteProcessorNewTest.java | 8 +--
.../bufferwrite/BufferWriteProcessorTest.java | 12 ++---
.../bufferwrite/RestorableTsFileIOWriterTest.java | 11 ++--
.../memcontrol/BufferwriteFileSizeControlTest.java | 2 +-
.../memcontrol/BufferwriteMetaSizeControlTest.java | 2 +-
.../db/engine/memcontrol/IoTDBMemControlTest.java | 2 +-
.../memcontrol/OverflowFileSizeControlTest.java | 2 +-
.../memcontrol/OverflowMetaSizeControlTest.java | 2 +-
.../engine/overflow/io/OverflowProcessorTest.java | 8 +--
.../apache/iotdb/db/integration/IOTDBFillIT.java | 8 +--
.../iotdb/db/integration/IOTDBGroupByIT.java | 8 +--
.../iotdb/db/integration/IoTDBAggregationIT.java | 6 +--
.../integration/IoTDBAggregationLargeDataIT.java | 2 +-
.../iotdb/db/integration/IoTDBDeletionIT.java | 2 +-
.../db/integration/IoTDBEngineTimeGeneratorIT.java | 2 +-
.../db/integration/IoTDBFloatPrecisionIT.java | 2 +-
.../db/integration/IoTDBFlushQueryMergeTest.java | 2 +-
.../iotdb/db/integration/IoTDBLargeDataIT.java | 4 +-
.../iotdb/db/integration/IoTDBMultiSeriesIT.java | 4 +-
.../db/integration/IoTDBSequenceDataQueryIT.java | 2 +-
.../iotdb/db/integration/IoTDBSeriesReaderIT.java | 4 +-
.../iotdb/db/integration/IoTDBVersionIT.java | 2 +-
.../iotdb/db/sync/sender/SingleClientSyncTest.java | 46 ++++++++--------
.../write/writer/NativeRestorableIOWriter.java | 2 +-
.../iotdb/tsfile/write/writer/TsFileIOWriter.java | 32 ++++-------
42 files changed, 172 insertions(+), 199 deletions(-)
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 4bf5ac8..275ee80 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -161,19 +161,19 @@ public class IoTDBConfig {
private long periodTimeForMerge = 7200;
/**
- * When set true, start timed flush and merge service. Else, stop timed flush and merge service.
+ * When set true, start timed flushMetadata and merge service. Else, stop timed flushMetadata and merge service.
* The default value is true. TODO: 'timed' better explains this than 'timing'.
*/
private boolean enableTimingCloseAndMerge = true;
/**
- * How many threads can concurrently flush. When <= 0, use CPU core number.
+ * How many threads can concurrently flushMetadata. When <= 0, use CPU core number.
*/
private int concurrentFlushThread = Runtime.getRuntime().availableProcessors();
private ZoneId zoneID = ZoneId.systemDefault();
/**
- * BufferWriteProcessor and OverflowProcessor will immediately flush if this threshold is
+ * BufferWriteProcessor and OverflowProcessor will immediately flushMetadata if this threshold is
* reached.
*/
private long memThresholdWarning = (long) (0.5 * Runtime.getRuntime().maxMemory());
@@ -183,7 +183,7 @@ public class IoTDBConfig {
private long memThresholdDangerous = (long) (0.6 * Runtime.getRuntime().maxMemory());
/**
* MemMonitorThread will check every such interval(in ms). If memThresholdWarning is reached,
- * MemMonitorThread will inform FileNodeManager to flush.
+ * MemMonitorThread will inform FileNodeManager to flushMetadata.
*/
private long memMonitorInterval = 1000;
/**
@@ -218,7 +218,7 @@ public class IoTDBConfig {
*/
private boolean enableSmallFlush = false;
/**
- * The interval of small flush in ms.
+ * The interval of small flushMetadata in ms.
*/
private long smallFlushInterval = 60L * 1000;
/**
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/Processor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/Processor.java
index 1b333db..b5bd219 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/Processor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/Processor.java
@@ -199,7 +199,7 @@ public abstract class Processor {
public abstract boolean canBeClosed();
/**
- * call flush operation asynchronously
+ * call flushMetadata operation asynchronously
* @return a future that returns true if successfully, otherwise false.
* @throws IOException
*/
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java
index 02ad6f5..ad884dd 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java
@@ -75,10 +75,11 @@ public class BufferWriteProcessor extends Processor {
private volatile Future<Boolean> closeFuture = new BWCloseFuture(new ImmediateFuture<>(true));
private ReentrantLock flushQueryLock = new ReentrantLock();
private AtomicLong memSize = new AtomicLong();
+ // do not use TsFileConfig.groupSizeInByte, it will ignore the config file
private long memThreshold = TSFileDescriptor.getInstance().getConfig().groupSizeInByte;
private IMemTable workMemTable;
- // each flush task has a flushId, IO task should scheduled by this id
+ // each flushMetadata task has a flushId, IO task should scheduled by this id
private long flushId = -1;
private List<IMemTable> flushingMemTables = new ArrayList<>();
@@ -265,8 +266,7 @@ public class BufferWriteProcessor extends Processor {
* @return corresponding chunk data and chunk metadata in memory
*/
public Pair<ReadOnlyMemChunk, List<ChunkMetaData>> queryBufferWriteData(String deviceId,
- String measurementId, TSDataType dataType, Map<String, String> props)
- throws BufferWriteProcessorException {
+ String measurementId, TSDataType dataType, Map<String, String> props) {
flushQueryLock.lock();
try {
MemSeriesLazyMerger memSeriesLazyMerger = new MemSeriesLazyMerger();
@@ -291,12 +291,11 @@ public class BufferWriteProcessor extends Processor {
/**
* return the memtable to MemTablePool and make
* @param memTable
- * @param tsFileIOWriter
*/
private void removeFlushedMemTable(IMemTable memTable, TsFileIOWriter tsFileIOWriter) {
long start = System.currentTimeMillis();
this.writeLock();
- tsFileIOWriter.mergeChunkGroupMetaData();
+ ((RestorableTsFileIOWriter) tsFileIOWriter).makeMetadataVisible();
try {
flushingMemTables.remove(memTable);
} finally {
@@ -328,7 +327,7 @@ public class BufferWriteProcessor extends Processor {
displayMessage);
try {
if (tmpMemTableToFlush != null && !tmpMemTableToFlush.isEmpty()) {
- // flush data
+ // flushMetadata data
MemTableFlushTask tableFlushTask = new MemTableFlushTask(writer, getProcessorName(), flushId,
this::removeFlushedMemTable);
tableFlushTask.flushMemTable(fileSchema, tmpMemTableToFlush, version);
@@ -340,15 +339,15 @@ public class BufferWriteProcessor extends Processor {
}
result = true;
} catch (Exception e) {
- LOGGER.error("The bufferwrite processor {} failed to flush {}.", getProcessorName(), displayMessage, e);
+ LOGGER.error("The bufferwrite processor {} failed to flushMetadata {}.", getProcessorName(), displayMessage, e);
result = false;
}
if (LOGGER.isInfoEnabled()) {
long flushEndTime = System.currentTimeMillis();
LOGGER.info(
- "The bufferwrite processor {} flush {}, start time is {}, flush end time is {}, "
- + "flush time consumption is {}ms",
+ "The bufferwrite processor {} flushMetadata {}, start time is {}, flushMetadata end time is {}, "
+ + "flushMetadata time consumption is {}ms",
getProcessorName(), displayMessage,
DatetimeUtils.convertMillsecondToZonedDateTime(flushStartTime),
DatetimeUtils.convertMillsecondToZonedDateTime(flushEndTime),
@@ -357,22 +356,22 @@ public class BufferWriteProcessor extends Processor {
return result;
}
- // keyword synchronized is added in this method, so that only one flush task can be submitted now.
+ // keyword synchronized is added in this method, so that only one flushMetadata task can be submitted now.
@Override
public synchronized Future<Boolean> flush() throws IOException {
return flush(false);
}
- // keyword synchronized is added in this method, so that only one flush task can be submitted now.
+ // keyword synchronized is added in this method, so that only one flushMetadata task can be submitted now.
private Future<Boolean> flush(boolean isCloseTaskCalled) throws IOException {
- // statistic information for flush
+ // statistic information for flushMetadata
if (lastFlushTime > 0) {
if (LOGGER.isInfoEnabled()) {
long thisFlushTime = System.currentTimeMillis();
LOGGER.info(
- "The bufferwrite processor {} will submit a flush task."
- + "The last flush time is {}, this flush time is {}, "
- + "flush time interval is {}s", getProcessorName(),
+ "The bufferwrite processor {} will submit a flushMetadata task."
+ + "The last flushMetadata time is {}, this flushMetadata time is {}, "
+ + "flushMetadata time interval is {}s", getProcessorName(),
DatetimeUtils.convertMillsecondToZonedDateTime(lastFlushTime),
DatetimeUtils.convertMillsecondToZonedDateTime(thisFlushTime),
(thisFlushTime - lastFlushTime) / 1000);
@@ -381,11 +380,11 @@ public class BufferWriteProcessor extends Processor {
lastFlushTime = System.currentTimeMillis();
// check value count
if (valueCount > 0) {
- // update the lastUpdatetime, prepare for flush
+ // update the lastUpdatetime, prepare for flushMetadata
try {
bufferwriteFlushAction.act();
} catch (Exception e) {
- LOGGER.error("Failed to flush bufferwrite row group when calling the action function.");
+ LOGGER.error("Failed to flushMetadata bufferwrite row group when calling the action function.");
throw new IOException(e);
}
final long walTaskId;
@@ -404,7 +403,7 @@ public class BufferWriteProcessor extends Processor {
start = System.currentTimeMillis() - start;
if (start > 1000) {
- LOGGER.info("BufferWriteProcessor.flush getEmptyMemtable cost: {}", start);
+ LOGGER.info("BufferWriteProcessor.flushMetadata getEmptyMemtable cost: {}", start);
}
flushId++;
@@ -414,7 +413,7 @@ public class BufferWriteProcessor extends Processor {
// switch
if (isCloseTaskCalled) {
LOGGER.info(
- "flush memtable for bufferwrite processor {} synchronously for close task.",
+ "flushMetadata memtable for bufferwrite processor {} synchronously for close task.",
getProcessorName(), FlushManager.getInstance().getWaitingTasksNumber(),
FlushManager.getInstance().getCorePoolSize());
flushTask("synchronously", tmpMemTableToFlush, version, walTaskId, flushId);
@@ -422,7 +421,7 @@ public class BufferWriteProcessor extends Processor {
} else {
if (LOGGER.isInfoEnabled()) {
LOGGER.info(
- "Begin to submit flush task for bufferwrite processor {}, current Flush Queue is {}, core pool size is {}.",
+ "Begin to submit flushMetadata task for bufferwrite processor {}, current Flush Queue is {}, core pool size is {}.",
getProcessorName(), FlushManager.getInstance().getWaitingTasksNumber(),
FlushManager.getInstance().getCorePoolSize());
}
@@ -452,9 +451,7 @@ public class BufferWriteProcessor extends Processor {
@Override
public synchronized void close() throws BufferWriteProcessorException {
try {
- // flush data (if there are flushing task, flush() will be blocked)
- //Future<Boolean> flush = flush();
- //and wait for finishing flush async
+ // flushMetadata data (if there are flushing task, flushMetadata() will be blocked) and wait for finishing flushMetadata async
LOGGER.info("Submit a BufferWrite ({}) close task.", getProcessorName());
closeFuture = new BWCloseFuture(FlushManager.getInstance().submit(() -> closeTask()));
//now, we omit the future of the closeTask.
@@ -468,19 +465,19 @@ public class BufferWriteProcessor extends Processor {
private boolean closeTask() {
long closeStartTime = System.currentTimeMillis();
try {
- LOGGER.info("Bufferwrite {} Close Task: begin to wait for the flush.", getProcessorName());
+ LOGGER.info("Bufferwrite {} Close Task: begin to wait for the flushMetadata.", getProcessorName());
flush(true);
- LOGGER.info("Bufferwrite {} Close Task: finishing the flush.", getProcessorName());
+ LOGGER.info("Bufferwrite {} Close Task: finishing the flushMetadata.", getProcessorName());
// end file
writer.endFile(fileSchema);
- //FIXME suppose the flush-thread-pool is 2.
- // then if a flush task and a close task are running in the same time
- // and the close task is faster, then writer == null, and the flush task will throw nullpointer
- // exception. Add "synchronized" keyword on both flush and close may solve the issue.
+ //FIXME suppose the flushMetadata-thread-pool is 2.
+ // then if a flushMetadata task and a close task are running in the same time
+ // and the close task is faster, then writer == null, and the flushMetadata task will throw nullpointer
+ // exception. Add "synchronized" keyword on both flushMetadata and close may solve the issue.
writer = null;
// update the IntervalFile for interval list
bufferwriteCloseConsumer.accept(this);
- // flush the changed information for filenode
+ // flushMetadata the changed information for filenode
filenodeFlushAction.act();
// delete the restore for this bufferwrite processor
if (LOGGER.isInfoEnabled()) {
@@ -563,8 +560,8 @@ public class BufferWriteProcessor extends Processor {
}
/**
- * used for test. We can know when the flush() is called.
- * @return the last flush() time. Time unit: millisecond.
+ * used for test. We can know when the flushMetadata() is called.
+ * @return the last flushMetadata() time. Time unit: millisecond.
*/
public long getLastFlushTime() {
return lastFlushTime;
@@ -572,7 +569,7 @@ public class BufferWriteProcessor extends Processor {
/**
* used for test. We can block to wait for finishing flushing.
- * @return the future of the flush() task.
+ * @return the future of the flushMetadata() task.
*/
public Future<Boolean> getFlushFuture() {
return flushFuture;
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriter.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriter.java
index d0591af..cb4007a 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriter.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriter.java
@@ -100,14 +100,14 @@ public class RestorableTsFileIOWriter extends TsFileIOWriter {
// cut off tsfile
this.out = new DefaultTsFileOutput(new FileOutputStream(insertFile, true));
out.truncate(position);
- this.flushedChunkGroupMetaDataList = existedMetadatas;
- lastFlushedChunkGroupIndex = flushedChunkGroupMetaDataList.size();
+ this.chunkGroupMetaDataList = existedMetadatas;
+ lastFlushedChunkGroupIndex = chunkGroupMetaDataList.size();
append = new ArrayList<>();
// recovery the metadata
recoverMetadata(existedMetadatas);
LOGGER.info(
"Recover the bufferwrite processor {}, the tsfile seriesPath is {}, "
- + "the position of last flush is {}, the size of rowGroupMetadata is {}",
+ + "the position of last flushMetadata is {}, the size of rowGroupMetadata is {}",
processorName, insertFilePath, position, existedMetadatas.size());
isNewResource = false;
} else {
@@ -122,8 +122,8 @@ public class RestorableTsFileIOWriter extends TsFileIOWriter {
LOGGER.info("remove unsealed tsfile restore file failed: ", e);
}
this.out = new DefaultTsFileOutput(new FileOutputStream(insertFile));
- this.flushedChunkGroupMetaDataList = new ArrayList<>();
- lastFlushedChunkGroupIndex = flushedChunkGroupMetaDataList.size();
+ this.chunkGroupMetaDataList = new ArrayList<>();
+ lastFlushedChunkGroupIndex = chunkGroupMetaDataList.size();
append = new ArrayList<>();
startFile();
isNewResource = true;
@@ -150,9 +150,9 @@ public class RestorableTsFileIOWriter extends TsFileIOWriter {
}
}
- private void writeRestoreInfo() throws IOException {
+ public void writeRestoreInfo() throws IOException {
long lastPosition = this.getPos();
- // TODO: no need to create a TsRowGroupBlockMetadata, flush RowGroupMetadata one by one is ok
+ // TODO: no need to create a TsRowGroupBlockMetadata, flushMetadata RowGroupMetadata one by one is ok
TsDeviceMetadata tsDeviceMetadata = new TsDeviceMetadata();
this.getAppendedRowGroupMetadata();
tsDeviceMetadata.setChunkGroupMetadataList(this.append);
@@ -250,19 +250,15 @@ public class RestorableTsFileIOWriter extends TsFileIOWriter {
this.isNewResource = isNewResource;
}
- public void flush() throws IOException {
- writeRestoreInfo();
- }
-
/**
* add all appendChunkGroupMetadatas into memory. After calling this method, other classes can
* read these metadata.
*/
- public void appendMetadata() {
+ public void makeMetadataVisible() {
if (!append.isEmpty()) {
for (ChunkGroupMetaData rowGroupMetaData : append) {
for (ChunkMetaData chunkMetaData : rowGroupMetaData.getChunkMetaDataList()) {
- addInsertMetadata(rowGroupMetaData.getDeviceID(), chunkMetaData.getMeasurementUid(),
+ addChunkMetadata(rowGroupMetaData.getDeviceID(), chunkMetaData.getMeasurementUid(),
chunkMetaData);
}
}
@@ -270,7 +266,7 @@ public class RestorableTsFileIOWriter extends TsFileIOWriter {
}
}
- private void addInsertMetadata(String deviceId, String measurementId,
+ private void addChunkMetadata(String deviceId, String measurementId,
ChunkMetaData chunkMetaData) {
if (!metadatas.containsKey(deviceId)) {
metadatas.put(deviceId, new HashMap<>());
@@ -298,11 +294,11 @@ public class RestorableTsFileIOWriter extends TsFileIOWriter {
* @return a list of chunkgroup metadata
*/
private List<ChunkGroupMetaData> getAppendedRowGroupMetadata() {
- if (lastFlushedChunkGroupIndex < flushedChunkGroupMetaDataList.size()) {
+ if (lastFlushedChunkGroupIndex < chunkGroupMetaDataList.size()) {
append.clear();
- append.addAll(flushedChunkGroupMetaDataList
- .subList(lastFlushedChunkGroupIndex, flushedChunkGroupMetaDataList.size()));
- lastFlushedChunkGroupIndex = flushedChunkGroupMetaDataList.size();
+ append.addAll(chunkGroupMetaDataList
+ .subList(lastFlushedChunkGroupIndex, chunkGroupMetaDataList.size()));
+ lastFlushedChunkGroupIndex = chunkGroupMetaDataList.size();
}
return append;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java
index 1e6746b..3a49da5 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java
@@ -510,7 +510,7 @@ public class FileNodeManager implements IStatistic, IService {
try {
long start2_3 = System.currentTimeMillis();
- // write tsrecord and check flush
+ // write tsrecord and check flushMetadata
if (!bufferWriteProcessor.write(tsRecord)) {
start2_3 = System.currentTimeMillis() - start2_3;
if (start2_3 > 1000) {
@@ -1180,39 +1180,39 @@ public class FileNodeManager implements IStatistic, IService {
}
/**
- * force flush to control memory usage.
+ * force flushMetadata to control memory usage.
*/
public void forceFlush(BasicMemController.UsageLevel level) {
// you may add some delicate process like below
// or you could provide multiple methods for different urgency
switch (level) {
// only select the most urgent (most active or biggest in size)
- // processors to flush
- // only select top 10% active memory user to flush
+ // processors to flushMetadata
+ // only select top 10% active memory user to flushMetadata
case WARNING:
try {
flushTop(0.1f);
} catch (IOException e) {
- LOGGER.error("force flush memory data error: {}", e);
+ LOGGER.error("force flushMetadata memory data error: {}", e);
}
break;
- // force all processors to flush
+ // force all processors to flushMetadata
case DANGEROUS:
try {
flushAll();
} catch (IOException e) {
- LOGGER.error("force flush memory data error: {}", e);
+ LOGGER.error("force flushMetadata memory data error: {}", e);
}
break;
- // if the flush thread pool is not full ( or half full), start a new
- // flush task
+ // if the flushMetadata thread pool is not full ( or half full), start a new
+ // flushMetadata task
case SAFE:
if (FlushManager.getInstance().getActiveCnt() < 0.5 * FlushManager.getInstance()
.getThreadCnt()) {
try {
flushTop(0.01f);
} catch (IOException e) {
- LOGGER.error("force flush memory data error: ", e);
+ LOGGER.error("force flushMetadata memory data error: ", e);
}
}
break;
@@ -1261,7 +1261,7 @@ public class FileNodeManager implements IStatistic, IService {
processor.writeUnlock();
}
start = System.currentTimeMillis() - start;
- LOGGER.info("flush Top cost: {}", start);
+ LOGGER.info("flushMetadata Top cost: {}", start);
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
index a1ee26a..c9d1077 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
@@ -653,7 +653,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
/**
- * get flush last update time.
+ * get flushMetadata last update time.
*/
public long getFlushLastUpdateTime(String deviceId) {
if (!flushLastUpdateTimeMap.containsKey(deviceId)) {
@@ -847,12 +847,8 @@ public class FileNodeProcessor extends Processor implements IStatistic {
+ "but the bufferwrite processor is null.",
newFileNodes.get(newFileNodes.size() - 1).getFile().getAbsolutePath(), getProcessorName()));
}
- try {
- bufferwritedata = bufferWriteProcessor
- .queryBufferWriteData(deviceId, measurementId, dataType, mSchema.getProps());
- } catch (BufferWriteProcessorException e) {
- throw new FileNodeProcessorException(e);
- }
+ bufferwritedata = bufferWriteProcessor
+ .queryBufferWriteData(deviceId, measurementId, dataType, mSchema.getProps());
try {
List<Modification> pathModifications = context.getPathModifications(
@@ -1105,7 +1101,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
fileNodeProcessorStore.setFileNodeProcessorStatus(isMerging);
fileNodeProcessorStore.setNewFileNodes(newFileNodes);
fileNodeProcessorStore.setEmptyTsFileResource(emptyTsFileResource);
- // flush this filenode information
+ // flushMetadata this filenode information
try {
writeStoreToDisk(fileNodeProcessorStore);
} catch (FileNodeProcessorException e) {
@@ -1634,7 +1630,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
writeOneSeries(path.getDevice(), seriesWriterImpl, dataType,
seriesReader,
startTimeMap, endTimeMap, timeValuePair);
- // flush the series data
+ // flushMetadata the series data
seriesWriterImpl.writeToFileWriter(mergeFileWriter);
}
} finally {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java
index 7c1eca4..0873d69 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java
@@ -32,8 +32,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
/**
* FileNodeProcessorStore is used to store information about FileNodeProcessor's status.
- * lastUpdateTime is changed and stored by BufferWrite flush or BufferWrite close.
- * emptyTsFileResource and newFileNodes are changed and stored by Overflow flush and
+ * lastUpdateTime is changed and stored by BufferWrite flushMetadata or BufferWrite close.
+ * emptyTsFileResource and newFileNodes are changed and stored by Overflow flushMetadata and
* Overflow close. fileNodeProcessorState is changed and stored by the change of FileNodeProcessor's
* status such as "work->merge merge->wait wait->work". numOfMergeFile is changed
* and stored when FileNodeProcessor's status changes from work to merge.
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/FlushPartialPolicy.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/FlushPartialPolicy.java
index 31a5860..5bef4b5 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/FlushPartialPolicy.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/FlushPartialPolicy.java
@@ -26,7 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This class only gives a hint to FilenodeManager that it may flush some data to avoid rush hour.
+ * This class only gives a hint to FilenodeManager that it may flushMetadata some data to avoid rush hour.
*/
public class FlushPartialPolicy implements Policy {
@@ -50,7 +50,7 @@ public class FlushPartialPolicy implements Policy {
workerThread.start();
} else {
if (workerThread.isAlive()) {
- LOGGER.debug("Last flush is ongoing...");
+ LOGGER.debug("Last flushMetadata is ongoing...");
} else {
workerThread = createWorkerThread();
workerThread.start();
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/ForceFLushAllPolicy.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/ForceFLushAllPolicy.java
index 53533d9..14cc637 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/ForceFLushAllPolicy.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/ForceFLushAllPolicy.java
@@ -45,7 +45,7 @@ public class ForceFLushAllPolicy implements Policy {
workerThread.start();
} else {
if (workerThread.isAlive()) {
- logger.info("Last flush is ongoing...");
+ logger.info("Last flushMetadata is ongoing...");
} else {
workerThread = createWorkerThread();
workerThread.start();
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTask.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTask.java
index 0c103bb..1c89439 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTask.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/memtable/MemTableFlushTask.java
@@ -55,7 +55,7 @@ public class MemTableFlushTask {
this.flushCallBack = callBack;
ioFlushThread.start();
memoryFlushThread.start();
- LOGGER.info("Processor {} flush task created, flushId: {}", processorName, flushId);
+ LOGGER.info("Processor {} flushMetadata task created, flushId: {}", processorName, flushId);
}
@@ -69,7 +69,7 @@ public class MemTableFlushTask {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
- LOGGER.error("BufferWrite Processor {}, io flush task is interrupted.", processorName, e);
+ LOGGER.error("BufferWrite Processor {}, io flushMetadata task is interrupted.", processorName, e);
}
} else {
if (task instanceof String) {
@@ -114,12 +114,12 @@ public class MemTableFlushTask {
long currWaitIdx = waitedTime / 2000;
if (currWaitIdx > lastWaitIdx) {
lastWaitIdx = currWaitIdx;
- LOGGER.info("tsFileIoWriter flushID: {}, flush task flushID: {} has waited {}ms", currentTsFileFlushId,
+ LOGGER.info("tsFileIoWriter flushID: {}, flushMetadata task flushID: {} has waited {}ms", currentTsFileFlushId,
flushId, waitedTime);
}
Thread.sleep(10);
} catch (InterruptedException e) {
- LOGGER.error("Processor {}, last flush io task is not finished.", processorName, e);
+ LOGGER.error("Processor {}, last flushMetadata io task is not finished.", processorName, e);
}
}
while (!stop) {
@@ -128,7 +128,7 @@ public class MemTableFlushTask {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
- LOGGER.error("BufferWrite Processor {}, io flush task is interrupted.", processorName, e);
+ LOGGER.error("BufferWrite Processor {}, io flushMetadata task is interrupted.", processorName, e);
}
} else {
long starTime = System.currentTimeMillis();
@@ -155,15 +155,13 @@ public class MemTableFlushTask {
flushCallBack.afterFlush(memTable, tsFileIoWriter);
if (tsFileIoWriter instanceof RestorableTsFileIOWriter) {
try {
- RestorableTsFileIOWriter restorableTsFileIOWriter = (RestorableTsFileIOWriter) tsFileIoWriter;
- restorableTsFileIOWriter.flush();
- restorableTsFileIOWriter.appendMetadata();
+ ((RestorableTsFileIOWriter) tsFileIoWriter).writeRestoreInfo();
} catch (IOException e) {
LOGGER.error("write restore file meet error", e);
}
}
- // enable next flush task to IO
+ // enable next flushMetadata task to IO
long newId = tsFileIoWriter.getFlushID().incrementAndGet();
LOGGER.info("BufferWrite Processor {}, flushing a memtable into disk: io cost {}ms, new flushID in tsFileIoWriter: {}.",
processorName, ioTime, newId);
@@ -237,7 +235,7 @@ public class MemTableFlushTask {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
- LOGGER.error("BufferWrite Processor {}, flush memtable table thread is interrupted.",
+ LOGGER.error("BufferWrite Processor {}, flushMetadata memtable table thread is interrupted.",
processorName, e);
throw new RuntimeException(e);
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowIO.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowIO.java
index 84dfe2c..90f0423 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowIO.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowIO.java
@@ -43,8 +43,7 @@ public class OverflowIO extends TsFileIOWriter {
}
public void clearRowGroupMetadatas() {
- super.flushingChunkGroupMetaDataList.clear();
- super.flushedChunkGroupMetaDataList.clear();
+ super.chunkGroupMetaDataList.clear();
}
@Override
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
index 4317cbc..2cf1294 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
@@ -488,7 +488,7 @@ public class OverflowProcessor extends Processor {
private void removeFlushedMemTable(IMemTable memTable, TsFileIOWriter overflowIOWriter) {
this.writeLock();
- overflowIOWriter.mergeChunkGroupMetaData();
+ //TODO check this implementation in BufferWriteProcessor
try {
overflowFlushMemTables.remove(memTable);
} finally {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowResource.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowResource.java
index a37653c..36f2f93 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowResource.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowResource.java
@@ -220,7 +220,7 @@ public class OverflowResource {
if (LOGGER.isInfoEnabled()) {
LOGGER.info(
"Overflow processor {} flushes overflow insert data, actual:{}, time consumption:{} ms,"
- + " flush rate:{}/s",
+ + " flushMetadata rate:{}/s",
processorName, MemUtils.bytesCntToStr(insertSize), timeInterval,
MemUtils.bytesCntToStr(insertSize / timeInterval * 1000));
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/pool/FlushManager.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/pool/FlushManager.java
index 1028014..ead8e89 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/pool/FlushManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/pool/FlushManager.java
@@ -64,7 +64,7 @@ public class FlushManager {
}
/**
- * Refuse new flush submits and exit when all RUNNING THREAD in the pool end.
+ * Refuse new flushMetadata submits and exit when all RUNNING THREAD in the pool end.
*
* @param block
* if set to true, this method will wait for timeOut milliseconds.
@@ -83,14 +83,14 @@ public class FlushManager {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new ProcessorException("Interrupted while waiting flush thread pool to exit. "
+ throw new ProcessorException("Interrupted while waiting flushMetadata thread pool to exit. "
, e);
}
}
}
/**
- * Block new flush submits and exit when all RUNNING THREADS AND TASKS IN THE QUEUE end.
+ * Block new flushMetadata submits and exit when all RUNNING THREADS AND TASKS IN THE QUEUE end.
*
* @param block
* if set to true, this method will wait for timeOut milliseconds.
@@ -109,7 +109,7 @@ public class FlushManager {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new ProcessorException("Interrupted while waiting flush thread pool to exit. ", e);
+ throw new ProcessorException("Interrupted while waiting flushMetadata thread pool to exit. ", e);
}
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index c502f46..20f9cdf 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -414,7 +414,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
statement = statement.toLowerCase();
switch (statement) {
- case "flush":
+ case "flushMetadata":
try {
FileNodeManager.getInstance().closeAll();
} catch (FileNodeManagerException e) {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
index b2e7fc3..fa27a54 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
@@ -57,13 +57,13 @@ public interface WriteLogNode {
void force() throws IOException;
/**
- * When a FileNode attempts to start a flush, this method must be called to rename log file.
+ * When a FileNode attempts to start a flushMetadata, this method must be called to rename log file.
* @return the task id ( being used in the renamed log file)
*/
long notifyStartFlush() throws IOException;
/**
- * When the flush of a FlieNode ends, this method must be called to check if log file needs
+ * When the flushMetadata of a FlieNode ends, this method must be called to check if log file needs
* cleaning.
* @param logPositions (deprecated)
* @param taskId the task id that notifyStartFlush() returns.
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java
index 65c29a8..d9a82fc 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java
@@ -316,7 +316,7 @@ public class ExclusiveLogRecoverPerformer implements RecoverPerformer {
try {
FileNodeManager.getInstance().closeOneFileNode(writeLogNode.getFileNodeName());
} catch (FileNodeManagerException e) {
- throw new RecoverException(String.format("Log node %s cannot perform flush"
+ throw new RecoverException(String.format("Log node %s cannot perform flushMetadata"
+ " after replaying logs!", writeLogNode.getIdentifier()), e);
}
currStage = CLEAN_UP;
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java
index a826495..400a4d8 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorNewTest.java
@@ -125,17 +125,17 @@ public class BufferWriteProcessorNewTest {
assertEquals(num, timeValuePair.getValue().getInt());
}
long lastFlushTime = bufferwrite.getLastFlushTime();
- // flush asynchronously
+ // flushMetadata asynchronously
bufferwrite.flush();
assertTrue(bufferwrite.getLastFlushTime() != lastFlushTime);
assertTrue(bufferwrite.canBeClosed());
- // waiting for the end of flush.
+ // waiting for the end of flushMetadata.
try {
bufferwrite.getFlushFuture().get(10, TimeUnit.SECONDS);
} catch (Exception e) {
- //because UT uses a mock flush operation, 10 seconds should be enough.
+ //because UT uses a mock flushMetadata operation, 10 seconds should be enough.
LOGGER.error(e.getMessage(), e);
- Assert.fail("mock flush spends more than 10 seconds... "
+ Assert.fail("mock flushMetadata spends more than 10 seconds... "
+ "Please modify the value or change a better test environment");
}
pair = bufferwrite.queryBufferWriteData(processorName, measurementId, dataType, props);
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java
index d3c462d..28ecf56 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessorTest.java
@@ -130,7 +130,7 @@ public class BufferWriteProcessorTest {
for (int i = 1; i < 100; i++) {
bufferwrite.write(deviceId, measurementId, i, dataType, String.valueOf(i));
}
- // waiting for the end of flush
+ // waiting for the end of flushMetadata
TimeUnit.SECONDS.sleep(2);
File dataFile = PathUtils.getBufferWriteDir(deviceId);
// check file
@@ -184,7 +184,7 @@ public class BufferWriteProcessorTest {
for (int i = 1; i < 100; i++) {
bufferwrite.write(deviceId, measurementId, i, dataType, String.valueOf(i));
}
- // waiting for the end of flush
+ // waiting for the end of flushMetadata
TimeUnit.SECONDS.sleep(2);
File dataFile = PathUtils.getBufferWriteDir(deviceId);
// check file
@@ -229,15 +229,15 @@ public class BufferWriteProcessorTest {
}
assertEquals(lastFlushTime, bufferwrite.getLastFlushTime());
bufferwrite.write(deviceId, measurementId, 86, dataType, String.valueOf(86));
- //assert a flush() is called.
+ //assert a flushMetadata() is called.
assertNotEquals(bufferwrite.getLastFlushTime(), lastFlushTime);
- // sleep to the end of flush
+ // sleep to the end of flushMetadata
try {
bufferwrite.getFlushFuture().get(10, TimeUnit.SECONDS);
} catch (Exception e) {
- //because UT uses a mock flush operation, 10 seconds should be enough.
+ //because UT uses a mock flushMetadata operation, 10 seconds should be enough.
LOGGER.error(e.getMessage(), e);
- Assert.fail("mock flush spends more than 10 seconds... "
+ Assert.fail("mock flushMetadata spends more than 10 seconds... "
+ "Please modify the value or change a better test environment");
}
assertEquals(0, bufferwrite.memoryUsage());
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriterTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriterTest.java
index 3c239c8..2b093c1 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriterTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/bufferwrite/RestorableTsFileIOWriterTest.java
@@ -32,7 +32,6 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.iotdb.db.engine.memtable.IMemTable;
import org.apache.iotdb.db.engine.memtable.MemTableFlushTask;
-import org.apache.iotdb.db.engine.memtable.MemTableFlushUtil;
import org.apache.iotdb.db.engine.memtable.MemTableTestUtils;
import org.apache.iotdb.db.engine.memtable.PrimitiveMemTable;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -169,8 +168,8 @@ public class RestorableTsFileIOWriterTest {
//MemTableFlushUtil.flushMemTable(schema, writer, memTable, 0);
MemTableFlushTask tableFlushTask = new MemTableFlushTask(writer, "test", 0L, (a,b) -> {});
tableFlushTask.flushMemTable(schema, memTable, 0);
- writer.flush();
- writer.appendMetadata();
+ writer.writeRestoreInfo();
+ writer.makeMetadataVisible();
writer.getOutput().close();
// recover
@@ -225,19 +224,19 @@ public class RestorableTsFileIOWriterTest {
MemTableFlushTask tableFlushTask = new MemTableFlushTask(writer, "test", 0L, (a,b) -> {});
tableFlushTask.flushMemTable(MemTableTestUtils.getFileSchema(), memTable, 0);
- writer.flush();
+ writer.writeRestoreInfo();
assertEquals(0,
writer.getMetadatas(MemTableTestUtils.deviceId0, MemTableTestUtils.measurementId0,
MemTableTestUtils.dataType0).size());
- writer.appendMetadata();
+ writer.makeMetadataVisible();
assertEquals(1,
writer.getMetadatas(MemTableTestUtils.deviceId0, MemTableTestUtils.measurementId0,
MemTableTestUtils.dataType0).size());
MemTableTestUtils.produceData(memTable, 200, 300, MemTableTestUtils.deviceId0,
MemTableTestUtils.measurementId0,
MemTableTestUtils.dataType0);
- writer.appendMetadata();
+ writer.makeMetadataVisible();
assertEquals(1,
writer.getMetadatas(MemTableTestUtils.deviceId0, MemTableTestUtils.measurementId0,
MemTableTestUtils.dataType0).size());
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java
index 3655273..227bc7b 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteFileSizeControlTest.java
@@ -154,7 +154,7 @@ public class BufferwriteFileSizeControlTest {
System.out.println(i + "," + MemUtils.bytesCntToStr(processor.getFileSize()));
}
}
- // wait to flush end
+ // wait to flushMetadata end
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java
index 7f1e4b7..9ec6019 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/BufferwriteMetaSizeControlTest.java
@@ -151,7 +151,7 @@ public class BufferwriteMetaSizeControlTest {
System.out.println(i + "," + MemUtils.bytesCntToStr(processor.getMetaSize()));
}
}
- // wait to flush end
+ // wait to flushMetadata end
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/IoTDBMemControlTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/IoTDBMemControlTest.java
index 0d39851..efde5de 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/IoTDBMemControlTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/IoTDBMemControlTest.java
@@ -121,7 +121,7 @@ public class IoTDBMemControlTest {
assertEquals(BasicMemController.UsageLevel.WARNING,
BasicMemController.getInstance().getCurrLevel());
- // test MemControlTread auto flush
+ // test MemControlTread auto flushMetadata
Thread.sleep(15000);
assertEquals(BasicMemController.UsageLevel.SAFE,
BasicMemController.getInstance().getCurrLevel());
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowFileSizeControlTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowFileSizeControlTest.java
index ddf7a3b..d5cdabf 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowFileSizeControlTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowFileSizeControlTest.java
@@ -129,7 +129,7 @@ public class OverflowFileSizeControlTest {
System.out.println(i + "," + MemUtils.bytesCntToStr(ofprocessor.getFileSize()));
}
}
- // wait to flush
+ // wait to flushMetadata
Thread.sleep(1000);
ofprocessor.close();
assertTrue(ofprocessor.getFileSize() < dbConfig.getOverflowFileSizeThreshold());
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowMetaSizeControlTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowMetaSizeControlTest.java
index 2334ee7..c687c1d 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowMetaSizeControlTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/memcontrol/OverflowMetaSizeControlTest.java
@@ -130,7 +130,7 @@ public class OverflowMetaSizeControlTest {
System.out.println(i + "," + MemUtils.bytesCntToStr(ofprocessor.getMetaSize()));
}
}
- // wait to flush
+ // wait to flushMetadata
Thread.sleep(1000);
assertTrue(ofprocessor.getMetaSize() < dbConfig.getOverflowMetaSizeThreshold());
ofprocessor.close();
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessorTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessorTest.java
index 9877f76..73a6bfc 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessorTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessorTest.java
@@ -52,14 +52,14 @@ public class OverflowProcessorTest {
private Action overflowflushaction = new Action() {
@Override
public void act() throws ActionException {
- System.out.println("overflow flush action");
+ System.out.println("overflow flushMetadata action");
}
};
private Action filenodeflushaction = new Action() {
@Override
public void act() throws ActionException {
- System.out.println("filenode flush action");
+ System.out.println("filenode flushMetadata action");
}
};
@@ -114,7 +114,7 @@ public class OverflowProcessorTest {
assertEquals(i, pair.getTimestamp());
assertEquals(i, pair.getValue().getInt());
}
- // flush synchronously
+ // flushMetadata synchronously
processor.close();
processor.reopen();
overflowSeriesDataSource = processor
@@ -169,7 +169,7 @@ public class OverflowProcessorTest {
processor = new OverflowProcessor(processorName, parameters, OverflowTestUtils.getFileSchema(),
SysTimeVersionController.INSTANCE);
processor.flush();
- // waiting for the end of flush.
+ // waiting for the end of flushMetadata.
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBFillIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBFillIT.java
index 261cc17..22e71ae 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBFillIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBFillIT.java
@@ -52,7 +52,7 @@ public class IOTDBFillIT {
+ "values(4, 4.4, false, 44)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(5, 5.5, false, 55)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(100, 100.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
@@ -63,7 +63,7 @@ public class IOTDBFillIT {
+ "values(250, 400.4, false, 440)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(300, 500.5, false, 550)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(10, 10.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
@@ -74,7 +74,7 @@ public class IOTDBFillIT {
+ "values(40, 40.4, false, 440)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(50, 50.5, false, 550)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(500, 100.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
@@ -85,7 +85,7 @@ public class IOTDBFillIT {
+ "values(530, 400.4, false, 440)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(540, 500.5, false, 550)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(580, 100.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBGroupByIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBGroupByIT.java
index 7d68679..103bc32 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBGroupByIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IOTDBGroupByIT.java
@@ -62,7 +62,7 @@ public class IOTDBGroupByIT {
+ "values(4, 4.4, false, 44)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(5, 5.5, false, 55)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(100, 100.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
@@ -73,7 +73,7 @@ public class IOTDBGroupByIT {
+ "values(250, 400.4, false, 440)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(300, 500.5, false, 550)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(10, 10.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
@@ -84,7 +84,7 @@ public class IOTDBGroupByIT {
+ "values(40, 40.4, false, 440)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(50, 50.5, false, 550)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(500, 100.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
@@ -95,7 +95,7 @@ public class IOTDBGroupByIT {
+ "values(530, 400.4, false, 440)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(540, 500.5, false, 550)",
- "flush",
+ "flushMetadata",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ "values(580, 100.1, false, 110)",
"INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationIT.java
index 0a7b4c4..81eb5d8 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationIT.java
@@ -582,16 +582,16 @@ public class IoTDBAggregationIT {
for (int i = 5000; i < 7000; i++) {
statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'", "true"));
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
for (int i = 7500; i < 8500; i++) {
statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'", "false"));
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// prepare Unseq-File
for (int i = 500; i < 1500; i++) {
statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'", "true"));
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
for (int i = 3000; i < 6500; i++) {
statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'", "false"));
}
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationLargeDataIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationLargeDataIT.java
index d0f782d..47f745e 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationLargeDataIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationLargeDataIT.java
@@ -853,7 +853,7 @@ public class IoTDBAggregationLargeDataIT {
d0s2sum += time % 22;
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
System.out.println("cnt = " + cnt);
// insert large amount of data
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBDeletionIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBDeletionIT.java
index e193f0a..fa4172c 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBDeletionIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBDeletionIT.java
@@ -163,7 +163,7 @@ public class IoTDBDeletionIT {
statement.execute("INSERT INTO root.ln.wf01.wt01(timestamp,status) VALUES(NOW(), false)");
statement.execute("delete from root.ln.wf01.wt01.status where time < NOW()");
- statement.execute("flush");
+ statement.execute("flushMetadata");
statement.execute("delete from root.ln.wf01.wt01.status where time < NOW()");
ResultSet resultSet = statement.executeQuery("select status from root.ln.wf01.wt01");
assertFalse(resultSet.next());
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBEngineTimeGeneratorIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBEngineTimeGeneratorIT.java
index e172971..7ef6080 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBEngineTimeGeneratorIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBEngineTimeGeneratorIT.java
@@ -136,7 +136,7 @@ public class IoTDBEngineTimeGeneratorIT {
}
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// insert data (time from 1200-1499)
for (long time = 1200; time < 1500; time++) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFloatPrecisionIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFloatPrecisionIT.java
index 5064658..c770fe6 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFloatPrecisionIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFloatPrecisionIT.java
@@ -146,7 +146,7 @@ public class IoTDBFloatPrecisionIT {
statement.close();
statement = connection.createStatement();
- statement.execute("flush");
+ statement.execute("flushMetadata");
resultSet = statement.executeQuery("select * from root");
cnt = 0;
while (resultSet.next()) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeTest.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeTest.java
index 39691bc..251fd63 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeTest.java
@@ -46,7 +46,7 @@ public class IoTDBFlushQueryMergeTest {
"insert into root.vehicle.d0(timestamp,s0) values(104,90)",
"insert into root.vehicle.d0(timestamp,s0) values(105,99)",
"insert into root.vehicle.d0(timestamp,s0) values(106,99)",
- "flush",
+ "flushMetadata",
"insert into root.vehicle.d0(timestamp,s0) values(2,10000)",
"insert into root.vehicle.d0(timestamp,s0) values(50,10000)",
"insert into root.vehicle.d0(timestamp,s0) values(1000,22222)",
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBLargeDataIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBLargeDataIT.java
index 6889253..1ccf0fb 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBLargeDataIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBLargeDataIT.java
@@ -135,7 +135,7 @@ public class IoTDBLargeDataIT {
statement.execute(sql);
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// statement.execute("merge");
Thread.sleep(5000);
@@ -154,7 +154,7 @@ public class IoTDBLargeDataIT {
statement.execute(sql);
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// bufferwrite data, memory data
for (int time = 200000; time < 201000; time++) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java
index f97e6c1..933f811 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java
@@ -136,7 +136,7 @@ public class IoTDBMultiSeriesIT {
statement.execute(sql);
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// statement.execute("merge");
Thread.sleep(5000);
@@ -155,7 +155,7 @@ public class IoTDBMultiSeriesIT {
statement.execute(sql);
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// bufferwrite data, memory data
for (int time = 200000; time < 201000; time++) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java
index a4b95a3..0d1a39d 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java
@@ -132,7 +132,7 @@ public class IoTDBSequenceDataQueryIT {
}
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// insert data (time from 1200-1499)
for (long time = 1200; time < 1500; time++) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java
index aa2ef7c..682368d 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java
@@ -138,7 +138,7 @@ public class IoTDBSeriesReaderIT {
statement.execute(sql);
}
- // statement.execute("flush");
+ // statement.execute("flushMetadata");
// insert large amount of data time range : 13700 ~ 24000
for (int time = 13700; time < 24000; time++) {
@@ -172,7 +172,7 @@ public class IoTDBSeriesReaderIT {
statement.execute(sql);
}
- statement.execute("flush");
+ statement.execute("flushMetadata");
// bufferwrite data, memory data
for (int time = 200000; time < 201000; time++) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBVersionIT.java b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBVersionIT.java
index ba42cd0..4978f33 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBVersionIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBVersionIT.java
@@ -63,7 +63,7 @@ public class IoTDBVersionIT {
statement.execute("CREATE TIMESERIES root.versionTest2.s0"
+ " WITH DATATYPE=INT32,ENCODING=PLAIN");
- // write and flush enough times to make the version file persist
+ // write and flushMetadata enough times to make the version file persist
for (int i = 0; i < 3 * SimpleFileVersionController.getSaveInterval(); i ++) {
for (int j = 1; j <= 100; j ++) {
statement.execute(String
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
index 4326580..79e13a3 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
@@ -85,7 +85,7 @@ public class SingleClientSyncTest {
"insert into root.test.d0(timestamp,s0) values(30,1006)",
"insert into root.test.d0(timestamp,s0,s1) values(34,1007,'1008')",
"insert into root.test.d0(timestamp,s1) values(36,'1090')",
- "insert into root.test.d1.g0(timestamp,s0) values(10,1100)", "merge", "flush",};
+ "insert into root.test.d1.g0(timestamp,s0) values(10,1100)", "merge", "flushMetadata",};
private static final String[] sqls2 = new String[]{
"insert into root.vehicle.d0(timestamp,s0) values(6,120)",
"insert into root.vehicle.d0(timestamp,s0,s1) values(38,121,'122')",
@@ -103,16 +103,16 @@ public class SingleClientSyncTest {
"insert into root.test.d0(timestamp,s0) values(150,426)",
"insert into root.test.d0(timestamp,s0,s1) values(80,427,'528')",
"insert into root.test.d0(timestamp,s1) values(2,'1209')",
- "insert into root.test.d1.g0(timestamp,s0) values(4,330)", "merge", "flush",};
+ "insert into root.test.d1.g0(timestamp,s0) values(4,330)", "merge", "flushMetadata",};
private static final String[] sqls3 = new String[]{"SET STORAGE GROUP TO root.iotdb",
- "SET STORAGE GROUP TO root.flush",
+ "SET STORAGE GROUP TO root.flushMetadata",
"CREATE TIMESERIES root.iotdb.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
"CREATE TIMESERIES root.iotdb.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN",
"CREATE TIMESERIES root.iotdb.d1.s2 WITH DATATYPE=FLOAT, ENCODING=RLE",
"CREATE TIMESERIES root.iotdb.d1.s3 WITH DATATYPE=BOOLEAN, ENCODING=PLAIN",
- "CREATE TIMESERIES root.flush.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
- "CREATE TIMESERIES root.flush.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN",
- "CREATE TIMESERIES root.flush.d1.g0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
+ "CREATE TIMESERIES root.flushMetadata.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
+ "CREATE TIMESERIES root.flushMetadata.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN",
+ "CREATE TIMESERIES root.flushMetadata.d1.g0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
"insert into root.iotdb.d0(timestamp,s0) values(3,100)",
"insert into root.iotdb.d0(timestamp,s0,s1) values(22,101,'102')",
"insert into root.iotdb.d0(timestamp,s1) values(24,'103')",
@@ -125,12 +125,12 @@ public class SingleClientSyncTest {
"insert into root.iotdb.d1(timestamp,s2) values(1,404.0)",
"insert into root.iotdb.d1(timestamp,s2,s3) values(250,10.0,true)",
"insert into root.iotdb.d1(timestamp,s3) values(207,false)",
- "insert into root.flush.d0(timestamp,s0) values(20,106)",
- "insert into root.flush.d0(timestamp,s0,s1) values(14,107,'108')",
- "insert into root.flush.d1.g0(timestamp,s0) values(1,110)",
- "insert into root.flush.d0(timestamp,s0) values(200,1006)",
- "insert into root.flush.d0(timestamp,s0,s1) values(1004,1007,'1080')",
- "insert into root.flush.d1.g0(timestamp,s0) values(1000,910)",
+ "insert into root.flushMetadata.d0(timestamp,s0) values(20,106)",
+ "insert into root.flushMetadata.d0(timestamp,s0,s1) values(14,107,'108')",
+ "insert into root.flushMetadata.d1.g0(timestamp,s0) values(1,110)",
+ "insert into root.flushMetadata.d0(timestamp,s0) values(200,1006)",
+ "insert into root.flushMetadata.d0(timestamp,s0,s1) values(1004,1007,'1080')",
+ "insert into root.flushMetadata.d1.g0(timestamp,s0) values(1000,910)",
"insert into root.vehicle.d0(timestamp,s0) values(209,130)",
"insert into root.vehicle.d0(timestamp,s0,s1) values(206,131,'132')",
"insert into root.vehicle.d0(timestamp,s1) values(70,'33')",
@@ -144,7 +144,7 @@ public class SingleClientSyncTest {
"insert into root.test.d0(timestamp,s0) values(1900,1316)",
"insert into root.test.d0(timestamp,s0,s1) values(700,1307,'1038')",
"insert into root.test.d0(timestamp,s1) values(3000,'1309')",
- "insert into root.test.d1.g0(timestamp,s0) values(400,1050)", "merge", "flush",};
+ "insert into root.test.d1.g0(timestamp,s0) values(400,1050)", "merge", "flushMetadata",};
private boolean testFlag = Constant.testFlag;
private static final String SYNC_CLIENT = Constans.SYNC_CLIENT;
private static final Logger logger = LoggerFactory.getLogger(SingleClientSyncTest.class);
@@ -490,14 +490,14 @@ public class SingleClientSyncTest {
+ res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0"));
}
}
- hasResultSet = statement.execute("select * from root.flush");
+ hasResultSet = statement.execute("select * from root.flushMetadata");
if (hasResultSet) {
ResultSet res = statement.getResultSet();
while (res.next()) {
- dataSender.add(res.getString("Time") + res.getString("root.flush.d0.s0")
- + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0"));
- logger.debug(res.getString("Time") + " | " + res.getString("root.flush.d0.s0")
- + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0"));
+ dataSender.add(res.getString("Time") + res.getString("root.flushMetadata.d0.s0")
+ + res.getString("root.flushMetadata.d0.s1") + res.getString("root.flushMetadata.d1.g0.s0"));
+ logger.debug(res.getString("Time") + " | " + res.getString("root.flushMetadata.d0.s0")
+ + res.getString("root.flushMetadata.d0.s1") + res.getString("root.flushMetadata.d1.g0.s0"));
}
}
hasResultSet = statement.execute("select * from root.iotdb");
@@ -551,14 +551,14 @@ public class SingleClientSyncTest {
+ res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0"));
}
}
- hasResultSet = statement.execute("select * from root.flush");
+ hasResultSet = statement.execute("select * from root.flushMetadata");
if (hasResultSet) {
ResultSet res = statement.getResultSet();
while (res.next()) {
- dataReceiver.add(res.getString("Time") + res.getString("root.flush.d0.s0")
- + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0"));
- logger.debug(res.getString("Time") + " | " + res.getString("root.flush.d0.s0")
- + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0"));
+ dataReceiver.add(res.getString("Time") + res.getString("root.flushMetadata.d0.s0")
+ + res.getString("root.flushMetadata.d0.s1") + res.getString("root.flushMetadata.d1.g0.s0"));
+ logger.debug(res.getString("Time") + " | " + res.getString("root.flushMetadata.d0.s0")
+ + res.getString("root.flushMetadata.d0.s1") + res.getString("root.flushMetadata.d1.g0.s0"));
}
}
hasResultSet = statement.execute("select * from root.iotdb");
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/NativeRestorableIOWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/NativeRestorableIOWriter.java
index 6fb4ee5..2496d42 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/NativeRestorableIOWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/NativeRestorableIOWriter.java
@@ -68,7 +68,7 @@ public class NativeRestorableIOWriter extends TsFileIOWriter {
out.close();
return;
}
- truncatedPosition = reader.selfCheck(knownSchemas, flushedChunkGroupMetaDataList, !append);
+ truncatedPosition = reader.selfCheck(knownSchemas, chunkGroupMetaDataList, !append);
if (truncatedPosition == TsFileCheckStatus.COMPLETE_FILE && !append) {
this.canWrite = false;
out.close();
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
index 75301e1..215a06e 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
@@ -67,8 +67,7 @@ public class TsFileIOWriter {
protected TsFileOutput out;
private AtomicLong flushID = new AtomicLong(0);
- protected List<ChunkGroupMetaData> flushedChunkGroupMetaDataList = new ArrayList<>();
- protected List<ChunkGroupMetaData> flushingChunkGroupMetaDataList = new ArrayList<>();
+ protected List<ChunkGroupMetaData> chunkGroupMetaDataList = new ArrayList<>();
private ChunkGroupMetaData currentChunkGroupMetaData;
private ChunkMetaData currentChunkMetaData;
protected boolean canWrite = true;
@@ -107,14 +106,14 @@ public class TsFileIOWriter {
* data in the TsFileOutput matches the given metadata list
*
* @param out the target output
- * @param flushedChunkGroupMetaDataList existing chunkgroups' metadata
+ * @param ChunkGroupMetaDataList existing chunkgroups' metadata
* @throws IOException if I/O error occurs
*/
- public TsFileIOWriter(TsFileOutput out, List<ChunkGroupMetaData> flushedChunkGroupMetaDataList)
+ public TsFileIOWriter(TsFileOutput out, List<ChunkGroupMetaData> ChunkGroupMetaDataList)
throws IOException {
this.out = out;
- this.flushedChunkGroupMetaDataList = flushedChunkGroupMetaDataList;
- if (flushedChunkGroupMetaDataList.isEmpty()) {
+ this.chunkGroupMetaDataList = ChunkGroupMetaDataList;
+ if (ChunkGroupMetaDataList.isEmpty()) {
startFile();
}
}
@@ -139,15 +138,6 @@ public class TsFileIOWriter {
}
/**
- * move ChunkGroupMetadata from flushingChunkGroupMetaDataList to flushedChunkGroupMetaDataList
- * only flushedChunkGroupMetaDataList is visible for query
- */
- public void mergeChunkGroupMetaData() {
- flushedChunkGroupMetaDataList.addAll(flushingChunkGroupMetaDataList);
- flushingChunkGroupMetaDataList.clear();
- }
-
- /**
* start a {@linkplain ChunkGroupMetaData ChunkGroupMetaData}.
*
* @param deviceId device id
@@ -167,7 +157,7 @@ public class TsFileIOWriter {
chunkGroupFooter.serializeTo(out.wrapAsStream());
currentChunkGroupMetaData.setEndOffsetOfChunkGroup(out.getPosition());
currentChunkGroupMetaData.setVersion(version);
- flushingChunkGroupMetaDataList.add(currentChunkGroupMetaData);
+ chunkGroupMetaDataList.add(currentChunkGroupMetaData);
LOG.debug("end chunk group:{}", currentChunkGroupMetaData);
currentChunkGroupMetaData = null;
}
@@ -238,8 +228,6 @@ public class TsFileIOWriter {
*/
public void endFile(FileSchema schema) throws IOException {
- mergeChunkGroupMetaData();
-
// serialize the SEPARATOR of MetaData and ChunkGroups
ReadWriteIOUtils.write(MetaMarker.SEPARATOR, out.wrapAsStream());
@@ -248,7 +236,7 @@ public class TsFileIOWriter {
LOG.debug("get time series list:{}", schemaDescriptors);
Map<String, TsDeviceMetadataIndex> tsDeviceMetadataIndexMap = flushTsDeviceMetaDataAndGetIndex(
- this.flushedChunkGroupMetaDataList);
+ this.chunkGroupMetaDataList);
TsFileMetaData tsFileMetaData = new TsFileMetaData(tsDeviceMetadataIndexMap, schemaDescriptors,
TSFileConfig.CURRENT_VERSION);
@@ -273,7 +261,7 @@ public class TsFileIOWriter {
}
/**
- * 1. group flushedChunkGroupMetaDataList to TsDeviceMetadata 2. flush TsDeviceMetadata 3. get
+ * 1. group chunkGroupMetaDataList to TsDeviceMetadata 2. flush TsDeviceMetadata 3. get
* TsDeviceMetadataIndex
*
* @param chunkGroupMetaDataList all chunk group metadata in memory
@@ -340,12 +328,12 @@ public class TsFileIOWriter {
}
/**
- * get flushedChunkGroupMetaDataList.
+ * get chunkGroupMetaDataList.
*
* @return - List of chunkGroupMetaData
*/
public List<ChunkGroupMetaData> getChunkGroupMetaDatas() {
- return flushedChunkGroupMetaDataList;
+ return chunkGroupMetaDataList;
}
public boolean canWrite() {