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..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 @@ -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(1L, TimeUnit.SECONDS) .atMost(timeoutSeconds, TimeUnit.SECONDS) .untilAsserted( () -> { @@ -720,6 +723,36 @@ 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() + .pollInSameThread() + .pollDelay(1L, TimeUnit.SECONDS) + .pollInterval(1L, TimeUnit.SECONDS) + .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); @@ -735,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( () -> { 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..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 @@ -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 count(at1) from root.db.d1 where time >= 2000 and time <= 10000", + new HashMap() { + { + 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 19c033af151ba..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 @@ -533,6 +533,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 +544,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 +582,7 @@ public void testTopicWithQueryMode() throws Exception { e.printStackTrace(); fail(e.getMessage()); } + assertTopicCount(1); // Subscription final AtomicInteger rowCount = new AtomicInteger(); @@ -652,6 +657,136 @@ 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(); + 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, 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"); + 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 AtomicBoolean result = new AtomicBoolean(false); + final List threads = new ArrayList<>(); + + // Subscribe on sender + threads.add( + new Thread( + () -> { + 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 (!result.get()) { + 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 + } + 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); + }, + 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 count(at1) from root.db.d1 where time >= 1500 and time <= 2500"), + new HashMap() { + { + put("count(root.db.d1.at1)", "2"); + } + })); + } + + result.set(true); + 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/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..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 @@ -40,6 +40,12 @@ 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_TIME_AND_PATH_VALUE = "time,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; }