You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2024/10/15 10:49:48 UTC
(iotdb) branch fix_insert_table_overlap created (now 0f1a206faf6)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a change to branch fix_insert_table_overlap
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 0f1a206faf6 Fix insertRelationalTablet may cause sequence data overlapped
This branch includes the following new commits:
new 0f1a206faf6 Fix insertRelationalTablet may cause sequence data overlapped
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Fix insertRelationalTablet may cause sequence data overlapped
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch fix_insert_table_overlap
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0f1a206faf6f705565dabb9df41646655e9f372a
Author: HTHou <hh...@outlook.com>
AuthorDate: Tue Oct 15 18:49:30 2024 +0800
Fix insertRelationalTablet may cause sequence data overlapped
---
.../relational/it/db/it/IoTDBInsertTableIT.java | 83 ++++++++++++++++++++++
.../db/storageengine/dataregion/DataRegion.java | 15 ++--
2 files changed, 94 insertions(+), 4 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java
index 6264fc26651..58883ecbdb8 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java
@@ -958,6 +958,89 @@ public class IoTDBInsertTableIT {
}
}
+ @Test
+ public void testInsertUnsequenceData()
+ throws IoTDBConnectionException, StatementExecutionException {
+ try (ISession session = EnvFactory.getEnv().getSessionConnection(BaseEnv.TABLE_SQL_DIALECT)) {
+ session.executeNonQueryStatement("USE \"test\"");
+ // the table is missing column "m2"
+ session.executeNonQueryStatement(
+ "CREATE TABLE table4 (id1 string id, attr1 string attribute, "
+ + "m1 double "
+ + "measurement)");
+
+ // the insertion contains "m2"
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("id1", TSDataType.STRING));
+ schemaList.add(new MeasurementSchema("attr1", TSDataType.STRING));
+ schemaList.add(new MeasurementSchema("m1", TSDataType.DOUBLE));
+ schemaList.add(new MeasurementSchema("m2", TSDataType.DOUBLE));
+ final List<Tablet.ColumnType> columnTypes =
+ Arrays.asList(
+ Tablet.ColumnType.ID,
+ Tablet.ColumnType.ATTRIBUTE,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT);
+
+ long timestamp = 0;
+ Tablet tablet = new Tablet("table4", schemaList, columnTypes, 15);
+
+ for (long row = 0; row < 15; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp + row);
+ tablet.addValue("id1", rowIndex, "id:" + row);
+ tablet.addValue("attr1", rowIndex, "attr:" + row);
+ tablet.addValue("m1", rowIndex, row * 1.0);
+ tablet.addValue("m2", rowIndex, row * 1.0);
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ try {
+ session.insertRelationalTablet(tablet, true);
+ } catch (StatementExecutionException e) {
+ // a partial insertion should be reported
+ if (!e.getMessage()
+ .equals(
+ "507: Fail to insert measurements [m2] caused by [Column m2 does not exists or fails to be created]")) {
+ throw e;
+ }
+ }
+ tablet.reset();
+ }
+ }
+
+ session.executeNonQueryStatement("FLush");
+
+ for (long row = 0; row < 15; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, 14 - row);
+ tablet.addValue("id1", rowIndex, "id:" + row);
+ tablet.addValue("attr1", rowIndex, "attr:" + row);
+ tablet.addValue("m1", rowIndex, row * 1.0);
+ tablet.addValue("m2", rowIndex, row * 1.0);
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ try {
+ session.insertRelationalTablet(tablet, true);
+ } catch (StatementExecutionException e) {
+ if (!e.getMessage()
+ .equals(
+ "507: Fail to insert measurements [m2] caused by [Column m2 does not exists or fails to be created]")) {
+ throw e;
+ }
+ }
+ tablet.reset();
+ }
+ }
+ session.executeNonQueryStatement("FLush");
+
+ int cnt = 0;
+ SessionDataSet dataSet = session.executeQueryStatement("select * from table4");
+ while (dataSet.hasNext()) {
+ dataSet.next();
+ cnt++;
+ }
+ assertEquals(29, cnt);
+ }
+ }
+
private List<Integer> checkHeader(
ResultSetMetaData resultSetMetaData, String expectedHeaderStrings, int[] expectedTypes)
throws SQLException {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index a96f7797b96..b5b1be00dc0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -1073,8 +1073,8 @@ public class DataRegion implements IDataRegionForQuery {
: Long.MAX_VALUE;
}
- private boolean splitAndInsert(InsertTabletNode insertTabletNode, int loc, TSStatus[] results)
- throws BatchProcessException, WriteProcessException {
+ private boolean splitAndInsert(
+ InsertTabletNode insertTabletNode, int loc, int endOffset, TSStatus[] results) {
boolean noFailure = true;
// before is first start point
@@ -1088,7 +1088,7 @@ public class DataRegion implements IDataRegionForQuery {
int insertCnt = 0;
// if is sequence
boolean isSequence = false;
- while (loc < insertTabletNode.getRowCount()) {
+ while (loc < endOffset) {
long time = insertTabletNode.getTimes()[loc];
final long timePartitionId = TimePartitionUtils.getTimePartitionId(time);
@@ -1199,7 +1199,14 @@ public class DataRegion implements IDataRegionForQuery {
boolean noFailure;
int loc = insertTabletNode.checkTTL(results, i -> getTTL(insertTabletNode));
noFailure = loc == 0;
- noFailure = noFailure && splitAndInsert(insertTabletNode, loc, results);
+ List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
+ insertTabletNode.splitByDevice(loc, insertTabletNode.getRowCount());
+ int start = loc;
+ for (Pair<IDeviceID, Integer> deviceEndOffsetPair : deviceEndOffsetPairs) {
+ int end = deviceEndOffsetPair.getRight();
+ noFailure = noFailure && splitAndInsert(insertTabletNode, start, end, results);
+ start = end;
+ }
if (CommonDescriptor.getInstance().getConfig().isLastCacheEnable()) {
if (!insertTabletNode.isGeneratedByRemoteConsensusLeader()) {