You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2023/10/11 01:24:49 UTC
[iotdb] branch master updated: Fix NPE when partially inserting tablet span multiple data partitions
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 c0260ea7b5c Fix NPE when partially inserting tablet span multiple data partitions
c0260ea7b5c is described below
commit c0260ea7b5c71e3b5f907101164979e166892085
Author: Chen YZ <43...@users.noreply.github.com>
AuthorDate: Wed Oct 11 09:24:43 2023 +0800
Fix NPE when partially inserting tablet span multiple data partitions
---
.../iotdb/db/it/IoTDBPartialInsertionIT.java | 71 ++++++++++++++++++++++
.../planner/plan/node/write/InsertTabletNode.java | 45 ++++++++------
2 files changed, 96 insertions(+), 20 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBPartialInsertionIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBPartialInsertionIT.java
index 0b284573a88..18bee17a734 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBPartialInsertionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBPartialInsertionIT.java
@@ -18,12 +18,21 @@
*/
package org.apache.iotdb.db.it;
+import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+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.common.RowRecord;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -35,6 +44,8 @@ import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -127,4 +138,64 @@ public class IoTDBPartialInsertionIT {
}
}
}
+
+ @Test
+ public void testPartialInsertTablet() {
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.createTimeseries(
+ "root.sg1.d1.s1", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.SNAPPY);
+ session.createTimeseries(
+ "root.sg1.d1.s2", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.SNAPPY);
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64));
+ Tablet tablet = new Tablet("root.sg1.d1", schemaList, 300);
+ long timestamp = 0;
+ for (long row = 0; row < 100; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ for (int s = 0; s < 3; s++) {
+ long value = timestamp;
+ tablet.addValue(schemaList.get(s).getMeasurementId(), rowIndex, value);
+ }
+ timestamp++;
+ }
+ timestamp = System.currentTimeMillis();
+ for (long row = 0; row < 100; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ for (int s = 0; s < 3; s++) {
+ long value = timestamp;
+ tablet.addValue(schemaList.get(s).getMeasurementId(), rowIndex, value);
+ }
+ timestamp++;
+ }
+ try {
+ session.insertTablet(tablet);
+ } catch (Exception e) {
+ if (!e.getMessage().contains("507")) {
+ fail(e.getMessage());
+ }
+ }
+ try (SessionDataSet dataSet = session.executeQueryStatement("SELECT * FROM root.sg1.d1")) {
+ assertEquals(dataSet.getColumnNames().size(), 3);
+ assertEquals(dataSet.getColumnNames().get(0), "Time");
+ assertEquals(dataSet.getColumnNames().get(1), "root.sg1.d1.s1");
+ assertEquals(dataSet.getColumnNames().get(2), "root.sg1.d1.s2");
+ int cnt = 0;
+ while (dataSet.hasNext()) {
+ RowRecord rowRecord = dataSet.next();
+ long time = rowRecord.getTimestamp();
+ assertEquals(time, rowRecord.getFields().get(0).getLongV());
+ assertEquals(time, rowRecord.getFields().get(1).getLongV());
+ cnt++;
+ }
+ Assert.assertEquals(200, cnt);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java
index 7bfae21daf1..4dbacc65971 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java
@@ -258,7 +258,9 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
BitMap[] bitMaps = this.bitMaps == null ? null : initBitmaps(dataTypes.length, count);
System.arraycopy(times, start, subTimes, destLoc, end - start);
for (int k = 0; k < values.length; k++) {
- System.arraycopy(columns[k], start, values[k], destLoc, end - start);
+ if (dataTypes[k] != null) {
+ System.arraycopy(columns[k], start, values[k], destLoc, end - start);
+ }
if (bitMaps != null && this.bitMaps[k] != null) {
BitMap.copyOfRange(this.bitMaps[k], start, bitMaps[k], destLoc, end - start);
}
@@ -275,6 +277,7 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
bitMaps,
values,
subTimes.length);
+ subNode.setFailedMeasurementNumber(getFailedMeasurementNumber());
subNode.setRange(locs);
subNode.setDataRegionReplicaSet(entry.getKey());
result.add(subNode);
@@ -303,25 +306,27 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
private Object[] initTabletValues(int columnSize, int rowSize, TSDataType[] dataTypes) {
Object[] values = new Object[columnSize];
for (int i = 0; i < values.length; i++) {
- switch (dataTypes[i]) {
- case TEXT:
- values[i] = new Binary[rowSize];
- break;
- case FLOAT:
- values[i] = new float[rowSize];
- break;
- case INT32:
- values[i] = new int[rowSize];
- break;
- case INT64:
- values[i] = new long[rowSize];
- break;
- case DOUBLE:
- values[i] = new double[rowSize];
- break;
- case BOOLEAN:
- values[i] = new boolean[rowSize];
- break;
+ if (dataTypes[i] != null) {
+ switch (dataTypes[i]) {
+ case TEXT:
+ values[i] = new Binary[rowSize];
+ break;
+ case FLOAT:
+ values[i] = new float[rowSize];
+ break;
+ case INT32:
+ values[i] = new int[rowSize];
+ break;
+ case INT64:
+ values[i] = new long[rowSize];
+ break;
+ case DOUBLE:
+ values[i] = new double[rowSize];
+ break;
+ case BOOLEAN:
+ values[i] = new boolean[rowSize];
+ break;
+ }
}
}
return values;