Merge branch 'master' into alter_column_datatype_draft

Author: zerolbsony

integration-test/pom.xml

@@ -183,7 +183,6 @@
         <dependency>
             <groupId>org.bouncycastle</groupId>
             <artifactId>bcprov-jdk18on</artifactId>
-            <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>junit</groupId>

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBFlushQueryIT.java

@@ -19,12 +19,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.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
 
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -34,6 +43,8 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Collections;
+import java.util.List;
 import java.util.Locale;
 
 import static org.junit.Assert.assertEquals;
@@ -181,4 +192,75 @@ public void testFlushNotExistGroupNoData() {
       fail(e.getMessage());
     }
   }
+
+  @Test
+  public void testStreamingQueryMemTableWithOverlappedData()
+      throws IoTDBConnectionException, StatementExecutionException {
+    String device = "root.stream1.d1";
+    try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      session.open();
+      generateTimeRangeWithTimestamp(session, device, 1, 10);
+
+      generateTimeRangeWithTimestamp(session, device, 500000, 510000);
+      session.executeNonQueryStatement("flush");
+      generateTimeRangeWithTimestamp(session, device, 100000, 350000);
+
+      SessionDataSet sessionDataSet =
+          session.executeQueryStatement("select count(*) from root.stream1.d1");
+      SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+      long count = 0;
+      while (iterator.next()) {
+        count = iterator.getLong(1);
+      }
+      Assert.assertEquals(10 + 10001 + 250001, count);
+    }
+  }
+
+  @Test
+  public void testStreamingQueryMemTableWithOverlappedData2()
+      throws IoTDBConnectionException, StatementExecutionException {
+    String device = "root.stream2.d1";
+    try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      session.open();
+      generateTimeRangeWithTimestamp(session, device, 1, 10);
+
+      generateTimeRangeWithTimestamp(session, device, 500000, 510000);
+      session.executeNonQueryStatement("flush");
+      generateTimeRangeWithTimestamp(session, device, 1, 20);
+      generateTimeRangeWithTimestamp(session, device, 100000, 210000);
+      session.executeNonQueryStatement("flush");
+
+      generateTimeRangeWithTimestamp(session, device, 150000, 450000);
+
+      SessionDataSet sessionDataSet =
+          session.executeQueryStatement("select count(*) from root.stream2.d1");
+      SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+      long count = 0;
+      while (iterator.next()) {
+        count = iterator.getLong(1);
+      }
+      Assert.assertEquals(20 + 10001 + 350001, count);
+    }
+  }
+
+  private static void generateTimeRangeWithTimestamp(
+      ISession session, String device, long start, long end)
+      throws IoTDBConnectionException, StatementExecutionException {
+    List<IMeasurementSchema> measurementSchemas =
+        Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT64));
+    Tablet tablet = new Tablet(device, measurementSchemas);
+    for (long currentTime = start; currentTime <= end; currentTime++) {
+      int rowIndex = tablet.getRowSize();
+      if (rowIndex == tablet.getMaxRowNumber()) {
+        session.insertTablet(tablet);
+        tablet.reset();
+        rowIndex = 0;
+      }
+      tablet.addTimestamp(rowIndex, currentTime);
+      tablet.addValue(rowIndex, 0, currentTime);
+    }
+    if (tablet.getRowSize() > 0) {
+      session.insertTablet(tablet);
+    }
+  }
 }

integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java

@@ -31,7 +31,7 @@ abstract class AbstractPipeSingleIT {
 
   @Before
   public void setUp() {
-    MultiEnvFactory.createEnv(2);
+    MultiEnvFactory.createEnv(1);
     env = MultiEnvFactory.getEnv(0);
     env.getConfig()
         .getCommonConfig()