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 210173bc4e02c..9605e92104e76 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 @@ -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 { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java index df050211626c9..5e88b02084d67 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java @@ -96,7 +96,10 @@ static Map 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; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerTest.java index 7dfdbde5c97f8..a1831e2d411b2 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerTest.java +++ b/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 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);