| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file; |
| |
| import org.apache.iotdb.commons.path.AlignedPath; |
| import org.apache.iotdb.commons.path.MeasurementPath; |
| import org.apache.iotdb.commons.path.PartialPath; |
| import org.apache.iotdb.db.conf.IoTDBDescriptor; |
| import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; |
| import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; |
| import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; |
| import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; |
| import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; |
| import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; |
| import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; |
| import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; |
| import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; |
| import org.apache.iotdb.db.storageengine.dataregion.wal.utils.TsFileUtilsForRecoverTest; |
| import org.apache.iotdb.db.utils.EnvironmentUtils; |
| import org.apache.iotdb.tsfile.common.conf.TSFileConfig; |
| import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; |
| import org.apache.iotdb.tsfile.exception.write.WriteProcessException; |
| import org.apache.iotdb.tsfile.file.metadata.IDeviceID; |
| import org.apache.iotdb.tsfile.file.metadata.PlainDeviceID; |
| import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; |
| import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; |
| import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; |
| import org.apache.iotdb.tsfile.read.TimeValuePair; |
| import org.apache.iotdb.tsfile.read.common.Path; |
| import org.apache.iotdb.tsfile.read.reader.IPointReader; |
| import org.apache.iotdb.tsfile.utils.Binary; |
| import org.apache.iotdb.tsfile.utils.BitMap; |
| import org.apache.iotdb.tsfile.utils.BytesUtils; |
| import org.apache.iotdb.tsfile.write.TsFileWriter; |
| import org.apache.iotdb.tsfile.write.record.TSRecord; |
| import org.apache.iotdb.tsfile.write.record.datapoint.BooleanDataPoint; |
| import org.apache.iotdb.tsfile.write.record.datapoint.DoubleDataPoint; |
| import org.apache.iotdb.tsfile.write.record.datapoint.FloatDataPoint; |
| import org.apache.iotdb.tsfile.write.record.datapoint.IntDataPoint; |
| import org.apache.iotdb.tsfile.write.record.datapoint.LongDataPoint; |
| import org.apache.iotdb.tsfile.write.record.datapoint.StringDataPoint; |
| import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; |
| |
| import org.junit.After; |
| import org.junit.Before; |
| import org.junit.Test; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.List; |
| |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| |
| public class TsFilePlanRedoerTest { |
| private static final String SG_NAME = "root.recover_sg"; |
| private static final IDeviceID DEVICE1_NAME = new PlainDeviceID(SG_NAME.concat(".d1")); |
| private static final IDeviceID DEVICE2_NAME = new PlainDeviceID(SG_NAME.concat(".d2")); |
| private static final IDeviceID DEVICE3_NAME = new PlainDeviceID(SG_NAME.concat(".d3")); |
| private static final String FILE_NAME = |
| TsFileUtilsForRecoverTest.getTestTsFilePath(SG_NAME, 0, 0, 1); |
| private TsFileResource tsFileResource; |
| private CompressionType compressionType; |
| boolean prevIsAutoCreateSchemaEnabled; |
| boolean prevIsEnablePartialInsert; |
| |
| @Before |
| public void setUp() throws Exception { |
| EnvironmentUtils.envSetUp(); |
| |
| // set recover config, avoid creating deleted time series when recovering wal |
| prevIsAutoCreateSchemaEnabled = |
| IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); |
| // IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(false); |
| prevIsEnablePartialInsert = IoTDBDescriptor.getInstance().getConfig().isEnablePartialInsert(); |
| IoTDBDescriptor.getInstance().getConfig().setEnablePartialInsert(true); |
| compressionType = TSFileDescriptor.getInstance().getConfig().getCompressor(); |
| } |
| |
| @After |
| public void tearDown() throws Exception { |
| if (tsFileResource != null) { |
| tsFileResource.close(); |
| } |
| File modsFile = new File(FILE_NAME.concat(ModificationFile.FILE_SUFFIX)); |
| if (modsFile.exists()) { |
| modsFile.delete(); |
| } |
| EnvironmentUtils.cleanEnv(); |
| // reset config |
| // IoTDBDescriptor.getInstance() |
| // .getConfig() |
| // .setAutoCreateSchemaEnabled(prevIsAutoCreateSchemaEnabled); |
| IoTDBDescriptor.getInstance().getConfig().setEnablePartialInsert(prevIsEnablePartialInsert); |
| } |
| |
| @Test |
| public void testRedoInsertRowPlan() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE1_NAME, 1); |
| tsFileResource.updateEndTime(DEVICE1_NAME, 2); |
| tsFileResource.updateStartTime(DEVICE2_NAME, 3); |
| tsFileResource.updateEndTime(DEVICE2_NAME, 4); |
| |
| // generate InsertRowPlan |
| long time = 5; |
| TSDataType[] dataTypes = new TSDataType[] {TSDataType.FLOAT, TSDataType.DOUBLE}; |
| Object[] columns = new Object[] {1f, 1.0d}; |
| InsertRowNode insertRowNode = |
| new InsertRowNode( |
| new PlanNodeId("0"), |
| new PartialPath(DEVICE2_NAME), |
| false, |
| new String[] {"s1", "s2"}, |
| dataTypes, |
| time, |
| columns, |
| false); |
| insertRowNode.setMeasurementSchemas( |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.FLOAT), |
| new MeasurementSchema("s2", TSDataType.DOUBLE), |
| }); |
| |
| // redo InsertTabletPlan, vsg processor is used to test IdTable, don't test IdTable here |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| planRedoer.redoInsert(insertRowNode); |
| |
| // check data in memTable |
| IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable(); |
| // check d2.s1 |
| MeasurementPath fullPath = |
| new MeasurementPath( |
| DEVICE2_NAME, "s1", new MeasurementSchema("s1", TSDataType.FLOAT, TSEncoding.RLE)); |
| ReadOnlyMemChunk memChunk = |
| recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| IPointReader iterator = memChunk.getPointReader(); |
| time = 5; |
| while (iterator.hasNextTimeValuePair()) { |
| TimeValuePair timeValuePair = iterator.nextTimeValuePair(); |
| assertEquals(time, timeValuePair.getTimestamp()); |
| assertEquals(1, timeValuePair.getValue().getFloat(), 0.0001); |
| ++time; |
| } |
| assertEquals(6, time); |
| // check d2.s2 |
| fullPath = |
| new MeasurementPath( |
| DEVICE2_NAME, "s2", new MeasurementSchema("s2", TSDataType.DOUBLE, TSEncoding.RLE)); |
| memChunk = recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| iterator = memChunk.getPointReader(); |
| time = 5; |
| while (iterator.hasNextTimeValuePair()) { |
| TimeValuePair timeValuePair = iterator.nextTimeValuePair(); |
| assertEquals(time, timeValuePair.getTimestamp()); |
| assertEquals(1, timeValuePair.getValue().getDouble(), 0.0001); |
| ++time; |
| } |
| assertEquals(6, time); |
| } |
| |
| @Test |
| public void testRedoInsertAlignedRowPlan() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE3_NAME, 5); |
| tsFileResource.updateEndTime(DEVICE3_NAME, 5); |
| |
| // generate InsertRowPlan |
| TSDataType[] dataTypes = |
| new TSDataType[] { |
| TSDataType.INT32, TSDataType.INT64, TSDataType.BOOLEAN, TSDataType.FLOAT, TSDataType.TEXT |
| }; |
| Object[] columns = |
| new Object[] {1, 1L, true, 1.0f, new Binary("1", TSFileConfig.STRING_CHARSET)}; |
| |
| InsertRowNode insertRowNode1 = |
| new InsertRowNode( |
| new PlanNodeId("0"), |
| new PartialPath(DEVICE3_NAME), |
| true, |
| new String[] {"s1", "s2", "s3", "s4", "s5"}, |
| dataTypes, |
| 5, |
| columns, |
| false); |
| insertRowNode1.setMeasurementSchemas( |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.INT32), |
| new MeasurementSchema("s2", TSDataType.INT64), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN), |
| new MeasurementSchema("s4", TSDataType.FLOAT), |
| new MeasurementSchema("s5", TSDataType.TEXT), |
| }); |
| |
| InsertRowNode insertRowNode2 = |
| new InsertRowNode( |
| new PlanNodeId("0"), |
| new PartialPath(DEVICE3_NAME), |
| true, |
| new String[] {"s1", "s2", "s3", "s4", "s5"}, |
| dataTypes, |
| 6, |
| columns, |
| false); |
| insertRowNode2.setMeasurementSchemas( |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.INT32), |
| new MeasurementSchema("s2", TSDataType.INT64), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN), |
| new MeasurementSchema("s4", TSDataType.FLOAT), |
| new MeasurementSchema("s5", TSDataType.TEXT), |
| }); |
| |
| // redo InsertTabletPlan, vsg processor is used to test IdTable, don't test IdTable here |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| planRedoer.redoInsert(insertRowNode1); |
| planRedoer.redoInsert(insertRowNode2); |
| |
| // check data in memTable |
| IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable(); |
| // check d3 |
| AlignedPath fullPath = |
| new AlignedPath( |
| ((PlainDeviceID) DEVICE3_NAME).toStringID(), |
| Arrays.asList("s1", "s2", "s3", "s4", "s5"), |
| Arrays.asList( |
| new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE), |
| new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN, TSEncoding.RLE), |
| new MeasurementSchema("s4", TSDataType.FLOAT, TSEncoding.RLE), |
| new MeasurementSchema("s5", TSDataType.TEXT, TSEncoding.PLAIN))); |
| ReadOnlyMemChunk memChunk = |
| recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| IPointReader iterator = memChunk.getPointReader(); |
| int time = 6; |
| while (iterator.hasNextTimeValuePair()) { |
| TimeValuePair timeValuePair = iterator.nextTimeValuePair(); |
| assertEquals(time, timeValuePair.getTimestamp()); |
| assertEquals(1, timeValuePair.getValue().getVector()[0].getInt()); |
| assertEquals(1L, timeValuePair.getValue().getVector()[1].getLong()); |
| assertTrue(timeValuePair.getValue().getVector()[2].getBoolean()); |
| assertEquals(1, timeValuePair.getValue().getVector()[3].getFloat(), 0.00001); |
| assertEquals(BytesUtils.valueOf("1"), timeValuePair.getValue().getVector()[4].getBinary()); |
| ++time; |
| } |
| assertEquals(7, time); |
| } |
| |
| @Test |
| public void testRedoInsertTabletPlan() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE1_NAME, 1); |
| tsFileResource.updateEndTime(DEVICE1_NAME, 2); |
| tsFileResource.updateStartTime(DEVICE2_NAME, 3); |
| tsFileResource.updateEndTime(DEVICE2_NAME, 4); |
| |
| // generate InsertTabletPlan |
| long[] times = {5, 6, 7, 8}; |
| List<Integer> dataTypes = new ArrayList<>(); |
| dataTypes.add(TSDataType.INT32.ordinal()); |
| dataTypes.add(TSDataType.INT64.ordinal()); |
| |
| Object[] columns = new Object[2]; |
| columns[0] = new int[times.length]; |
| columns[1] = new long[times.length]; |
| |
| for (int r = 0; r < times.length; r++) { |
| ((int[]) columns[0])[r] = 100; |
| ((long[]) columns[1])[r] = 10000; |
| } |
| |
| BitMap[] bitMaps = new BitMap[dataTypes.size()]; |
| for (int i = 0; i < dataTypes.size(); i++) { |
| if (bitMaps[i] == null) { |
| bitMaps[i] = new BitMap(times.length); |
| } |
| // mark value of time=8 as null |
| bitMaps[i].mark(3); |
| } |
| |
| InsertTabletNode insertTabletNode = |
| new InsertTabletNode( |
| new PlanNodeId("0"), |
| new PartialPath(DEVICE1_NAME), |
| false, |
| new String[] {"s1", "s2"}, |
| new TSDataType[] {TSDataType.INT32, TSDataType.INT64}, |
| null, |
| times, |
| bitMaps, |
| columns, |
| times.length); |
| insertTabletNode.setMeasurementSchemas( |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.INT32), |
| new MeasurementSchema("s2", TSDataType.INT64), |
| }); |
| |
| // redo InsertTabletPlan, vsg processor is used to test IdTable, don't test IdTable here |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| planRedoer.redoInsert(insertTabletNode); |
| |
| // check data in memTable |
| IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable(); |
| // check d1.s1 |
| MeasurementPath fullPath = |
| new MeasurementPath( |
| DEVICE1_NAME, "s1", new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE)); |
| ReadOnlyMemChunk memChunk = |
| recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| IPointReader iterator = memChunk.getPointReader(); |
| int time = 5; |
| while (iterator.hasNextTimeValuePair()) { |
| TimeValuePair timeValuePair = iterator.nextTimeValuePair(); |
| assertEquals(time, timeValuePair.getTimestamp()); |
| assertEquals(100, timeValuePair.getValue().getInt()); |
| ++time; |
| } |
| assertEquals(8, time); |
| // check d1.s2 |
| fullPath = |
| new MeasurementPath( |
| DEVICE1_NAME, "s2", new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); |
| memChunk = recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| iterator = memChunk.getPointReader(); |
| time = 5; |
| while (iterator.hasNextTimeValuePair()) { |
| TimeValuePair timeValuePair = iterator.nextTimeValuePair(); |
| assertEquals(time, timeValuePair.getTimestamp()); |
| assertEquals(10000, timeValuePair.getValue().getLong()); |
| ++time; |
| } |
| assertEquals(8, time); |
| } |
| |
| @Test |
| public void testRedoInsertAlignedTabletPlan() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE3_NAME, 5); |
| tsFileResource.updateStartTime(DEVICE3_NAME, 5); |
| |
| // generate InsertTabletPlan |
| long[] times = {6, 7, 8, 9}; |
| List<Integer> dataTypes = new ArrayList<>(); |
| dataTypes.add(TSDataType.INT32.ordinal()); |
| dataTypes.add(TSDataType.INT64.ordinal()); |
| dataTypes.add(TSDataType.BOOLEAN.ordinal()); |
| dataTypes.add(TSDataType.FLOAT.ordinal()); |
| dataTypes.add(TSDataType.TEXT.ordinal()); |
| |
| Object[] columns = new Object[5]; |
| columns[0] = new int[times.length]; |
| columns[1] = new long[times.length]; |
| columns[2] = new boolean[times.length]; |
| columns[3] = new float[times.length]; |
| columns[4] = new Binary[times.length]; |
| |
| for (int r = 0; r < times.length; r++) { |
| ((int[]) columns[0])[r] = (r + 1) * 100; |
| ((long[]) columns[1])[r] = (r + 1) * 100; |
| ((boolean[]) columns[2])[r] = true; |
| ((float[]) columns[3])[r] = (r + 1) * 100; |
| ((Binary[]) columns[4])[r] = BytesUtils.valueOf((r + 1) * 100 + ""); |
| } |
| |
| BitMap[] bitMaps = new BitMap[dataTypes.size()]; |
| for (int i = 0; i < dataTypes.size(); i++) { |
| if (bitMaps[i] == null) { |
| bitMaps[i] = new BitMap(times.length); |
| } |
| // mark value of time=9 as null |
| bitMaps[i].mark(3); |
| } |
| |
| InsertTabletNode insertTabletNode = |
| new InsertTabletNode( |
| new PlanNodeId("0"), |
| new PartialPath(DEVICE3_NAME), |
| true, |
| new String[] {"s1", "s2", "s3", "s4", "s5"}, |
| new TSDataType[] { |
| TSDataType.INT32, |
| TSDataType.INT64, |
| TSDataType.BOOLEAN, |
| TSDataType.FLOAT, |
| TSDataType.TEXT |
| }, |
| null, |
| times, |
| bitMaps, |
| columns, |
| times.length); |
| insertTabletNode.setMeasurementSchemas( |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.INT32), |
| new MeasurementSchema("s2", TSDataType.INT64), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN), |
| new MeasurementSchema("s4", TSDataType.FLOAT), |
| new MeasurementSchema("s5", TSDataType.TEXT), |
| }); |
| |
| // redo InsertTabletPlan, vsg processor is used to test IdTable, don't test IdTable here |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| planRedoer.redoInsert(insertTabletNode); |
| |
| // check data in memTable |
| IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable(); |
| // check d3 |
| AlignedPath fullPath = |
| new AlignedPath( |
| ((PlainDeviceID) DEVICE3_NAME).toStringID(), |
| Arrays.asList("s1", "s2", "s3", "s4", "s5"), |
| Arrays.asList( |
| new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE), |
| new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN, TSEncoding.RLE), |
| new MeasurementSchema("s4", TSDataType.FLOAT, TSEncoding.RLE), |
| new MeasurementSchema("s5", TSDataType.TEXT, TSEncoding.PLAIN))); |
| ReadOnlyMemChunk memChunk = |
| recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| IPointReader iterator = memChunk.getPointReader(); |
| int time = 6; |
| while (iterator.hasNextTimeValuePair()) { |
| TimeValuePair timeValuePair = iterator.nextTimeValuePair(); |
| assertEquals(time, timeValuePair.getTimestamp()); |
| assertEquals((time - 5) * 100, timeValuePair.getValue().getVector()[0].getInt()); |
| assertEquals((time - 5) * 100L, timeValuePair.getValue().getVector()[1].getLong()); |
| assertEquals(true, timeValuePair.getValue().getVector()[2].getBoolean()); |
| assertEquals((time - 5) * 100, timeValuePair.getValue().getVector()[3].getFloat(), 0.00001); |
| assertEquals( |
| BytesUtils.valueOf((time - 5) * 100 + ""), |
| timeValuePair.getValue().getVector()[4].getBinary()); |
| ++time; |
| } |
| assertEquals(9, time); |
| } |
| |
| @Test |
| public void testRedoOverLapPlanIntoSeqFile() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE1_NAME, 1); |
| tsFileResource.updateEndTime(DEVICE1_NAME, 2); |
| tsFileResource.updateStartTime(DEVICE2_NAME, 3); |
| tsFileResource.updateEndTime(DEVICE2_NAME, 4); |
| |
| // generate InsertTabletPlan, time=3 and time=4 are overlap |
| long[] times = {1, 2}; |
| List<Integer> dataTypes = new ArrayList<>(); |
| dataTypes.add(TSDataType.INT32.ordinal()); |
| dataTypes.add(TSDataType.INT64.ordinal()); |
| |
| Object[] columns = new Object[2]; |
| columns[0] = new int[times.length]; |
| columns[1] = new long[times.length]; |
| |
| for (int r = 0; r < times.length; r++) { |
| ((int[]) columns[0])[r] = 100; |
| ((long[]) columns[1])[r] = 10000; |
| } |
| |
| InsertTabletNode insertTabletNode = |
| new InsertTabletNode( |
| new PlanNodeId("0"), |
| new PartialPath(DEVICE1_NAME), |
| false, |
| new String[] {"s1", "s2"}, |
| new TSDataType[] {TSDataType.INT32, TSDataType.INT64}, |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.INT32), |
| new MeasurementSchema("s2", TSDataType.INT64), |
| }, |
| times, |
| null, |
| columns, |
| times.length); |
| |
| // redo InsertTabletPlan, vsg processor is used to test IdTable, don't test IdTable here |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| planRedoer.redoInsert(insertTabletNode); |
| |
| // check data in memTable |
| IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable(); |
| assertTrue(recoveryMemTable.isEmpty()); |
| } |
| |
| @Test |
| public void testRedoOverLapPlanIntoUnseqFile() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE1_NAME, 1); |
| tsFileResource.updateEndTime(DEVICE1_NAME, 2); |
| tsFileResource.updateStartTime(DEVICE2_NAME, 3); |
| tsFileResource.updateEndTime(DEVICE2_NAME, 4); |
| |
| // generate InsertTabletPlan, time=3 and time=4 are overlap |
| long[] times = {1, 2}; |
| List<Integer> dataTypes = new ArrayList<>(); |
| dataTypes.add(TSDataType.INT32.ordinal()); |
| dataTypes.add(TSDataType.INT64.ordinal()); |
| |
| Object[] columns = new Object[2]; |
| columns[0] = new int[times.length]; |
| columns[1] = new long[times.length]; |
| |
| for (int r = 0; r < times.length; r++) { |
| ((int[]) columns[0])[r] = 100; |
| ((long[]) columns[1])[r] = 10000; |
| } |
| |
| InsertTabletNode insertTabletNode = |
| new InsertTabletNode( |
| new PlanNodeId("0"), |
| new PartialPath(DEVICE1_NAME), |
| false, |
| new String[] {"s1", "s2"}, |
| new TSDataType[] {TSDataType.INT32, TSDataType.INT64}, |
| new MeasurementSchema[] { |
| new MeasurementSchema("s1", TSDataType.INT32), |
| new MeasurementSchema("s2", TSDataType.INT64), |
| }, |
| times, |
| null, |
| columns, |
| times.length); |
| |
| // redo InsertTabletPlan, vsg processor is used to test IdTable, don't test IdTable here |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| planRedoer.redoInsert(insertTabletNode); |
| |
| // check data in memTable |
| IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable(); |
| // check d1.s1 |
| MeasurementPath fullPath = |
| new MeasurementPath( |
| DEVICE1_NAME, "s1", new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE)); |
| ReadOnlyMemChunk memChunk = |
| recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| assertTrue(memChunk == null || memChunk.isEmpty()); |
| // check d1.s2 |
| fullPath = |
| new MeasurementPath( |
| DEVICE1_NAME, "s2", new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); |
| memChunk = recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| assertTrue(memChunk == null || memChunk.isEmpty()); |
| } |
| |
| @Test |
| public void testRedoDeleteDataNode() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE1_NAME, 1); |
| tsFileResource.updateEndTime(DEVICE1_NAME, 2); |
| tsFileResource.updateStartTime(DEVICE2_NAME, 3); |
| tsFileResource.updateEndTime(DEVICE2_NAME, 4); |
| |
| // generate DeleteDataNode |
| DeleteDataNode deleteDataNode = |
| new DeleteDataNode( |
| new PlanNodeId(""), |
| Collections.singletonList(new PartialPath(DEVICE1_NAME)), |
| Long.MIN_VALUE, |
| Long.MAX_VALUE); |
| |
| // redo DeleteDataNode, vsg processor is used to test IdTable, don't test IdTable here |
| File modsFile = new File(FILE_NAME.concat(ModificationFile.FILE_SUFFIX)); |
| assertFalse(modsFile.exists()); |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| planRedoer.redoDelete(deleteDataNode); |
| assertTrue(modsFile.exists()); |
| } |
| |
| @Test |
| public void testRedoAlignedInsertAfterDeleteTimeseries() throws Exception { |
| // generate .tsfile and update resource in memory |
| File file = new File(FILE_NAME); |
| generateCompleteFile(file); |
| tsFileResource = new TsFileResource(file); |
| tsFileResource.updateStartTime(DEVICE3_NAME, 5); |
| tsFileResource.updateStartTime(DEVICE3_NAME, 5); |
| |
| // generate InsertTabletPlan |
| long[] times = {6, 7, 8, 9}; |
| List<Integer> dataTypes = |
| Arrays.asList( |
| TSDataType.INT32.ordinal(), |
| TSDataType.INT64.ordinal(), |
| TSDataType.BOOLEAN.ordinal(), |
| TSDataType.FLOAT.ordinal(), |
| TSDataType.TEXT.ordinal()); |
| Object[] columns = |
| new Object[] { |
| new int[times.length], |
| new long[times.length], |
| new boolean[times.length], |
| new float[times.length], |
| new Binary[times.length] |
| }; |
| for (int r = 0; r < times.length; r++) { |
| ((int[]) columns[0])[r] = (r + 1) * 100; |
| ((long[]) columns[1])[r] = (r + 1) * 100; |
| ((boolean[]) columns[2])[r] = true; |
| ((float[]) columns[3])[r] = (r + 1) * 100; |
| ((Binary[]) columns[4])[r] = BytesUtils.valueOf((r + 1) * 100 + ""); |
| } |
| BitMap[] bitMaps = new BitMap[dataTypes.size()]; |
| for (int i = 0; i < dataTypes.size(); i++) { |
| if (bitMaps[i] == null) { |
| bitMaps[i] = new BitMap(times.length); |
| } |
| // mark value of time=9 as null |
| bitMaps[i].mark(3); |
| } |
| MeasurementSchema[] schemas = |
| new MeasurementSchema[] { |
| null, |
| new MeasurementSchema("s2", TSDataType.INT64), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN), |
| new MeasurementSchema("s4", TSDataType.FLOAT), |
| null |
| }; |
| InsertTabletNode insertTabletNode = |
| new InsertTabletNode( |
| new PlanNodeId(""), |
| new PartialPath(DEVICE3_NAME), |
| true, |
| new String[] {null, "s2", "s3", "s4", null}, |
| new TSDataType[] { |
| TSDataType.INT32, |
| TSDataType.INT64, |
| TSDataType.BOOLEAN, |
| TSDataType.FLOAT, |
| TSDataType.TEXT |
| }, |
| schemas, |
| times, |
| bitMaps, |
| columns, |
| times.length); |
| // redo InsertTabletPlan, data region is used to test IdTable, don't test IdTable here |
| TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); |
| insertTabletNode.setMeasurementSchemas(schemas); |
| planRedoer.redoInsert(insertTabletNode); |
| |
| // generate InsertRowPlan |
| int time = 9; |
| TSDataType[] dataTypes2 = |
| new TSDataType[] { |
| TSDataType.INT32, TSDataType.INT64, TSDataType.BOOLEAN, TSDataType.FLOAT, TSDataType.TEXT |
| }; |
| Object[] columns2 = |
| new Object[] {400, 400L, true, 400.0f, new Binary("400", TSFileConfig.STRING_CHARSET)}; |
| // redo InsertTabletPlan, data region is used to test IdTable, don't test IdTable here |
| InsertRowNode insertRowNode = |
| new InsertRowNode( |
| new PlanNodeId(""), |
| new PartialPath(DEVICE3_NAME), |
| true, |
| new String[] {null, "s2", "s3", "s4", null}, |
| dataTypes2, |
| time, |
| columns2, |
| false); |
| insertRowNode.setMeasurementSchemas(schemas); |
| planRedoer.redoInsert(insertRowNode); |
| |
| // check data in memTable |
| IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable(); |
| // check d3 |
| AlignedPath fullPath = |
| new AlignedPath( |
| ((PlainDeviceID) DEVICE3_NAME).toStringID(), |
| Arrays.asList("s1", "s2", "s3", "s4", "s5"), |
| Arrays.asList( |
| new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE), |
| new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN, TSEncoding.RLE), |
| new MeasurementSchema("s4", TSDataType.FLOAT, TSEncoding.RLE), |
| new MeasurementSchema("s5", TSDataType.TEXT, TSEncoding.PLAIN))); |
| ReadOnlyMemChunk memChunk = |
| recoveryMemTable.query(new QueryContext(), fullPath, Long.MIN_VALUE, null); |
| IPointReader iterator = memChunk.getPointReader(); |
| time = 6; |
| while (iterator.hasNextTimeValuePair()) { |
| TimeValuePair timeValuePair = iterator.nextTimeValuePair(); |
| assertEquals(time, timeValuePair.getTimestamp()); |
| assertEquals(null, timeValuePair.getValue().getVector()[0]); |
| assertEquals((time - 5) * 100L, timeValuePair.getValue().getVector()[1].getLong()); |
| assertEquals(true, timeValuePair.getValue().getVector()[2].getBoolean()); |
| assertEquals((time - 5) * 100, timeValuePair.getValue().getVector()[3].getFloat(), 0.00001); |
| assertEquals(null, timeValuePair.getValue().getVector()[4]); |
| time++; |
| } |
| assertEquals(10, time); |
| } |
| |
| private void generateCompleteFile(File tsFile) throws IOException, WriteProcessException { |
| try (TsFileWriter writer = new TsFileWriter(tsFile)) { |
| writer.registerTimeseries( |
| new Path(DEVICE1_NAME), new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE)); |
| writer.registerTimeseries( |
| new Path(DEVICE1_NAME), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); |
| writer.registerTimeseries( |
| new Path(DEVICE2_NAME), new MeasurementSchema("s1", TSDataType.FLOAT, TSEncoding.RLE)); |
| writer.registerTimeseries( |
| new Path(DEVICE2_NAME), new MeasurementSchema("s2", TSDataType.DOUBLE, TSEncoding.RLE)); |
| writer.registerAlignedTimeseries( |
| new Path(DEVICE3_NAME), |
| Arrays.asList( |
| new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE), |
| new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE), |
| new MeasurementSchema("s3", TSDataType.BOOLEAN, TSEncoding.RLE), |
| new MeasurementSchema("s4", TSDataType.FLOAT, TSEncoding.RLE), |
| new MeasurementSchema("s5", TSDataType.TEXT, TSEncoding.PLAIN))); |
| writer.write( |
| new TSRecord(1, DEVICE1_NAME) |
| .addTuple(new IntDataPoint("s1", 1)) |
| .addTuple(new LongDataPoint("s2", 1))); |
| writer.write( |
| new TSRecord(2, DEVICE1_NAME) |
| .addTuple(new IntDataPoint("s1", 2)) |
| .addTuple(new LongDataPoint("s2", 2))); |
| writer.write( |
| new TSRecord(3, DEVICE2_NAME) |
| .addTuple(new FloatDataPoint("s1", 3)) |
| .addTuple(new DoubleDataPoint("s2", 3))); |
| writer.write( |
| new TSRecord(4, DEVICE2_NAME) |
| .addTuple(new FloatDataPoint("s1", 4)) |
| .addTuple(new DoubleDataPoint("s2", 4))); |
| writer.writeAligned( |
| new TSRecord(5, DEVICE3_NAME) |
| .addTuple(new IntDataPoint("s1", 5)) |
| .addTuple(new LongDataPoint("s2", 5)) |
| .addTuple(new BooleanDataPoint("s3", true)) |
| .addTuple(new FloatDataPoint("s4", 5)) |
| .addTuple(new StringDataPoint("s5", BytesUtils.valueOf("5")))); |
| } |
| } |
| } |
