Clone value array when insert non aligned tablet with null (#14643)

* Clone value array when insert non aligned tablet with null * Fix UT * Add recover wal IT
apache · Jan 10, 2025 · 5bd1ead · 5bd1ead
1 parent 937df76
commit 5bd1ead
Show file tree

Hide file tree

Showing 12 changed files with 121 additions and 11 deletions.
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
@@ -1127,6 +1127,11 @@ public void shutdownAllDataNodes() {
     dataNodeWrapperList.forEach(AbstractNodeWrapper::stop);
   }
 
+  @Override
+  public void shutdownForciblyAllDataNodes() {
+    dataNodeWrapperList.forEach(AbstractNodeWrapper::stopForcibly);
+  }
+
   @Override
   public void ensureNodeStatus(
       final List<BaseNodeWrapper> nodes, final List<NodeStatus> targetStatus)

diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
@@ -440,6 +440,11 @@ public void shutdownAllDataNodes() {
     throw new UnsupportedOperationException();
   }
 
+  @Override
+  public void shutdownForciblyAllDataNodes() {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public int getMqttPort() {
     throw new UnsupportedOperationException();

diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
@@ -300,6 +300,9 @@ void ensureNodeStatus(List<BaseNodeWrapper> nodes, List<NodeStatus> targetStatus
   /** Shutdown all existed DataNodes. */
   void shutdownAllDataNodes();
 
+  /** Shutdown forcibly all existed DataNodes. */
+  void shutdownForciblyAllDataNodes();
+
   int getMqttPort();
 
   String getIP();

diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
@@ -1318,4 +1318,27 @@ public static void restartDataNodes() {
       }
     }
   }
+
+  public static void stopForciblyAndRestartDataNodes() {
+    EnvFactory.getEnv().shutdownForciblyAllDataNodes();
+    EnvFactory.getEnv().startAllDataNodes();
+    long waitStartMS = System.currentTimeMillis();
+    long maxWaitMS = 60_000L;
+    long retryIntervalMS = 1000;
+    while (true) {
+      try (Connection connection = EnvFactory.getEnv().getConnection()) {
+        break;
+      } catch (Exception e) {
+        try {
+          Thread.sleep(retryIntervalMS);
+        } catch (InterruptedException ex) {
+          break;
+        }
+      }
+      long waited = System.currentTimeMillis() - waitStartMS;
+      if (waited > maxWaitMS) {
+        fail("Timeout while waiting for datanodes restart");
+      }
+    }
+  }
 }
diff --git a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java
@@ -20,6 +20,7 @@
 
 import org.apache.iotdb.common.rpc.thrift.TAggregationType;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.it.utils.TestUtils;
 import org.apache.iotdb.db.protocol.thrift.OperationType;
 import org.apache.iotdb.isession.ISession;
 import org.apache.iotdb.isession.SessionDataSet;
@@ -64,6 +65,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
@@ -479,7 +481,7 @@ public void insertTabletWithAlignedTimeseriesTest() {
 
   @Test
   @Category({LocalStandaloneIT.class, ClusterIT.class})
-  public void insertTabletWithNullValuesTest() {
+  public void insertTabletWithNullValuesTest() throws InterruptedException {
     try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
       List<IMeasurementSchema> schemaList = new ArrayList<>();
       schemaList.add(new MeasurementSchema("s0", TSDataType.DOUBLE, TSEncoding.RLE));
@@ -527,6 +529,38 @@ public void insertTabletWithNullValuesTest() {
           assertEquals(9L, field.getLongV());
         }
       }
+      dataSet = session.executeQueryStatement("select s3 from root.sg1.d1");
+      int result = 0;
+      assertTrue(dataSet.hasNext());
+      while (dataSet.hasNext()) {
+        RowRecord rowRecord = dataSet.next();
+        Field field = rowRecord.getFields().get(0);
+        // skip null value
+        if (result == 3) {
+          result++;
+        }
+        assertEquals(result++, field.getIntV());
+      }
+    } catch (Exception e) {
+      e.printStackTrace();
+      fail(e.getMessage());
+    }
+    TimeUnit.MILLISECONDS.sleep(2000);
+
+    TestUtils.stopForciblyAndRestartDataNodes();
+
+    try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      SessionDataSet dataSet = session.executeQueryStatement("select s3 from root.sg1.d1");
+      int result = 0;
+      while (dataSet.hasNext()) {
+        RowRecord rowRecord = dataSet.next();
+        Field field = rowRecord.getFields().get(0);
+        // skip null value
+        if (result == 3) {
+          result++;
+        }
+        assertEquals(result++, field.getIntV());
+      }
     } catch (Exception e) {
       e.printStackTrace();
       fail(e.getMessage());

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
@@ -210,11 +210,15 @@ public void putBinaries(long[] time, Binary[] value, BitMap bitMap, int start, i
     // constraint: time.length + timeIdxOffset == value.length
     int timeIdxOffset = 0;
     if (bitMap != null && !bitMap.isAllUnmarked()) {
-      // time array is a reference, should clone necessary time values
+      // time array is a reference, should clone necessary time array
       long[] clonedTime = new long[end - start];
       System.arraycopy(time, start, clonedTime, 0, end - start);
       time = clonedTime;
       timeIdxOffset = start;
+      // value array is a reference, should clone necessary value array
+      Binary[] clonedValue = new Binary[value.length];
+      System.arraycopy(value, 0, clonedValue, 0, value.length);
+      value = clonedValue;
       // drop null at the end of value array
       int nullCnt =
           dropNullValThenUpdateMaxTimeAndSorted(time, value, bitMap, start, end, timeIdxOffset);

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
@@ -169,11 +169,15 @@ public void putBooleans(long[] time, boolean[] value, BitMap bitMap, int start,
     // constraint: time.length + timeIdxOffset == value.length
     int timeIdxOffset = 0;
     if (bitMap != null && !bitMap.isAllUnmarked()) {
-      // time array is a reference, should clone necessary time values
+      // time array is a reference, should clone necessary time array
       long[] clonedTime = new long[end - start];
       System.arraycopy(time, start, clonedTime, 0, end - start);
       time = clonedTime;
       timeIdxOffset = start;
+      // value array is a reference, should clone necessary value array
+      boolean[] clonedValue = new boolean[value.length];
+      System.arraycopy(value, 0, clonedValue, 0, value.length);
+      value = clonedValue;
       // drop null at the end of value array
       int nullCnt =
           dropNullValThenUpdateMaxTimeAndSorted(time, value, bitMap, start, end, timeIdxOffset);

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
@@ -174,11 +174,15 @@ public void putDoubles(long[] time, double[] value, BitMap bitMap, int start, in
     // constraint: time.length + timeIdxOffset == value.length
     int timeIdxOffset = 0;
     if (bitMap != null && !bitMap.isAllUnmarked()) {
-      // time array is a reference, should clone necessary time values
+      // time array is a reference, should clone necessary time array
       long[] clonedTime = new long[end - start];
       System.arraycopy(time, start, clonedTime, 0, end - start);
       time = clonedTime;
       timeIdxOffset = start;
+      // value array is a reference, should clone necessary value array
+      double[] clonedValue = new double[value.length];
+      System.arraycopy(value, 0, clonedValue, 0, value.length);
+      value = clonedValue;
       // drop null at the end of value array
       int nullCnt =
           dropNullValThenUpdateMaxTimeAndSorted(time, value, bitMap, start, end, timeIdxOffset);

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
@@ -174,11 +174,15 @@ public void putFloats(long[] time, float[] value, BitMap bitMap, int start, int
     // constraint: time.length + timeIdxOffset == value.length
     int timeIdxOffset = 0;
     if (bitMap != null && !bitMap.isAllUnmarked()) {
-      // time array is a reference, should clone necessary time values
+      // time array is a reference, should clone necessary time array
       long[] clonedTime = new long[end - start];
       System.arraycopy(time, start, clonedTime, 0, end - start);
       time = clonedTime;
       timeIdxOffset = start;
+      // value array is a reference, should clone necessary value array
+      float[] clonedValue = new float[value.length];
+      System.arraycopy(value, 0, clonedValue, 0, value.length);
+      value = clonedValue;
       // drop null at the end of value array
       int nullCnt =
           dropNullValThenUpdateMaxTimeAndSorted(time, value, bitMap, start, end, timeIdxOffset);

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
@@ -167,11 +167,15 @@ public void putInts(long[] time, int[] value, BitMap bitMap, int start, int end)
     // constraint: time.length + timeIdxOffset == value.length
     int timeIdxOffset = 0;
     if (bitMap != null && !bitMap.isAllUnmarked()) {
-      // time array is a reference, should clone necessary time values
+      // time array is a reference, should clone necessary time array
       long[] clonedTime = new long[end - start];
       System.arraycopy(time, start, clonedTime, 0, end - start);
       time = clonedTime;
       timeIdxOffset = start;
+      // value array is a reference, should clone necessary value array
+      int[] clonedValue = new int[value.length];
+      System.arraycopy(value, 0, clonedValue, 0, value.length);
+      value = clonedValue;
       // drop null at the end of value array
       int nullCnt =
           dropNullValThenUpdateMaxTimeAndSorted(time, value, bitMap, start, end, timeIdxOffset);

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
@@ -167,11 +167,15 @@ public void putLongs(long[] time, long[] value, BitMap bitMap, int start, int en
     // constraint: time.length + timeIdxOffset == value.length
     int timeIdxOffset = 0;
     if (bitMap != null && !bitMap.isAllUnmarked()) {
-      // time array is a reference, should clone necessary time values
+      // time array is a reference, should clone necessary time array
       long[] clonedTime = new long[end - start];
       System.arraycopy(time, start, clonedTime, 0, end - start);
       time = clonedTime;
       timeIdxOffset = start;
+      // value array is a reference, should clone necessary value array
+      long[] clonedValue = new long[value.length];
+      System.arraycopy(value, 0, clonedValue, 0, value.length);
+      value = clonedValue;
       // drop null at the end of value array
       int nullCnt =
           dropNullValThenUpdateMaxTimeAndSorted(time, value, bitMap, start, end, timeIdxOffset);

diff --git a/...e/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/...e/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java
@@ -90,6 +90,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -385,8 +386,19 @@ public void testIoTDBTabletWriteAndSyncClose()
 
     for (int r = 0; r < 100; r++) {
       times[r] = r;
-      ((int[]) columns[0])[r] = 1;
-      ((long[]) columns[1])[r] = 1;
+      ((int[]) columns[0])[r] = r;
+      ((long[]) columns[1])[r] = r;
+    }
+
+    BitMap[] bitMaps = new BitMap[2];
+    bitMaps[0] = new BitMap(100);
+    bitMaps[1] = new BitMap(100);
+    for (int r = 0; r < 100; r++) {
+      if (r % 2 == 0) {
+        bitMaps[0].mark(r);
+      } else {
+        bitMaps[1].mark(r);
+      }
     }
 
     InsertTabletNode insertTabletNode1 =
@@ -398,11 +410,15 @@ public void testIoTDBTabletWriteAndSyncClose()
             dataTypes,
             measurementSchemas,
             times,
-            null,
+            bitMaps,
             columns,
             times.length);
-
+    int hashCode1 = Arrays.hashCode((int[]) columns[0]);
+    int hashCode2 = Arrays.hashCode((long[]) columns[1]);
     dataRegion.insertTablet(insertTabletNode1);
+    // the hashCode should not be changed when insert
+    Assert.assertEquals(hashCode1, Arrays.hashCode((int[]) columns[0]));
+    Assert.assertEquals(hashCode2, Arrays.hashCode((long[]) columns[1]));
     dataRegion.syncCloseAllWorkingTsFileProcessors();
 
     for (int r = 50; r < 149; r++) {