This is an automated email from the ASF dual-hosted git repository.
xuekaifeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 2e7d902 [IOTDB-863]: add a switch to drop ouf-of-order data that is
recognized by IoTDB (#1693)
2e7d902 is described below
commit 2e7d902176229f77c8c56e3d0bbd57a83835484b
Author: Haimei Guo <[email protected]>
AuthorDate: Sun Sep 27 20:07:20 2020 +0800
[IOTDB-863]: add a switch to drop ouf-of-order data that is recognized by
IoTDB (#1693)
* server/src/assembly/resources/conf/logback.xml
* fix loginfo appender ref to FILEINFO
* initial code
* edit name
* changed code position
* edit test case
* add test cases for different time paritition
Co-authored-by: root <[email protected]>
---
.../resources/conf/iotdb-engine.properties | 4 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 10 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 4 +
.../engine/storagegroup/StorageGroupProcessor.java | 35 ++-
.../storagegroup/StorageGroupProcessorTest.java | 284 ++++++++++++++++++++-
5 files changed, 325 insertions(+), 12 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index 5f1b0b7..07bf45d 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -46,6 +46,10 @@ rpc_max_concurrent_client_num=65535
# Is insert ahead log enable
enable_wal=true
+# Add a switch to drop ouf-of-order data
+# Out-of-order data will impact the aggregation query a lot. Users may not
care about discarding some out-of-order data.
+enable_discard_out_of_order_data=false
+
# When a certain amount of insert ahead log is reached, it will be flushed to
disk
# It is possible to lose at most flush_wal_threshold operations
flush_wal_threshold=10000
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 8ded20b..ab4d3cb 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -145,6 +145,8 @@ public class IoTDBConfig {
private volatile boolean readOnly = false;
+ private boolean enableDiscardOutOfOrderData = false;
+
/**
* When a certain amount of write ahead logs is reached, they will be
flushed to the disk. It is
* possible to lose at most flush_wal_threshold operations.
@@ -858,6 +860,14 @@ public class IoTDBConfig {
this.enableWal = enableWal;
}
+ public boolean isEnableDiscardOutOfOrderData() {
+ return enableDiscardOutOfOrderData ;
+ }
+
+ public void setEnableDiscardOutOfOrderData(boolean
enableDiscardOutOfOrderData ) {
+ this.enableDiscardOutOfOrderData = enableDiscardOutOfOrderData ;
+ }
+
public int getFlushWalThreshold() {
return flushWalThreshold;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index bdaa420..747916a 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -540,6 +540,10 @@ public class IoTDBDescriptor {
.parseLong(properties.getProperty("force_wal_period_in_ms",
Long.toString(conf.getForceWalPeriodInMs()))));
+ conf.setEnableDiscardOutOfOrderData(Boolean.parseBoolean(
+ properties.getProperty("enable_discard_out_of_order_data",
+ Boolean.toString(conf.isEnableDiscardOutOfOrderData()))));
+
}
private void loadAutoCreateSchemaProps(Properties properties) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index ac9005d..347fd25 100755
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -626,14 +626,20 @@ public class StorageGroupProcessor {
// init map
long timePartitionId =
StorageEngine.getTimePartition(insertRowPlan.getTime());
- latestTimeForEachDevice.computeIfAbsent(timePartitionId, l -> new
HashMap<>());
partitionLatestFlushedTimeForEachDevice
.computeIfAbsent(timePartitionId, id -> new HashMap<>());
+ boolean isSequence = insertRowPlan.getTime() >
partitionLatestFlushedTimeForEachDevice.get(timePartitionId)
+ .getOrDefault(insertRowPlan.getDeviceId().getFullPath(),
Long.MIN_VALUE);
+
+ //is unsequence and user set config to discard out of order data
+ if (!isSequence &&
IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData()) {
+ return;
+ }
+
+ latestTimeForEachDevice.computeIfAbsent(timePartitionId, l -> new
HashMap<>());
// insert to sequence or unSequence file
- insertToTsFileProcessor(insertRowPlan,
- insertRowPlan.getTime() >
partitionLatestFlushedTimeForEachDevice.get(timePartitionId)
- .getOrDefault(insertRowPlan.getDeviceId().getFullPath(),
Long.MIN_VALUE));
+ insertToTsFileProcessor(insertRowPlan, isSequence);
} finally {
writeUnlock();
@@ -690,9 +696,11 @@ public class StorageGroupProcessor {
// start next partition
if (curTimePartition != beforeTimePartition) {
// insert last time partition
- noFailure = insertTabletToTsFileProcessor(insertTabletPlan, before,
loc, isSequence,
- results,
- beforeTimePartition) && noFailure;
+ if (isSequence ||
!IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData()) {
+ noFailure = insertTabletToTsFileProcessor(insertTabletPlan,
before, loc, isSequence,
+ results,
+ beforeTimePartition) && noFailure;
+ }
// re initialize
before = loc;
beforeTimePartition = curTimePartition;
@@ -706,8 +714,11 @@ public class StorageGroupProcessor {
// judge if we should insert sequence
if (!isSequence && time > lastFlushTime) {
// insert into unsequence and then start sequence
- noFailure = insertTabletToTsFileProcessor(insertTabletPlan,
before, loc, false, results,
- beforeTimePartition) && noFailure;
+ if
(!IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData()) {
+ noFailure =
+ insertTabletToTsFileProcessor(insertTabletPlan, before, loc,
false, results,
+ beforeTimePartition) && noFailure;
+ }
before = loc;
isSequence = true;
}
@@ -717,8 +728,10 @@ public class StorageGroupProcessor {
// do not forget last part
if (before < loc) {
- noFailure = insertTabletToTsFileProcessor(insertTabletPlan, before,
loc, isSequence,
- results, beforeTimePartition) && noFailure;
+ if (isSequence ||
!IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData()) {
+ noFailure = insertTabletToTsFileProcessor(insertTabletPlan, before,
loc, isSequence,
+ results, beforeTimePartition) && noFailure;
+ }
}
long globalLatestFlushedTime =
globalLatestFlushedTimeForEachDevice.getOrDefault(
insertTabletPlan.getDeviceId().getFullPath(), Long.MIN_VALUE);
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
index b1452cd..cfa7c64 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
@@ -26,6 +26,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.adapter.ActiveTimeSeriesCounter;
import org.apache.iotdb.db.constant.TestConstant;
@@ -40,7 +41,6 @@ import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
-import org.apache.iotdb.db.metadata.mnode.MNode;
import org.apache.iotdb.db.metadata.mnode.MeasurementMNode;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
@@ -268,6 +268,288 @@ public class StorageGroupProcessorTest {
}
@Test
+ public void testEnableDiscardOutOfOrderDataForInsertRowPlan()
+ throws WriteProcessException, QueryProcessException,
IllegalPathException, IOException {
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ boolean defaultValue = config.isEnableDiscardOutOfOrderData();
+ config.setEnableDiscardOutOfOrderData(true);
+
+ for (int j = 21; j <= 30; j++) {
+ TSRecord record = new TSRecord(j, deviceId);
+ record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId,
String.valueOf(j)));
+ insertToStorageGroupProcessor(record);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+ }
+ processor.syncCloseAllWorkingTsFileProcessors();
+
+ for (int j = 10; j >= 1; j--) {
+ TSRecord record = new TSRecord(j, deviceId);
+ record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId,
String.valueOf(j)));
+ insertToStorageGroupProcessor(record);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+ }
+
+ processor.syncCloseAllWorkingTsFileProcessors();
+
+ for (TsFileProcessor tsfileProcessor :
processor.getWorkUnsequenceTsFileProcessor()) {
+ tsfileProcessor.syncFlush();
+ }
+
+ QueryDataSource queryDataSource = processor.query(new
PartialPath(deviceId), measurementId, context,
+ null, null);
+ Assert.assertEquals(10, queryDataSource.getSeqResources().size());
+ Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
+ for (TsFileResource resource : queryDataSource.getSeqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+ for (TsFileResource resource : queryDataSource.getUnseqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+
+ config.setEnableDiscardOutOfOrderData(defaultValue);
+ }
+
+ @Test
+ public void testEnableDiscardOutOfOrderDataForInsertTablet1()
+ throws QueryProcessException, IllegalPathException, IOException {
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ boolean defaultEnableDiscard = config.isEnableDiscardOutOfOrderData();
+ long defaultTimePartition = config.getPartitionInterval();
+ boolean defaultEnablePartition = config.isEnablePartition();
+ config.setEnableDiscardOutOfOrderData(true);
+ config.setEnablePartition(true);
+ config.setPartitionInterval(100);
+
+ String[] measurements = new String[2];
+ measurements[0] = "s0";
+ measurements[1] = "s1";
+ List<Integer> dataTypes = new ArrayList<>();
+ dataTypes.add(TSDataType.INT32.ordinal());
+ dataTypes.add(TSDataType.INT64.ordinal());
+
+ MeasurementMNode[] measurementMNodes = new MeasurementMNode[2];
+ measurementMNodes[0] = new MeasurementMNode(null, "s0",
+ new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.PLAIN), null);
+ measurementMNodes[1] = new MeasurementMNode(null, "s1",
+ new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), null);
+
+ InsertTabletPlan insertTabletPlan1 = new InsertTabletPlan(new
PartialPath("root.vehicle.d0"), measurements,
+ dataTypes);
+
+ 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;
+ }
+ insertTabletPlan1.setTimes(times);
+ insertTabletPlan1.setColumns(columns);
+ insertTabletPlan1.setRowCount(times.length);
+ insertTabletPlan1.setMeasurementMNodes(measurementMNodes);
+
+ processor.insertTablet(insertTabletPlan1);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+
+ InsertTabletPlan insertTabletPlan2 = new InsertTabletPlan(new
PartialPath("root.vehicle.d0"), measurements,
+ dataTypes);
+
+ for (int r = 149; r >= 50; r--) {
+ times[r - 50] = r;
+ ((int[]) columns[0])[r - 50] = 1;
+ ((long[]) columns[1])[r - 50] = 1;
+ }
+ insertTabletPlan2.setTimes(times);
+ insertTabletPlan2.setColumns(columns);
+ insertTabletPlan2.setRowCount(times.length);
+ insertTabletPlan2.setMeasurementMNodes(measurementMNodes);
+
+ processor.insertTablet(insertTabletPlan2);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+ processor.syncCloseAllWorkingTsFileProcessors();
+
+ for (TsFileProcessor tsfileProcessor :
processor.getWorkUnsequenceTsFileProcessor()) {
+ tsfileProcessor.syncFlush();
+ }
+
+ QueryDataSource queryDataSource = processor.query(new
PartialPath(deviceId), measurementId, context,
+ null, null);
+
+ Assert.assertEquals(2, queryDataSource.getSeqResources().size());
+ Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
+ for (TsFileResource resource : queryDataSource.getSeqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+
+ config.setEnableDiscardOutOfOrderData(defaultEnableDiscard);
+ config.setPartitionInterval(defaultTimePartition);
+ config.setEnablePartition(defaultEnablePartition);
+ }
+
+ @Test
+ public void testEnableDiscardOutOfOrderDataForInsertTablet2()
+ throws QueryProcessException, IllegalPathException, IOException {
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ boolean defaultEnableDiscard = config.isEnableDiscardOutOfOrderData();
+ long defaultTimePartition = config.getPartitionInterval();
+ boolean defaultEnablePartition = config.isEnablePartition();
+ config.setEnableDiscardOutOfOrderData(true);
+ config.setEnablePartition(true);
+ config.setPartitionInterval(1200);
+
+ String[] measurements = new String[2];
+ measurements[0] = "s0";
+ measurements[1] = "s1";
+ List<Integer> dataTypes = new ArrayList<>();
+ dataTypes.add(TSDataType.INT32.ordinal());
+ dataTypes.add(TSDataType.INT64.ordinal());
+
+ MeasurementMNode[] measurementMNodes = new MeasurementMNode[2];
+ measurementMNodes[0] = new MeasurementMNode(null, "s0",
+ new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.PLAIN), null);
+ measurementMNodes[1] = new MeasurementMNode(null, "s1",
+ new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), null);
+
+ InsertTabletPlan insertTabletPlan1 = new InsertTabletPlan(new
PartialPath("root.vehicle.d0"), measurements,
+ dataTypes);
+
+ long[] times = new long[1200];
+ Object[] columns = new Object[2];
+ columns[0] = new int[1200];
+ columns[1] = new long[1200];
+
+ for (int r = 0; r < 1200; r++) {
+ times[r] = r;
+ ((int[]) columns[0])[r] = 1;
+ ((long[]) columns[1])[r] = 1;
+ }
+ insertTabletPlan1.setTimes(times);
+ insertTabletPlan1.setColumns(columns);
+ insertTabletPlan1.setRowCount(times.length);
+ insertTabletPlan1.setMeasurementMNodes(measurementMNodes);
+
+ processor.insertTablet(insertTabletPlan1);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+
+ InsertTabletPlan insertTabletPlan2 = new InsertTabletPlan(new
PartialPath("root.vehicle.d0"), measurements,
+ dataTypes);
+
+ for (int r = 1249; r >= 50; r--) {
+ times[r - 50] = r;
+ ((int[]) columns[0])[r - 50] = 1;
+ ((long[]) columns[1])[r - 50] = 1;
+ }
+ insertTabletPlan2.setTimes(times);
+ insertTabletPlan2.setColumns(columns);
+ insertTabletPlan2.setRowCount(times.length);
+ insertTabletPlan2.setMeasurementMNodes(measurementMNodes);
+
+ processor.insertTablet(insertTabletPlan2);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+ processor.syncCloseAllWorkingTsFileProcessors();
+
+ for (TsFileProcessor tsfileProcessor :
processor.getWorkUnsequenceTsFileProcessor()) {
+ tsfileProcessor.syncFlush();
+ }
+
+ QueryDataSource queryDataSource = processor.query(new
PartialPath(deviceId), measurementId, context,
+ null, null);
+
+ Assert.assertEquals(2, queryDataSource.getSeqResources().size());
+ Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
+ for (TsFileResource resource : queryDataSource.getSeqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+
+ config.setEnableDiscardOutOfOrderData(defaultEnableDiscard);
+ config.setPartitionInterval(defaultTimePartition);
+ config.setEnablePartition(defaultEnablePartition);
+ }
+
+ @Test
+ public void testEnableDiscardOutOfOrderDataForInsertTablet3()
+ throws QueryProcessException, IllegalPathException, IOException {
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ boolean defaultEnableDiscard = config.isEnableDiscardOutOfOrderData();
+ long defaultTimePartition = config.getPartitionInterval();
+ boolean defaultEnablePartition = config.isEnablePartition();
+ config.setEnableDiscardOutOfOrderData(true);
+ config.setEnablePartition(true);
+ config.setPartitionInterval(1000);
+
+ String[] measurements = new String[2];
+ measurements[0] = "s0";
+ measurements[1] = "s1";
+ List<Integer> dataTypes = new ArrayList<>();
+ dataTypes.add(TSDataType.INT32.ordinal());
+ dataTypes.add(TSDataType.INT64.ordinal());
+
+ MeasurementMNode[] measurementMNodes = new MeasurementMNode[2];
+ measurementMNodes[0] = new MeasurementMNode(null, "s0",
+ new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.PLAIN), null);
+ measurementMNodes[1] = new MeasurementMNode(null, "s1",
+ new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), null);
+
+ InsertTabletPlan insertTabletPlan1 = new InsertTabletPlan(new
PartialPath("root.vehicle.d0"), measurements,
+ dataTypes);
+
+ long[] times = new long[1200];
+ Object[] columns = new Object[2];
+ columns[0] = new int[1200];
+ columns[1] = new long[1200];
+
+ for (int r = 0; r < 1200; r++) {
+ times[r] = r;
+ ((int[]) columns[0])[r] = 1;
+ ((long[]) columns[1])[r] = 1;
+ }
+ insertTabletPlan1.setTimes(times);
+ insertTabletPlan1.setColumns(columns);
+ insertTabletPlan1.setRowCount(times.length);
+ insertTabletPlan1.setMeasurementMNodes(measurementMNodes);
+
+ processor.insertTablet(insertTabletPlan1);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+
+ InsertTabletPlan insertTabletPlan2 = new InsertTabletPlan(new
PartialPath("root.vehicle.d0"), measurements,
+ dataTypes);
+
+ for (int r = 1249; r >= 50; r--) {
+ times[r - 50] = r;
+ ((int[]) columns[0])[r - 50] = 1;
+ ((long[]) columns[1])[r - 50] = 1;
+ }
+ insertTabletPlan2.setTimes(times);
+ insertTabletPlan2.setColumns(columns);
+ insertTabletPlan2.setRowCount(times.length);
+ insertTabletPlan2.setMeasurementMNodes(measurementMNodes);
+
+ processor.insertTablet(insertTabletPlan2);
+ processor.asyncCloseAllWorkingTsFileProcessors();
+ processor.syncCloseAllWorkingTsFileProcessors();
+
+ for (TsFileProcessor tsfileProcessor :
processor.getWorkUnsequenceTsFileProcessor()) {
+ tsfileProcessor.syncFlush();
+ }
+
+ QueryDataSource queryDataSource = processor.query(new
PartialPath(deviceId), measurementId, context,
+ null, null);
+
+ Assert.assertEquals(2, queryDataSource.getSeqResources().size());
+ Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
+ for (TsFileResource resource : queryDataSource.getSeqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+
+ config.setEnableDiscardOutOfOrderData(defaultEnableDiscard);
+ config.setPartitionInterval(defaultTimePartition);
+ config.setEnablePartition(defaultEnablePartition);
+ }
+
+ @Test
public void testMerge() throws WriteProcessException, QueryProcessException,
IllegalPathException {
mergeLock = new AtomicLong(0);
