From bc7d9745db0576cc221ac3deb9c41af7473cbfd7 Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Tue, 18 Jun 2024 17:14:48 +0800 Subject: [PATCH 1/5] setup --- .../rpc/subscription/config/TopicConfig.java | 34 +++++++++++-------- .../subscription/config/TopicConstant.java | 5 +++ .../subscription/meta/topic/TopicMeta.java | 2 ++ 3 files changed, 27 insertions(+), 14 deletions(-) diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java index c3b5f24622c43..6dc42ed634aea 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java @@ -28,7 +28,10 @@ import java.nio.ByteBuffer; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; public class TopicConfig extends PipeParameters { @@ -40,14 +43,6 @@ public TopicConfig(final Map attributes) { super(attributes); } - private static final Map LOOSE_RANGE_TIME_CONFIG = - new HashMap() { - { - put("history.loose-range", "time"); - put("realtime.loose-range", "time"); - } - }; - private static final Map REALTIME_BATCH_MODE_CONFIG = Collections.singletonMap("realtime.mode", "batch"); private static final Map REALTIME_STREAM_MODE_CONFIG = @@ -58,6 +53,15 @@ public TopicConfig(final Map attributes) { private static final Map SUBSCRIBE_MODE_CONFIG = Collections.singletonMap("mode", "subscribe"); + private static final Set LOOSE_RANGE_KEY_SET = + Collections.unmodifiableSet( + new HashSet() { + { + add("history.loose-range"); + add("realtime.loose-range"); + } + }); + /////////////////////////////// de/ser /////////////////////////////// public void serialize(final DataOutputStream stream) throws IOException { @@ -102,12 +106,6 @@ public Map getAttributesWithTimeRange(final long creationTime) { attributesWithTimeRange.put(TopicConstant.END_TIME_KEY, endTime); } - // enable loose range when using tsfile format - if (TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE.equals( - attributes.getOrDefault(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_DEFAULT_VALUE))) { - attributesWithTimeRange.putAll(LOOSE_RANGE_TIME_CONFIG); - } - return attributesWithTimeRange; } @@ -125,6 +123,14 @@ public Map getAttributesWithSourceMode() { : SUBSCRIBE_MODE_CONFIG; } + public Map getAttributesWithSourceLooseRange() { + final String looseRangeValue = + attributes.getOrDefault( + TopicConstant.LOOSE_RANGE_KEY, TopicConstant.LOOSE_RANGE_DEFAULT_VALUE); + return LOOSE_RANGE_KEY_SET.stream() + .collect(Collectors.toMap(key -> key, key -> looseRangeValue)); + } + public Map getAttributesWithProcessorPrefix() { final Map attributesWithProcessorPrefix = new HashMap<>(); attributes.forEach( diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java index a6e8ec901bf9f..37d3a98dcc3aa 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java @@ -40,6 +40,11 @@ public class TopicConstant { public static final String FORMAT_TS_FILE_HANDLER_VALUE = "TsFileHandler"; public static final String FORMAT_DEFAULT_VALUE = FORMAT_SESSION_DATA_SETS_HANDLER_VALUE; + public static final String LOOSE_RANGE_KEY = "loose-range"; + public static final String LOOSE_RANGE_TIME_VALUE = "time"; + public static final String LOOSE_RANGE_PATH_VALUE = "path"; + public static final String LOOSE_RANGE_DEFAULT_VALUE = ""; + private TopicConstant() { throw new IllegalStateException("Utility class"); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java index 70666a0718249..14c48c85da33b 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java @@ -187,6 +187,8 @@ public Map generateExtractorAttributes() { extractorAttributes.putAll(config.getAttributesWithRealtimeMode()); // source mode extractorAttributes.putAll(config.getAttributesWithSourceMode()); + // loose range + extractorAttributes.putAll(config.getAttributesWithSourceLooseRange()); return extractorAttributes; } From d601ba3048d038cb195dd02e6caf14876caeb765 Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Tue, 18 Jun 2024 19:14:06 +0800 Subject: [PATCH 2/5] tmp --- .../apache/iotdb/db/it/utils/TestUtils.java | 30 ++++ .../it/autocreate/IoTDBPipeExtractorIT.java | 15 +- .../it/dual/IoTDBSubscriptionTopicIT.java | 143 ++++++++++++++++++ .../subscription/config/TopicConstant.java | 1 + 4 files changed, 187 insertions(+), 2 deletions(-) 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 index 176d0158dfce7..49925d29b2a10 100644 --- 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 @@ -704,6 +704,9 @@ public static void assertDataEventuallyOnEnv( Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures await() + .pollInSameThread() + .pollDelay(1L, TimeUnit.SECONDS) + .pollInterval(2L, TimeUnit.SECONDS) .atMost(timeoutSeconds, TimeUnit.SECONDS) .untilAsserted( () -> { @@ -720,6 +723,33 @@ public static void assertDataEventuallyOnEnv( } } + public static void assertDataEventuallyOnEnv( + BaseEnv env, String sql, Map expectedHeaderWithResult) { + assertDataEventuallyOnEnv(env, sql, expectedHeaderWithResult, 600); + } + + public static void assertDataEventuallyOnEnv( + BaseEnv env, String sql, Map expectedHeaderWithResult, long timeoutSeconds) { + try (Connection connection = env.getConnection(); + Statement statement = connection.createStatement()) { + // Keep retrying if there are execution failures + await() + .atMost(timeoutSeconds, TimeUnit.SECONDS) + .untilAsserted( + () -> { + try { + TestUtils.assertSingleResultSetEqual( + executeQueryWithRetry(statement, sql), expectedHeaderWithResult); + } catch (Exception e) { + Assert.fail(); + } + }); + } catch (Exception e) { + e.printStackTrace(); + fail(); + } + } + public static void assertDataAlwaysOnEnv( BaseEnv env, String sql, String expectedHeader, Set expectedResSet) { assertDataAlwaysOnEnv(env, sql, expectedHeader, expectedResSet, 10); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java index 8ce885935b8fd..916055bdbd5a1 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java @@ -1013,12 +1013,23 @@ public void testRealtimeLooseRange() throws Exception { return; } - TestUtils.tryExecuteNonQueriesWithRetry( + if (!TestUtils.tryExecuteNonQueriesWithRetry( senderEnv, Arrays.asList( "insert into root.db.d1 (time, at1)" + " values (5000, 1), (16000, 3)", "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)", - "flush")); + "flush"))) { + return; + } + + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select * from root.db.d1.at1 where time >= 2000 and time <= 10000", + new HashMap() { + { + put("count(root.*.*.*)", "4"); + } + }); } } diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java index 19c033af151ba..8780060aefe0d 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java @@ -51,6 +51,7 @@ import java.sql.Connection; import java.sql.Statement; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -533,6 +534,8 @@ public void testTopicInvalidTimeRangeConfig() throws Exception { @Test public void testTopicInvalidPathConfig() throws Exception { // Test invalid path when using tsfile format + // NOTE: Delete this test after the restriction "on path/time range/processor when subscribing + // to tsfile" is removed. final Properties config = new Properties(); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE); config.put(TopicConstant.PATH_KEY, "root.db.*.s"); @@ -542,6 +545,8 @@ public void testTopicInvalidPathConfig() throws Exception { @Test public void testTopicInvalidProcessorConfig() throws Exception { // Test invalid processor when using tsfile format + // NOTE: Delete this test after the restriction "on path/time range/processor when subscribing + // to tsfile" is removed. final Properties config = new Properties(); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE); config.put("processor", "tumbling-time-sampling-processor"); @@ -578,6 +583,7 @@ public void testTopicWithQueryMode() throws Exception { e.printStackTrace(); fail(e.getMessage()); } + assertTopicCount(1); // Subscription final AtomicInteger rowCount = new AtomicInteger(); @@ -652,6 +658,143 @@ public void testTopicWithQueryMode() throws Exception { } } + @Test + public void testTopicWithLooseRange() throws Exception { + // Create topic + final String topicName = "topic12"; + final String host = senderEnv.getIP(); + final int port = Integer.parseInt(senderEnv.getPort()); + try (final SubscriptionSession session = new SubscriptionSession(host, port)) { + session.open(); + final Properties config = new Properties(); + config.put( + TopicConstant.LOOSE_RANGE_KEY, + String.join( + TopicConstant.LOOSE_RANGE_VALUE_SEPARATOR, + TopicConstant.LOOSE_RANGE_TIME_VALUE, + TopicConstant.LOOSE_RANGE_PATH_VALUE)); + config.put(TopicConstant.PATH_KEY, "root.db.d1.at1"); + config.put(TopicConstant.START_TIME_KEY, "1500"); + config.put(TopicConstant.END_TIME_KEY, "2500"); + session.createTopic(topicName, config); + } catch (final Exception e) { + e.printStackTrace(); + fail(e.getMessage()); + } + assertTopicCount(1); + + final AtomicBoolean dataPrepared = new AtomicBoolean(false); + final AtomicBoolean topicSubscribed = new AtomicBoolean(false); + final List threads = new ArrayList<>(); + + // Subscribe on sender + threads.add( + new Thread( + () -> { + final AtomicInteger retryCount = new AtomicInteger(); + final int maxRetryTimes = 32; + try (final SubscriptionPullConsumer consumer = + new SubscriptionPullConsumer.Builder() + .host(host) + .port(port) + .consumerId("c1") + .consumerGroupId("cg1") + .autoCommit(false) + .buildPullConsumer(); + final ISession session = receiverEnv.getSessionConnection()) { + consumer.open(); + consumer.subscribe(topicName); + topicSubscribed.set(true); + while (retryCount.getAndIncrement() < maxRetryTimes) { + LockSupport.parkNanos(IoTDBSubscriptionITConstant.SLEEP_NS); // wait some time + if (dataPrepared.get()) { + final List messages = + consumer.poll(IoTDBSubscriptionITConstant.POLL_TIMEOUT_MS); + for (final SubscriptionMessage message : messages) { + for (final Iterator it = + message.getSessionDataSetsHandler().tabletIterator(); + it.hasNext(); ) { + final Tablet tablet = it.next(); + session.insertTablet(tablet); + } + } + consumer.commitSync(messages); + } + } + consumer.unsubscribe(topicName); + } catch (final Exception e) { + e.printStackTrace(); + // Avoid failure + } finally { + LOGGER.info("consumer exiting..."); + } + }, + String.format("%s - consumer", testName.getMethodName()))); + + // Insert some realtime data on sender + threads.add( + new Thread( + () -> { + while (!topicSubscribed.get()) { + LockSupport.parkNanos(IoTDBSubscriptionITConstant.SLEEP_NS); // wait some time + } + if (!TestUtils.tryExecuteNonQueriesWithRetry( + senderEnv, + Arrays.asList( + // TsFile 1, extracted without parse + "insert into root.db.d1 (time, at1, at2)" + + " values (1000, 1, 2), (2000, 3, 4)", + // TsFile 2, not extracted because pattern not overlapped + "insert into root.db1.d1 (time, at1, at2)" + + " values (1000, 1, 2), (2000, 3, 4)", + "flush"))) { + fail(); + } + if (!TestUtils.tryExecuteNonQueriesWithRetry( + senderEnv, + Arrays.asList( + // TsFile 3, not extracted because time range not overlapped + "insert into root.db.d1 (time, at1, at2)" + + " values (3000, 1, 2), (4000, 3, 4)", + "flush"))) { + fail(); + } + dataPrepared.set(true); + }, + String.format("%s - data inserter", testName.getMethodName()))); + + for (final Thread thread : threads) { + thread.start(); + } + + try (final Connection connection = receiverEnv.getConnection(); + final Statement statement = connection.createStatement()) { + // Keep retrying if there are execution failures + AWAIT.untilAsserted( + () -> + // TestUtils.assertSingleResultSetEqual( + // TestUtils.executeQueryWithRetry( + // statement, "select * from root.**"), + // new HashMap() { + // { + // put("count(root.*.*.*)", "4"); + // } + // })); + TestUtils.assertSingleResultSetEqual( + TestUtils.executeQueryWithRetry( + statement, "select count(*) from root.** group by level=0"), + new HashMap() { + { + put("count(root.*.*.*)", "4"); + } + })); + } + + for (final Thread thread : threads) { + thread.join(); + } + } + private void testTopicInvalidRuntimeConfigTemplate( final String topicName, final Properties config) throws Exception { // Create topic diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java index 37d3a98dcc3aa..c1c3712d54943 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java @@ -44,6 +44,7 @@ public class TopicConstant { public static final String LOOSE_RANGE_TIME_VALUE = "time"; public static final String LOOSE_RANGE_PATH_VALUE = "path"; public static final String LOOSE_RANGE_DEFAULT_VALUE = ""; + public static final String LOOSE_RANGE_VALUE_SEPARATOR = ","; private TopicConstant() { throw new IllegalStateException("Utility class"); From b9b758e425631fea9657f7ddd5c0da5dee4e25e4 Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Tue, 18 Jun 2024 19:45:31 +0800 Subject: [PATCH 3/5] add IT --- .../apache/iotdb/db/it/utils/TestUtils.java | 5 +- .../it/autocreate/IoTDBPipeExtractorIT.java | 4 +- .../it/IoTDBSubscriptionITConstant.java | 2 +- .../it/dual/IoTDBSubscriptionTopicIT.java | 65 +++++++++---------- 4 files changed, 38 insertions(+), 38 deletions(-) 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 index 49925d29b2a10..35210c02aad19 100644 --- 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 @@ -706,7 +706,7 @@ public static void assertDataEventuallyOnEnv( await() .pollInSameThread() .pollDelay(1L, TimeUnit.SECONDS) - .pollInterval(2L, TimeUnit.SECONDS) + .pollInterval(1L, TimeUnit.SECONDS) .atMost(timeoutSeconds, TimeUnit.SECONDS) .untilAsserted( () -> { @@ -734,6 +734,9 @@ public static void assertDataEventuallyOnEnv( Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures await() + .pollInSameThread() + .pollDelay(1L, TimeUnit.SECONDS) + .pollInterval(1L, TimeUnit.SECONDS) .atMost(timeoutSeconds, TimeUnit.SECONDS) .untilAsserted( () -> { diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java index 916055bdbd5a1..2a4439eb605c9 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java @@ -1024,10 +1024,10 @@ public void testRealtimeLooseRange() throws Exception { TestUtils.assertDataEventuallyOnEnv( receiverEnv, - "select * from root.db.d1.at1 where time >= 2000 and time <= 10000", + "select count(at1) from root.db.d1 where time >= 2000 and time <= 10000", new HashMap() { { - put("count(root.*.*.*)", "4"); + put("count(root.db.d1.at1)", "4"); } }); } diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java index 840abb4ccb5b5..5b8ec39327414 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/IoTDBSubscriptionITConstant.java @@ -27,7 +27,7 @@ public class IoTDBSubscriptionITConstant { private static final long AWAITILITY_POLL_DELAY_SECOND = 1L; - private static final long AWAITILITY_POLL_INTERVAL_SECOND = 2L; + private static final long AWAITILITY_POLL_INTERVAL_SECOND = 1L; private static final long AWAITILITY_AT_MOST_SECOND = 600L; public static final ConditionFactory AWAIT = diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java index 8780060aefe0d..a6567d0ba32c4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java @@ -51,7 +51,6 @@ import java.sql.Connection; import java.sql.Statement; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -660,6 +659,20 @@ public void testTopicWithQueryMode() throws Exception { @Test public void testTopicWithLooseRange() throws Exception { + // Insert some historical data on sender + try (final ISession session = senderEnv.getSessionConnection()) { + session.executeNonQueryStatement( + "insert into root.db.d1 (time, at1, at2) values (1000, 1, 2), (2000, 3, 4)"); + session.executeNonQueryStatement( + "insert into root.db1.d1 (time, at1, at2) values (1000, 1, 2), (2000, 3, 4)"); + session.executeNonQueryStatement( + "insert into root.db.d1 (time, at1, at2) values (3000, 1, 2), (4000, 3, 4)"); + session.executeNonQueryStatement("flush"); + } catch (final Exception e) { + e.printStackTrace(); + fail(e.getMessage()); + } + // Create topic final String topicName = "topic12"; final String host = senderEnv.getIP(); @@ -685,14 +698,13 @@ public void testTopicWithLooseRange() throws Exception { final AtomicBoolean dataPrepared = new AtomicBoolean(false); final AtomicBoolean topicSubscribed = new AtomicBoolean(false); + final AtomicBoolean result = new AtomicBoolean(false); final List threads = new ArrayList<>(); // Subscribe on sender threads.add( new Thread( () -> { - final AtomicInteger retryCount = new AtomicInteger(); - final int maxRetryTimes = 32; try (final SubscriptionPullConsumer consumer = new SubscriptionPullConsumer.Builder() .host(host) @@ -705,7 +717,7 @@ public void testTopicWithLooseRange() throws Exception { consumer.open(); consumer.subscribe(topicName); topicSubscribed.set(true); - while (retryCount.getAndIncrement() < maxRetryTimes) { + while (!result.get()) { LockSupport.parkNanos(IoTDBSubscriptionITConstant.SLEEP_NS); // wait some time if (dataPrepared.get()) { final List messages = @@ -738,26 +750,17 @@ public void testTopicWithLooseRange() throws Exception { while (!topicSubscribed.get()) { LockSupport.parkNanos(IoTDBSubscriptionITConstant.SLEEP_NS); // wait some time } - if (!TestUtils.tryExecuteNonQueriesWithRetry( - senderEnv, - Arrays.asList( - // TsFile 1, extracted without parse - "insert into root.db.d1 (time, at1, at2)" - + " values (1000, 1, 2), (2000, 3, 4)", - // TsFile 2, not extracted because pattern not overlapped - "insert into root.db1.d1 (time, at1, at2)" - + " values (1000, 1, 2), (2000, 3, 4)", - "flush"))) { - fail(); - } - if (!TestUtils.tryExecuteNonQueriesWithRetry( - senderEnv, - Arrays.asList( - // TsFile 3, not extracted because time range not overlapped - "insert into root.db.d1 (time, at1, at2)" - + " values (3000, 1, 2), (4000, 3, 4)", - "flush"))) { - fail(); + try (final ISession session = senderEnv.getSessionConnection()) { + session.executeNonQueryStatement( + "insert into root.db.d1 (time, at1, at2) values (1001, 1, 2), (2001, 3, 4)"); + session.executeNonQueryStatement( + "insert into root.db1.d1 (time, at1, at2) values (1001, 1, 2), (2001, 3, 4)"); + session.executeNonQueryStatement( + "insert into root.db.d1 (time, at1, at2) values (3001, 1, 2), (4001, 3, 4)"); + session.executeNonQueryStatement("flush"); + } catch (final Exception e) { + e.printStackTrace(); + fail(e.getMessage()); } dataPrepared.set(true); }, @@ -772,24 +775,18 @@ public void testTopicWithLooseRange() throws Exception { // Keep retrying if there are execution failures AWAIT.untilAsserted( () -> - // TestUtils.assertSingleResultSetEqual( - // TestUtils.executeQueryWithRetry( - // statement, "select * from root.**"), - // new HashMap() { - // { - // put("count(root.*.*.*)", "4"); - // } - // })); TestUtils.assertSingleResultSetEqual( TestUtils.executeQueryWithRetry( - statement, "select count(*) from root.** group by level=0"), + statement, + "select count(at1) from root.db.d1 where time >= 1500 and time <= 2500"), new HashMap() { { - put("count(root.*.*.*)", "4"); + put("count(root.db.d1.at1)", "2"); } })); } + result.set(true); for (final Thread thread : threads) { thread.join(); } From ee2dc711174aca210db444d46c30ac99a0d8dbf0 Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Tue, 18 Jun 2024 19:58:28 +0800 Subject: [PATCH 4/5] remove LOOSE_RANGE_VALUE_SEPARATOR --- .../subscription/it/dual/IoTDBSubscriptionTopicIT.java | 7 +------ .../iotdb/rpc/subscription/config/TopicConstant.java | 2 +- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java index a6567d0ba32c4..d4aee45c482ad 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java @@ -680,12 +680,7 @@ public void testTopicWithLooseRange() throws Exception { try (final SubscriptionSession session = new SubscriptionSession(host, port)) { session.open(); final Properties config = new Properties(); - config.put( - TopicConstant.LOOSE_RANGE_KEY, - String.join( - TopicConstant.LOOSE_RANGE_VALUE_SEPARATOR, - TopicConstant.LOOSE_RANGE_TIME_VALUE, - TopicConstant.LOOSE_RANGE_PATH_VALUE)); + config.put(TopicConstant.LOOSE_RANGE_KEY, TopicConstant.LOOSE_RANGE_TIME_AND_PATH_VALUE); config.put(TopicConstant.PATH_KEY, "root.db.d1.at1"); config.put(TopicConstant.START_TIME_KEY, "1500"); config.put(TopicConstant.END_TIME_KEY, "2500"); diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java index c1c3712d54943..a3226c1d64063 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java @@ -43,8 +43,8 @@ public class TopicConstant { public static final String LOOSE_RANGE_KEY = "loose-range"; public static final String LOOSE_RANGE_TIME_VALUE = "time"; public static final String LOOSE_RANGE_PATH_VALUE = "path"; + public static final String LOOSE_RANGE_TIME_AND_PATH_VALUE = "time,path"; public static final String LOOSE_RANGE_DEFAULT_VALUE = ""; - public static final String LOOSE_RANGE_VALUE_SEPARATOR = ","; private TopicConstant() { throw new IllegalStateException("Utility class"); From e890d0e6fa041b2ef0df1358de2b7713cbbb0fea Mon Sep 17 00:00:00 2001 From: VGalaxies Date: Wed, 19 Jun 2024 16:08:32 +0800 Subject: [PATCH 5/5] improve --- .../src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java | 3 +++ 1 file changed, 3 insertions(+) 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 index 35210c02aad19..372c9200c5e6c 100644 --- 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 @@ -768,6 +768,9 @@ public static void assertDataAlwaysOnEnv( Statement statement = connection.createStatement()) { // Keep retrying if there are execution failures await() + .pollInSameThread() + .pollDelay(1L, TimeUnit.SECONDS) + .pollInterval(1L, TimeUnit.SECONDS) .atMost(consistentSeconds, TimeUnit.SECONDS) .failFast( () -> {