This is an automated email from the ASF dual-hosted git repository. haonan pushed a commit to branch empty_tablet in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3aa725765953f3b355b68a3b4a8fd555fe886920 Author: HTHou <[email protected]> AuthorDate: Fri Oct 27 16:35:09 2023 +0800 Fix TsFileresource error when insert empty tablet and flush --- .../db/storageengine/dataregion/DataRegion.java | 2 +- .../dataregion/memtable/AbstractMemTable.java | 6 +- .../dataregion/tsfile/TsFileResource.java | 4 ++ .../storageengine/dataregion/DataRegionTest.java | 78 ++++++++++++++++++++++ 4 files changed, 88 insertions(+), 2 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 8bce9f4f6ee..0eed0ceefcf 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 @@ -2053,7 +2053,7 @@ public class DataRegion implements IDataRegionForQuery { closeQueryLock.writeLock().lock(); try { tsFileProcessor.close(); - if (tsFileProcessor.isEmpty()) { + if (tsFileProcessor.isEmpty() || tsFileProcessor.getTsFileResource().isEmpty()) { try { fsFactory.deleteIfExists(tsFileProcessor.getTsFileResource().getTsFile()); tsFileManager.remove(tsFileProcessor.getTsFileResource(), tsFileProcessor.isSequence()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index f3264c09738..56c80dc308c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -660,7 +660,11 @@ public abstract class AbstractMemTable implements IMemTable { public Map<String, Long> getMaxTime() { Map<String, Long> latestTimeForEachDevice = new HashMap<>(); for (Entry<IDeviceID, IWritableMemChunkGroup> entry : memTableMap.entrySet()) { - latestTimeForEachDevice.put(entry.getKey().toStringID(), entry.getValue().getMaxTime()); + // When insert null values in to IWritableMemChunkGroup, the maxTime will not be updated. + // In this scenario, the maxTime will be Long.MIN_VALUE. We shouldn't return this device. + if (entry.getValue().getMaxTime() != Long.MIN_VALUE) { + latestTimeForEachDevice.put(entry.getKey().toStringID(), entry.getValue().getMaxTime()); + } } return latestTimeForEachDevice; } 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 cd8eb0b65f9..2875fabd178 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 @@ -1167,6 +1167,10 @@ public class TsFileResource { return maxProgressIndex == null ? MinimumProgressIndex.INSTANCE : maxProgressIndex; } + public boolean isEmpty() { + return getDevices().isEmpty(); + } + public String getDatabaseName() { return file.getParentFile().getParentFile().getParentFile().getName(); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java index 4e7b678298c..5f219927e4d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java @@ -302,6 +302,84 @@ public class DataRegionTest { } } + @Test + public void testIoTDBTabletWriteAndSyncClose() + throws QueryProcessException, IllegalPathException, WriteProcessException { + String[] measurements = new String[2]; + measurements[0] = "s0"; + measurements[1] = "s1"; + TSDataType[] dataTypes = new TSDataType[2]; + dataTypes[0] = TSDataType.INT32; + dataTypes[1] = TSDataType.INT64; + + MeasurementSchema[] measurementSchemas = new MeasurementSchema[2]; + measurementSchemas[0] = new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.PLAIN); + measurementSchemas[1] = new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN); + + long[] times = new long[100]; + Object[] columns = new Object[2]; + columns[0] = new int[100]; + columns[1] = new long[100]; + + for (int r = 0; r < 100; r++) { + times[r] = r; + ((int[]) columns[0])[r] = 1; + ((long[]) columns[1])[r] = 1; + } + + InsertTabletNode insertTabletNode1 = + new InsertTabletNode( + new QueryId("test_write").genPlanNodeId(), + new PartialPath("root.vehicle.d0"), + false, + measurements, + dataTypes, + measurementSchemas, + times, + null, + columns, + times.length); + + dataRegion.insertTablet(insertTabletNode1); + dataRegion.asyncCloseAllWorkingTsFileProcessors(); + + for (int r = 50; r < 149; r++) { + times[r - 50] = r; + ((int[]) columns[0])[r - 50] = 1; + ((long[]) columns[1])[r - 50] = 1; + } + + InsertTabletNode insertTabletNode2 = + new InsertTabletNode( + new QueryId("test_write").genPlanNodeId(), + new PartialPath("root.vehicle.d0"), + false, + measurements, + dataTypes, + measurementSchemas, + times, + null, + columns, + times.length); + + dataRegion.insertTablet(insertTabletNode2); + dataRegion.asyncCloseAllWorkingTsFileProcessors(); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + + QueryDataSource queryDataSource = + dataRegion.query( + Collections.singletonList(new PartialPath(deviceId, measurementId)), + deviceId, + context, + null); + + Assert.assertEquals(2, queryDataSource.getSeqResources().size()); + Assert.assertEquals(1, queryDataSource.getUnseqResources().size()); + for (TsFileResource resource : queryDataSource.getSeqResources()) { + Assert.assertTrue(resource.isClosed()); + } + } + @Test public void testSeqAndUnSeqSyncClose() throws WriteProcessException, QueryProcessException, IllegalPathException {
