add new logs
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java index 829e0ed..3a4a0ba1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
@@ -273,10 +273,9 @@ @Override public String toString() { return "PipeBatchTsFileInsertionEvent{" - + "resources=" + + resources.size() + + " resources=" + resources - + ", tsFiles=" - + tsFiles + ", totalSize=" + totalSize + '}';
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java index d2754c4..15b6694 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
@@ -101,7 +101,11 @@ event.setExtractorOnConnectorTimeout(this::onConnectorTimeout); event.setExtractorOnConnectorSuccess(this::onConnectorSuccess); - LOGGER.info("Generated a TsFile event: {}", event); + LOGGER.info( + "Generated a TsFile event: {}, max file num: {}, max file size: {}", + event, + maxBatchSize, + maxFileTotalSize); return event; }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/MergedTsFileSplitter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/MergedTsFileSplitter.java index c8a7aa4..bdc894c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/MergedTsFileSplitter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/MergedTsFileSplitter.java
@@ -111,7 +111,7 @@ File tsFile = tsFiles.get(fileIndex); SplitTask splitTask = new SplitTask(tsFile, asyncExecutor, fileIndex); - logger.info("Start to split {}", tsFiles.get(fileIndex)); + logger.debug("Start to split {}", tsFiles.get(fileIndex)); if (splitTask.hasNext()) { taskPriorityQueue.add(splitTask); }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/nodesplit/ClusteringMeasurementSplitter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/nodesplit/ClusteringMeasurementSplitter.java index f79a06b..8b5840f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/nodesplit/ClusteringMeasurementSplitter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/nodesplit/ClusteringMeasurementSplitter.java
@@ -97,7 +97,9 @@ loadTsFilePieceNodes.stream() .map(l -> l.getAllTsFileData().size()) .collect(Collectors.toList())); - refineClusters(loadTsFilePieceNodes); + } + refineClusters(loadTsFilePieceNodes); + if (logger.isDebugEnabled()) { logger.debug( "Split distribution after refinement: {}", loadTsFilePieceNodes.stream()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java index c1721c9..bb34437 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -570,7 +570,8 @@ @Override public String toString() { - return String.format("file is %s, status: %s", file.toString(), getStatus()); + return String.format( + "file is %s, status: %s, size: %s", file.toString(), getStatus(), getTsFileSize()); } @Override
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSenderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSenderTest.java index a682263..21754a6 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSenderTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSenderTest.java
@@ -73,7 +73,7 @@ private long dummyDelayMS = 0; private double packetLossRatio = 0.00; private Random random = new Random(); - private long maxSplitSize = 128 * 1024 * 1024; + private long maxSplitSize = 128 * 1024 * 1024L; // simulating jvm stall like GC private long minStuckIntervalMS = 50000; private long maxStuckIntervalMS = 100000; @@ -109,27 +109,35 @@ }); thread.start(); - LoadTsFileNode loadTsFileNode = - new LoadTsFileNode(new PlanNodeId("testPlanNode"), tsFileResources); - DataPartitionBatchFetcher partitionBatchFetcher = dummyDataPartitionBatchFetcher(); - TsFileSplitSender splitSender = - new TsFileSplitSender( - loadTsFileNode, - partitionBatchFetcher, - TimePartitionUtils.getTimePartitionInterval(), - internalServiceClientManager, - false, - maxSplitSize, - 100, - "root", - "root"); + int filesPerNode = 2; + int startFileIndex = 0; long start = System.currentTimeMillis(); - splitSender.start(); + long transmissionTime = 0; + while (startFileIndex < tsFileResources.size()) { + int endFileIndex = Math.min(tsFileResources.size(), startFileIndex + filesPerNode); + LoadTsFileNode loadTsFileNode = + new LoadTsFileNode(new PlanNodeId("testPlanNode"), tsFileResources.subList(startFileIndex, endFileIndex)); + DataPartitionBatchFetcher partitionBatchFetcher = dummyDataPartitionBatchFetcher(); + TsFileSplitSender splitSender = + new TsFileSplitSender( + loadTsFileNode, + partitionBatchFetcher, + TimePartitionUtils.getTimePartitionInterval(), + internalServiceClientManager, + false, + maxSplitSize, + 100, + "root", + "root"); + splitSender.start(); + transmissionTime += splitSender.getStatistic().getCompressedSize().get() / nodeThroughput; + startFileIndex = endFileIndex; + } + long timeConsumption = System.currentTimeMillis() - start; thread.interrupt(); printPhaseResult(); - long transmissionTime = splitSender.getStatistic().getCompressedSize().get() / nodeThroughput; System.out.printf( "Split ends after %dms + %dms (Transmission) = %dms\n", timeConsumption, transmissionTime, timeConsumption + transmissionTime);