You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by sh...@apache.org on 2025/06/20 03:49:58 UTC
(iotdb) branch compation-log updated (b1c1058c68c -> 29496df2097)
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a change to branch compation-log
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from b1c1058c68c Update logback-datanode.xml
new ea15c88f290 modify TsFileProcessor
new fc8dac9ab52 modify DataRegion
new 29496df2097 modify wal log
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/storageengine/dataregion/DataRegion.java | 149 +++++++++++----------
.../dataregion/flush/MemTableFlushTask.java | 3 +-
.../dataregion/memtable/TsFileProcessor.java | 145 ++++++++++----------
.../storageengine/dataregion/wal/WALManager.java | 2 +-
.../allocation/AbstractNodeAllocationStrategy.java | 3 +-
.../dataregion/wal/buffer/AbstractWALBuffer.java | 3 +-
.../dataregion/wal/buffer/WALBuffer.java | 3 +-
.../dataregion/wal/buffer/WALEntry.java | 4 +-
.../wal/checkpoint/CheckpointManager.java | 3 +-
.../storageengine/dataregion/wal/io/LogWriter.java | 3 +-
.../dataregion/wal/io/WALMetaData.java | 3 +-
.../storageengine/dataregion/wal/io/WALReader.java | 3 +-
.../storageengine/dataregion/wal/node/WALNode.java | 3 +-
.../dataregion/wal/utils/WALEntryHandler.java | 3 +-
14 files changed, 180 insertions(+), 150 deletions(-)
(iotdb) 01/03: modify TsFileProcessor
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch compation-log
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ea15c88f290ec80d063c44fcf303ac3971f59c14
Author: shuwenwei <s1...@gmail.com>
AuthorDate: Fri Jun 20 11:34:33 2025 +0800
modify TsFileProcessor
---
.../dataregion/flush/MemTableFlushTask.java | 3 +-
.../dataregion/memtable/TsFileProcessor.java | 141 +++++++++++----------
2 files changed, 74 insertions(+), 70 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
index d1a3f3c5bab..8d586dfc619 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.storageengine.dataregion.flush;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -60,7 +61,7 @@ import java.util.concurrent.TimeUnit;
*/
public class MemTableFlushTask {
- private static final Logger LOGGER = LoggerFactory.getLogger(MemTableFlushTask.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
private static final FlushSubTaskPoolManager SUB_TASK_POOL_MANAGER =
FlushSubTaskPoolManager.getInstance();
private static final WritingMetrics WRITING_METRICS = WritingMetrics.getInstance();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index bd4c11e969b..9cc35f11d3a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.memtable;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.AlignedPath;
@@ -120,7 +121,9 @@ import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.WORK
public class TsFileProcessor {
/** Logger fot this class. */
- private static final Logger logger = LoggerFactory.getLogger(TsFileProcessor.class);
+ private static final Logger writeLogger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
+ private static final Logger queryLogger = LoggerFactory.getLogger(IoTDBConstant.QUERY_LOGGER_NAME);
+ private static final Logger otherLogger = LoggerFactory.getLogger(TsFileProcessor.class);
private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS =
PerformanceOverviewMetrics.getInstance();
@@ -226,7 +229,7 @@ public class TsFileProcessor {
flushListeners.add(FlushListener.DefaultMemTableFLushListener.INSTANCE);
flushListeners.add(this.walNode);
closeFileListeners.add(closeUnsealedTsFileProcessor);
- logger.info("create a new tsfile processor {}", tsfile.getAbsolutePath());
+ writeLogger.info("create a new tsfile processor {}", tsfile.getAbsolutePath());
}
@SuppressWarnings("java:S107") // ignore number of arguments
@@ -250,7 +253,7 @@ public class TsFileProcessor {
flushListeners.add(FlushListener.DefaultMemTableFLushListener.INSTANCE);
flushListeners.add(this.walNode);
closeFileListeners.add(closeUnsealedTsFileProcessor);
- logger.info("reopen a tsfile processor {}", tsFileResource.getTsFile());
+ writeLogger.info("reopen a tsfile processor {}", tsFileResource.getTsFile());
}
/**
@@ -296,7 +299,7 @@ public class TsFileProcessor {
}
} catch (Exception e) {
rollbackMemoryInfo(memIncrements);
- logger.warn("Exception during wal flush", e);
+ writeLogger.warn("Exception during wal flush", e);
throw new WriteProcessException(
String.format(
"%s: %s write WAL failed: %s",
@@ -392,7 +395,7 @@ public class TsFileProcessor {
}
} catch (Exception e) {
rollbackMemoryInfo(memIncrements);
- logger.warn("Exception during wal flush", e);
+ writeLogger.warn("Exception during wal flush", e);
throw new WriteProcessException(
String.format(
"%s: %s write WAL failed: %s",
@@ -1028,7 +1031,7 @@ public class TsFileProcessor {
logFlushQueryWriteLocked();
try {
if (workMemTable != null) {
- logger.info(
+ writeLogger.info(
"[Deletion] Deletion with path: {}, time:{}-{} in workMemTable",
deletion.getPath(),
deletion.getStartTime(),
@@ -1069,7 +1072,7 @@ public class TsFileProcessor {
@TestOnly
public void syncClose() throws ExecutionException {
- logger.info(
+ writeLogger.info(
"Sync close file: {}, will firstly async close it",
tsFileResource.getTsFile().getAbsolutePath());
if (shouldClose) {
@@ -1077,7 +1080,7 @@ public class TsFileProcessor {
}
try {
asyncClose().get();
- logger.info("Start to wait until file {} is closed", tsFileResource);
+ writeLogger.info("Start to wait until file {} is closed", tsFileResource);
// if this TsFileProcessor is closing, asyncClose().get() of this thread will return quickly,
// but the TsFileProcessor may be not closed. Therefore, we need to check whether the writer
// is null.
@@ -1087,7 +1090,7 @@ public class TsFileProcessor {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
- logger.info("File {} is closed synchronously", tsFileResource.getTsFile().getAbsolutePath());
+ writeLogger.info("File {} is closed synchronously", tsFileResource.getTsFile().getAbsolutePath());
}
/** async close one tsfile, register and close it by another thread */
@@ -1095,9 +1098,9 @@ public class TsFileProcessor {
flushQueryLock.writeLock().lock();
logFlushQueryWriteLocked();
try {
- if (logger.isDebugEnabled()) {
+ if (writeLogger.isDebugEnabled()) {
if (workMemTable != null) {
- logger.debug(
+ writeLogger.debug(
"{}: flush a working memtable in async close tsfile {}, memtable size: {}, tsfile "
+ "size: {}, plan index: [{}, {}], progress index: {}",
storageGroupName,
@@ -1108,7 +1111,7 @@ public class TsFileProcessor {
workMemTable.getMaxPlanIndex(),
tsFileResource.getMaxProgressIndex());
} else {
- logger.debug(
+ writeLogger.debug(
"{}: flush a NotifyFlushMemTable in async close tsfile {}, tsfile size: {}",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1145,7 +1148,7 @@ public class TsFileProcessor {
shouldClose = true;
return future;
} catch (Exception e) {
- logger.error(
+ writeLogger.error(
"{}: {} async close failed, because",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1169,8 +1172,8 @@ public class TsFileProcessor {
logFlushQueryWriteLocked();
try {
tmpMemTable = workMemTable == null ? new NotifyFlushMemTable() : workMemTable;
- if (logger.isDebugEnabled() && tmpMemTable.isSignalMemTable()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled() && tmpMemTable.isSignalMemTable()) {
+ writeLogger.debug(
"{}: {} add a signal memtable into flushing memtable list when sync flush",
storageGroupName,
tsFileResource.getTsFile().getName());
@@ -1188,14 +1191,14 @@ public class TsFileProcessor {
flushingMemTables.wait(1000);
if ((System.currentTimeMillis() - startWait) > 60_000) {
- logger.warn(
+ writeLogger.warn(
"has waited for synced flushing a memtable in {} for 60 seconds.",
this.tsFileResource.getTsFile().getAbsolutePath());
startWait = System.currentTimeMillis();
}
}
} catch (InterruptedException e) {
- logger.error(
+ writeLogger.error(
"{}: {} wait flush finished meets error",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1213,11 +1216,11 @@ public class TsFileProcessor {
if (workMemTable == null) {
return;
}
- logger.info(
+ writeLogger.info(
"Async flush a memtable to tsfile: {}", tsFileResource.getTsFile().getAbsolutePath());
addAMemtableIntoFlushingList(workMemTable);
} catch (Exception e) {
- logger.error(
+ writeLogger.error(
"{}: {} add a memtable into flushing list failed",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1257,8 +1260,8 @@ public class TsFileProcessor {
SystemInfo.getInstance().addFlushingMemTableCost(tobeFlushed.getTVListsRamCost());
flushingMemTables.addLast(tobeFlushed);
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} Memtable (signal = {}) is added into the flushing Memtable, queue size = {}",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1284,13 +1287,13 @@ public class TsFileProcessor {
try {
writer.makeMetadataVisible();
if (!flushingMemTables.remove(memTable)) {
- logger.warn(
+ writeLogger.warn(
"{}: {} put the memtable (signal={}) out of flushingMemtables but it is not in the queue.",
storageGroupName,
tsFileResource.getTsFile().getName(),
memTable.isSignalMemTable());
- } else if (logger.isDebugEnabled()) {
- logger.debug(
+ } else if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} memtable (signal={}) is removed from the queue. {} left.",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1301,8 +1304,8 @@ public class TsFileProcessor {
MemTableManager.getInstance().decreaseMemtableNumber();
// Reset the mem cost in StorageGroupProcessorInfo
dataRegionInfo.releaseStorageGroupMemCost(memTable.getTVListsRamCost());
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"[mem control] {}: {} flush finished, try to reset system mem cost, "
+ "flushing memtable list size: {}",
storageGroupName,
@@ -1312,8 +1315,8 @@ public class TsFileProcessor {
// Report to System
SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo);
SystemInfo.getInstance().resetFlushingMemTableCost(memTable.getTVListsRamCost());
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} flush finished, remove a memtable from flushing list, "
+ "flushing memtable list size: {}",
storageGroupName,
@@ -1321,7 +1324,7 @@ public class TsFileProcessor {
flushingMemTables.size());
}
} catch (Exception e) {
- logger.error("{}: {}", storageGroupName, tsFileResource.getTsFile().getName(), e);
+ writeLogger.error("{}: {}", storageGroupName, tsFileResource.getTsFile().getName(), e);
} finally {
flushQueryLock.writeLock().unlock();
logFlushQueryWriteUnlocked();
@@ -1333,8 +1336,8 @@ public class TsFileProcessor {
synchronized (flushingMemTables) {
releaseFlushedMemTable(memTable);
flushingMemTables.notifyAll();
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} released a memtable (signal={}), flushingMemtables size ={}",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1355,7 +1358,7 @@ public class TsFileProcessor {
// Signal memtable only may appear when calling asyncClose()
if (!memTableToFlush.isSignalMemTable()) {
if (memTableToFlush.isEmpty()) {
- logger.info(
+ writeLogger.info(
"This normal memtable is empty, skip flush. {}: {}",
storageGroupName,
tsFileResource.getTsFile().getName());
@@ -1372,7 +1375,7 @@ public class TsFileProcessor {
memTableFlushPointCount = memTableToFlush.getTotalPointsNum();
} catch (Throwable e) {
if (writer == null) {
- logger.info(
+ writeLogger.info(
"{}: {} is closed during flush, abandon flush task",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath());
@@ -1380,21 +1383,21 @@ public class TsFileProcessor {
flushingMemTables.notifyAll();
}
} else {
- logger.error(
+ writeLogger.error(
"{}: {} meet error when flushing a memtable, change system mode to error",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
e);
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
try {
- logger.error(
+ writeLogger.error(
"{}: {} IOTask meets error, truncate the corrupted data",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
e);
writer.reset();
} catch (IOException e1) {
- logger.error(
+ writeLogger.error(
"{}: {} Truncate corrupted data meets error",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1416,7 +1419,7 @@ public class TsFileProcessor {
flushingMemTables.notifyAll();
}
} catch (Exception e1) {
- logger.error(
+ writeLogger.error(
"{}: {} Release resource meets error",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1438,7 +1441,7 @@ public class TsFileProcessor {
this.tsFileResource.getModFile().write(entry.left);
tsFileResource.getModFile().close();
iterator.remove();
- logger.info(
+ otherLogger.info(
"[Deletion] Deletion with path: {}, time:{}-{} written when flush memtable",
entry.left.getPath(),
((Deletion) (entry.left)).getStartTime(),
@@ -1446,7 +1449,7 @@ public class TsFileProcessor {
}
}
} catch (IOException e) {
- logger.error(
+ otherLogger.error(
"Meet error when writing into ModificationFile file of {} ",
tsFileResource.getTsFile().getAbsolutePath(),
e);
@@ -1454,8 +1457,8 @@ public class TsFileProcessor {
flushQueryLock.writeLock().unlock();
}
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} try get lock to release a memtable (signal={})",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1467,7 +1470,7 @@ public class TsFileProcessor {
try {
writer.getTsFileOutput().force();
} catch (IOException e) {
- logger.error("fsync memTable data to disk error,", e);
+ writeLogger.error("fsync memTable data to disk error,", e);
}
// Call flushed listener after memtable is released safely
@@ -1483,19 +1486,19 @@ public class TsFileProcessor {
} else {
writer.mark();
updateCompressionRatio();
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} flushingMemtables is empty and will close the file",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath());
}
endFile();
}
- if (logger.isDebugEnabled()) {
- logger.debug("{} flushingMemtables is clear", storageGroupName);
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug("{} flushingMemtables is clear", storageGroupName);
}
} catch (Exception e) {
- logger.error(
+ writeLogger.error(
"{}: {} marking or ending file meet error",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1504,7 +1507,7 @@ public class TsFileProcessor {
try {
writer.reset();
} catch (IOException e1) {
- logger.error(
+ writeLogger.error(
"{}: {} truncate corrupted data meets error",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1512,7 +1515,7 @@ public class TsFileProcessor {
}
// Retry or set read-only
if (retryCnt < 3) {
- logger.warn(
+ writeLogger.warn(
"{} meet error when flush FileMetadata to {}, retry it again",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1520,7 +1523,7 @@ public class TsFileProcessor {
retryCnt++;
continue;
} else {
- logger.error(
+ writeLogger.error(
"{} meet error when flush FileMetadata to {}, change system mode to error",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -1530,8 +1533,8 @@ public class TsFileProcessor {
}
}
// For sync close
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} try to get flushingMemtables lock.",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath());
@@ -1545,7 +1548,7 @@ public class TsFileProcessor {
private void updateCompressionRatio() {
try {
double compressionRatio = ((double) totalMemTableSize) / writer.getPos();
- logger.info(
+ writeLogger.info(
"The compression ratio of tsfile {} is {}, totalMemTableSize: {}, the file size: {}",
writer.getFile().getAbsolutePath(),
String.format("%.2f", compressionRatio),
@@ -1556,7 +1559,7 @@ public class TsFileProcessor {
.recordTsFileCompressionRatioOfFlushingMemTable(dataRegionId, compressionRatio);
CompressionRatio.getInstance().updateRatio(totalMemTableSize, writer.getPos());
} catch (IOException e) {
- logger.error(
+ writeLogger.error(
"{}: {} update compression ratio failed",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1566,14 +1569,14 @@ public class TsFileProcessor {
/** end file and write some meta */
private void endFile() throws IOException, TsFileProcessorException {
- if (logger.isDebugEnabled()) {
- logger.debug("Start to end file {}", tsFileResource);
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug("Start to end file {}", tsFileResource);
}
writer.endFile();
tsFileResource.serialize();
FileTimeIndexCacheRecorder.getInstance().logFileTimeIndex(tsFileResource);
- if (logger.isDebugEnabled()) {
- logger.debug("Ended file {}", tsFileResource);
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug("Ended file {}", tsFileResource);
}
// Remove this processor from Closing list in StorageGroupProcessor,
// Mark the TsFileResource closed, no need writer anymore
@@ -1589,7 +1592,7 @@ public class TsFileProcessor {
/** End empty file and remove it from file system */
private void endEmptyFile() throws TsFileProcessorException, IOException {
- logger.info("Start to end empty file {}", tsFileResource);
+ writeLogger.info("Start to end empty file {}", tsFileResource);
// Remove this processor from Closing list in DataRegion,
// Mark the TsFileResource closed, no need writer anymore
@@ -1599,7 +1602,7 @@ public class TsFileProcessor {
}
tsFileProcessorInfo.clear();
dataRegionInfo.closeTsFileProcessorAndReportToSystem(this);
- logger.info(
+ writeLogger.info(
"Storage group {} close and remove empty file {}",
storageGroupName,
tsFileResource.getTsFile().getAbsoluteFile());
@@ -1856,7 +1859,7 @@ public class TsFileProcessor {
}
}
} catch (QueryProcessException | MetadataException | IOException e) {
- logger.error(
+ queryLogger.error(
"{}: {} get ReadOnlyMemChunk has error",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1940,7 +1943,7 @@ public class TsFileProcessor {
}
}
} catch (QueryProcessException | MetadataException | IOException e) {
- logger.error(
+ queryLogger.error(
"{}: {} get ReadOnlyMemChunk has error",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -2019,7 +2022,7 @@ public class TsFileProcessor {
}
}
} catch (QueryProcessException | MetadataException e) {
- logger.error(
+ queryLogger.error(
"{}: {} get ReadOnlyMemChunk has error",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -2141,22 +2144,22 @@ public class TsFileProcessor {
}
private void logFlushQueryWriteLocked() {
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
FLUSH_QUERY_WRITE_LOCKED, storageGroupName, tsFileResource.getTsFile().getName());
}
}
private void logFlushQueryWriteUnlocked() {
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
}
}
private void logFlushQueryReadUnlocked() {
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"{}: {} release flushQueryLock", storageGroupName, tsFileResource.getTsFile().getName());
}
}
(iotdb) 02/03: modify DataRegion
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch compation-log
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit fc8dac9ab520151d56450088aeba360cc0e2ee5a
Author: shuwenwei <s1...@gmail.com>
AuthorDate: Fri Jun 20 11:44:19 2025 +0800
modify DataRegion
---
.../db/storageengine/dataregion/DataRegion.java | 149 +++++++++++----------
.../dataregion/memtable/TsFileProcessor.java | 10 +-
2 files changed, 87 insertions(+), 72 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 86523df44d6..5cf91910866 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -201,8 +201,6 @@ import static org.apache.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
public class DataRegion implements IDataRegionForQuery {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- private static final Logger QUERY_LOGGER =
- LoggerFactory.getLogger(IoTDBConstant.QUERY_LOGGER_NAME);
/**
* All newly generated chunks after merge have version number 0, so we set merged Modification
@@ -210,7 +208,13 @@ public class DataRegion implements IDataRegionForQuery {
*/
private static final int MERGE_MOD_START_VERSION_NUM = 1;
- private static final Logger logger = LoggerFactory.getLogger(DataRegion.class);
+ private static final Logger writeLogger =
+ LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
+ private static final Logger queryLogger =
+ LoggerFactory.getLogger(IoTDBConstant.QUERY_LOGGER_NAME);
+ private static final Logger compactionLogger =
+ LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private static final Logger otherLogger = LoggerFactory.getLogger(DataRegion.class);
/**
* A read write lock for guaranteeing concurrent safety when accessing all fields in this class
@@ -332,10 +336,10 @@ public class DataRegion implements IDataRegionForQuery {
dataRegionSysDir = SystemFileFactory.INSTANCE.getFile(systemDir, dataRegionId);
this.tsFileManager = new TsFileManager(databaseName, dataRegionId, dataRegionSysDir.getPath());
if (dataRegionSysDir.mkdirs()) {
- logger.info(
+ otherLogger.info(
"Database system Directory {} doesn't exist, create it", dataRegionSysDir.getPath());
} else if (!dataRegionSysDir.exists()) {
- logger.error("create database system Directory {} failed", dataRegionSysDir.getPath());
+ otherLogger.error("create database system Directory {} failed", dataRegionSysDir.getPath());
}
lastFlushTimeMap = new HashLastFlushTimeMap();
@@ -343,7 +347,7 @@ public class DataRegion implements IDataRegionForQuery {
// recover tsfiles unless consensus protocol is ratis and storage engine is not ready
if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)
&& !StorageEngine.getInstance().isReadyForReadAndWrite()) {
- logger.debug(
+ otherLogger.debug(
"Skip recovering data region {}[{}] when consensus protocol is ratis and storage engine is not ready.",
databaseName,
dataRegionId);
@@ -353,7 +357,7 @@ public class DataRegion implements IDataRegionForQuery {
try {
fsFactory.deleteDirectory(dataRegionFolder.getPath());
} catch (IOException e) {
- logger.error(
+ otherLogger.error(
"Exception occurs when deleting data region folder for {}-{}",
databaseName,
dataRegionId,
@@ -361,10 +365,10 @@ public class DataRegion implements IDataRegionForQuery {
}
if (FSUtils.getFSType(dataRegionFolder) == FSType.LOCAL) {
if (dataRegionFolder.mkdirs()) {
- logger.info(
+ otherLogger.info(
"Data region directory {} doesn't exist, create it", dataRegionFolder.getPath());
} else if (!dataRegionFolder.exists()) {
- logger.error("create data region directory {} failed", dataRegionFolder.getPath());
+ otherLogger.error("create data region directory {} failed", dataRegionFolder.getPath());
}
}
}
@@ -456,7 +460,7 @@ public class DataRegion implements IDataRegionForQuery {
recoveredFilesNum++;
if (recoveredFilesNum < numOfFilesToRecover) {
if (System.currentTimeMillis() - lastLogTime > config.getRecoveryLogIntervalInMs()) {
- logger.info(
+ otherLogger.info(
"The TsFiles of data region {}[{}] has recovered {}/{}.",
databaseName,
dataRegionId,
@@ -465,7 +469,7 @@ public class DataRegion implements IDataRegionForQuery {
lastLogTime = System.currentTimeMillis();
}
} else {
- logger.info(
+ otherLogger.info(
"The TsFiles of data region {}[{}] has recovered completely {}/{}.",
databaseName,
dataRegionId,
@@ -629,7 +633,7 @@ public class DataRegion implements IDataRegionForQuery {
// wait until all unsealed TsFiles have been recovered
for (WALRecoverListener recoverListener : recoverListeners) {
if (recoverListener.waitForResult() == WALRecoverListener.Status.FAILURE) {
- logger.error(
+ otherLogger.error(
"Fail to recover unsealed TsFile {}, skip it.",
recoverListener.getFilePath(),
recoverListener.getCause());
@@ -673,9 +677,11 @@ public class DataRegion implements IDataRegionForQuery {
.equals(ConsensusFactory.IOT_CONSENSUS_V2)) {
WALManager.getInstance().applyForWALNode(databaseName + FILE_NAME_SEPARATOR + dataRegionId);
}
- logger.info("The data region {}[{}] is created successfully", databaseName, dataRegionId);
+ otherLogger.info(
+ "The data region {}[{}] is created successfully", databaseName, dataRegionId);
} else {
- logger.info("The data region {}[{}] is recovered successfully", databaseName, dataRegionId);
+ otherLogger.info(
+ "The data region {}[{}] is recovered successfully", databaseName, dataRegionId);
}
}
@@ -758,7 +764,7 @@ public class DataRegion implements IDataRegionForQuery {
if (subFiles != null) {
for (File partitionFolder : subFiles) {
if (!partitionFolder.isDirectory()) {
- logger.warn("{} is not a directory.", partitionFolder.getAbsolutePath());
+ otherLogger.warn("{} is not a directory.", partitionFolder.getAbsolutePath());
} else {
// some TsFileResource may be being persisted when the system crashed, try recovering
// such resources
@@ -838,7 +844,8 @@ public class DataRegion implements IDataRegionForQuery {
try {
tsFileResource.close();
} catch (IOException e) {
- logger.error("Fail to close TsFile {} when recovering", tsFileResource.getTsFile(), e);
+ otherLogger.error(
+ "Fail to close TsFile {} when recovering", tsFileResource.getTsFile(), e);
}
if (!TsFileValidator.getInstance().validateTsFile(tsFileResource)) {
tsFileResource.remove();
@@ -898,7 +905,7 @@ public class DataRegion implements IDataRegionForQuery {
}
tsFileManager.add(tsFileResource, recoverPerformer.isSequence());
} catch (Throwable e) {
- logger.error(
+ otherLogger.error(
"Fail to recover unsealed TsFile {}, skip it.",
recoverPerformer.getTsFileAbsolutePath(),
e);
@@ -914,7 +921,8 @@ public class DataRegion implements IDataRegionForQuery {
sealedTsFile.close();
tsFileResourceManager.registerSealedTsFileResource(sealedTsFile);
} catch (Throwable e) {
- logger.error("Fail to recover sealed TsFile {}, skip it.", sealedTsFile.getTsFilePath(), e);
+ otherLogger.error(
+ "Fail to recover sealed TsFile {}, skip it.", sealedTsFile.getTsFilePath(), e);
} finally {
// update recovery context
context.incrementRecoveredFilesNum();
@@ -982,7 +990,7 @@ public class DataRegion implements IDataRegionForQuery {
recoverPerformer.recover();
tsFileResourceManager.registerSealedTsFileResource(tsFileResource);
} catch (Throwable e) {
- logger.error(
+ otherLogger.error(
"Fail to recover sealed TsFile {}, skip it.", tsFileResource.getTsFilePath(), e);
} finally {
// update recovery context
@@ -1108,7 +1116,7 @@ public class DataRegion implements IDataRegionForQuery {
PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() - startTime);
try {
if (deleted) {
- logger.info(
+ writeLogger.info(
"Won't insert tablet {}, because region is deleted", insertTabletNode.getSearchIndex());
return;
}
@@ -1248,8 +1256,8 @@ public class DataRegion implements IDataRegionForQuery {
long timePartitionId) {
// return when start >= end or all measurement failed
if (start >= end || insertTabletNode.allMeasurementFailed()) {
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (writeLogger.isDebugEnabled()) {
+ writeLogger.debug(
"Won't insert tablet {}, because {}",
insertTabletNode.getSearchIndex(),
start >= end ? "start >= end" : "insertTabletNode allMeasurementFailed");
@@ -1271,10 +1279,10 @@ public class DataRegion implements IDataRegionForQuery {
try {
tsFileProcessor.insertTablet(insertTabletNode, start, end, results);
} catch (WriteProcessRejectException e) {
- logger.warn("insert to TsFileProcessor rejected, {}", e.getMessage());
+ writeLogger.warn("insert to TsFileProcessor rejected, {}", e.getMessage());
return false;
} catch (WriteProcessException e) {
- logger.error("insert to TsFileProcessor error ", e);
+ writeLogger.error("insert to TsFileProcessor error ", e);
return false;
}
@@ -1526,23 +1534,23 @@ public class DataRegion implements IDataRegionForQuery {
getOrCreateTsFileProcessorIntern(timeRangeId, workUnsequenceTsFileProcessors, false);
}
} catch (DiskSpaceInsufficientException e) {
- logger.error(
+ writeLogger.error(
"disk space is insufficient when creating TsFile processor, change system mode to read-only",
e);
CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly);
break;
} catch (IOException e) {
if (retryCnt < 3) {
- logger.warn("meet IOException when creating TsFileProcessor, retry it again", e);
+ writeLogger.warn("meet IOException when creating TsFileProcessor, retry it again", e);
retryCnt++;
} else {
- logger.error(
+ writeLogger.error(
"meet IOException when creating TsFileProcessor, change system mode to error", e);
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
break;
}
} catch (ExceedQuotaException e) {
- logger.error(e.getMessage());
+ writeLogger.error(e.getMessage());
break;
}
} while (tsFileProcessor == null);
@@ -1656,7 +1664,7 @@ public class DataRegion implements IDataRegionForQuery {
workUnsequenceTsFileProcessors.remove(tsFileProcessor.getTimeRangeId());
}
TsFileResource resource = tsFileProcessor.getTsFileResource();
- logger.info(
+ writeLogger.info(
"Async close tsfile: {}, file start time: {}, file end time: {}",
resource.getTsFile().getAbsolutePath(),
resource.getFileStartTime(),
@@ -1674,7 +1682,7 @@ public class DataRegion implements IDataRegionForQuery {
* @param systemDir system dir
*/
public void deleteFolder(String systemDir) {
- logger.info(
+ otherLogger.info(
"{} will close all files for deleting data folder {}",
databaseName + "-" + dataRegionId,
systemDir);
@@ -1698,21 +1706,21 @@ public class DataRegion implements IDataRegionForQuery {
try {
tsFileResource.close();
} catch (IOException e) {
- logger.error("Cannot close a TsFileResource {}", tsFileResource, e);
+ writeLogger.error("Cannot close a TsFileResource {}", tsFileResource, e);
}
}
for (TsFileResource tsFileResource : tsFileManager.getTsFileList(true)) {
try {
tsFileResource.close();
} catch (IOException e) {
- logger.error("Cannot close a TsFileResource {}", tsFileResource, e);
+ writeLogger.error("Cannot close a TsFileResource {}", tsFileResource, e);
}
}
}
/** delete tsfile */
public void syncDeleteDataFiles() throws TsFileProcessorException {
- logger.info(
+ otherLogger.info(
"{} will close all files for deleting data files", databaseName + "-" + dataRegionId);
writeLock("syncDeleteDataFiles");
try {
@@ -1740,7 +1748,7 @@ public class DataRegion implements IDataRegionForQuery {
TimePartitionManager.getInstance()
.removeTimePartitionInfo(new DataRegionId(Integer.parseInt(dataRegionId)));
} catch (InterruptedException e) {
- logger.error(
+ otherLogger.error(
"CloseFileNodeCondition error occurs while waiting for closing the storage " + "group {}",
databaseName + "-" + dataRegionId,
e);
@@ -1758,7 +1766,7 @@ public class DataRegion implements IDataRegionForQuery {
try {
fsFactory.deleteDirectory(dataRegionDataFolder.getPath());
} catch (IOException e) {
- logger.error("Fail to delete data region folder {}", dataRegionDataFolder);
+ otherLogger.error("Fail to delete data region folder {}", dataRegionDataFolder);
}
} else {
if (dataRegionDataFolder.exists()) {
@@ -1779,7 +1787,7 @@ public class DataRegion implements IDataRegionForQuery {
long timeLowerBound = System.currentTimeMillis() - config.getSeqMemtableFlushInterval();
for (TsFileProcessor tsFileProcessor : tsFileProcessors) {
if (tsFileProcessor.getWorkMemTableUpdateTime() < timeLowerBound) {
- logger.info(
+ writeLogger.info(
"Exceed sequence memtable flush interval, so flush working memtable of time partition {} in database {}[{}]",
tsFileProcessor.getTimeRangeId(),
databaseName,
@@ -1805,7 +1813,7 @@ public class DataRegion implements IDataRegionForQuery {
for (TsFileProcessor tsFileProcessor : tsFileProcessors) {
if (tsFileProcessor.getWorkMemTableUpdateTime() < timeLowerBound) {
- logger.info(
+ writeLogger.info(
"Exceed unsequence memtable flush interval, so flush working memtable of time partition {} in database {}[{}]",
tsFileProcessor.getTimeRangeId(),
databaseName,
@@ -1830,7 +1838,7 @@ public class DataRegion implements IDataRegionForQuery {
}
}
} catch (InterruptedException | ExecutionException e) {
- logger.error(
+ writeLogger.error(
"CloseFileNodeCondition error occurs while waiting for closing tsfile processors of {}",
databaseName + "-" + dataRegionId,
e);
@@ -1864,7 +1872,7 @@ public class DataRegion implements IDataRegionForQuery {
}
}
} catch (InterruptedException | ExecutionException e) {
- logger.error(
+ writeLogger.error(
"CloseFileNodeCondition error occurs while waiting for closing tsfile processors of {}",
databaseName + "-" + dataRegionId,
e);
@@ -1884,7 +1892,7 @@ public class DataRegion implements IDataRegionForQuery {
}
}
if (System.currentTimeMillis() - startTime > 60_000) {
- logger.warn(
+ writeLogger.warn(
"{} has spent {}s to wait for closing all TsFiles.",
databaseName + "-" + this.dataRegionId,
(System.currentTimeMillis() - startTime) / 1000);
@@ -1898,7 +1906,8 @@ public class DataRegion implements IDataRegionForQuery {
List<Future<?>> futures = new ArrayList<>();
int count = 0;
try {
- logger.info("async force close all files in database: {}", databaseName + "-" + dataRegionId);
+ writeLogger.info(
+ "async force close all files in database: {}", databaseName + "-" + dataRegionId);
// to avoid concurrent modification problem, we need a new array list
for (TsFileProcessor tsFileProcessor :
new ArrayList<>(workSequenceTsFileProcessors.values())) {
@@ -1922,7 +1931,8 @@ public class DataRegion implements IDataRegionForQuery {
public void forceCloseAllWorkingTsFileProcessors() throws TsFileProcessorException {
writeLock("forceCloseAllWorkingTsFileProcessors");
try {
- logger.info("force close all processors in database: {}", databaseName + "-" + dataRegionId);
+ writeLogger.info(
+ "force close all processors in database: {}", databaseName + "-" + dataRegionId);
// to avoid concurrent modification problem, we need a new array list
List<TsFileResource> closedTsFileResources = new ArrayList<>();
for (TsFileProcessor tsFileProcessor :
@@ -2157,7 +2167,7 @@ public class DataRegion implements IDataRegionForQuery {
throws MetadataException {
if (context.isDebug()) {
- QUERY_LOGGER.info(
+ queryLogger.info(
"Path: {}, get tsfile list: {} isSeq: {} time filter: {}",
pathList,
tsFileResources,
@@ -2239,7 +2249,7 @@ public class DataRegion implements IDataRegionForQuery {
for (WALFlushListener walFlushListener : walListeners) {
if (walFlushListener.waitForResult() == WALFlushListener.Status.FAILURE) {
- logger.error("Fail to log delete to wal.", walFlushListener.getCause());
+ otherLogger.error("Fail to log delete to wal.", walFlushListener.getCause());
throw walFlushListener.getCause();
}
}
@@ -2268,7 +2278,7 @@ public class DataRegion implements IDataRegionForQuery {
public void deleteDataDirectly(
PartialPath pathToDelete, long startTime, long endTime, long searchIndex) throws IOException {
- logger.info(
+ otherLogger.info(
"{} will delete data files directly for deleting data between {} and {}",
databaseName + "-" + dataRegionId,
startTime,
@@ -2288,7 +2298,7 @@ public class DataRegion implements IDataRegionForQuery {
for (WALFlushListener walFlushListener : walListeners) {
if (walFlushListener.waitForResult() == WALFlushListener.Status.FAILURE) {
- logger.error("Fail to log delete to wal.", walFlushListener.getCause());
+ otherLogger.error("Fail to log delete to wal.", walFlushListener.getCause());
throw walFlushListener.getCause();
}
}
@@ -2490,7 +2500,7 @@ public class DataRegion implements IDataRegionForQuery {
}
throw t;
}
- logger.info(
+ otherLogger.info(
"[Deletion] Deletion with path:{}, time:{}-{} written into mods file:{}.",
deletion.getPath(),
deletion.getStartTime(),
@@ -2566,7 +2576,7 @@ public class DataRegion implements IDataRegionForQuery {
}
throw t;
}
- logger.info(
+ otherLogger.info(
"[Deletion] Deletion with path:{}, time:{}-{} written into mods file:{}.",
deletion.getPath(),
deletion.getStartTime(),
@@ -2593,7 +2603,8 @@ public class DataRegion implements IDataRegionForQuery {
FileMetrics.getInstance().decreaseModFileSize(tsFileResource.getModFile().getSize());
}
tsFileResource.remove();
- logger.info("Remove tsfile {} directly when delete data", tsFileResource.getTsFilePath());
+ otherLogger.info(
+ "Remove tsfile {} directly when delete data", tsFileResource.getTsFilePath());
} finally {
tsFileResource.writeUnlock();
}
@@ -2739,7 +2750,7 @@ public class DataRegion implements IDataRegionForQuery {
} catch (InterruptedException e) {
throw e;
} catch (Throwable e) {
- logger.error("Meet error in compaction schedule.", e);
+ compactionLogger.error("Meet error in compaction schedule.", e);
} finally {
isCompactionSelecting.set(false);
}
@@ -2752,7 +2763,7 @@ public class DataRegion implements IDataRegionForQuery {
// wait until success
Thread.sleep(500);
}
- logger.info("[TTL] {}-{} Start ttl checking.", databaseName, dataRegionId);
+ compactionLogger.info("[TTL] {}-{} Start ttl checking.", databaseName, dataRegionId);
int trySubmitCount = 0;
try {
CompactionScheduleContext context = new CompactionScheduleContext();
@@ -2774,7 +2785,7 @@ public class DataRegion implements IDataRegionForQuery {
if (context.hasSubmitTask()) {
CompactionMetrics.getInstance().updateCompactionTaskSelectionNum(context);
}
- logger.info(
+ compactionLogger.info(
"[TTL] {}-{} Totally select {} all-outdated files and {} partial-outdated files.",
databaseName,
dataRegionId,
@@ -2783,7 +2794,7 @@ public class DataRegion implements IDataRegionForQuery {
} catch (InterruptedException e) {
throw e;
} catch (Throwable e) {
- logger.error("Meet error in ttl check.", e);
+ compactionLogger.error("Meet error in ttl check.", e);
} finally {
isCompactionSelecting.set(false);
}
@@ -2814,7 +2825,7 @@ public class DataRegion implements IDataRegionForQuery {
} catch (InterruptedException e) {
throw e;
} catch (Throwable e) {
- logger.error("Meet error in insertion compaction schedule.", e);
+ compactionLogger.error("Meet error in insertion compaction schedule.", e);
} finally {
context.clearTimePartitionDeviceInfoCache();
CompactionScheduler.sharedUnlockCompactionSelection();
@@ -2850,7 +2861,7 @@ public class DataRegion implements IDataRegionForQuery {
oldTsFileResource.setSettleTsFileCallBack(null);
SettleService.getINSTANCE().getFilesToBeSettledCount().addAndGet(-1);
} catch (IOException e) {
- logger.error("Exception to move new tsfile in settling", e);
+ compactionLogger.error("Exception to move new tsfile in settling", e);
throw new WriteProcessException(
"Meet error when settling file: " + oldTsFileResource.getTsFile().getAbsolutePath(), e);
} finally {
@@ -2927,7 +2938,7 @@ public class DataRegion implements IDataRegionForQuery {
0);
if (!newFileName.equals(tsfileToBeInserted.getName())) {
- logger.info(
+ writeLogger.info(
"TsFile {} must be renamed to {} for loading into the unsequence list.",
tsfileToBeInserted.getName(),
newFileName);
@@ -2963,9 +2974,9 @@ public class DataRegion implements IDataRegionForQuery {
false);
}
- logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName);
+ writeLogger.info("TsFile {} is successfully loaded in unsequence list.", newFileName);
} catch (final DiskSpaceInsufficientException e) {
- logger.error(
+ writeLogger.error(
"Failed to append the tsfile {} to database processor {} because the disk space is insufficient.",
tsfileToBeInserted.getAbsolutePath(),
tsfileToBeInserted.getParentFile().getName());
@@ -3019,11 +3030,11 @@ public class DataRegion implements IDataRegionForQuery {
tsFileResource.setFile(targetFile);
if (tsFileManager.contains(tsFileResource, false)) {
- logger.warn("The file {} has already been loaded in unsequence list", tsFileResource);
+ writeLogger.warn("The file {} has already been loaded in unsequence list", tsFileResource);
return false;
}
- logger.info(
+ writeLogger.info(
"Load tsfile in unsequence list, move file from {} to {}",
tsFileToLoad.getAbsolutePath(),
targetFile.getAbsolutePath());
@@ -3049,7 +3060,7 @@ public class DataRegion implements IDataRegionForQuery {
});
}
} catch (final IOException e) {
- logger.warn(
+ writeLogger.warn(
"File renaming failed when loading tsfile. Origin: {}, Target: {}",
tsFileToLoad.getAbsolutePath(),
targetFile.getAbsolutePath(),
@@ -3079,7 +3090,7 @@ public class DataRegion implements IDataRegionForQuery {
});
}
} catch (final IOException e) {
- logger.warn(
+ otherLogger.warn(
"File renaming failed when loading .resource file. Origin: {}, Target: {}",
resourceFileToLoad.getAbsolutePath(),
targetResourceFile.getAbsolutePath(),
@@ -3107,7 +3118,7 @@ public class DataRegion implements IDataRegionForQuery {
return null;
});
} catch (final IOException e) {
- logger.warn("Cannot delete localModFile {}", targetModFile, e);
+ writeLogger.warn("Cannot delete localModFile {}", targetModFile, e);
}
try {
final long modFileSize = modFileToLoad.length();
@@ -3128,7 +3139,7 @@ public class DataRegion implements IDataRegionForQuery {
FileMetrics.getInstance().increaseModFileNum(1);
FileMetrics.getInstance().increaseModFileSize(modFileSize);
} catch (final IOException e) {
- logger.warn(
+ writeLogger.warn(
"File renaming failed when loading .mod file. Origin: {}, Target: {}",
modFileToLoad.getAbsolutePath(),
targetModFile.getAbsolutePath(),
@@ -3172,7 +3183,7 @@ public class DataRegion implements IDataRegionForQuery {
tsFileResourceToBeRemoved.writeLock();
try {
tsFileResourceToBeRemoved.remove();
- logger.info("Remove tsfile {} successfully.", tsFileResourceToBeRemoved.getTsFile());
+ otherLogger.info("Remove tsfile {} successfully.", tsFileResourceToBeRemoved.getTsFile());
} finally {
tsFileResourceToBeRemoved.writeUnlock();
}
@@ -3197,7 +3208,7 @@ public class DataRegion implements IDataRegionForQuery {
tsFileResourceToBeMoved.writeLock();
try {
tsFileResourceToBeMoved.moveTo(targetDir);
- logger.info(
+ otherLogger.info(
"Move tsfile {} to target dir {} successfully.",
tsFileResourceToBeMoved.getTsFile(),
targetDir.getPath());
@@ -3280,7 +3291,7 @@ public class DataRegion implements IDataRegionForQuery {
try {
TimeUnit.MILLISECONDS.sleep(10);
} catch (InterruptedException e) {
- logger.error("Thread get interrupted when waiting compaction to finish", e);
+ compactionLogger.error("Thread get interrupted when waiting compaction to finish", e);
Thread.currentThread().interrupt();
}
}
@@ -3646,7 +3657,7 @@ public class DataRegion implements IDataRegionForQuery {
deletedCondition.await();
}
} catch (InterruptedException e) {
- logger.error("Interrupted When waiting for data region deleted.");
+ otherLogger.error("Interrupted When waiting for data region deleted.");
Thread.currentThread().interrupt();
} finally {
writeUnlock();
@@ -3714,7 +3725,7 @@ public class DataRegion implements IDataRegionForQuery {
Files.move(originFile.toPath(), Paths.get(newFileName));
}
} catch (IOException e) {
- logger.error("Failed to rename {} to {},", originFileName, newFileName, e);
+ otherLogger.error("Failed to rename {} to {},", originFileName, newFileName, e);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 9cc35f11d3a..01ff6152fdf 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -121,8 +121,11 @@ import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.WORK
public class TsFileProcessor {
/** Logger fot this class. */
- private static final Logger writeLogger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
- private static final Logger queryLogger = LoggerFactory.getLogger(IoTDBConstant.QUERY_LOGGER_NAME);
+ private static final Logger writeLogger =
+ LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
+
+ private static final Logger queryLogger =
+ LoggerFactory.getLogger(IoTDBConstant.QUERY_LOGGER_NAME);
private static final Logger otherLogger = LoggerFactory.getLogger(TsFileProcessor.class);
private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS =
@@ -1090,7 +1093,8 @@ public class TsFileProcessor {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
- writeLogger.info("File {} is closed synchronously", tsFileResource.getTsFile().getAbsolutePath());
+ writeLogger.info(
+ "File {} is closed synchronously", tsFileResource.getTsFile().getAbsolutePath());
}
/** async close one tsfile, register and close it by another thread */
(iotdb) 03/03: modify wal log
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch compation-log
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 29496df2097104835ef9019371aad344f35b4f7f
Author: shuwenwei <s1...@gmail.com>
AuthorDate: Fri Jun 20 11:49:45 2025 +0800
modify wal log
---
.../org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java | 2 +-
.../dataregion/wal/allocation/AbstractNodeAllocationStrategy.java | 3 ++-
.../db/storageengine/dataregion/wal/buffer/AbstractWALBuffer.java | 3 ++-
.../iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java | 3 ++-
.../apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java | 4 +++-
.../db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java | 3 ++-
.../apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java | 3 ++-
.../apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java | 3 ++-
.../apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java | 3 ++-
.../apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java | 3 ++-
.../iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java | 3 ++-
11 files changed, 22 insertions(+), 11 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
index 96d4931fa43..794fb7a8b0d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
@@ -56,7 +56,7 @@ import java.util.concurrent.atomic.AtomicLong;
/** This class is used to manage and allocate wal nodes. */
public class WALManager implements IService {
- private static final Logger logger = LoggerFactory.getLogger(WALManager.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
// manage all wal nodes and decide how to allocate them
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java
index e98086e5146..97c4c7aa68c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.allocation;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALFakeNode;
@@ -37,7 +38,7 @@ import java.util.Arrays;
public abstract class AbstractNodeAllocationStrategy implements NodeAllocationStrategy {
private static final Logger logger =
- LoggerFactory.getLogger(AbstractNodeAllocationStrategy.class);
+ LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
// manage wal folders
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/AbstractWALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/AbstractWALBuffer.java
index c8f68523032..d8ca0a5f4f6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/AbstractWALBuffer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/AbstractWALBuffer.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter;
@@ -35,7 +36,7 @@ import java.nio.file.StandardCopyOption;
import java.util.Arrays;
public abstract class AbstractWALBuffer implements IWALBuffer {
- private static final Logger logger = LoggerFactory.getLogger(AbstractWALBuffer.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
// WALNode identifier of this buffer
protected final String identifier;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
index 872a116d5c9..fe187123465 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -72,7 +73,7 @@ import static org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode.DEFA
* writes and avoid waiting for buffer syncing to disk.
*/
public class WALBuffer extends AbstractWALBuffer {
- private static final Logger logger = LoggerFactory.getLogger(WALBuffer.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
public static final int ONE_THIRD_WAL_BUFFER_SIZE = config.getWalBufferSize() / 3;
private static final double FSYNC_BUFFER_RATIO = 0.95;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java
index a8e94e2780a..f0dde4d2b49 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ContinuousSameSearchIndexSeparatorNode;
@@ -45,7 +47,7 @@ import java.util.Objects;
* value(physical plan or memTable snapshot).
*/
public abstract class WALEntry implements SerializedSize {
- private static final Logger logger = LoggerFactory.getLogger(WALEntry.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
// type of value
protected final WALEntryType type;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java
index 1c859f1eae4..8fd22bd9de4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -49,7 +50,7 @@ import java.util.concurrent.locks.ReentrantLock;
/** This class is used to manage checkpoints of one wal node. */
public class CheckpointManager implements AutoCloseable {
- private static final Logger logger = LoggerFactory.getLogger(CheckpointManager.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final WritingMetrics WRITING_METRICS = WritingMetrics.getInstance();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
index 95721f846cc..4bbb4231d15 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.service.metrics.WritingMetrics;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
@@ -41,7 +42,7 @@ import java.nio.channels.FileChannel;
* and writing {@link Checkpoint} into .checkpoint file.
*/
public abstract class LogWriter implements ILogWriter {
- private static final Logger logger = LoggerFactory.getLogger(LogWriter.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
protected final File logFile;
protected final FileOutputStream logStream;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
index ba9211656ef..c7bd86b6bb1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.BrokenWALFileException;
import org.apache.iotdb.db.utils.SerializedSize;
@@ -42,7 +43,7 @@ import java.util.Set;
*/
public class WALMetaData implements SerializedSize {
- private static final Logger logger = LoggerFactory.getLogger(WALMetaData.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
// search index 8 byte, wal entries' number 4 bytes
private static final int FIXED_SERIALIZED_SIZE = Long.BYTES + Integer.BYTES;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java
index 261e95eb52f..072244c0343 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
@@ -38,7 +39,7 @@ import java.util.NoSuchElementException;
* This reader returns {@link WALEntry} directly, the usage of WALReader is like {@link Iterator}.
*/
public class WALReader implements Closeable {
- private static final Logger logger = LoggerFactory.getLogger(WALReader.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
// use 1/10 of .wal file size as buffer size
private static final int STREAM_BUFFER_SIZE =
(int) IoTDBDescriptor.getInstance().getConfig().getWalFileSizeThresholdInByte() / 10;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
index 35432a8fd6f..7b101e28542 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.node;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.utils.TestOnly;
@@ -89,7 +90,7 @@ import java.util.stream.Collectors;
* 1.
*/
public class WALNode implements IWALNode {
- private static final Logger logger = LoggerFactory.getLogger(WALNode.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
// no iot consensus, all insert nodes can be safely deleted
public static final long DEFAULT_SAFELY_DELETED_SEARCH_INDEX = Long.MAX_VALUE;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java
index f5d7406f5a6..d04331b695e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.utils;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue;
@@ -40,7 +41,7 @@ import java.util.concurrent.atomic.AtomicReference;
*/
public class WALEntryHandler {
- private static final Logger logger = LoggerFactory.getLogger(WALEntryHandler.class);
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBConstant.WRITE_LOGGER_NAME);
private long memTableId = -1;