From c7f3120f362c1895e0935d3a471db629fca8dcc6 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:48:52 +0800 Subject: [PATCH 1/5] Primitive --- .../relational/it/db/it/IoTDBInsertTableIT.java | 16 ++++++++++++++++ .../realtime/epoch/TsFileEpochManager.java | 5 ++++- 2 files changed, 20 insertions(+), 1 deletion(-) 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..84f205ffa1376 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 @@ -1187,6 +1187,22 @@ public void testInsertWithTTL() { } } + @Test + public void testInsertWithPipeAndTTL() { + try (final Connection connection = + EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("use \"test\""); + 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) (1793772199053, 'd1', 14);"); + fail(); + } catch (Exception e) { + Assert.assertTrue(e.getMessage().contains("less than ttl time bound")); + } + } + @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; From a0bfd6aca42f40645b2e37db6c032b243d64abe9 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:12:10 +0800 Subject: [PATCH 2/5] Update TsFileEpochManagerTest.java --- .../realtime/epoch/TsFileEpochManagerTest.java | 11 ----------- 1 file changed, 11 deletions(-) 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); From f07e4a46a0d891d722618d57c56e2664363229b7 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:14:17 +0800 Subject: [PATCH 3/5] Update IoTDBInsertTableIT.java --- .../apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 84f205ffa1376..ddda281809c76 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 @@ -1196,7 +1196,7 @@ public void testInsertWithPipeAndTTL() { 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) (1793772199053, 'd1', 14);"); + "insert into t_big (time, dev, v) values (1793772199053, 'd1', 13);"); fail(); } catch (Exception e) { Assert.assertTrue(e.getMessage().contains("less than ttl time bound")); From dbab4d1830b75088a40ae57d078780428ca8dfb6 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:14:26 +0800 Subject: [PATCH 4/5] Update IoTDBInsertTableIT.java --- .../apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java | 3 +-- 1 file changed, 1 insertion(+), 2 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 ddda281809c76..e458e8100364b 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 @@ -1195,8 +1195,7 @@ public void testInsertWithPipeAndTTL() { statement.execute("use \"test\""); 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);"); + statement.execute("insert into t_big (time, dev, v) values (1793772199053, 'd1', 13);"); fail(); } catch (Exception e) { Assert.assertTrue(e.getMessage().contains("less than ttl time bound")); From 4a6222a39fcbea4549f331a4396249603c335d25 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 4 Nov 2026 18:28:25 +0800 Subject: [PATCH 5/5] Update IoTDBInsertTableIT.java --- .../it/db/it/IoTDBInsertTableIT.java | 27 ++++++++++++------- 1 file changed, 17 insertions(+), 10 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 e458e8100364b..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()) { @@ -1188,17 +1190,22 @@ public void testInsertWithTTL() { } @Test - public void testInsertWithPipeAndTTL() { - try (final Connection connection = - EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - final Statement statement = connection.createStatement()) { + 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 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);"); - fail(); - } catch (Exception e) { + 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(); } }