This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new f34cfb3 [IOTDB-1355] Support updating aligned timeseries values when
insert partially (#3128)
f34cfb3 is described below
commit f34cfb3c55d8b29f4c23ac9827c558cb41630acb
Author: Haonan <[email protected]>
AuthorDate: Thu May 6 20:17:19 2021 +0800
[IOTDB-1355] Support updating aligned timeseries values when insert
partially (#3128)
* [IOTDB-1355] Support updating aligned timeseries values when insert
partially
---
.../iotdb/db/engine/flush/MemTableFlushTask.java | 20 +++++-
.../iotdb/db/utils/datastructure/TVList.java | 31 ++++++++-
.../iotdb/db/utils/datastructure/VectorTVList.java | 57 ++++++++++++++--
.../db/integration/IOTDBInsertAlignedValuesIT.java | 78 ++++++++++++++++++++--
4 files changed, 172 insertions(+), 14 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
index 633dd6e..1d76495 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
@@ -39,6 +39,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
@@ -161,12 +162,21 @@ public class MemTableFlushTask {
new Runnable() {
private void writeOneSeries(
TVList tvPairs, IChunkWriter seriesWriterImpl, TSDataType
dataType) {
+ List<Integer> timeDuplicatedVectorRowIndexList = null;
for (int sortedRowIndex = 0; sortedRowIndex < tvPairs.size();
sortedRowIndex++) {
long time = tvPairs.getTime(sortedRowIndex);
// skip duplicated data
if ((sortedRowIndex + 1 < tvPairs.size()
&& (time == tvPairs.getTime(sortedRowIndex + 1)))) {
+ // record the time duplicated row index list for vector type
+ if (dataType == TSDataType.VECTOR) {
+ if (timeDuplicatedVectorRowIndexList == null) {
+ timeDuplicatedVectorRowIndexList = new ArrayList<>();
+
timeDuplicatedVectorRowIndexList.add(tvPairs.getValueIndex(sortedRowIndex));
+ }
+
timeDuplicatedVectorRowIndexList.add(tvPairs.getValueIndex(sortedRowIndex + 1));
+ }
continue;
}
@@ -199,6 +209,13 @@ public class MemTableFlushTask {
List<TSDataType> dataTypes = vectorTvPairs.getTsDataTypes();
int originRowIndex =
vectorTvPairs.getValueIndex(sortedRowIndex);
for (int columnIndex = 0; columnIndex < dataTypes.size();
columnIndex++) {
+ // write the time duplicated rows
+ if (timeDuplicatedVectorRowIndexList != null
+ && !timeDuplicatedVectorRowIndexList.isEmpty()) {
+ originRowIndex =
+ vectorTvPairs.getValidRowIndexForTimeDuplicatedRows(
+ timeDuplicatedVectorRowIndexList, columnIndex);
+ }
boolean isNull = vectorTvPairs.isValueMarked(originRowIndex,
columnIndex);
switch (dataTypes.get(columnIndex)) {
case BOOLEAN:
@@ -241,11 +258,12 @@ public class MemTableFlushTask {
LOGGER.error(
"Storage group {} does not support data type: {}",
storageGroup,
- dataType);
+ dataTypes.get(columnIndex));
break;
}
}
seriesWriterImpl.write(time);
+ timeDuplicatedVectorRowIndexList = null;
break;
default:
LOGGER.error(
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
index 8ac571f..988995b 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -225,6 +225,10 @@ public abstract class TVList {
throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
}
+ public int getValueIndex(int index) {
+ throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
+ }
+
public abstract void sort();
public long getMinTime() {
@@ -523,6 +527,14 @@ public abstract class TVList {
protected abstract TimeValuePair getTimeValuePair(
int index, long time, Integer floatPrecision, TSEncoding encoding);
+ public TimeValuePair getTimeValuePairForTimeDuplicatedRows(
+ List<Integer> timeDuplicatedVectorRowIndexList,
+ long time,
+ Integer floatPrecision,
+ TSEncoding encoding) {
+ throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
+ }
+
@TestOnly
public IPointReader getIterator() {
return new Ite();
@@ -565,13 +577,30 @@ public abstract class TVList {
return true;
}
+ List<Integer> timeDuplicatedVectorRowIndexList = null;
while (cur < iteSize) {
long time = getTime(cur);
if (isPointDeleted(time) || (cur + 1 < size() && (time == getTime(cur
+ 1)))) {
+ // record the time duplicated row index list for vector type
+ if (getDataType() == TSDataType.VECTOR) {
+ if (timeDuplicatedVectorRowIndexList == null) {
+ timeDuplicatedVectorRowIndexList = new ArrayList<>();
+ timeDuplicatedVectorRowIndexList.add(getValueIndex(cur));
+ }
+ timeDuplicatedVectorRowIndexList.add(getValueIndex(cur + 1));
+ }
cur++;
continue;
}
- TimeValuePair tvPair = getTimeValuePair(cur, time, floatPrecision,
encoding);
+ TimeValuePair tvPair;
+ if (getDataType() == TSDataType.VECTOR &&
timeDuplicatedVectorRowIndexList != null) {
+ tvPair =
+ getTimeValuePairForTimeDuplicatedRows(
+ timeDuplicatedVectorRowIndexList, time, floatPrecision,
encoding);
+ timeDuplicatedVectorRowIndexList = null;
+ } else {
+ tvPair = getTimeValuePair(cur, time, floatPrecision, encoding);
+ }
cur++;
if (tvPair.getValue() != null) {
cachedTimeValuePair = tvPair;
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/VectorTVList.java
b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/VectorTVList.java
index 0a3b405..b7726f0 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/VectorTVList.java
+++
b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/VectorTVList.java
@@ -69,8 +69,6 @@ public class VectorTVList extends TVList {
List<Object> columnValues = values.get(i);
if (columnValue == null) {
markNullValue(i, arrayIndex, elementIndex);
- } else if (isValueMarked(indices.get(arrayIndex)[elementIndex], i)) {
- bitMaps.get(i).get(arrayIndex).unmark(elementIndex);
}
switch (dataTypes.get(i)) {
case TEXT:
@@ -116,10 +114,21 @@ public class VectorTVList extends TVList {
int arrayIndex = index / ARRAY_SIZE;
int elementIndex = index % ARRAY_SIZE;
int valueIndex = indices.get(arrayIndex)[elementIndex];
- return getVectorByValueIndex(valueIndex);
+ return getVectorByValueIndex(valueIndex, null);
}
- private Object getVectorByValueIndex(int valueIndex) {
+ public Object getVector(List<Integer> timeDuplicatedIndexList) {
+ int[] validIndexesForTimeDuplicatedRows = new int[values.size()];
+ for (int i = 0; i < values.size(); i++) {
+ validIndexesForTimeDuplicatedRows[i] =
+ getValidRowIndexForTimeDuplicatedRows(timeDuplicatedIndexList, i);
+ }
+ return getVectorByValueIndex(
+ timeDuplicatedIndexList.get(timeDuplicatedIndexList.size() - 1),
+ validIndexesForTimeDuplicatedRows);
+ }
+
+ private Object getVectorByValueIndex(int valueIndex, int[]
validIndexesForTimeDuplicatedRows) {
if (valueIndex >= size) {
throw new ArrayIndexOutOfBoundsException(valueIndex);
}
@@ -128,6 +137,10 @@ public class VectorTVList extends TVList {
TsPrimitiveType[] vector = new TsPrimitiveType[values.size()];
for (int i = 0; i < values.size(); i++) {
List<Object> columnValues = values.get(i);
+ if (validIndexesForTimeDuplicatedRows != null) {
+ arrayIndex = validIndexesForTimeDuplicatedRows[i] / ARRAY_SIZE;
+ elementIndex = validIndexesForTimeDuplicatedRows[i] % ARRAY_SIZE;
+ }
if (bitMaps != null
&& bitMaps.get(i) != null
&& bitMaps.get(i).get(arrayIndex).isMarked(elementIndex)) {
@@ -498,6 +511,7 @@ public class VectorTVList extends TVList {
}
/* Get the row index value in index column. */
+ @Override
public int getValueIndex(int index) {
if (index >= size) {
throw new ArrayIndexOutOfBoundsException(index);
@@ -507,6 +521,27 @@ public class VectorTVList extends TVList {
return indices.get(arrayIndex)[elementIndex];
}
+ /**
+ * Get the valid original row index in a column by a given time duplicated
original row index
+ * list.
+ *
+ * @param timeDuplicatedOriginRowIndexList The row index list that the time
of all indexes are
+ * same.
+ * @param columnIndex The index of a given column.
+ * @return The original row index of the latest non-null value, or the first
row index if all
+ * values in given columns are null.
+ */
+ public int getValidRowIndexForTimeDuplicatedRows(
+ List<Integer> timeDuplicatedOriginRowIndexList, int columnIndex) {
+ int validRowIndex = timeDuplicatedOriginRowIndexList.get(0);
+ for (int originRowIndex : timeDuplicatedOriginRowIndexList) {
+ if (!isValueMarked(originRowIndex, columnIndex)) {
+ validRowIndex = originRowIndex;
+ }
+ }
+ return validRowIndex;
+ }
+
@Override
protected void setPivotTo(int pos) {
set(pos, pivotTime, pivotIndex);
@@ -525,9 +560,19 @@ public class VectorTVList extends TVList {
protected TimeValuePair getTimeValuePair(
int index, long time, Integer floatPrecision, TSEncoding encoding) {
if (this.dataTypes.size() == 1) {
- return new TimeValuePair(getTime(index), ((TsPrimitiveType)
getVector(index)).getVector()[0]);
+ return new TimeValuePair(time, ((TsPrimitiveType)
getVector(index)).getVector()[0]);
} else {
- return new TimeValuePair(getTime(index), (TsPrimitiveType)
getVector(index));
+ return new TimeValuePair(time, (TsPrimitiveType) getVector(index));
+ }
+ }
+
+ @Override
+ public TimeValuePair getTimeValuePairForTimeDuplicatedRows(
+ List<Integer> indexList, long time, Integer floatPrecision, TSEncoding
encoding) {
+ if (this.dataTypes.size() == 1) {
+ return new TimeValuePair(time, ((TsPrimitiveType)
getVector(indexList)).getVector()[0]);
+ } else {
+ return new TimeValuePair(time, (TsPrimitiveType) getVector(indexList));
}
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/IOTDBInsertAlignedValuesIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/IOTDBInsertAlignedValuesIT.java
index fadfd7f..4c15d83 100644
---
a/server/src/test/java/org/apache/iotdb/db/integration/IOTDBInsertAlignedValuesIT.java
+++
b/server/src/test/java/org/apache/iotdb/db/integration/IOTDBInsertAlignedValuesIT.java
@@ -23,9 +23,9 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.jdbc.Config;
-import org.junit.AfterClass;
+import org.junit.After;
import org.junit.Assert;
-import org.junit.BeforeClass;
+import org.junit.Before;
import org.junit.Test;
import java.sql.Connection;
@@ -38,8 +38,8 @@ import java.util.Objects;
public class IOTDBInsertAlignedValuesIT {
private static Connection connection;
- @BeforeClass
- public static void setUp() throws Exception {
+ @Before
+ public void setUp() throws Exception {
EnvironmentUtils.closeStatMonitor();
EnvironmentUtils.envSetUp();
IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(true);
@@ -48,8 +48,8 @@ public class IOTDBInsertAlignedValuesIT {
DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root");
}
- @AfterClass
- public static void tearDown() throws Exception {
+ @After
+ public void tearDown() throws Exception {
close();
EnvironmentUtils.cleanEnv();
}
@@ -134,6 +134,72 @@ public class IOTDBInsertAlignedValuesIT {
st1.close();
}
+ @Test
+ public void testUpdatingAlignedValues() throws SQLException {
+ Statement st0 = connection.createStatement();
+ st0.execute(
+ "insert into root.t1.wf01.wt01(time, (status, temperature)) values
(4000, (true, 17.1))");
+ st0.execute(
+ "insert into root.t1.wf01.wt01(time, (status, temperature)) values
(5000, (true, null))");
+ st0.execute(
+ "insert into root.t1.wf01.wt01(time, (status, temperature)) values
(5000, (NULL, 20.1))");
+ st0.execute(
+ "insert into root.t1.wf01.wt01(time, (status, temperature)) values
(6000, (null, 22))");
+ st0.close();
+
+ Statement st1 = connection.createStatement();
+
+ ResultSet rs1 = st1.executeQuery("select status from root.t1.wf01.wt01");
+ rs1.next();
+ Assert.assertEquals(true, rs1.getBoolean(2));
+ rs1.next();
+ Assert.assertEquals(true, rs1.getBoolean(2));
+ rs1.next();
+ Assert.assertEquals(null, rs1.getObject(2));
+
+ ResultSet rs2 = st1.executeQuery("select * from root.t1.wf01.wt01");
+ rs2.next();
+ Assert.assertEquals(4000, rs2.getLong(1));
+ Assert.assertEquals(true, rs2.getBoolean(2));
+ Assert.assertEquals(17.1, rs2.getFloat(3), 0.1);
+
+ rs2.next();
+ Assert.assertEquals(5000, rs2.getLong(1));
+ Assert.assertEquals(true, rs2.getObject(2));
+ Assert.assertEquals(20.1, rs2.getFloat(3), 0.1);
+
+ rs2.next();
+ Assert.assertEquals(6000, rs2.getLong(1));
+ Assert.assertEquals(null, rs2.getObject(2));
+ Assert.assertEquals(22.0f, rs2.getObject(3));
+
+ st1.execute("flush");
+ ResultSet rs3 = st1.executeQuery("select status from root.t1.wf01.wt01");
+ rs3.next();
+ Assert.assertEquals(true, rs3.getBoolean(2));
+ rs3.next();
+ Assert.assertEquals(true, rs3.getBoolean(2));
+ rs3.next();
+ Assert.assertEquals(null, rs3.getObject(2));
+
+ ResultSet rs4 = st1.executeQuery("select * from root.t1.wf01.wt01");
+ rs4.next();
+ Assert.assertEquals(4000, rs4.getLong(1));
+ Assert.assertEquals(true, rs4.getBoolean(2));
+ Assert.assertEquals(17.1, rs4.getFloat(3), 0.1);
+
+ rs4.next();
+ Assert.assertEquals(5000, rs4.getLong(1));
+ Assert.assertEquals(true, rs4.getObject(2));
+ Assert.assertEquals(20.1, rs4.getFloat(3), 0.1);
+
+ rs4.next();
+ Assert.assertEquals(6000, rs4.getLong(1));
+ Assert.assertEquals(null, rs4.getObject(2));
+ Assert.assertEquals(22.0f, rs4.getObject(3));
+ st1.close();
+ }
+
@Test(expected = Exception.class)
public void testInsertWithWrongMeasurementNum1() throws SQLException {
Statement st1 = connection.createStatement();
