LogReplayer { public void replayLogs() { WriteLogNode logNode = MultiFileLogNodeManager.getInstance().getNode( logNodePrefix + FSFactoryProducer.getFSFactory().getFile(insertFilePath).getName()); ILogReader logReader = logNode.getLogReader(); try { while (logReader.hasNext()) { try { PhysicalPlan plan = logReader.next(); if (plan instanceof InsertPlan) { replayInsert((InsertPlan) plan); } else if (plan instanceof DeletePlan) { replayDelete((DeletePlan) plan); } else if (plan instanceof UpdatePlan) { replayUpdate((UpdatePlan) plan); } } catch (Exception e) { logger.error("recover wal of {} failed", insertFilePath, e); } } } catch (IOException e) { logger.error("meet error when redo wal of {}", insertFilePath, e); } finally { logReader.close(); try { modFile.close(); } catch (IOException e) { logger.error("Canno close the modifications file {}", modFile.getFilePath(), e); } } tempStartTimeMap.forEach((k, v) -> currentTsFileResource.updateStartTime(k, v)); tempEndTimeMap.forEach((k, v) -> currentTsFileResource.updateEndTime(k, v)); } LogReplayer(String logNodePrefix, String insertFilePath, ModificationFile modFile,
      VersionController versionController, TsFileResource currentTsFileResource,
      IMemTable memTable, boolean sequence); void replayLogs();  }
@Test public void test() throws IOException, StorageGroupProcessorException, QueryProcessException, MetadataException { String logNodePrefix = "testLogNode"; File tsFile = SystemFileFactory.INSTANCE.getFile("temp", "1-1-1.tsfile"); File modF = SystemFileFactory.INSTANCE.getFile("test.mod"); ModificationFile modFile = new ModificationFile(modF.getPath()); VersionController versionController = new VersionController() { @Override public long nextVersion() { return 5; } @Override public long currVersion() { return 5; } }; TsFileResource tsFileResource = new TsFileResource(tsFile); IMemTable memTable = new PrimitiveMemTable(); IoTDB.metaManager.setStorageGroup(new PartialPath("root.sg")); try { for (int i = 0; i <= 5; i++) { for (int j = 0; j <= 5; j++) { IoTDB.metaManager .createTimeseries(new PartialPath("root.sg.device" + i + ".sensor" + j), TSDataType.INT64, TSEncoding.PLAIN, TSFileDescriptor.getInstance().getConfig().getCompressor(), Collections.emptyMap()); } } LogReplayer replayer = new LogReplayer(logNodePrefix, tsFile.getPath(), modFile, versionController, tsFileResource, memTable, false); WriteLogNode node = MultiFileLogNodeManager.getInstance().getNode(logNodePrefix + tsFile.getName()); node.write( new InsertRowPlan(new PartialPath("root.sg.device0"), 100, "sensor0", TSDataType.INT64, String.valueOf(0))); node.write( new InsertRowPlan(new PartialPath("root.sg.device0"), 2, "sensor1", TSDataType.INT64, String.valueOf(0))); for (int i = 1; i < 5; i++) { node.write(new InsertRowPlan(new PartialPath("root.sg.device" + i), i, "sensor" + i, TSDataType.INT64, String.valueOf(i))); } node.write(insertTablePlan()); DeletePlan deletePlan = new DeletePlan(0, 200, new PartialPath("root.sg.device0.sensor0")); node.write(deletePlan); node.close(); replayer.replayLogs(); for (int i = 0; i < 5; i++) { ReadOnlyMemChunk memChunk = memTable .query("root.sg.device" + i, "sensor" + i, TSDataType.INT64, TSEncoding.RLE, Collections.emptyMap(), Long.MIN_VALUE); IPointReader iterator = memChunk.getPointReader(); if (i == 0) { assertFalse(iterator.hasNextTimeValuePair()); } else { assertTrue(iterator.hasNextTimeValuePair()); TimeValuePair timeValuePair = iterator.nextTimeValuePair(); assertEquals(i, timeValuePair.getTimestamp()); assertEquals(i, timeValuePair.getValue().getLong()); assertFalse(iterator.hasNextTimeValuePair()); } } Modification[] mods = modFile.getModifications().toArray(new Modification[0]); assertEquals(1, mods.length); assertEquals("root.sg.device0.sensor0", mods[0].getPathString()); assertEquals(5, mods[0].getVersionNum()); assertEquals(((Deletion) mods[0]).getEndTime(), 200); assertEquals(2, tsFileResource.getStartTime("root.sg.device0")); assertEquals(100, tsFileResource.getEndTime("root.sg.device0")); for (int i = 1; i < 5; i++) { assertEquals(i, tsFileResource.getStartTime("root.sg.device" + i)); assertEquals(i, tsFileResource.getEndTime("root.sg.device" + i)); } for (int i = 0; i < 2 ; i++) { ReadOnlyMemChunk memChunk = memTable .query("root.sg.device5", "sensor" + i, TSDataType.INT64, TSEncoding.PLAIN, Collections.emptyMap(), Long.MIN_VALUE); if (i == 0) { assertNull(memChunk); } else { IPointReader iterator = memChunk.getPointReader(); iterator.hasNextTimeValuePair(); for (int time = 0; time < 100; time++) { TimeValuePair timeValuePair = iterator.nextTimeValuePair(); assertEquals(time, timeValuePair.getTimestamp()); assertEquals(time, timeValuePair.getValue().getLong()); } } } } finally { modFile.close(); MultiFileLogNodeManager.getInstance().deleteNode(logNodePrefix + tsFile.getName()); modF.delete(); tsFile.delete(); tsFile.getParentFile().delete(); } }