Fixed the insertion NPE bug with out of ttl and pipe

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

@@ -87,7 +87,9 @@ public static void setUp() throws Exception {
     EnvFactory.getEnv()
         .getConfig()
         .getDataNodeCommonConfig()
-        .setWriteMemoryProportion("10000000:1");
+        .setWriteMemoryProportion("10000000:1")
+        .setPipeMemoryManagementEnabled(false)
+        .setIsPipeEnableMemoryCheck(false);
     EnvFactory.getEnv().initClusterEnvironment();
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
@@ -1187,6 +1189,26 @@ public void testInsertWithTTL() {
     }
   }
 
+  @Test
+  public void testInsertWithPipeAndTTL() throws Exception {
+    final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+    final Statement statement = connection.createStatement();
+    try {
+      statement.execute("use \"test\"");
+      statement.execute("create pipe placeholder ('sink'='do-nothing-sink')");
+      statement.execute("create table t_big (dev string tag, v double field) with (ttl=100000)");
+      statement.execute("insert into t_big (time, dev, v) values (now(), 'd1', 11)");
+      statement.execute("insert into t_big (time, dev, v) values (1793772199053, 'd1', 13)");
+    } catch (final Exception e) {
+      Assert.assertTrue(e.getMessage().contains("less than ttl time bound"));
+    } finally {
+      statement.execute("drop pipe if exists placeholder_history");
+      statement.execute("drop pipe placeholder_realtime");
+      statement.close();
+      connection.close();
+    }
+  }
+
   @Test
   public void testInsertTabletWithTTL()
       throws IoTDBConnectionException, StatementExecutionException {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java

@@ -96,7 +96,10 @@ static Map<IDeviceID, String[]> getDevice2MeasurementsMapFromInsertRowsNode(
     // Keep one set per repeated device instead and materialize its array only once.
     for (final InsertNode insertRowNode : insertRowsNode.getInsertRowNodeList()) {
       final IDeviceID deviceID = insertRowNode.getDeviceID();
-      final String[] measurements = Objects.requireNonNull(insertRowNode.getMeasurements());
+      final String[] measurements = insertRowNode.getMeasurements();
+      if (measurements == null) {
+        continue;
+      }
       final String[] firstMeasurements = device2Measurements.putIfAbsent(deviceID, measurements);
       if (firstMeasurements == null) {
         continue;

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerTest.java

@@ -81,17 +81,6 @@ public void testSubsequenceAggregationPreservesEncounterOrder() {
         new String[] {"s1", "s3", null, "s2"}, device2Measurements.get(device));
   }
 
-  @Test(expected = NullPointerException.class)
-  public void testNullMeasurementsAreRejected() {
-    final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.test.d1");
-    final InsertRowsNode rows = mock(InsertRowsNode.class);
-    final List<InsertRowNode> insertRowNodes =
-        Arrays.asList(mockRow(device, "s1"), mockRow(device, (String[]) null));
-    when(rows.getInsertRowNodeList()).thenReturn(insertRowNodes);
-
-    TsFileEpochManager.getDevice2MeasurementsMapFromInsertRowsNode(rows);
-  }
-
   private static InsertRowNode mockRow(final IDeviceID device, final String... measurements) {
     final InsertRowNode row = mock(InsertRowNode.class);
     when(row.getDeviceID()).thenReturn(device);