You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/03/17 01:33:38 UTC
(iotdb) branch master updated: Subscription: refactor payload APIs to use ResultSet and tsfile readers (#17225)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 80399a6f120 Subscription: refactor payload APIs to use ResultSet and tsfile readers (#17225)
80399a6f120 is described below
commit 80399a6f120f8c8970f5af50cbece661e58eec1e
Author: VGalaxies <vg...@apache.org>
AuthorDate: Tue Mar 17 09:33:32 2026 +0800
Subscription: refactor payload APIs to use ResultSet and tsfile readers (#17225)
* refactor(subscription)!: replace dataset handlers with record and tsfile APIs
Reuse tsfile ResultSet in subscription payloads, rename message accessors to getRecords()/getTsFile(), and return ITsFileReader from tsfile handler. Also rename format constants to SubscriptionRecordHandler/SubscriptionTsFileHandler with legacy TsFileHandler compatibility.
BREAKING CHANGE: SubscriptionMessage.getSessionDataSetsHandler() and getTsFileHandler() are removed in favor of getRecords() and getTsFile(). SubscriptionSessionDataSet/SubscriptionSessionDataSetsHandler are removed and format default/value names are updated.
* refactor(subscription)!: align payload APIs with record and tsfile readers
- migrate subscription message usage to getRecords/getTsFile across tests and examples
- add record helpers (tablet iterator, row iteration) for ResultSet-based payloads
- unify tsfile access through openReader() with tree/table model-specific implementations
BREAKING CHANGE: subscription payload consumers should use getRecords()/getTsFile() and the new message type names (RECORD_HANDLER/TS_FILE); legacy handler-style payload APIs are no longer used.
* refactor(subscription): build tsfile readers via builders
* refactor(subscription)!: split tsfile readers by model
Replace SubscriptionTsFileHandler.openReader() with openTreeReader() and openTableReader(). Update tree-model examples and integration tests to use ITsFileTreeReader and a v4 query adapter.
BREAKING CHANGE: tree consumers must call openTreeReader(), and table consumers must call openTableReader().
* refactor(subscription)!: rename record payload result set APIs
Rename SubscriptionMessage.getRecords() to getResultSets() and SubscriptionRecordHandler.SubscriptionRecord to SubscriptionResultSet. Update examples and integration tests to use the new record payload names.
BREAKING CHANGE: callers must replace getRecords() with getResultSets() and SubscriptionRecord with SubscriptionResultSet.
* refactor(subscription): remove full payload type names in tests
---
.../apache/iotdb/SubscriptionSessionExample.java | 45 +--
.../TableModelSubscriptionSessionExample.java | 22 +-
.../it/SubscriptionTreeReaderTestUtils.java | 164 +++++++++
.../it/cluster/IoTDBSubscriptionRestartIT.java | 47 ++-
.../dual/tablemodel/IoTDBSubscriptionTopicIT.java | 18 +-
.../IoTDBSubscriptionConsumerGroupIT.java | 43 +--
.../IoTDBSubscriptionTimePrecisionIT.java | 3 +-
.../dual/treemodel/IoTDBSubscriptionTopicIT.java | 40 +--
.../it/local/IoTDBSubscriptionBasicIT.java | 68 ++--
.../it/local/IoTDBSubscriptionDataTypeIT.java | 48 ++-
.../tablemodel/IoTDBSubscriptionPermissionIT.java | 66 ++--
.../treemodel/IoTDBSubscriptionSharingIT.java | 146 ++++----
.../AbstractSubscriptionTreeRegressionIT.java | 40 +--
.../IoTDBDefaultTsfilePushConsumerIT.java | 14 +-
.../IoTDBRootDatasetPushConsumerIT.java | 9 +-
...merPullConsumerWith1TopicShareProcessMixIT.java | 9 +-
.../param/IoTDBTestParamPullConsumerIT.java | 23 +-
.../param/IoTDBTestParamPushConsumerIT.java | 19 +-
...DBTestAutoCommitFalseDataSetPullConsumerIT.java | 3 +-
...TDBTestAutoCommitTrueDataSetPullConsumerIT.java | 3 +-
.../multi/IoTDBOneConsumerMultiTopicsMixIT.java | 39 +-
.../format/IoTDBTestPushConsumeDataSetIT.java | 9 +-
.../IoTDBTestPushConsumeNoTargetDirTsfileIT.java | 12 +-
.../format/IoTDBTestPushConsumeTsfileIT.java | 12 +-
.../IoTDBLooseAllTsDatasetPushConsumerIT.java | 9 +-
...TDBLooseAllTsDatasetPushConsumerSnapshotIT.java | 9 +-
.../IoTDBLooseAllTsfilePushConsumerIT.java | 16 +-
.../IoTDBPathLooseDeviceTsfilePushConsumerIT.java | 14 +-
.../IoTDBPathLooseTsDatasetPushConsumerIT.java | 9 +-
.../IoTDBPathLooseTsfilePushConsumerIT.java | 10 +-
.../IoTDBPathTsLooseDatasetPushConsumerIT.java | 9 +-
.../IoTDBTimeLooseTsDatasetPushConsumerIT.java | 9 +-
.../IoTDBTimeLooseTsTsfilePushConsumerIT.java | 14 +-
.../IoTDBTimeLooseTsfilePushConsumerIT.java | 10 +-
.../IoTDBTimeTsLooseDatasetPushConsumerIT.java | 9 +-
...oTDBSnapshotTSPatternDatasetPushConsumerIT.java | 9 +-
...IoTDBSnapshotTSPatternTsfilePushConsumerIT.java | 14 +-
...DBConsumer2With1TopicShareProcessDataSetIT.java | 15 +-
...TDBConsumer2With1TopicShareProcessTsfileIT.java | 15 +-
.../multi/IoTDBMultiGroupVsMultiConsumerIT.java | 125 ++++---
.../IoTDBOneConsumerMultiTopicsDatasetIT.java | 9 +-
.../multi/IoTDBOneConsumerMultiTopicsMixIT.java | 36 +-
.../multi/IoTDBOneConsumerMultiTopicsTsfileIT.java | 19 +-
.../IoTDBDBPatternDatasetPushConsumerIT.java | 9 +-
.../IoTDBDBPatternTsfilePushConsumerIT.java | 14 +-
.../IoTDBDefaultPatternTsfilePushConsumerIT.java | 14 +-
.../IoTDBDevicePatternDatasetPushConsumerIT.java | 9 +-
.../IoTDBDevicePatternTsfilePushConsumerIT.java | 14 +-
.../IoTDBRootPatternDatasetPushConsumerIT.java | 9 +-
.../IoTDBTSPatternDatasetPushConsumerIT.java | 9 +-
.../IoTDBTSPatternTsfilePushConsumerIT.java | 14 +-
.../IoTDBHistoryRootDatasetPushConsumerIT.java | 9 +-
.../time/IoTDBHistoryRootTsFilePushConsumerIT.java | 10 +-
.../time/IoTDBRealTimeDBDatasetPushConsumerIT.java | 9 +-
.../time/IoTDBRealTimeDBTsfilePushConsumerIT.java | 10 +-
...DBTimeRangeAccurateDBDataSetPushConsumerIT.java | 9 +-
.../IoTDBTimeRangeDBDataSetPushConsumerIT.java | 9 +-
.../time/IoTDBTimeRangeDBTsfilePushConsumerIT.java | 10 +-
.../subscription/SubscriptionTableTsFile.java | 2 +-
.../subscription/SubscriptionTreeTsFile.java | 2 +-
.../iotdb/rpc/subscription/config/TopicConfig.java | 10 +-
.../rpc/subscription/config/TopicConstant.java | 11 +-
.../base/AbstractSubscriptionConsumer.java | 5 +-
.../subscription/payload/SubscriptionMessage.java | 31 +-
.../payload/SubscriptionMessageType.java | 4 +-
.../payload/SubscriptionRecordHandler.java | 392 +++++++++++++++++++++
.../payload/SubscriptionSessionDataSet.java | 255 --------------
.../SubscriptionSessionDataSetsHandler.java | 91 -----
.../payload/SubscriptionTsFileHandler.java | 28 +-
69 files changed, 1259 insertions(+), 992 deletions(-)
diff --git a/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java b/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java
index 6f4d503bce0..1202c533fe6 100644
--- a/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java
@@ -30,19 +30,16 @@ import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
-import org.apache.tsfile.read.TsFileReader;
-import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import java.io.IOException;
import java.net.URLEncoder;
import java.nio.file.Paths;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -149,11 +146,13 @@ public class SubscriptionSessionExample {
}
}
for (final SubscriptionMessage message : messages) {
- for (final SubscriptionSessionDataSet dataSet : message.getSessionDataSetsHandler()) {
- System.out.println(dataSet.getColumnNames());
- System.out.println(dataSet.getColumnTypes());
- while (dataSet.hasNext()) {
- System.out.println(dataSet.next());
+ for (final ResultSet dataSet : message.getResultSets()) {
+ final SubscriptionRecordHandler.SubscriptionResultSet record =
+ (SubscriptionRecordHandler.SubscriptionResultSet) dataSet;
+ System.out.println(record.getColumnNames());
+ System.out.println(record.getColumnTypes());
+ while (dataSet.next()) {
+ System.out.println("Time=" + dataSet.getLong(1));
}
}
}
@@ -180,7 +179,7 @@ public class SubscriptionSessionExample {
final Properties config = new Properties();
config.put(TopicConstant.START_TIME_KEY, CURRENT_TIME + 33);
config.put(TopicConstant.END_TIME_KEY, CURRENT_TIME + 66);
- config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+ config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_VALUE);
subscriptionSession.createTopic(TOPIC_2, config);
}
@@ -210,17 +209,8 @@ public class SubscriptionSessionExample {
}
}
for (final SubscriptionMessage message : messages) {
- try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
- final QueryDataSet dataSet =
- reader.query(
- QueryExpression.create(
- Arrays.asList(
- new Path("root.db.d2", "s2", true),
- new Path("root.sg.d3", "s1", true)),
- null));
- while (dataSet.hasNext()) {
- System.out.println(dataSet.next());
- }
+ try (final ITsFileTreeReader reader = message.getTsFile().openTreeReader()) {
+ reader.getAllDeviceIds().forEach(System.out::println);
}
}
consumer2.commitSync(messages);
@@ -245,7 +235,7 @@ public class SubscriptionSessionExample {
new SubscriptionTreeSession(HOST, PORT)) {
subscriptionSession.open();
final Properties config = new Properties();
- config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+ config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_VALUE);
config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_SNAPSHOT_VALUE);
subscriptionSession.createTopic(TOPIC_3, config);
}
@@ -265,8 +255,7 @@ public class SubscriptionSessionExample {
.consumeListener(
message -> {
// do something for SubscriptionTsFileHandler
- System.out.println(
- message.getTsFileHandler().getFile().getAbsolutePath());
+ System.out.println(message.getTsFile().getFile().getAbsolutePath());
return ConsumeResult.SUCCESS;
})
.buildPushConsumer()) {
@@ -292,7 +281,7 @@ public class SubscriptionSessionExample {
new SubscriptionTreeSession(HOST, PORT)) {
subscriptionSession.open();
final Properties config = new Properties();
- config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+ config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_VALUE);
config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_SNAPSHOT_VALUE);
subscriptionSession.createTopic(TOPIC_4, config);
}
@@ -316,7 +305,7 @@ public class SubscriptionSessionExample {
consumer4.subscribe(TOPIC_4);
while (!consumer4.allTopicMessagesHaveBeenConsumed()) {
for (final SubscriptionMessage message : consumer4.poll(POLL_TIMEOUT_MS)) {
- final SubscriptionTsFileHandler handler = message.getTsFileHandler();
+ final SubscriptionTsFileHandler handler = message.getTsFile();
handler.moveFile(
Paths.get(System.getProperty("user.dir"), "exported-tsfiles")
.resolve(
diff --git a/example/session/src/main/java/org/apache/iotdb/TableModelSubscriptionSessionExample.java b/example/session/src/main/java/org/apache/iotdb/TableModelSubscriptionSessionExample.java
index 363ff971662..176530dcb8f 100644
--- a/example/session/src/main/java/org/apache/iotdb/TableModelSubscriptionSessionExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/TableModelSubscriptionSessionExample.java
@@ -29,7 +29,9 @@ import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
import org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
+
+import org.apache.tsfile.read.query.dataset.ResultSet;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
@@ -131,14 +133,16 @@ public class TableModelSubscriptionSessionExample {
}
}
for (final SubscriptionMessage message : messages) {
- for (final SubscriptionSessionDataSet dataSet : message.getSessionDataSetsHandler()) {
- System.out.println(dataSet.getDatabaseName());
- System.out.println(dataSet.getTableName());
- System.out.println(dataSet.getColumnNames());
- System.out.println(dataSet.getColumnTypes());
- System.out.println(dataSet.getColumnCategories());
- while (dataSet.hasNext()) {
- System.out.println(dataSet.next());
+ for (final ResultSet dataSet : message.getResultSets()) {
+ final SubscriptionRecordHandler.SubscriptionResultSet record =
+ (SubscriptionRecordHandler.SubscriptionResultSet) dataSet;
+ System.out.println(record.getDatabaseName());
+ System.out.println(record.getTableName());
+ System.out.println(record.getColumnNames());
+ System.out.println(record.getColumnTypes());
+ System.out.println(record.getColumnCategories());
+ while (dataSet.next()) {
+ System.out.println("Time=" + dataSet.getLong(1));
}
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/SubscriptionTreeReaderTestUtils.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/SubscriptionTreeReaderTestUtils.java
new file mode 100644
index 00000000000..741e8baeb64
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/SubscriptionTreeReaderTestUtils.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.subscription.it;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
+import org.apache.tsfile.utils.Binary;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+
+public final class SubscriptionTreeReaderTestUtils {
+
+ private SubscriptionTreeReaderTestUtils() {
+ // utility class
+ }
+
+ public static QueryDataSetAdapter query(final ITsFileTreeReader reader, final Path path)
+ throws IOException {
+ return query(reader, Collections.singletonList(path));
+ }
+
+ public static QueryDataSetAdapter query(final ITsFileTreeReader reader, final List<Path> paths)
+ throws IOException {
+ final Set<String> devices = new LinkedHashSet<>();
+ final Set<String> measurements = new LinkedHashSet<>();
+ for (final Path path : paths) {
+ devices.add(path.getDeviceString());
+ measurements.add(path.getMeasurement());
+ }
+
+ final List<String> deviceList = new ArrayList<>(devices);
+ final List<String> measurementList = new ArrayList<>(measurements);
+ final List<Integer> selectedColumnIndexes = new ArrayList<>(paths.size());
+ for (final Path path : paths) {
+ final int deviceIndex = deviceList.indexOf(path.getDeviceString());
+ final int measurementIndex = measurementList.indexOf(path.getMeasurement());
+ selectedColumnIndexes.add(2 + deviceIndex * measurementList.size() + measurementIndex);
+ }
+
+ return new QueryDataSetAdapter(
+ reader.query(deviceList, measurementList, Long.MIN_VALUE, Long.MAX_VALUE),
+ Collections.unmodifiableList(new ArrayList<>(paths)),
+ selectedColumnIndexes);
+ }
+
+ public static final class QueryDataSetAdapter implements AutoCloseable {
+
+ private final ResultSet resultSet;
+
+ private final List<Path> paths;
+
+ private final List<Integer> selectedColumnIndexes;
+
+ private RowRecord nextRowRecord;
+
+ private boolean initialized;
+
+ private boolean hasNext;
+
+ private QueryDataSetAdapter(
+ final ResultSet resultSet,
+ final List<Path> paths,
+ final List<Integer> selectedColumnIndexes) {
+ this.resultSet = resultSet;
+ this.paths = paths;
+ this.selectedColumnIndexes = selectedColumnIndexes;
+ }
+
+ public boolean hasNext() throws IOException {
+ if (!initialized) {
+ advance();
+ }
+ return hasNext;
+ }
+
+ public RowRecord next() throws IOException {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final RowRecord current = nextRowRecord;
+ advance();
+ return current;
+ }
+
+ public List<Path> getPaths() {
+ return paths;
+ }
+
+ @Override
+ public void close() {
+ resultSet.close();
+ }
+
+ private void advance() throws IOException {
+ initialized = true;
+ hasNext = resultSet.next();
+ if (!hasNext) {
+ nextRowRecord = null;
+ return;
+ }
+
+ nextRowRecord = new RowRecord(resultSet.getLong(1), selectedColumnIndexes.size());
+ for (int i = 0; i < selectedColumnIndexes.size(); ++i) {
+ final int columnIndex = selectedColumnIndexes.get(i);
+ final TSDataType dataType = resultSet.getMetadata().getColumnType(columnIndex);
+ nextRowRecord.setField(
+ i,
+ resultSet.isNull(columnIndex) ? null : getValue(resultSet, columnIndex, dataType),
+ dataType);
+ }
+ }
+ }
+
+ private static Object getValue(
+ final ResultSet resultSet, final int columnIndex, final TSDataType dataType) {
+ switch (dataType) {
+ case BOOLEAN:
+ return resultSet.getBoolean(columnIndex);
+ case INT32:
+ return resultSet.getInt(columnIndex);
+ case INT64:
+ case TIMESTAMP:
+ return resultSet.getLong(columnIndex);
+ case FLOAT:
+ return resultSet.getFloat(columnIndex);
+ case DOUBLE:
+ return resultSet.getDouble(columnIndex);
+ case DATE:
+ return resultSet.getDate(columnIndex);
+ case TEXT:
+ case BLOB:
+ case STRING:
+ return new Binary(resultSet.getBinary(columnIndex));
+ default:
+ return resultSet.getString(columnIndex);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java
index 885087136e0..44afe26da9a 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/cluster/IoTDBSubscriptionRestartIT.java
@@ -35,10 +35,11 @@ import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.session.subscription.SubscriptionTreeSession;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.AbstractSubscriptionIT;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.utils.Pair;
import org.junit.After;
import org.junit.Assert;
@@ -202,10 +203,13 @@ public class IoTDBSubscriptionRestartIT extends AbstractSubscriptionIT {
continue;
}
for (final SubscriptionMessage message : messages) {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- final long timestamp = dataSet.next().getTimestamp();
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ final long timestamp =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .nextRecord()
+ .getTimestamp();
timestamps.put(new Pair<>(timestamp, consumerRef1.toString()), timestamp);
}
}
@@ -236,10 +240,13 @@ public class IoTDBSubscriptionRestartIT extends AbstractSubscriptionIT {
continue;
}
for (final SubscriptionMessage message : messages) {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- final long timestamp = dataSet.next().getTimestamp();
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ final long timestamp =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .nextRecord()
+ .getTimestamp();
timestamps.put(new Pair<>(timestamp, consumerRef2.toString()), timestamp);
}
}
@@ -354,10 +361,13 @@ public class IoTDBSubscriptionRestartIT extends AbstractSubscriptionIT {
continue;
}
for (final SubscriptionMessage message : messages) {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- final long timestamp = dataSet.next().getTimestamp();
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ final long timestamp =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .nextRecord()
+ .getTimestamp();
timestamps.put(timestamp, timestamp);
}
}
@@ -482,10 +492,13 @@ public class IoTDBSubscriptionRestartIT extends AbstractSubscriptionIT {
continue;
}
for (final SubscriptionMessage message : messages) {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- final long timestamp = dataSet.next().getTimestamp();
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ final long timestamp =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .nextRecord()
+ .getTimestamp();
timestamps.put(timestamp, timestamp);
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionTopicIT.java
index e60d8792303..f3c0dd29f46 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionTopicIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionTopicIT.java
@@ -37,11 +37,12 @@ import org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullCons
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
import org.apache.iotdb.subscription.it.dual.AbstractSubscriptionDualIT;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
@@ -401,15 +402,18 @@ public class IoTDBSubscriptionTopicIT extends AbstractSubscriptionDualIT {
continue;
}
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet : message.getSessionDataSetsHandler()) {
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
session.executeNonQueryStatement(
- "use " + Objects.requireNonNull(dataSet.getDatabaseName()));
- session.insert(dataSet.getTablet());
+ "use "
+ + Objects.requireNonNull(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getDatabaseName()));
+ session.insert(((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
}
break;
- case TS_FILE_HANDLER:
- final SubscriptionTsFileHandler tsFileHandler = message.getTsFileHandler();
+ case TS_FILE:
+ final SubscriptionTsFileHandler tsFileHandler = message.getTsFile();
session.executeNonQueryStatement(
"use " + Objects.requireNonNull(tsFileHandler.getDatabaseName()));
session.executeNonQueryStatement(
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionConsumerGroupIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionConsumerGroupIT.java
index 79485cfe7db..25cd576a518 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionConsumerGroupIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionConsumerGroupIT.java
@@ -33,15 +33,15 @@ import org.apache.iotdb.session.subscription.SubscriptionTreeSession;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.dual.AbstractSubscriptionDualIT;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.utils.Pair;
import org.junit.Assert;
import org.junit.Before;
@@ -1054,30 +1054,33 @@ public class IoTDBSubscriptionConsumerGroupIT extends AbstractSubscriptionDualIT
continue;
}
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
+ case RECORD_HANDLER:
{
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- final List<String> columnNameList = dataSet.getColumnNames();
- while (dataSet.hasNext()) {
- final RowRecord record = dataSet.next();
+ for (final ResultSet dataSet : message.getResultSets()) {
+ final List<String> columnNameList =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getColumnNames();
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ final RowRecord record =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .nextRecord();
insertRowRecordEnrichedByConsumerGroupId(
columnNameList, record.getTimestamp(), consumerGroupId);
}
}
break;
}
- case TS_FILE_HANDLER:
+ case TS_FILE:
{
- try (final TsFileReader tsFileReader =
- message.getTsFileHandler().openReader()) {
- final QueryDataSet dataSet =
- tsFileReader.query(
- QueryExpression.create(
- Arrays.asList(
- new Path("root.topic1", "s", true),
- new Path("root.topic2", "s", true)),
- null));
+ try (final ITsFileTreeReader tsFileReader =
+ message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataSet =
+ SubscriptionTreeReaderTestUtils.query(
+ tsFileReader,
+ Arrays.asList(
+ new Path("root.topic1", "s", true),
+ new Path("root.topic2", "s", true)));
while (dataSet.hasNext()) {
final RowRecord record = dataSet.next();
for (final Path path : dataSet.getPaths()) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTimePrecisionIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTimePrecisionIT.java
index 436ae19a612..b272f2321a0 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTimePrecisionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTimePrecisionIT.java
@@ -163,8 +163,7 @@ public class IoTDBSubscriptionTimePrecisionIT extends AbstractSubscriptionDualIT
final List<SubscriptionMessage> messages =
consumer.poll(IoTDBSubscriptionITConstant.POLL_TIMEOUT_MS);
for (final SubscriptionMessage message : messages) {
- for (final Iterator<Tablet> it =
- message.getSessionDataSetsHandler().tabletIterator();
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator();
it.hasNext(); ) {
final Tablet tablet = it.next();
session.insertTablet(tablet);
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTopicIT.java
index 7bb5c08b570..6c2f84b4d68 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTopicIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/treemodel/IoTDBSubscriptionTopicIT.java
@@ -34,15 +34,15 @@ import org.apache.iotdb.session.subscription.SubscriptionTreeSession;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.dual.AbstractSubscriptionDualIT;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.junit.Assert;
import org.junit.Before;
@@ -494,8 +494,7 @@ public class IoTDBSubscriptionTopicIT extends AbstractSubscriptionDualIT {
final List<SubscriptionMessage> messages =
consumer.poll(IoTDBSubscriptionITConstant.POLL_TIMEOUT_MS);
for (final SubscriptionMessage message : messages) {
- for (final Iterator<Tablet> it =
- message.getSessionDataSetsHandler().tabletIterator();
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator();
it.hasNext(); ) {
final Tablet tablet = it.next();
session.insertTablet(tablet);
@@ -650,22 +649,22 @@ public class IoTDBSubscriptionTopicIT extends AbstractSubscriptionDualIT {
continue;
}
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
dataSet.next();
rowCount.addAndGet(1);
}
}
break;
- case TS_FILE_HANDLER:
- try (final TsFileReader tsFileReader =
- message.getTsFileHandler().openReader()) {
+ case TS_FILE:
+ try (final ITsFileTreeReader tsFileReader =
+ message.getTsFile().openTreeReader()) {
final Path path = new Path("root.db.d1", "s1", true);
- final QueryDataSet dataSet =
- tsFileReader.query(
- QueryExpression.create(Collections.singletonList(path), null));
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataSet =
+ SubscriptionTreeReaderTestUtils.query(
+ tsFileReader, Collections.singletonList(path));
while (dataSet.hasNext()) {
dataSet.next();
rowCount.addAndGet(1);
@@ -945,15 +944,14 @@ public class IoTDBSubscriptionTopicIT extends AbstractSubscriptionDualIT {
continue;
}
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final Iterator<Tablet> it = message.getSessionDataSetsHandler().tabletIterator();
- it.hasNext(); ) {
+ case RECORD_HANDLER:
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator(); it.hasNext(); ) {
final Tablet tablet = it.next();
session.insertTablet(tablet);
}
break;
- case TS_FILE_HANDLER:
- final SubscriptionTsFileHandler tsFileHandler = message.getTsFileHandler();
+ case TS_FILE:
+ final SubscriptionTsFileHandler tsFileHandler = message.getTsFile();
session.executeNonQueryStatement(
String.format("load '%s'", tsFileHandler.getFile().getAbsolutePath()));
break;
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java
index fe66e3e9fa7..bed9721f3aa 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java
@@ -36,13 +36,13 @@ import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullC
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
import org.apache.iotdb.session.subscription.model.Subscription;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
@@ -134,9 +134,9 @@ public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT {
}
for (final SubscriptionMessage message : messages) {
int rowCountInOneMessage = 0;
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
dataSet.next();
rowCount.addAndGet(1);
rowCountInOneMessage++;
@@ -270,8 +270,7 @@ public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT {
message -> {
onReceiveCount.getAndIncrement();
message
- .getSessionDataSetsHandler()
- .tabletIterator()
+ .getRecordTabletIterator()
.forEachRemaining(tablet -> rowCount.addAndGet(tablet.getRowSize()));
return ConsumeResult.SUCCESS;
})
@@ -390,10 +389,13 @@ public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT {
final List<SubscriptionMessage> messages =
consumer.poll(IoTDBSubscriptionITConstant.POLL_TIMEOUT_MS);
for (final SubscriptionMessage message : messages) {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- timestampSum.getAndAdd(dataSet.next().getTimestamp());
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ timestampSum.getAndAdd(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .nextRecord()
+ .getTimestamp());
rowCount.addAndGet(1);
}
}
@@ -470,12 +472,12 @@ public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT {
.consumeListener(
message -> {
onReceiveCount.getAndIncrement();
- try (final TsFileReader tsFileReader = message.getTsFileHandler().openReader()) {
- final QueryDataSet dataSet =
- tsFileReader.query(
- QueryExpression.create(
- Collections.singletonList(new Path("root.db.d1", "s1", true)),
- null));
+ try (final ITsFileTreeReader tsFileReader =
+ message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataSet =
+ SubscriptionTreeReaderTestUtils.query(
+ tsFileReader,
+ Collections.singletonList(new Path("root.db.d1", "s1", true)));
while (dataSet.hasNext()) {
dataSet.next();
rowCount.addAndGet(1);
@@ -544,11 +546,15 @@ public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
+ for (final ResultSet dataSet : message.getResultSets()) {
+ try {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ dataSet.next();
+ rowCount.addAndGet(1);
+ }
+ } catch (final IOException e) {
+ throw new RuntimeException(e);
}
}
return ConsumeResult.SUCCESS;
@@ -609,11 +615,15 @@ public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
+ for (final ResultSet dataSet : message.getResultSets()) {
+ try {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ dataSet.next();
+ rowCount.addAndGet(1);
+ }
+ } catch (final IOException e) {
+ throw new RuntimeException(e);
}
}
return ConsumeResult.SUCCESS;
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionDataTypeIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionDataTypeIT.java
index 1dd20785361..5b54a520e9c 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionDataTypeIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionDataTypeIT.java
@@ -28,16 +28,16 @@ import org.apache.iotdb.session.subscription.SubscriptionTreeSession;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.write.record.Tablet;
import org.junit.Assert;
@@ -308,28 +308,38 @@ public class IoTDBSubscriptionDataTypeIT extends AbstractSubscriptionLocalIT {
continue;
}
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- final RowRecord record = dataSet.next();
- Assert.assertEquals(type.toString(), dataSet.getColumnTypes().get(1));
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .hasNext()) {
+ final RowRecord record =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .nextRecord();
+ Assert.assertEquals(
+ type.toString(),
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getColumnTypes()
+ .get(1));
Assert.assertEquals(type, record.getFields().get(0).getDataType());
- Assert.assertEquals(expectedData, getValue(type, dataSet.getTablet()));
+ Assert.assertEquals(
+ expectedData,
+ getValue(
+ type,
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet()));
Assert.assertEquals(
expectedData, record.getFields().get(0).getObjectValue(type));
rowCount.addAndGet(1);
}
}
break;
- case TS_FILE_HANDLER:
- try (final TsFileReader tsFileReader =
- message.getTsFileHandler().openReader()) {
- final QueryDataSet dataSet =
- tsFileReader.query(
- QueryExpression.create(
- Collections.singletonList(new Path("root.db.d1", "s1", true)),
- null));
+ case TS_FILE:
+ try (final ITsFileTreeReader tsFileReader =
+ message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataSet =
+ SubscriptionTreeReaderTestUtils.query(
+ tsFileReader,
+ Collections.singletonList(new Path("root.db.d1", "s1", true)));
while (dataSet.hasNext()) {
final RowRecord record = dataSet.next();
Assert.assertEquals(type, record.getFields().get(0).getDataType());
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java
index f3f95642917..43c3ea7aebf 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java
@@ -31,9 +31,11 @@ import org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePushCons
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePushConsumerBuilder;
import org.apache.iotdb.session.subscription.model.Subscription;
import org.apache.iotdb.session.subscription.model.Topic;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.local.AbstractSubscriptionLocalIT;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
@@ -195,13 +197,7 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
- }
- }
+ countRowsFromMessage(message, rowCount);
return ConsumeResult.SUCCESS;
})
.build();
@@ -216,13 +212,7 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
- }
- }
+ countRowsFromMessage(message, rowCount);
return ConsumeResult.SUCCESS;
})
.build();
@@ -237,13 +227,7 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
- }
- }
+ countRowsFromMessage(message, rowCount);
return ConsumeResult.SUCCESS;
})
.build()) {
@@ -291,13 +275,7 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
- }
- }
+ countRowsFromMessage(message, rowCount);
return ConsumeResult.SUCCESS;
})
.build();
@@ -312,13 +290,7 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
- }
- }
+ countRowsFromMessage(message, rowCount);
return ConsumeResult.SUCCESS;
})
.build();
@@ -333,13 +305,7 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT {
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
- while (dataSet.hasNext()) {
- dataSet.next();
- rowCount.addAndGet(1);
- }
- }
+ countRowsFromMessage(message, rowCount);
return ConsumeResult.SUCCESS;
})
.build()) {
@@ -353,6 +319,20 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT {
}
}
+ private static void countRowsFromMessage(
+ final SubscriptionMessage message, final AtomicInteger rowCount) {
+ for (final ResultSet dataSet : message.getResultSets()) {
+ try {
+ while (((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).hasNext()) {
+ dataSet.next();
+ rowCount.addAndGet(1);
+ }
+ } catch (final Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
@Ignore
@Test
public void testTablePermission() {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/IoTDBSubscriptionSharingIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/IoTDBSubscriptionSharingIT.java
index c0f4c9cb987..d33adcbdcdb 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/IoTDBSubscriptionSharingIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/IoTDBSubscriptionSharingIT.java
@@ -31,15 +31,15 @@ import org.apache.iotdb.session.subscription.SubscriptionTreeSession;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.AbstractSubscriptionTripleIT;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.pushconsumer.multi.IoTDBMultiGroupVsMultiConsumerIT;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -199,13 +199,12 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumeListener(
message -> {
try {
- try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
- final QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ try (final ITsFileTreeReader reader = message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount00.addAndGet(1);
dataset.next();
@@ -226,13 +225,12 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumeListener(
message -> {
try {
- try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
- final QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ try (final ITsFileTreeReader reader = message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount10.addAndGet(1);
dataset.next();
@@ -252,10 +250,13 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumerGroupId("push_group_id_1")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- receiver1.getSessionConnection().insertTablet(dataSet.getTablet());
+ receiver1
+ .getSessionConnection()
+ .insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (final StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
@@ -271,10 +272,13 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumerGroupId("push_group_id_1")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- receiver2.getSessionConnection().insertTablet(dataSet.getTablet());
+ receiver2
+ .getSessionConnection()
+ .insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (final StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
@@ -290,10 +294,13 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumerGroupId("push_group_id_2")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- receiver1.getSessionConnection().insertTablet(dataSet.getTablet());
+ receiver1
+ .getSessionConnection()
+ .insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (final StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
@@ -309,10 +316,13 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumerGroupId("push_group_id_2")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- receiver2.getSessionConnection().insertTablet(dataSet.getTablet());
+ receiver2
+ .getSessionConnection()
+ .insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (final StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
@@ -328,10 +338,13 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumerGroupId("push_group_id_2")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- receiver1.getSessionConnection().insertTablet(dataSet.getTablet());
+ receiver1
+ .getSessionConnection()
+ .insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (final StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
@@ -350,24 +363,27 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
final short messageType = message.getMessageType();
if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- receiver1.getSessionConnection().insertTablet(dataSet.getTablet());
+ receiver1
+ .getSessionConnection()
+ .insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (final StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
}
break;
- case TS_FILE_HANDLER:
- try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
- final QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ case TS_FILE:
+ try (final ITsFileTreeReader reader =
+ message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount70.addAndGet(1);
dataset.next();
@@ -389,13 +405,12 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
.consumerGroupId("push_group_id_3")
.consumeListener(
message -> {
- try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
- final QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "6.d_0", "s_0", true)),
- null));
+ try (final ITsFileTreeReader reader = message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "6.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount6.addAndGet(1);
dataset.next();
@@ -417,24 +432,27 @@ public class IoTDBSubscriptionSharingIT extends AbstractSubscriptionTripleIT {
final short messageType = message.getMessageType();
if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- receiver2.getSessionConnection().insertTablet(dataSet.getTablet());
+ receiver2
+ .getSessionConnection()
+ .insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (final StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
}
break;
- case TS_FILE_HANDLER:
- try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
- final QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ case TS_FILE:
+ try (final ITsFileTreeReader reader =
+ message.getTsFile().openTreeReader()) {
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount90.addAndGet(1);
dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/AbstractSubscriptionTreeRegressionIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/AbstractSubscriptionTreeRegressionIT.java
index 15e317fb3a0..078c190c28d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/AbstractSubscriptionTreeRegressionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/AbstractSubscriptionTreeRegressionIT.java
@@ -30,14 +30,13 @@ import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullC
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.WrappedVoidSupplier;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.AbstractSubscriptionTripleIT;
import org.apache.thrift.TException;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.junit.After;
@@ -300,8 +299,7 @@ public abstract class AbstractSubscriptionTreeRegressionIT extends AbstractSubsc
break;
}
for (final SubscriptionMessage message : messages) {
- for (final Iterator<Tablet> it = message.getSessionDataSetsHandler().tabletIterator();
- it.hasNext(); ) {
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator(); it.hasNext(); ) {
final Tablet tablet = it.next();
LOGGER.info(
"Inserting a tablet, device {}, times {}, measurements {}",
@@ -343,12 +341,13 @@ public abstract class AbstractSubscriptionTreeRegressionIT extends AbstractSubsc
for (final SubscriptionMessage message : messages) {
onReceived.incrementAndGet();
// System.out.println(FORMAT.format(new Date()) + " onReceived=" + onReceived.get());
- final SubscriptionTsFileHandler tsFileHandler = message.getTsFileHandler();
- try (final TsFileReader tsFileReader = tsFileHandler.openReader()) {
+ final SubscriptionTsFileHandler tsFileHandler = message.getTsFile();
+ try (final ITsFileTreeReader tsFileReader = tsFileHandler.openTreeReader()) {
for (int i = 0; i < devices.size(); i++) {
final Path path = new Path(devices.get(i), "s_0", true);
- final QueryDataSet dataSet =
- tsFileReader.query(QueryExpression.create(Collections.singletonList(path), null));
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataSet =
+ SubscriptionTreeReaderTestUtils.query(
+ tsFileReader, Collections.singletonList(path));
while (dataSet.hasNext()) {
RowRecord next = dataSet.next();
rowCounts.get(i).addAndGet(1);
@@ -382,8 +381,7 @@ public abstract class AbstractSubscriptionTreeRegressionIT extends AbstractSubsc
session_src.executeNonQueryStatement("flush");
}
for (final SubscriptionMessage message : messages) {
- for (final Iterator<Tablet> it = message.getSessionDataSetsHandler().tabletIterator();
- it.hasNext(); ) {
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator(); it.hasNext(); ) {
final Tablet tablet = it.next();
session.insertTablet(tablet);
}
@@ -411,13 +409,13 @@ public abstract class AbstractSubscriptionTreeRegressionIT extends AbstractSubsc
session_src.executeNonQueryStatement("flush");
}
for (final SubscriptionMessage message : messages) {
- final SubscriptionTsFileHandler tsFileHandler = message.getTsFileHandler();
- try (final TsFileReader tsFileReader = tsFileHandler.openReader()) {
+ final SubscriptionTsFileHandler tsFileHandler = message.getTsFile();
+ try (final ITsFileTreeReader tsFileReader = tsFileHandler.openTreeReader()) {
for (int i = 0; i < devices.size(); i++) {
final Path path = new Path(devices.get(i), "s_0", true);
- final QueryDataSet dataSet =
- tsFileReader.query(
- QueryExpression.create(Collections.singletonList(path), null));
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataSet =
+ SubscriptionTreeReaderTestUtils.query(
+ tsFileReader, Collections.singletonList(path));
while (dataSet.hasNext()) {
dataSet.next();
counters.get(i).addAndGet(1);
@@ -462,13 +460,13 @@ public abstract class AbstractSubscriptionTreeRegressionIT extends AbstractSubsc
}
for (final SubscriptionMessage message : messages) {
onReceived.incrementAndGet();
- final SubscriptionTsFileHandler tsFileHandler = message.getTsFileHandler();
- try (final TsFileReader tsFileReader = tsFileHandler.openReader()) {
+ final SubscriptionTsFileHandler tsFileHandler = message.getTsFile();
+ try (final ITsFileTreeReader tsFileReader = tsFileHandler.openTreeReader()) {
for (int i = 0; i < devices.size(); i++) {
final Path path = new Path(devices.get(i), "s_0", true);
- final QueryDataSet dataSet =
- tsFileReader.query(
- QueryExpression.create(Collections.singletonList(path), null));
+ final SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataSet =
+ SubscriptionTreeReaderTestUtils.query(
+ tsFileReader, Collections.singletonList(path));
while (dataSet.hasNext()) {
dataSet.next();
counters.get(i).addAndGet(1);
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBDefaultTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBDefaultTsfilePushConsumerIT.java
index 35582f7feae..42939f37258 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBDefaultTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBDefaultTsfilePushConsumerIT.java
@@ -27,15 +27,14 @@ import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -159,12 +158,11 @@ public class IoTDBDefaultTsfilePushConsumerIT extends AbstractSubscriptionTreeRe
System.out.println(
FORMAT.format(new Date()) + " ######## onReceived: " + onReceiveCount.get());
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
for (int i = 0; i < deviceCount; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBRootDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBRootDatasetPushConsumerIT.java
index b3ff0b8f4c7..4ff28f1c0e5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBRootDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/auto_create_db/IoTDBRootDatasetPushConsumerIT.java
@@ -26,11 +26,12 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -129,10 +130,10 @@ public class IoTDBRootDatasetPushConsumerIT extends AbstractSubscriptionTreeRegr
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/mix/IoTDBPushConsumerPullConsumerWith1TopicShareProcessMixIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/mix/IoTDBPushConsumerPullConsumerWith1TopicShareProcessMixIT.java
index a99cfb855b3..9501862ba8b 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/mix/IoTDBPushConsumerPullConsumerWith1TopicShareProcessMixIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/mix/IoTDBPushConsumerPullConsumerWith1TopicShareProcessMixIT.java
@@ -27,13 +27,14 @@ import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -148,10 +149,10 @@ public class IoTDBPushConsumerPullConsumerWith1TopicShareProcessMixIT
.ackStrategy(AckStrategy.BEFORE_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPullConsumerIT.java
index aeea7b5f41e..d8f855059fb 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPullConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPullConsumerIT.java
@@ -29,17 +29,16 @@ import org.apache.iotdb.rpc.subscription.exception.SubscriptionIdentifierSemanti
import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -506,11 +505,10 @@ public class IoTDBTestParamPullConsumerIT extends AbstractSubscriptionTreeRegres
break;
}
for (final SubscriptionMessage message : messages) {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device, "s_0", true)), null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(new Path(device, "s_0", true)));
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
@@ -552,11 +550,10 @@ public class IoTDBTestParamPullConsumerIT extends AbstractSubscriptionTreeRegres
break;
}
for (final SubscriptionMessage message : messages) {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device, "s_0", true)), null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(new Path(device, "s_0", true)));
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPushConsumerIT.java
index 979524b8c90..baf129e19b3 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/param/IoTDBTestParamPushConsumerIT.java
@@ -30,12 +30,13 @@ import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalEx
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -144,11 +145,13 @@ public class IoTDBTestParamPushConsumerIT extends AbstractSubscriptionTreeRegres
.ackStrategy(AckStrategy.BEFORE_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- // System.out.println("#### " + dataSet.getTablet().rowSize);
- session_dest.insertTablet(dataSet.getTablet());
+ // System.out.println("#### " +
+ // ((SubscriptionRecordHandler.SubscriptionResultSet)
+ // dataSet).getTablet().rowSize);
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -182,10 +185,10 @@ public class IoTDBTestParamPushConsumerIT extends AbstractSubscriptionTreeRegres
.autoPollIntervalMs(10)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitFalseDataSetPullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitFalseDataSetPullConsumerIT.java
index c7d49aed906..50946a631f5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitFalseDataSetPullConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitFalseDataSetPullConsumerIT.java
@@ -137,8 +137,7 @@ public class IoTDBTestAutoCommitFalseDataSetPullConsumerIT
break;
}
for (final SubscriptionMessage message : messages) {
- for (final Iterator<Tablet> it = message.getSessionDataSetsHandler().tabletIterator();
- it.hasNext(); ) {
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator(); it.hasNext(); ) {
final Tablet tablet = it.next();
LOGGER.info(
"Inserting a tablet, device {}, times {}, measurements {}",
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitTrueDataSetPullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitTrueDataSetPullConsumerIT.java
index 7cd749a91c0..09682650c53 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitTrueDataSetPullConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitTrueDataSetPullConsumerIT.java
@@ -128,8 +128,7 @@ public class IoTDBTestAutoCommitTrueDataSetPullConsumerIT
break;
}
for (final SubscriptionMessage message : messages) {
- for (final Iterator<Tablet> it = message.getSessionDataSetsHandler().tabletIterator();
- it.hasNext(); ) {
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator(); it.hasNext(); ) {
final Tablet tablet = it.next();
session.insertTablet(tablet);
System.out.println(
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java
index 7fd364c1f96..4bbc33d1ea6 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java
@@ -29,17 +29,16 @@ import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
import org.apache.iotdb.subscription.it.Retry;
import org.apache.iotdb.subscription.it.RetryRule;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -189,9 +188,8 @@ public class IoTDBOneConsumerMultiTopicsMixIT extends AbstractSubscriptionTreeRe
final short messageType = message.getMessageType();
if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final Iterator<Tablet> it =
- message.getSessionDataSetsHandler().tabletIterator();
+ case RECORD_HANDLER:
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator();
it.hasNext(); ) {
final Tablet tablet = it.next();
try {
@@ -205,14 +203,12 @@ public class IoTDBOneConsumerMultiTopicsMixIT extends AbstractSubscriptionTreeRe
}
}
break;
- case TS_FILE_HANDLER:
+ case TS_FILE:
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device, "s_1", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(new Path(device, "s_1", true)));
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
@@ -238,9 +234,8 @@ public class IoTDBOneConsumerMultiTopicsMixIT extends AbstractSubscriptionTreeRe
final short messageType = message.getMessageType();
if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final Iterator<Tablet> it =
- message.getSessionDataSetsHandler().tabletIterator();
+ case RECORD_HANDLER:
+ for (final Iterator<Tablet> it = message.getRecordTabletIterator();
it.hasNext(); ) {
final Tablet tablet = it.next();
try {
@@ -254,14 +249,12 @@ public class IoTDBOneConsumerMultiTopicsMixIT extends AbstractSubscriptionTreeRe
}
}
break;
- case TS_FILE_HANDLER:
+ case TS_FILE:
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device, "s_1", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(new Path(device, "s_1", true)));
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeDataSetIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeDataSetIT.java
index afde2f7376a..d446da83a51 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeDataSetIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeDataSetIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -130,10 +131,10 @@ public class IoTDBTestPushConsumeDataSetIT extends AbstractSubscriptionTreeRegre
.fileSaveDir("target/iotdb-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeNoTargetDirTsfileIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeNoTargetDirTsfileIT.java
index d041647d23e..64a66499b42 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeNoTargetDirTsfileIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeNoTargetDirTsfileIT.java
@@ -26,16 +26,15 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -138,10 +137,11 @@ public class IoTDBTestPushConsumeNoTargetDirTsfileIT extends AbstractSubscriptio
message -> {
onReceiveCount.incrementAndGet();
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
Path path = new Path(device, "s_0", true);
- QueryDataSet dataset =
- reader.query(QueryExpression.create(Collections.singletonList(path), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(path));
while (dataset.hasNext()) {
rowCount.addAndGet(1);
dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeTsfileIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeTsfileIT.java
index 6445e53e9dc..bb9946c88de 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeTsfileIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/format/IoTDBTestPushConsumeTsfileIT.java
@@ -26,17 +26,16 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -137,10 +136,11 @@ public class IoTDBTestPushConsumeTsfileIT extends AbstractSubscriptionTreeRegres
message -> {
onReceiveCount.incrementAndGet();
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
Path path = new Path(device, "s_0", true);
- QueryDataSet dataset =
- reader.query(QueryExpression.create(Collections.singletonList(path), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(path));
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java
index ad5c43290e2..c30e9d8493d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java
@@ -27,13 +27,14 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -168,10 +169,10 @@ public class IoTDBLooseAllTsDatasetPushConsumerIT extends AbstractSubscriptionTr
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerSnapshotIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerSnapshotIT.java
index 40391992d02..177c84de88c 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerSnapshotIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerSnapshotIT.java
@@ -27,13 +27,14 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -169,10 +170,10 @@ public class IoTDBLooseAllTsDatasetPushConsumerSnapshotIT
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java
index 7978c597d4c..91fb1a281e5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java
@@ -27,17 +27,16 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -174,13 +173,12 @@ public class IoTDBLooseAllTsfilePushConsumerIT extends AbstractSubscriptionTreeR
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
for (int i = 0; i < 2; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(paths.get(i), "s_0", true)),
- null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(new Path(paths.get(i), "s_0", true)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java
index 439924b71c1..c647160493a 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java
@@ -27,17 +27,16 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -173,12 +172,11 @@ public class IoTDBPathLooseDeviceTsfilePushConsumerIT extends AbstractSubscripti
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
for (int i = 0; i < 2; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java
index 8de488366f8..3e144c137fd 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java
@@ -27,13 +27,14 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -164,10 +165,10 @@ public class IoTDBPathLooseTsDatasetPushConsumerIT extends AbstractSubscriptionT
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java
index ea70ca9628c..b8c438bfb9d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java
@@ -27,17 +27,16 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -160,12 +159,13 @@ public class IoTDBPathLooseTsfilePushConsumerIT extends AbstractSubscriptionTree
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
List<Path> paths = new ArrayList<>(2);
for (int i = 0; i < 2; i++) {
paths.add(new Path(device, "s_" + i, true));
}
- QueryDataSet dataset = reader.query(QueryExpression.create(paths, null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(reader, paths);
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java
index f66714def3d..5a2a5d6a351 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java
@@ -27,13 +27,14 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -160,10 +161,10 @@ public class IoTDBPathTsLooseDatasetPushConsumerIT extends AbstractSubscriptionT
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java
index 239312a1101..ecbd661e5e0 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java
@@ -27,13 +27,14 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -166,10 +167,10 @@ public class IoTDBTimeLooseTsDatasetPushConsumerIT extends AbstractSubscriptionT
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java
index f4ad3cddd02..0311bc0a0b5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java
@@ -27,17 +27,16 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -178,12 +177,11 @@ public class IoTDBTimeLooseTsTsfilePushConsumerIT extends AbstractSubscriptionTr
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
for (int i = 0; i < 4; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java
index d0257d72cbb..53207ee7fe6 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java
@@ -27,17 +27,16 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -162,12 +161,13 @@ public class IoTDBTimeLooseTsfilePushConsumerIT extends AbstractSubscriptionTree
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
List<Path> paths = new ArrayList<>(2);
for (int i = 0; i < 2; i++) {
paths.add(new Path(device, "s_" + i, true));
}
- QueryDataSet dataset = reader.query(QueryExpression.create(paths, null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(reader, paths);
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java
index 66172c0013b..4f358ddb8c4 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java
@@ -27,13 +27,14 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -163,10 +164,10 @@ public class IoTDBTimeTsLooseDatasetPushConsumerIT extends AbstractSubscriptionT
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternDatasetPushConsumerIT.java
index b42cd494dab..dab48a7180a 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternDatasetPushConsumerIT.java
@@ -27,7 +27,7 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
import org.apache.iotdb.subscription.it.Retry;
import org.apache.iotdb.subscription.it.RetryRule;
@@ -37,6 +37,7 @@ import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -169,10 +170,10 @@ public class IoTDBSnapshotTSPatternDatasetPushConsumerIT
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternTsfilePushConsumerIT.java
index 2905e39cf50..6fb60c944a2 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/mode/IoTDBSnapshotTSPatternTsfilePushConsumerIT.java
@@ -29,17 +29,16 @@ import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
import org.apache.iotdb.subscription.it.Retry;
import org.apache.iotdb.subscription.it.RetryRule;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -194,12 +193,11 @@ public class IoTDBSnapshotTSPatternTsfilePushConsumerIT
onReceiveCount.incrementAndGet();
System.out.println("onReceiveCount=" + onReceiveCount.get());
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
for (int i = 0; i < 4; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessDataSetIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessDataSetIT.java
index 3e5f3755286..583feb244ba 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessDataSetIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessDataSetIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -153,10 +154,10 @@ public class IoTDBConsumer2With1TopicShareProcessDataSetIT
.ackStrategy(AckStrategy.BEFORE_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -177,10 +178,10 @@ public class IoTDBConsumer2With1TopicShareProcessDataSetIT
.ackStrategy(AckStrategy.BEFORE_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest2.insertTablet(dataSet.getTablet());
+ session_dest2.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessTsfileIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessTsfileIT.java
index 721a346835c..9bed540d613 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessTsfileIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBConsumer2With1TopicShareProcessTsfileIT.java
@@ -26,17 +26,16 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -152,12 +151,13 @@ public class IoTDBConsumer2With1TopicShareProcessTsfileIT
.consumeListener(
message -> {
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
List<Path> paths = new ArrayList<>(2);
for (int i = 0; i < 2; i++) {
paths.add(new Path(device, "s_" + i, true));
}
- QueryDataSet dataset = reader.query(QueryExpression.create(paths, null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(reader, paths);
while (dataset.hasNext()) {
rowCount1.addAndGet(1);
RowRecord next = dataset.next();
@@ -183,12 +183,13 @@ public class IoTDBConsumer2With1TopicShareProcessTsfileIT
.consumeListener(
message -> {
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
List<Path> paths = new ArrayList<>(2);
for (int i = 0; i < 2; i++) {
paths.add(new Path(device, "s_" + i, true));
}
- QueryDataSet dataset = reader.query(QueryExpression.create(paths, null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(reader, paths);
while (dataset.hasNext()) {
rowCount2.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBMultiGroupVsMultiConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBMultiGroupVsMultiConsumerIT.java
index 73f06d272d6..5d2e10ac453 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBMultiGroupVsMultiConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBMultiGroupVsMultiConsumerIT.java
@@ -26,17 +26,17 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -211,13 +211,12 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.consumeListener(
message -> {
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount00.addAndGet(1);
RowRecord next = dataset.next();
@@ -242,13 +241,12 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.consumeListener(
message -> {
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount10.addAndGet(1);
RowRecord next = dataset.next();
@@ -273,10 +271,10 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -295,10 +293,10 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest2.insertTablet(dataSet.getTablet());
+ session_dest2.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -317,10 +315,10 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -339,10 +337,10 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest2.insertTablet(dataSet.getTablet());
+ session_dest2.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -361,10 +359,10 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -386,11 +384,12 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
final short messageType = message.getMessageType();
if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -398,15 +397,14 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
}
}
break;
- case TS_FILE_HANDLER:
+ case TS_FILE:
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount70.addAndGet(1);
RowRecord next = dataset.next();
@@ -433,13 +431,12 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
.consumeListener(
message -> {
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "6.d_0", "s_0", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "6.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount6.addAndGet(1);
RowRecord next = dataset.next();
@@ -466,11 +463,12 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
final short messageType = message.getMessageType();
if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest2.insertTablet(dataSet.getTablet());
+ session_dest2.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -478,15 +476,14 @@ public class IoTDBMultiGroupVsMultiConsumerIT extends AbstractSubscriptionTreeRe
}
}
break;
- case TS_FILE_HANDLER:
+ case TS_FILE:
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(
- new Path(databasePrefix + "0.d_0", "s_0", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(
+ new Path(databasePrefix + "0.d_0", "s_0", true)));
while (dataset.hasNext()) {
rowCount90.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsDatasetIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsDatasetIT.java
index f23e66a6369..dc90fc2aaba 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsDatasetIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsDatasetIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -154,10 +155,10 @@ public class IoTDBOneConsumerMultiTopicsDatasetIT extends AbstractSubscriptionTr
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java
index 92baad95f3c..31fd010c4b7 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsMixIT.java
@@ -27,20 +27,20 @@ import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.Retry;
import org.apache.iotdb.subscription.it.RetryRule;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -173,11 +173,12 @@ public class IoTDBOneConsumerMultiTopicsMixIT extends AbstractSubscriptionTreeRe
final short messageType = message.getMessageType();
if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
switch (SubscriptionMessageType.valueOf(messageType)) {
- case SESSION_DATA_SETS_HANDLER:
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ case RECORD_HANDLER:
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet)
+ .getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
@@ -185,14 +186,12 @@ public class IoTDBOneConsumerMultiTopicsMixIT extends AbstractSubscriptionTreeRe
}
}
break;
- case TS_FILE_HANDLER:
+ case TS_FILE:
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device, "s_0", true)),
- null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(new Path(device, "s_0", true)));
while (dataset.hasNext()) {
rowCount1.addAndGet(1);
RowRecord next = dataset.next();
@@ -200,10 +199,9 @@ public class IoTDBOneConsumerMultiTopicsMixIT extends AbstractSubscriptionTreeRe
device + ":" + next.getTimestamp() + "," + next.getFields());
}
dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device2, "s_0", true)),
- null));
+ SubscriptionTreeReaderTestUtils.query(
+ reader,
+ Collections.singletonList(new Path(device2, "s_0", true)));
while (dataset.hasNext()) {
rowCount2.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsTsfileIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsTsfileIT.java
index 10f66c2ebd0..74bce3f9669 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsTsfileIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/multi/IoTDBOneConsumerMultiTopicsTsfileIT.java
@@ -29,17 +29,16 @@ import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushC
import org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant;
import org.apache.iotdb.subscription.it.Retry;
import org.apache.iotdb.subscription.it.RetryRule;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -175,11 +174,10 @@ public class IoTDBOneConsumerMultiTopicsTsfileIT extends AbstractSubscriptionTre
.consumeListener(
message -> {
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device, "s_0", true)), null));
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(new Path(device, "s_0", true)));
while (dataset.hasNext()) {
rowCount1.addAndGet(1);
RowRecord next = dataset.next();
@@ -187,9 +185,8 @@ public class IoTDBOneConsumerMultiTopicsTsfileIT extends AbstractSubscriptionTre
device + ":" + next.getTimestamp() + "," + next.getFields());
}
dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(new Path(device2, "s_0", true)), null));
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(new Path(device2, "s_0", true)));
while (dataset.hasNext()) {
rowCount2.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternDatasetPushConsumerIT.java
index b7e192ff6bc..3e3c44e219f 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternDatasetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -149,10 +150,10 @@ public class IoTDBDBPatternDatasetPushConsumerIT extends AbstractSubscriptionTre
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternTsfilePushConsumerIT.java
index 1d22dbaa260..30d907df8cc 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDBPatternTsfilePushConsumerIT.java
@@ -26,16 +26,15 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -165,7 +164,7 @@ public class IoTDBDBPatternTsfilePushConsumerIT extends AbstractSubscriptionTree
message -> {
onReceiveCount.incrementAndGet();
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
Path path_d0s0 = new Path(device, "s_0", true);
Path path_d0s1 = new Path(device, "s_1", true);
Path path_d1s0 = new Path(database + ".d_1", "s_0", true);
@@ -176,10 +175,9 @@ public class IoTDBDBPatternTsfilePushConsumerIT extends AbstractSubscriptionTree
paths.add(path_d1s0);
paths.add(path_other_d2);
for (int i = 0; i < 4; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDefaultPatternTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDefaultPatternTsfilePushConsumerIT.java
index 85104349658..a3c38507245 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDefaultPatternTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDefaultPatternTsfilePushConsumerIT.java
@@ -26,16 +26,15 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -164,7 +163,7 @@ public class IoTDBDefaultPatternTsfilePushConsumerIT extends AbstractSubscriptio
message -> {
onReceiveCount.incrementAndGet();
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
Path path_d0s0 = new Path(device, "s_0", true);
Path path_d0s1 = new Path(device, "s_1", true);
Path path_d1s0 = new Path(database + ".d_1", "s_0", true);
@@ -175,10 +174,9 @@ public class IoTDBDefaultPatternTsfilePushConsumerIT extends AbstractSubscriptio
paths.add(path_d1s0);
paths.add(path_other_d2);
for (int i = 0; i < 4; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternDatasetPushConsumerIT.java
index 34c798df9f9..9e29aabb587 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternDatasetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -148,10 +149,10 @@ public class IoTDBDevicePatternDatasetPushConsumerIT extends AbstractSubscriptio
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternTsfilePushConsumerIT.java
index bd3af70afc1..83a2e9072df 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBDevicePatternTsfilePushConsumerIT.java
@@ -26,17 +26,16 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -167,7 +166,7 @@ public class IoTDBDevicePatternTsfilePushConsumerIT extends AbstractSubscription
message -> {
onReceiveCount.incrementAndGet();
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
Path path_d0s0 = new Path(device, "s_0", true);
Path path_d0s1 = new Path(device, "s_1", true);
Path path_d1s0 = new Path(database + ".d_1", "s_0", true);
@@ -178,10 +177,9 @@ public class IoTDBDevicePatternTsfilePushConsumerIT extends AbstractSubscription
paths.add(path_d1s0);
paths.add(path_other_d2);
for (int i = 0; i < 4; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBRootPatternDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBRootPatternDatasetPushConsumerIT.java
index 7746d3b49b0..ff7b1cd4fb3 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBRootPatternDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBRootPatternDatasetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -149,10 +150,10 @@ public class IoTDBRootPatternDatasetPushConsumerIT extends AbstractSubscriptionT
.fileSaveDir("target/push-subscription")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternDatasetPushConsumerIT.java
index bc91825a4d4..65f73a1b9c8 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternDatasetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -148,10 +149,10 @@ public class IoTDBTSPatternDatasetPushConsumerIT extends AbstractSubscriptionTre
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- session_dest.insertTablet(dataSet.getTablet());
+ session_dest.insertTablet(
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternTsfilePushConsumerIT.java
index 4af5cc8c892..e2774b053c5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/pattern/IoTDBTSPatternTsfilePushConsumerIT.java
@@ -26,16 +26,15 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -166,7 +165,7 @@ public class IoTDBTSPatternTsfilePushConsumerIT extends AbstractSubscriptionTree
onReceiveCount.incrementAndGet();
System.out.println("onReceiveCount=" + onReceiveCount.get());
try {
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
Path path_d0s0 = new Path(device, "s_0", true);
Path path_d0s1 = new Path(device, "s_1", true);
Path path_d1s0 = new Path(database + ".d_1", "s_0", true);
@@ -177,10 +176,9 @@ public class IoTDBTSPatternTsfilePushConsumerIT extends AbstractSubscriptionTree
paths.add(path_d1s0);
paths.add(path_other_d2);
for (int i = 0; i < 4; i++) {
- QueryDataSet dataset =
- reader.query(
- QueryExpression.create(
- Collections.singletonList(paths.get(i)), null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(
+ reader, Collections.singletonList(paths.get(i)));
while (dataset.hasNext()) {
rowCounts.get(i).addAndGet(1);
dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootDatasetPushConsumerIT.java
index e7b5cbe560a..2ed8596fff1 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootDatasetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -132,10 +133,10 @@ public class IoTDBHistoryRootDatasetPushConsumerIT extends AbstractSubscriptionT
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- Tablet tablet = dataSet.getTablet();
+ Tablet tablet =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet();
session_dest.insertTablet(tablet);
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootTsFilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootTsFilePushConsumerIT.java
index fea7eff45fa..25ae05db46f 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootTsFilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBHistoryRootTsFilePushConsumerIT.java
@@ -26,17 +26,16 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -140,12 +139,13 @@ public class IoTDBHistoryRootTsFilePushConsumerIT extends AbstractSubscriptionTr
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
List<Path> paths = new ArrayList<>(2);
for (int i = 0; i < 2; i++) {
paths.add(new Path(device, "s_" + i, true));
}
- QueryDataSet dataset = reader.query(QueryExpression.create(paths, null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(reader, paths);
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBDatasetPushConsumerIT.java
index c2ed24c5d4a..31bd8a581ff 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBDatasetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBDatasetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -125,10 +126,10 @@ public class IoTDBRealTimeDBDatasetPushConsumerIT extends AbstractSubscriptionTr
.fileSaveDir("target")
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- Tablet tablet = dataSet.getTablet();
+ Tablet tablet =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet();
session_dest.insertTablet(tablet);
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBTsfilePushConsumerIT.java
index 81343a1acff..89276531f65 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBRealTimeDBTsfilePushConsumerIT.java
@@ -26,17 +26,16 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -135,12 +134,13 @@ public class IoTDBRealTimeDBTsfilePushConsumerIT extends AbstractSubscriptionTre
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
List<Path> paths = new ArrayList<>(2);
for (int i = 0; i < 2; i++) {
paths.add(new Path(device, "s_" + i, true));
}
- QueryDataSet dataset = reader.query(QueryExpression.create(paths, null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(reader, paths);
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeAccurateDBDataSetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeAccurateDBDataSetPushConsumerIT.java
index 2e37fa64ff7..d8d456ec74b 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeAccurateDBDataSetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeAccurateDBDataSetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -126,10 +127,10 @@ public class IoTDBTimeRangeAccurateDBDataSetPushConsumerIT
.ackStrategy(AckStrategy.BEFORE_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- Tablet tablet = dataSet.getTablet();
+ Tablet tablet =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet();
session_dest.insertTablet(tablet);
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBDataSetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBDataSetPushConsumerIT.java
index 047888c25d9..aa9108e70bd 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBDataSetPushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBDataSetPushConsumerIT.java
@@ -26,13 +26,14 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
-import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -128,10 +129,10 @@ public class IoTDBTimeRangeDBDataSetPushConsumerIT extends AbstractSubscriptionT
.ackStrategy(AckStrategy.AFTER_CONSUME)
.consumeListener(
message -> {
- for (final SubscriptionSessionDataSet dataSet :
- message.getSessionDataSetsHandler()) {
+ for (final ResultSet dataSet : message.getResultSets()) {
try {
- Tablet tablet = dataSet.getTablet();
+ Tablet tablet =
+ ((SubscriptionRecordHandler.SubscriptionResultSet) dataSet).getTablet();
session_dest.insertTablet(tablet);
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBTsfilePushConsumerIT.java
index 5654e6f07e5..5216e5b79f5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBTsfilePushConsumerIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/time/IoTDBTimeRangeDBTsfilePushConsumerIT.java
@@ -26,17 +26,16 @@ import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumer;
+import org.apache.iotdb.subscription.it.SubscriptionTreeReaderTestUtils;
import org.apache.iotdb.subscription.it.triple.treemodel.regression.AbstractSubscriptionTreeRegressionIT;
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.read.TsFileReader;
import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.read.expression.QueryExpression;
-import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -139,12 +138,13 @@ public class IoTDBTimeRangeDBTsfilePushConsumerIT extends AbstractSubscriptionTr
message -> {
try {
onReceive.addAndGet(1);
- TsFileReader reader = message.getTsFileHandler().openReader();
+ ITsFileTreeReader reader = message.getTsFile().openTreeReader();
List<Path> paths = new ArrayList<>(2);
for (int i = 0; i < 2; i++) {
paths.add(new Path(device, "s_" + i, true));
}
- QueryDataSet dataset = reader.query(QueryExpression.create(paths, null));
+ SubscriptionTreeReaderTestUtils.QueryDataSetAdapter dataset =
+ SubscriptionTreeReaderTestUtils.query(reader, paths);
while (dataset.hasNext()) {
rowCount.addAndGet(1);
RowRecord next = dataset.next();
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTableTsFile.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTableTsFile.java
index 5ed446bd751..02ee2da2ed0 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTableTsFile.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTableTsFile.java
@@ -151,7 +151,7 @@ public class SubscriptionTableTsFile extends AbstractSubscriptionTsFile {
try {
for (final SubscriptionMessage message :
consumer.poll(Constants.POLL_MESSAGE_TIMEOUT)) {
- final SubscriptionTsFileHandler handler = message.getTsFileHandler();
+ final SubscriptionTsFileHandler handler = message.getTsFile();
ioTPrinter.println(handler.getFile().getName());
try {
handler.moveFile(
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTreeTsFile.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTreeTsFile.java
index ec15b01365d..61274ed59d2 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTreeTsFile.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/subscription/SubscriptionTreeTsFile.java
@@ -147,7 +147,7 @@ public class SubscriptionTreeTsFile extends AbstractSubscriptionTsFile {
try {
for (final SubscriptionMessage message :
consumer.poll(Constants.POLL_MESSAGE_TIMEOUT)) {
- final SubscriptionTsFileHandler handler = message.getTsFileHandler();
+ final SubscriptionTsFileHandler handler = message.getTsFile();
ioTPrinter.println(handler.getFile().getName());
try {
handler.moveFile(
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 f71980d629f..ea588f12763 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
@@ -61,6 +61,8 @@ public class TopicConfig extends PipeParameters {
private static final Map<String, String> SINK_HYBRID_FORMAT_CONFIG =
Collections.singletonMap("format", "hybrid");
+ private static final String LEGACY_FORMAT_TS_FILE_HANDLER_VALUE = "TsFileHandler";
+
private static final Map<String, String> SNAPSHOT_MODE_CONFIG =
Collections.singletonMap("mode", TopicConstant.MODE_SNAPSHOT_VALUE);
private static final Map<String, String> LIVE_MODE_CONFIG =
@@ -196,12 +198,18 @@ public class TopicConfig extends PipeParameters {
public Map<String, String> getAttributesWithSinkFormat() {
// refer to
// org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector.parseAndCollectEvent(org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent)
- return TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE.equalsIgnoreCase(
+ return isTsFileFormat(
attributes.getOrDefault(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_DEFAULT_VALUE))
? SINK_TS_FILE_FORMAT_CONFIG
: SINK_TABLET_FORMAT_CONFIG;
}
+ private boolean isTsFileFormat(final String formatValue) {
+ return TopicConstant.FORMAT_TS_FILE_VALUE.equalsIgnoreCase(formatValue)
+ || TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE.equalsIgnoreCase(formatValue)
+ || LEGACY_FORMAT_TS_FILE_HANDLER_VALUE.equalsIgnoreCase(formatValue);
+ }
+
public Map<String, String> getAttributesWithSinkPrefix() {
final Map<String, String> 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 4e9df56ec4b..bb84358648e 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
@@ -41,9 +41,14 @@ public class TopicConstant {
public static final String MODE_DEFAULT_VALUE = MODE_LIVE_VALUE;
public static final String FORMAT_KEY = "format";
- public static final String FORMAT_SESSION_DATA_SETS_HANDLER_VALUE = "SessionDataSetsHandler";
- 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 FORMAT_RECORD_HANDLER_VALUE = "SubscriptionRecordHandler";
+ public static final String FORMAT_TS_FILE_VALUE = "SubscriptionTsFileHandler";
+ public static final String FORMAT_DEFAULT_VALUE = FORMAT_RECORD_HANDLER_VALUE;
+
+ @Deprecated
+ public static final String FORMAT_SESSION_DATA_SETS_HANDLER_VALUE = FORMAT_RECORD_HANDLER_VALUE;
+
+ @Deprecated public static final String FORMAT_TS_FILE_HANDLER_VALUE = FORMAT_TS_FILE_VALUE;
public static final String LOOSE_RANGE_KEY = "loose-range";
public static final String LOOSE_RANGE_TIME_VALUE = "time";
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
index e27f182549e..c290af2b67b 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -1163,13 +1163,12 @@ abstract class AbstractSubscriptionConsumer implements AutoCloseable {
new HashMap<>();
for (final SubscriptionMessage message : messages) {
// make every effort to delete stale intermediate file
- if (Objects.equals(
- SubscriptionMessageType.TS_FILE_HANDLER.getType(), message.getMessageType())
+ if (Objects.equals(SubscriptionMessageType.TS_FILE.getType(), message.getMessageType())
&&
// do not delete file that can resume from breakpoint
!inFlightFilesCommitContextSet.contains(message.getCommitContext())) {
try {
- message.getTsFileHandler().deleteFile();
+ message.getTsFile().deleteFile();
} catch (final Exception ignored) {
}
}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessage.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessage.java
index f48fa485f7d..f41d34f0ab8 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessage.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessage.java
@@ -23,8 +23,10 @@ import org.apache.iotdb.rpc.subscription.exception.SubscriptionIncompatibleHandl
import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
import org.apache.thrift.annotation.Nullable;
+import org.apache.tsfile.read.query.dataset.ResultSet;
import org.apache.tsfile.write.record.Tablet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -40,8 +42,8 @@ public class SubscriptionMessage implements Comparable<SubscriptionMessage> {
public SubscriptionMessage(
final SubscriptionCommitContext commitContext, final Map<String, List<Tablet>> tablets) {
this.commitContext = commitContext;
- this.messageType = SubscriptionMessageType.SESSION_DATA_SETS_HANDLER.getType();
- this.handler = new SubscriptionSessionDataSetsHandler(tablets);
+ this.messageType = SubscriptionMessageType.RECORD_HANDLER.getType();
+ this.handler = new SubscriptionRecordHandler(tablets);
}
public SubscriptionMessage(
@@ -49,7 +51,7 @@ public class SubscriptionMessage implements Comparable<SubscriptionMessage> {
final String absolutePath,
@Nullable final String databaseName) {
this.commitContext = commitContext;
- this.messageType = SubscriptionMessageType.TS_FILE_HANDLER.getType();
+ this.messageType = SubscriptionMessageType.TS_FILE.getType();
this.handler = new SubscriptionTsFileHandler(absolutePath, databaseName);
}
@@ -98,20 +100,31 @@ public class SubscriptionMessage implements Comparable<SubscriptionMessage> {
/////////////////////////////// handlers ///////////////////////////////
- public SubscriptionSessionDataSetsHandler getSessionDataSetsHandler() {
- if (handler instanceof SubscriptionSessionDataSetsHandler) {
- return (SubscriptionSessionDataSetsHandler) handler;
+ public List<ResultSet> getResultSets() {
+ if (handler instanceof SubscriptionRecordHandler) {
+ return ((SubscriptionRecordHandler) handler).getResultSets();
+ }
+ throw new SubscriptionIncompatibleHandlerException(
+ String.format("%s do not support getResultSets().", handler.getClass().getSimpleName()));
+ }
+
+ public Iterator<Tablet> getRecordTabletIterator() {
+ if (handler instanceof SubscriptionRecordHandler) {
+ final List<ResultSet> resultSets = ((SubscriptionRecordHandler) handler).getResultSets();
+ return resultSets.stream()
+ .map(record -> ((SubscriptionRecordHandler.SubscriptionResultSet) record).getTablet())
+ .iterator();
}
throw new SubscriptionIncompatibleHandlerException(
String.format(
- "%s do not support getSessionDataSetsHandler().", handler.getClass().getSimpleName()));
+ "%s do not support getRecordTabletIterator().", handler.getClass().getSimpleName()));
}
- public SubscriptionTsFileHandler getTsFileHandler() {
+ public SubscriptionTsFileHandler getTsFile() {
if (handler instanceof SubscriptionTsFileHandler) {
return (SubscriptionTsFileHandler) handler;
}
throw new SubscriptionIncompatibleHandlerException(
- String.format("%s do not support getTsFileHandler().", handler.getClass().getSimpleName()));
+ String.format("%s do not support getTsFile().", handler.getClass().getSimpleName()));
}
}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessageType.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessageType.java
index 5dabf3711cc..34189c2fa9b 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessageType.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionMessageType.java
@@ -24,8 +24,8 @@ import java.util.HashMap;
import java.util.Map;
public enum SubscriptionMessageType {
- SESSION_DATA_SETS_HANDLER((short) 0),
- TS_FILE_HANDLER((short) 1),
+ RECORD_HANDLER((short) 0),
+ TS_FILE((short) 1),
;
private final short type;
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java
new file mode 100644
index 00000000000..c7f74db7695
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java
@@ -0,0 +1,392 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.session.subscription.payload;
+
+import org.apache.iotdb.rpc.subscription.annotation.TableModel;
+
+import org.apache.thrift.annotation.Nullable;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.query.dataset.AbstractResultSet;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.DateUtils;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+import org.apache.tsfile.write.record.TSRecord;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+
+import java.io.IOException;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class SubscriptionRecordHandler implements Iterable<ResultSet>, SubscriptionMessageHandler {
+
+ private final List<ResultSet> resultSets;
+
+ public SubscriptionRecordHandler(final Map<String, List<Tablet>> tablets) {
+ final List<ResultSet> resultSets = new ArrayList<>();
+ for (final Map.Entry<String, List<Tablet>> entry : tablets.entrySet()) {
+ final String databaseName = entry.getKey();
+ final List<Tablet> tabletList = entry.getValue();
+ if (Objects.isNull(tabletList)) {
+ continue;
+ }
+ for (final Tablet tablet : tabletList) {
+ if (Objects.isNull(tablet)) {
+ continue;
+ }
+ resultSets.add(new SubscriptionResultSet(tablet, databaseName));
+ }
+ }
+ this.resultSets = Collections.unmodifiableList(resultSets);
+ }
+
+ public List<ResultSet> getResultSets() {
+ return resultSets;
+ }
+
+ @Override
+ public Iterator<ResultSet> iterator() {
+ return resultSets.iterator();
+ }
+
+ public static class SubscriptionResultSet extends AbstractResultSet {
+
+ private Tablet tablet;
+
+ @Nullable private final String databaseName;
+
+ private final List<RowPosition> sortedRowPositions;
+
+ private int rowIndex = -1;
+
+ @TableModel private List<ColumnCategory> columnCategoryList;
+
+ private SubscriptionResultSet(final Tablet tablet, @Nullable final String databaseName) {
+ super(generateColumnNames(tablet, databaseName), generateColumnTypes(tablet));
+ this.tablet = tablet;
+ this.databaseName = databaseName;
+ this.sortedRowPositions = generateSortedRowPositions(tablet);
+ }
+
+ @TableModel
+ public String getDatabaseName() {
+ return databaseName;
+ }
+
+ @TableModel
+ public String getTableName() {
+ return tablet.getTableName();
+ }
+
+ @TableModel
+ public List<ColumnCategory> getColumnCategories() {
+ if (Objects.nonNull(columnCategoryList)) {
+ return columnCategoryList;
+ }
+
+ if (!isTableData()) {
+ return Collections.emptyList();
+ }
+
+ return columnCategoryList =
+ Stream.concat(
+ Stream.of(ColumnCategory.TIME),
+ tablet.getColumnTypes().stream()
+ .map(
+ columnCategory -> {
+ switch (columnCategory) {
+ case FIELD:
+ return ColumnCategory.FIELD;
+ case TAG:
+ return ColumnCategory.TAG;
+ case ATTRIBUTE:
+ return ColumnCategory.ATTRIBUTE;
+ default:
+ throw new IllegalArgumentException(
+ "Unknown column category: " + columnCategory);
+ }
+ }))
+ .collect(Collectors.toList());
+ }
+
+ public Tablet getTablet() {
+ return tablet;
+ }
+
+ public boolean hasNext() {
+ return Objects.nonNull(tablet) && rowIndex + 1 < sortedRowPositions.size();
+ }
+
+ @Nullable
+ public RowRecord nextRecord() throws IOException {
+ return next() ? currentRow : null;
+ }
+
+ public int getColumnCount() {
+ return tablet.getSchemas().size() + 1;
+ }
+
+ public List<String> getColumnNames() {
+ final int columnCount = getColumnCount();
+ final List<String> columnNames = new ArrayList<>(columnCount);
+ for (int i = 1; i <= columnCount; ++i) {
+ columnNames.add(resultSetMetadata.getColumnName(i));
+ }
+ return columnNames;
+ }
+
+ public List<String> getColumnTypes() {
+ final int columnCount = getColumnCount();
+ final List<String> columnTypes = new ArrayList<>(columnCount);
+ for (int i = 1; i <= columnCount; ++i) {
+ columnTypes.add(resultSetMetadata.getColumnType(i).toString());
+ }
+ return columnTypes;
+ }
+
+ @Override
+ public boolean next() throws IOException {
+ if (Objects.isNull(tablet)) {
+ return false;
+ }
+
+ ++rowIndex;
+ if (rowIndex >= sortedRowPositions.size()) {
+ return false;
+ }
+
+ final RowPosition position = sortedRowPositions.get(rowIndex);
+ currentRow = generateRowRecord(position.timestamp, position.rowIndex);
+ return true;
+ }
+
+ @Override
+ public void close() {
+ tablet = null;
+ currentRow = null;
+ }
+
+ @Override
+ public Iterator<TSRecord> iterator() {
+ final Tablet currentTablet = this.tablet;
+ if (Objects.isNull(currentTablet)) {
+ return Collections.emptyIterator();
+ }
+ return new Iterator<TSRecord>() {
+ private int index = 0;
+
+ @Override
+ public boolean hasNext() {
+ return index < sortedRowPositions.size();
+ }
+
+ @Override
+ public TSRecord next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final RowPosition position = sortedRowPositions.get(index++);
+ return generateTsRecord(currentTablet, position.timestamp, position.rowIndex);
+ }
+ };
+ }
+
+ public enum ColumnCategory {
+ TIME,
+ TAG,
+ FIELD,
+ ATTRIBUTE
+ }
+
+ private boolean isTableData() {
+ return Objects.nonNull(databaseName);
+ }
+
+ private static List<String> generateColumnNames(
+ final Tablet tablet, @Nullable final String databaseName) {
+ final List<IMeasurementSchema> schemas = tablet.getSchemas();
+ if (Objects.nonNull(databaseName)) {
+ return schemas.stream()
+ .map(IMeasurementSchema::getMeasurementName)
+ .collect(Collectors.toList());
+ }
+ final String deviceId = tablet.getDeviceId();
+ return schemas.stream()
+ .map(schema -> deviceId + "." + schema.getMeasurementName())
+ .collect(Collectors.toList());
+ }
+
+ private static List<TSDataType> generateColumnTypes(final Tablet tablet) {
+ return tablet.getSchemas().stream()
+ .map(IMeasurementSchema::getType)
+ .collect(Collectors.toList());
+ }
+
+ private static List<RowPosition> generateSortedRowPositions(final Tablet tablet) {
+ final int rowSize = tablet.getRowSize();
+ final List<RowPosition> positions = new ArrayList<>(rowSize);
+ for (int i = 0; i < rowSize; ++i) {
+ positions.add(new RowPosition(tablet.getTimestamp(i), i));
+ }
+ positions.sort(
+ (left, right) -> {
+ final int timeComparison = Long.compare(left.timestamp, right.timestamp);
+ return timeComparison != 0
+ ? timeComparison
+ : Integer.compare(left.rowIndex, right.rowIndex);
+ });
+ return positions;
+ }
+
+ private RowRecord generateRowRecord(final long timestamp, final int rowPosition) {
+ final int columnSize = tablet.getSchemas().size();
+ final List<Field> fields = new ArrayList<>(columnSize);
+
+ final BitMap[] bitMaps = tablet.getBitMaps();
+ for (int columnIndex = 0; columnIndex < columnSize; ++columnIndex) {
+ final Field field;
+ if (bitMaps != null
+ && bitMaps[columnIndex] != null
+ && bitMaps[columnIndex].isMarked(rowPosition)) {
+ field = new Field(null);
+ } else {
+ final TSDataType dataType = tablet.getSchemas().get(columnIndex).getType();
+ field =
+ generateFieldFromTabletValue(dataType, tablet.getValues()[columnIndex], rowPosition);
+ }
+ fields.add(field);
+ }
+ return new RowRecord(timestamp, fields);
+ }
+
+ private TSRecord generateTsRecord(
+ final Tablet currentTablet, final long timestamp, final int currentRowIndex) {
+ final TSRecord record =
+ isTableData()
+ ? new TSRecord(currentTablet.getTableName(), timestamp)
+ : new TSRecord(currentTablet.getDeviceId(), timestamp);
+
+ final BitMap[] bitMaps = currentTablet.getBitMaps();
+ for (int columnIndex = 0; columnIndex < currentTablet.getSchemas().size(); ++columnIndex) {
+ if (bitMaps != null
+ && bitMaps[columnIndex] != null
+ && bitMaps[columnIndex].isMarked(currentRowIndex)) {
+ continue;
+ }
+
+ final String measurement = currentTablet.getSchemas().get(columnIndex).getMeasurementName();
+ final TSDataType dataType = currentTablet.getSchemas().get(columnIndex).getType();
+ final Object value = currentTablet.getValues()[columnIndex];
+ switch (dataType) {
+ case BOOLEAN:
+ record.addPoint(measurement, ((boolean[]) value)[currentRowIndex]);
+ break;
+ case INT32:
+ record.addPoint(measurement, ((int[]) value)[currentRowIndex]);
+ break;
+ case DATE:
+ record.addPoint(measurement, ((LocalDate[]) value)[currentRowIndex]);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ record.addPoint(measurement, ((long[]) value)[currentRowIndex]);
+ break;
+ case FLOAT:
+ record.addPoint(measurement, ((float[]) value)[currentRowIndex]);
+ break;
+ case DOUBLE:
+ record.addPoint(measurement, ((double[]) value)[currentRowIndex]);
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ case OBJECT:
+ final Binary binary = ((Binary[]) value)[currentRowIndex];
+ if (Objects.nonNull(binary)) {
+ record.addPoint(measurement, binary.getValues());
+ }
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.", dataType));
+ }
+ }
+ return record;
+ }
+
+ private static Field generateFieldFromTabletValue(
+ final TSDataType dataType, final Object value, final int index) {
+ final Field field = new Field(dataType);
+ switch (dataType) {
+ case BOOLEAN:
+ field.setBoolV(((boolean[]) value)[index]);
+ break;
+ case INT32:
+ field.setIntV(((int[]) value)[index]);
+ break;
+ case DATE:
+ field.setIntV(DateUtils.parseDateExpressionToInt(((LocalDate[]) value)[index]));
+ break;
+ case INT64:
+ case TIMESTAMP:
+ field.setLongV(((long[]) value)[index]);
+ break;
+ case FLOAT:
+ field.setFloatV(((float[]) value)[index]);
+ break;
+ case DOUBLE:
+ field.setDoubleV(((double[]) value)[index]);
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ case OBJECT:
+ field.setBinaryV(new Binary((((Binary[]) value)[index]).getValues()));
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.", dataType));
+ }
+ return field;
+ }
+
+ private static class RowPosition {
+ private final long timestamp;
+ private final int rowIndex;
+
+ private RowPosition(final long timestamp, final int rowIndex) {
+ this.timestamp = timestamp;
+ this.rowIndex = rowIndex;
+ }
+ }
+ }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionSessionDataSet.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionSessionDataSet.java
deleted file mode 100644
index a3f01b3e11e..00000000000
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionSessionDataSet.java
+++ /dev/null
@@ -1,255 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.session.subscription.payload;
-
-import org.apache.iotdb.isession.ISessionDataSet;
-import org.apache.iotdb.rpc.subscription.annotation.TableModel;
-
-import org.apache.thrift.annotation.Nullable;
-import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.common.Field;
-import org.apache.tsfile.read.common.RowRecord;
-import org.apache.tsfile.utils.Binary;
-import org.apache.tsfile.utils.BitMap;
-import org.apache.tsfile.utils.DateUtils;
-import org.apache.tsfile.write.UnSupportedDataTypeException;
-import org.apache.tsfile.write.record.Tablet;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
-
-import java.time.LocalDate;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.TreeMap;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-public class SubscriptionSessionDataSet implements ISessionDataSet {
-
- private Tablet tablet;
-
- @Nullable private final String databaseName;
-
- public Tablet getTablet() {
- return tablet;
- }
-
- public SubscriptionSessionDataSet(final Tablet tablet, @Nullable final String databaseName) {
- this.tablet = tablet;
- this.databaseName = databaseName;
- generateRowIterator();
- }
-
- /////////////////////////////// table model ///////////////////////////////
-
- @TableModel private List<ColumnCategory> columnCategoryList;
-
- @TableModel
- public String getDatabaseName() {
- return databaseName;
- }
-
- @TableModel
- public String getTableName() {
- return tablet.getTableName();
- }
-
- @TableModel
- public List<ColumnCategory> getColumnCategories() {
- if (Objects.nonNull(columnCategoryList)) {
- return columnCategoryList;
- }
-
- if (!isTableData()) {
- return Collections.emptyList();
- }
-
- return columnCategoryList =
- Stream.concat(
- Stream.of(ColumnCategory.TIME),
- tablet.getColumnTypes().stream()
- .map(
- columnCategory -> {
- switch (columnCategory) {
- case FIELD:
- return ColumnCategory.FIELD;
- case TAG:
- return ColumnCategory.TAG;
- case ATTRIBUTE:
- return ColumnCategory.ATTRIBUTE;
- default:
- throw new IllegalArgumentException(
- "Unknown column category: " + columnCategory);
- }
- }))
- .collect(Collectors.toList());
- }
-
- @TableModel
- public enum ColumnCategory {
- TIME,
- TAG,
- FIELD,
- ATTRIBUTE
- }
-
- private boolean isTableData() {
- return Objects.nonNull(databaseName);
- }
-
- /////////////////////////////// override ///////////////////////////////
-
- private List<String> columnNameList;
- private List<String> columnTypeList;
-
- @Override
- public List<String> getColumnNames() {
- if (Objects.nonNull(columnNameList)) {
- return columnNameList;
- }
-
- List<IMeasurementSchema> schemas = tablet.getSchemas();
- String deviceId = tablet.getDeviceId();
- return columnNameList =
- isTableData()
- ? Stream.concat(
- Stream.of("time"), schemas.stream().map(IMeasurementSchema::getMeasurementName))
- .collect(Collectors.toList())
- : Stream.concat(
- Stream.of("Time"),
- schemas.stream().map(schema -> deviceId + "." + schema.getMeasurementName()))
- .collect(Collectors.toList());
- }
-
- @Override
- public List<String> getColumnTypes() {
- if (Objects.nonNull(columnTypeList)) {
- return columnTypeList;
- }
-
- List<IMeasurementSchema> schemas = tablet.getSchemas();
- return columnTypeList =
- Stream.concat(
- Stream.of(TSDataType.INT64.toString()),
- schemas.stream().map(schema -> schema.getType().toString()))
- .collect(Collectors.toList());
- }
-
- public boolean hasNext() {
- return rowIterator.hasNext();
- }
-
- @Override
- public RowRecord next() {
- final Map.Entry<Long, Integer> entry = this.rowIterator.next();
- final int columnSize = getColumnSize();
-
- final List<Field> fields = new ArrayList<>();
- final long timestamp = entry.getKey();
- final int rowIndex = entry.getValue();
-
- BitMap[] bitMaps = tablet.getBitMaps();
- for (int columnIndex = 0; columnIndex < columnSize; ++columnIndex) {
- final Field field;
- if (bitMaps != null
- && bitMaps[columnIndex] != null
- && bitMaps[columnIndex].isMarked(rowIndex)) {
- field = new Field(null);
- } else {
- final TSDataType dataType = tablet.getSchemas().get(columnIndex).getType();
- field = generateFieldFromTabletValue(dataType, tablet.getValues()[columnIndex], rowIndex);
- }
- fields.add(field);
- }
- return new RowRecord(timestamp, fields);
- }
-
- @Override
- public void close() throws Exception {
- // maybe friendly for gc
- tablet = null;
- }
-
- /////////////////////////////// utility ///////////////////////////////
-
- private Iterator<Map.Entry<Long, Integer>> rowIterator;
-
- private int getColumnSize() {
- return tablet.getSchemas().size();
- }
-
- private void generateRowIterator() {
- // timestamp -> row index
- final long[] timestamps = tablet.getTimestamps();
- final TreeMap<Long, Integer> timestampToRowIndex = new TreeMap<>();
- final int rowSize = timestamps.length;
- for (int rowIndex = 0; rowIndex < rowSize; ++rowIndex) {
- final Long timestamp = timestamps[rowIndex];
- timestampToRowIndex.put(timestamp, rowIndex);
- }
- this.rowIterator = timestampToRowIndex.entrySet().iterator();
- }
-
- private static Field generateFieldFromTabletValue(
- final TSDataType dataType, final Object value, final int index) {
- final Field field = new Field(dataType);
- switch (dataType) {
- case BOOLEAN:
- final boolean booleanValue = ((boolean[]) value)[index];
- field.setBoolV(booleanValue);
- break;
- case INT32:
- final int intValue = ((int[]) value)[index];
- field.setIntV(intValue);
- break;
- case DATE:
- final LocalDate dateValue = ((LocalDate[]) value)[index];
- field.setIntV(DateUtils.parseDateExpressionToInt(dateValue));
- break;
- case INT64:
- case TIMESTAMP:
- final long longValue = ((long[]) value)[index];
- field.setLongV(longValue);
- break;
- case FLOAT:
- final float floatValue = ((float[]) value)[index];
- field.setFloatV(floatValue);
- break;
- case DOUBLE:
- final double doubleValue = ((double[]) value)[index];
- field.setDoubleV(doubleValue);
- break;
- case TEXT:
- case STRING:
- case BLOB:
- case OBJECT:
- final Binary binaryValue = new Binary((((Binary[]) value)[index]).getValues());
- field.setBinaryV(binaryValue);
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format("Data type %s is not supported.", dataType));
- }
- return field;
- }
-}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionSessionDataSetsHandler.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionSessionDataSetsHandler.java
deleted file mode 100644
index bfe5d83d5fa..00000000000
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionSessionDataSetsHandler.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.session.subscription.payload;
-
-import org.apache.tsfile.write.record.Tablet;
-
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-
-public class SubscriptionSessionDataSetsHandler
- implements Iterable<SubscriptionSessionDataSet>, SubscriptionMessageHandler {
-
- private final Map<String, List<Tablet>> tablets;
-
- public SubscriptionSessionDataSetsHandler(final Map<String, List<Tablet>> tablets) {
- this.tablets = tablets;
- }
-
- @Override
- public Iterator<SubscriptionSessionDataSet> iterator() {
- return new Iterator<SubscriptionSessionDataSet>() {
- // Iterator over map entries: databaseName -> list of tablets
- private final Iterator<Map.Entry<String, List<Tablet>>> entryIterator =
- tablets.entrySet().iterator();
- // Current databaseName
- private String currentDatabase;
- // Iterator over the current list of tablets
- private Iterator<Tablet> tabletIterator = Collections.emptyIterator();
-
- @Override
- public boolean hasNext() {
- // Advance to next non-empty tablet list if needed
- while (!tabletIterator.hasNext() && entryIterator.hasNext()) {
- Map.Entry<String, List<Tablet>> entry = entryIterator.next();
- currentDatabase = entry.getKey();
- List<Tablet> list = entry.getValue();
- tabletIterator = (list != null ? list.iterator() : Collections.emptyIterator());
- }
- return tabletIterator.hasNext();
- }
-
- @Override
- public SubscriptionSessionDataSet next() {
- if (!hasNext()) {
- throw new NoSuchElementException();
- }
- Tablet tablet = tabletIterator.next();
- return new SubscriptionSessionDataSet(tablet, currentDatabase);
- }
- };
- }
-
- public Iterator<Tablet> tabletIterator() {
- return new Iterator<Tablet>() {
- final Iterator<SubscriptionSessionDataSet> iterator = iterator();
-
- @Override
- public boolean hasNext() {
- return iterator.hasNext();
- }
-
- @Override
- public Tablet next() {
- if (!hasNext()) {
- throw new NoSuchElementException();
- }
- return iterator.next().getTablet();
- }
- };
- }
-}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionTsFileHandler.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionTsFileHandler.java
index 749f1bcb8b7..cd6f3270891 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionTsFileHandler.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionTsFileHandler.java
@@ -20,10 +20,13 @@
package org.apache.iotdb.session.subscription.payload;
import org.apache.iotdb.rpc.subscription.annotation.TableModel;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionIncompatibleHandlerException;
import org.apache.thrift.annotation.Nullable;
-import org.apache.tsfile.read.TsFileReader;
-import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.v4.ITsFileReader;
+import org.apache.tsfile.read.v4.ITsFileTreeReader;
+import org.apache.tsfile.read.v4.TsFileReaderBuilder;
+import org.apache.tsfile.read.v4.TsFileTreeReaderBuilder;
import java.io.IOException;
@@ -36,8 +39,25 @@ public class SubscriptionTsFileHandler extends SubscriptionFileHandler {
this.databaseName = databaseName;
}
- public TsFileReader openReader() throws IOException {
- return new TsFileReader(new TsFileSequenceReader(absolutePath));
+ public ITsFileTreeReader openTreeReader() throws IOException {
+ if (databaseName != null) {
+ throw new SubscriptionIncompatibleHandlerException(
+ String.format(
+ "%s does not support openTreeReader() for table model tsfile.",
+ getClass().getSimpleName()));
+ }
+ return new TsFileTreeReaderBuilder().file(getFile()).build();
+ }
+
+ @TableModel
+ public ITsFileReader openTableReader() throws IOException {
+ if (databaseName == null) {
+ throw new SubscriptionIncompatibleHandlerException(
+ String.format(
+ "%s does not support openTableReader() for tree model tsfile.",
+ getClass().getSimpleName()));
+ }
+ return new TsFileReaderBuilder().file(getFile()).build();
}
@TableModel