This is an automated email from the ASF dual-hosted git repository.
CRZbulabula pushed a commit to branch research/cluster-compress
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/research/cluster-compress by
this push:
new 915ca4ffbee Integrate research Cluster pages and SQL validation
(#18809)
915ca4ffbee is described below
commit 915ca4ffbeef718f9dd479cf06aab627b9208377
Author: SamM755 <[email protected]>
AuthorDate: Fri Oct 9 15:51:57 2026 +0800
Integrate research Cluster pages and SQL validation (#18809)
---
iotdb-client/session/CLUSTER_TABLE.md | 62 ++++++
iotdb-client/session/pom.xml | 2 +
.../apache/iotdb/session/ClusterTableSession.java | 161 ++++++++++++++
.../apache/iotdb/session/ClusterAlignedSqlIT.java | 232 +++++++++++++++++++++
.../iotdb/session/ClusterTableSessionIT.java | 77 +++++++
.../iotdb/session/ClusterTableSessionTest.java | 167 +++++++++++++++
pom.xml | 2 +-
7 files changed, 702 insertions(+), 1 deletion(-)
diff --git a/iotdb-client/session/CLUSTER_TABLE.md
b/iotdb-client/session/CLUSTER_TABLE.md
new file mode 100644
index 00000000000..0607e720fd0
--- /dev/null
+++ b/iotdb-client/session/CLUSTER_TABLE.md
@@ -0,0 +1,62 @@
+<!--
+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.
+-->
+
+# Complete numeric table compression (research)
+
+`ClusterTableSession` adds the complete column-selection + ACluster/KCluster +
ALP
+pipeline using ordinary IoTDB BLOB storage. The Session module depends on the
companion
+TsFile `research/cluster-compress` branch's `2.1.0-SNAPSHOT`; install that
branch's Java
+common and tsfile modules locally before building this module. This BLOB API
itself
+does not require a modified server. The research branch now also uses that
TsFile
+version on the server for a separate native aligned/scalar numeric SQL path.
+`ClusterAlignedSqlIT` tests that path without calling this BLOB API.
+
+```java
+ClusterTableSession storage = new ClusterTableSession(
+ session, "root.research.numeric_table", ClusterTableOptions.aCluster());
+storage.createSchema();
+int blocks = storage.writeTable(firstUnusedBlockId, numericTable);
+storage.readAll(table -> { /* consume complete records */ });
+storage.readTimeRange(startInclusive, endInclusive, table -> { /* original
timestamps */ });
+```
+
+Choose `ClusterTableOptions.kCluster()` for KCluster; both use the same
selector and ALP.
+See `ClusterTable` for the typed INT32/INT64/FLOAT/DOUBLE plus null record API.
+
+Use one dedicated device per table stream, and supply unique increasing block
IDs across
+writers/restarts. The outer IoTDB timestamp is the block ID, not the original
timestamp.
+Ordinary SQL sees payload, min_time, max_time and row_count; original numeric
fields require
+this decoder. Original-time queries prune with min_time/max_time and then
filter decoded rows.
+Do not apply original-time filters or retention policies directly to the outer
block ID.
+The caller owns the Session and must close it.
+
+Every independent block preserves complete records and duplicate
multiplicities, including
+timestamps and exact numeric bits, while permitting physical row reordering.
+The writer is single-threaded and refuses further writes after an uncertain
insert failure.
+Reconcile that block's outcome before restarting; do not blindly retry at a
new block ID.
+
+Tests:
+- `ClusterTableSessionTest`: RPC payloads, complete-record filtering, invalid
paths and failures.
+- `ClusterTableSessionIT host port dedicatedDevice write|read`: opt-in local
service check.
+ Write mode writes 1,200 records for each method, flushes, and verifies
full/range queries.
+ Read mode performs the same verification without writing, for persistence
testing after restart.
+ Run only against a dedicated test device. It uses the local test credentials
root/root.
+
+This is an opt-in client/block integration, not a new SQL encoding option or a
server storage-page
+rewrite. Its correctness checks do not establish a performance advantage.
diff --git a/iotdb-client/session/pom.xml b/iotdb-client/session/pom.xml
index 40ed5ab0a96..c327e2030c4 100644
--- a/iotdb-client/session/pom.xml
+++ b/iotdb-client/session/pom.xml
@@ -29,6 +29,8 @@
<artifactId>iotdb-session</artifactId>
<name>IoTDB: Client: Session</name>
<properties>
+ <!-- The complete-record codec lives in the companion TsFile research
branch. -->
+ <tsfile.version>2.1.0-SNAPSHOT</tsfile.version>
<session.it.skip>${session.test.skip}</session.it.skip>
<session.test.skip>false</session.test.skip>
<session.ut.skip>${session.test.skip}</session.ut.skip>
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/ClusterTableSession.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/ClusterTableSession.java
new file mode 100644
index 00000000000..3a9d5c95552
--- /dev/null
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/ClusterTableSession.java
@@ -0,0 +1,161 @@
+/*
+ * 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;
+
+import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+
+import org.apache.tsfile.encoding.table.ClusterTable;
+import org.apache.tsfile.encoding.table.ClusterTableCodec;
+import org.apache.tsfile.encoding.table.ClusterTableEncoder;
+import org.apache.tsfile.encoding.table.ClusterTableOptions;
+import org.apache.tsfile.encoding.table.ClusterTableTsFile;
+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.common.RowRecord;
+import org.apache.tsfile.utils.Binary;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Complete-record clustering over ordinary IoTDB BLOB storage. One dedicated
device is one numeric
+ * table stream. The outer timestamp is a unique BLOCK identifier, never an
original row timestamp.
+ * Use this reader to decode original fields; ordinary SQL sees only block
metadata.
+ *
+ * <p>The caller owns the session and supplies collision-free increasing block
IDs, including across
+ * restarts. Do not apply original-row time filters or TTL to the outer block
timestamp. This writer
+ * is single-threaded. A failed insert may have committed remotely: retry the
same block/table
+ * through a new writer after checking its outcome, rather than choosing a new
ID.
+ */
+public final class ClusterTableSession {
+ private static final List<String> MEASUREMENTS =
+ Collections.unmodifiableList(Arrays.asList("payload", "min_time",
"max_time", "row_count"));
+ private static final List<TSDataType> TYPES =
+ Collections.unmodifiableList(
+ Arrays.asList(TSDataType.BLOB, TSDataType.INT64, TSDataType.INT64,
TSDataType.INT32));
+ private final ISession session;
+ private final String device;
+ private final ClusterTableEncoder encoder;
+ private boolean written;
+ private boolean failed;
+ private long lastBlockId;
+
+ public ClusterTableSession(ISession session, String device,
ClusterTableOptions options) {
+ if (session == null
+ || device == null
+ || !device.matches("root(?:\\.[A-Za-z_][A-Za-z0-9_]*)+")) {
+ throw new IllegalArgumentException("Expected a dedicated simple
root.device path");
+ }
+ this.session = session;
+ this.device = device;
+ this.encoder = new ClusterTableEncoder(options);
+ }
+
+ public void createSchema() throws IoTDBConnectionException,
StatementExecutionException {
+ for (int i = 0; i < MEASUREMENTS.size(); i++) {
+ String path = device + "." + MEASUREMENTS.get(i);
+ if (!session.checkTimeseriesExists(path)) {
+ session.createTimeseries(path, TYPES.get(i), TSEncoding.PLAIN,
CompressionType.LZ4);
+ }
+ }
+ }
+
+ /** Writes one independently decodable page; returns its actual serialized
byte count. */
+ public int writeBlock(long blockId, ClusterTable page)
+ throws IOException, IoTDBConnectionException,
StatementExecutionException {
+ if (failed)
+ throw new IOException("Prior insert outcome is uncertain; reconcile
before resuming");
+ if (written && blockId <= lastBlockId) {
+ throw new IllegalArgumentException("Block IDs must increase within this
writer");
+ }
+ if (page.rowCount() == 0) throw new IllegalArgumentException("Cannot store
an empty page");
+ ClusterTableCodec.Encoded encoded = encoder.encode(page);
+ long min = Long.MAX_VALUE, max = Long.MIN_VALUE;
+ for (int r = 0; r < page.rowCount(); r++) {
+ min = Math.min(min, page.timestamp(r));
+ max = Math.max(max, page.timestamp(r));
+ }
+ byte[] bytes = encoded.bytes();
+ try {
+ session.insertRecord(
+ device,
+ blockId,
+ MEASUREMENTS,
+ TYPES,
+ Arrays.asList(new Binary(bytes), min, max, page.rowCount()));
+ } catch (IoTDBConnectionException | StatementExecutionException error) {
+ failed = true;
+ throw error;
+ }
+ written = true;
+ lastBlockId = blockId;
+ return bytes.length;
+ }
+
+ /** Fits on leading rows, splits into bounded pages, and returns the number
of blocks written. */
+ public int writeTable(long firstBlockId, ClusterTable table)
+ throws IOException, IoTDBConnectionException,
StatementExecutionException {
+ encoder.fit(table);
+ int pageSize = ClusterTableEncoder.pageSize(table);
+ int blocks = (table.rowCount() + pageSize - 1) / pageSize;
+ if (blocks > 0) Math.addExact(firstBlockId, blocks - 1L);
+ for (int i = 0; i < blocks; i++) {
+ int from = i * pageSize;
+ writeBlock(firstBlockId + i, table.slice(from,
Math.min(table.rowCount(), from + pageSize)));
+ }
+ return blocks;
+ }
+
+ public void readAll(ClusterTableTsFile.BlockConsumer consumer)
+ throws IOException, IoTDBConnectionException,
StatementExecutionException {
+ read("select payload from " + device, Long.MIN_VALUE, Long.MAX_VALUE,
consumer);
+ }
+
+ /** Inclusive original-time filter: block bounds prune candidates, then
exact row filtering. */
+ public void readTimeRange(long start, long end,
ClusterTableTsFile.BlockConsumer consumer)
+ throws IOException, IoTDBConnectionException,
StatementExecutionException {
+ if (start > end) throw new IllegalArgumentException("Reversed time range");
+ String sql =
+ "select payload from " + device + " where min_time <= " + end + " and
max_time >= " + start;
+ read(sql, start, end, consumer);
+ }
+
+ private void read(String sql, long start, long end,
ClusterTableTsFile.BlockConsumer consumer)
+ throws IOException, IoTDBConnectionException,
StatementExecutionException {
+ try (SessionDataSet result = session.executeQueryStatement(sql)) {
+ while (result.hasNext()) {
+ RowRecord record = result.next();
+ if (record.getFields().isEmpty() ||
record.getFields().get(0).getDataType() == null) {
+ throw new IOException("Missing cluster block payload");
+ }
+ ClusterTable table =
+
ClusterTableCodec.decode(record.getFields().get(0).getBinaryV().getValues());
+ if (start != Long.MIN_VALUE || end != Long.MAX_VALUE) table =
table.timeRange(start, end);
+ if (table.rowCount() != 0) consumer.accept(table);
+ }
+ }
+ }
+}
diff --git
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterAlignedSqlIT.java
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterAlignedSqlIT.java
new file mode 100644
index 00000000000..9b094112e76
--- /dev/null
+++
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterAlignedSqlIT.java
@@ -0,0 +1,232 @@
+/*
+ * 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;
+
+import org.apache.iotdb.isession.SessionDataSet;
+
+import org.apache.tsfile.read.common.RowRecord;
+
+/** Live native SQL check against an isolated server. Arguments: host port
database write|read. */
+public final class ClusterAlignedSqlIT {
+ private static final int ROWS = 1200;
+ private static final long START = 1700000000000L;
+
+ private ClusterAlignedSqlIT() {}
+
+ public static void main(String[] args) throws Exception {
+ if (args.length != 4
+ || !args[2].matches("root\\.[A-Za-z_][A-Za-z0-9_]*")
+ || (!args[3].equals("write") && !args[3].equals("read"))) {
+ throw new IllegalArgumentException("host port dedicatedDatabase
write|read");
+ }
+ try (Session session =
+ new Session.Builder()
+ .host(args[0])
+ .port(Integer.parseInt(args[1]))
+ .username("root")
+ .password("root")
+ .enableRedirection(false)
+ .build()) {
+ session.open(false);
+ if (args[3].equals("write")) {
+ for (String method : new String[] {"ACLUSTER", "KCLUSTER"}) {
+ String device = args[2] + "." +
method.toLowerCase(java.util.Locale.ROOT);
+ session.executeNonQueryStatement(
+ "CREATE ALIGNED TIMESERIES "
+ + device
+ + " ("
+ + "a DOUBLE ENCODING="
+ + method
+ + ", b INT32 ENCODING="
+ + method
+ + ", c DOUBLE ENCODING="
+ + method
+ + ", d INT64 ENCODING="
+ + method
+ + ", empty INT32 ENCODING="
+ + method
+ + ")");
+ insert(session, device, 0, 1000);
+ session.executeNonQueryStatement("FLUSH");
+ verify(session, device, 1000);
+ }
+ for (String method : new String[] {"acluster", "kcluster"}) {
+ String device = args[2] + ".scalar_" + method;
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES " + device + ".v WITH DATATYPE=FLOAT
ENCODING=" + method);
+ for (int start = 0; start < 400; start += 50) {
+ StringBuilder sql =
+ new StringBuilder("INSERT INTO
").append(device).append("(time,v) VALUES ");
+ for (int r = start; r < start + 50; r++) {
+ if (r != start) sql.append(',');
+ sql.append('(').append(START + r).append(',').append((float)
c(r)).append(')');
+ }
+ session.executeNonQueryStatement(sql.toString());
+ }
+ }
+ session.executeNonQueryStatement("FLUSH");
+ // Acknowledge ordinary inserts without FLUSH, then test WAL recovery
after process death.
+ for (String method : new String[] {"acluster", "kcluster"})
+ insert(session, args[2] + "." + method, 1000, ROWS);
+ }
+ for (String method : new String[] {"acluster", "kcluster"}) {
+ verify(session, args[2] + "." + method, ROWS);
+ try (SessionDataSet scalar =
+ session.executeQueryStatement("SELECT v FROM " + args[2] +
".scalar_" + method)) {
+ int r = 0;
+ while (scalar.hasNext()) {
+ RowRecord row = scalar.next();
+ require(row.getTimestamp() == START + r, "scalar timestamp", r);
+ require(
+ Float.floatToRawIntBits(row.getFields().get(0).getFloatV())
+ == Float.floatToRawIntBits((float) c(r)),
+ "scalar value",
+ r++);
+ }
+ require(r == 400, "scalar count", r);
+ }
+ System.out.println(
+ method
+ + ": "
+ + ROWS
+ + " aligned SQL rows plus 400 scalar SQL rows, projection,
value/time predicates, descending scan, aggregates passed; mode="
+ + args[3]);
+ }
+ }
+ }
+
+ private static double a(int r) {
+ return ((r * 17) % 37) / 10.0;
+ }
+
+ private static int b(int r) {
+ return ((r * 17) % 37) * 2;
+ }
+
+ private static double c(int r) {
+ return r % 13 == 0 ? -0.0 : ((r * 3) % 19) / 10.0;
+ }
+
+ private static long d(int r) {
+ return r % 11 == 0 ? Long.MAX_VALUE - r : 9007199254740993L + r;
+ }
+
+ private static void insert(Session session, String device, int from, int to)
throws Exception {
+ for (int start = from; start < to; start += 50) {
+ StringBuilder sql =
+ new StringBuilder("INSERT INTO
").append(device).append("(time,a,b,c,d) ALIGNED VALUES ");
+ for (int r = start; r < Math.min(start + 50, to); r++) {
+ if (r != start) sql.append(',');
+ sql.append('(')
+ .append(START + r)
+ .append(',')
+ .append(a(r))
+ .append(',')
+ .append(b(r))
+ .append(',')
+ .append(r % 7 == 0 ? "null" : Double.toString(c(r)))
+ .append(',')
+ .append(d(r))
+ .append(')');
+ }
+ session.executeNonQueryStatement(sql.toString());
+ }
+ }
+
+ private static void verify(Session session, String device, int rows) throws
Exception {
+ try (SessionDataSet data = session.executeQueryStatement("SELECT a,b,c,d
FROM " + device)) {
+ int r = 0;
+ while (data.hasNext()) {
+ RowRecord row = data.next();
+ require(row.getTimestamp() == START + r, "timestamp", r);
+ require(
+ Double.doubleToRawLongBits(row.getFields().get(0).getDoubleV())
+ == Double.doubleToRawLongBits(a(r)),
+ "a",
+ r);
+ require(row.getFields().get(1).getIntV() == b(r), "b", r);
+ if (r % 7 == 0) require(row.getFields().get(2).getDataType() == null,
"null", r);
+ else
+ require(
+ Double.doubleToRawLongBits(row.getFields().get(2).getDoubleV())
+ == Double.doubleToRawLongBits(c(r)),
+ "c",
+ r);
+ require(row.getFields().get(3).getLongV() == d(r), "d", r);
+ r++;
+ }
+ require(r == rows, "row count", r);
+ }
+ try (SessionDataSet data = session.executeQueryStatement("SELECT b FROM "
+ device)) {
+ int r = 0;
+ while (data.hasNext()) {
+ RowRecord row = data.next();
+ require(
+ row.getTimestamp() == START + r &&
row.getFields().get(0).getIntV() == b(r),
+ "projection",
+ r++);
+ }
+ require(r == rows, "projection count", r);
+ }
+ try (SessionDataSet data =
+ session.executeQueryStatement(
+ "SELECT a,d FROM "
+ + device
+ + " WHERE time >= "
+ + (START + 125)
+ + " AND time < "
+ + (START + 377)
+ + " AND b >= 40 ORDER BY TIME DESC")) {
+ int r = 376, count = 0;
+ while (data.hasNext()) {
+ while (r >= 125 && b(r) < 40) r--;
+ RowRecord row = data.next();
+ require(r >= 125 && row.getTimestamp() == START + r, "filtered time",
r);
+ require(
+ row.getFields().get(0).getDoubleV() == a(r)
+ && row.getFields().get(1).getLongV() == d(r),
+ "filtered value",
+ r);
+ r--;
+ count++;
+ }
+ int expected = 0;
+ for (int i = 125; i < 377; i++) if (b(i) >= 40) expected++;
+ require(count == expected, "predicate count", count);
+ }
+ try (SessionDataSet data =
+ session.executeQueryStatement(
+ "SELECT count(a),count(c),sum(b),first_value(b),last_value(b) FROM
" + device)) {
+ require(data.hasNext(), "aggregate row", 0);
+ RowRecord row = data.next();
+ long sum = 0;
+ for (int r = 0; r < rows; r++) sum += b(r);
+ require(row.getFields().get(0).getLongV() == rows, "count(a)", 0);
+ require(row.getFields().get(1).getLongV() == rows - (rows + 6) / 7,
"count(c)", 0);
+ require(row.getFields().get(2).getDoubleV() == sum, "sum(b)", 0);
+ require(row.getFields().get(3).getIntV() == b(0), "first_value", 0);
+ require(row.getFields().get(4).getIntV() == b(rows - 1), "last_value",
0);
+ }
+ }
+
+ private static void require(boolean condition, String what, int row) {
+ if (!condition) throw new AssertionError(what + " at row " + row);
+ }
+}
diff --git
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterTableSessionIT.java
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterTableSessionIT.java
new file mode 100644
index 00000000000..03d86915867
--- /dev/null
+++
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterTableSessionIT.java
@@ -0,0 +1,77 @@
+/*
+ * 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;
+
+import org.apache.tsfile.encoding.table.ClusterTable;
+import org.apache.tsfile.encoding.table.ClusterTableOptions;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Standalone local integration check. Arguments: host port dedicatedDevice
write|read. */
+public final class ClusterTableSessionIT {
+ private ClusterTableSessionIT() {}
+
+ public static void main(String[] args) throws Exception {
+ if (args.length != 4 || (!args[3].equals("write") &&
!args[3].equals("read"))) {
+ throw new IllegalArgumentException("host port dedicatedDevice
write|read");
+ }
+ try (Session session =
+ new Session.Builder()
+ .host(args[0])
+ .port(Integer.parseInt(args[1]))
+ .username("root")
+ .password("root")
+ .enableRedirection(false)
+ .build()) {
+ session.open(false);
+ for (ClusterTableOptions.Method method :
ClusterTableOptions.Method.values()) {
+ ClusterTableOptions options = new ClusterTableOptions(method, 8, 3,
42);
+ ClusterTableSession tableSession =
+ new ClusterTableSession(
+ session,
+ args[2] + (method == ClusterTableOptions.Method.ACLUSTER ?
".a" : ".k"),
+ options);
+ ClusterTable table = ClusterTableSessionTest.sample(1200);
+ if (args[3].equals("write")) {
+ tableSession.createSchema();
+ tableSession.writeBlock(1, table.slice(0, 600));
+ tableSession.writeBlock(2, table.slice(600, 1200));
+ session.executeNonQueryStatement("flush");
+ }
+ Map<String, Integer> expected = new HashMap<>(), actual = new
HashMap<>();
+ ClusterTableSessionTest.records(expected, table);
+ tableSession.readAll(page -> ClusterTableSessionTest.records(actual,
page));
+ if (!expected.equals(actual))
+ throw new AssertionError("Complete-record mismatch: " + method);
+ expected.clear();
+ actual.clear();
+ long start = 1700000000010L, end = 1700000000020L;
+ ClusterTableSessionTest.records(expected, table.timeRange(start, end));
+ tableSession.readTimeRange(
+ start, end, page -> ClusterTableSessionTest.records(actual, page));
+ if (!expected.equals(actual))
+ throw new AssertionError("Original-time query mismatch: " + method);
+ System.out.println(
+ method + ": 1200 exact records; original-time range passed; mode="
+ args[3]);
+ }
+ }
+ }
+}
diff --git
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterTableSessionTest.java
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterTableSessionTest.java
new file mode 100644
index 00000000000..5712ed5c41a
--- /dev/null
+++
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/ClusterTableSessionTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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;
+
+import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.rpc.StatementExecutionException;
+
+import org.apache.tsfile.encoding.table.ClusterTable;
+import org.apache.tsfile.encoding.table.ClusterTableCodec;
+import org.apache.tsfile.encoding.table.ClusterTableOptions;
+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.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ClusterTableSessionTest {
+ static ClusterTable sample(int n) {
+ long[] times = new long[n];
+ Object[][] rows = new Object[n][4];
+ for (int r = 0; r < n; r++) {
+ times[r] = 1700000000000L + (r % 97);
+ rows[r][0] = (double) ((r * 17) % 31);
+ rows[r][1] = ((Double) rows[r][0]) * 3.0;
+ rows[r][2] = r % 11 == 0 ? -0.0 : Math.sin(r);
+ rows[r][3] = r % 13 == 0 ? null : Long.MAX_VALUE - r;
+ }
+ return new ClusterTable(
+ times,
+ new String[] {"x", "y", "z", "counter"},
+ new TSDataType[] {
+ TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.DOUBLE,
TSDataType.INT64
+ },
+ rows);
+ }
+
+ static void records(Map<String, Integer> result, ClusterTable table) {
+ for (int r = 0; r < table.rowCount(); r++) {
+ StringBuilder key = new StringBuilder().append(table.timestamp(r));
+ for (int c = 0; c < table.columnCount(); c++) {
+ key.append('/').append(table.isNull(r, c) ? "null" :
Long.toHexString(table.rawBits(r, c)));
+ }
+ result.merge(key.toString(), 1, Integer::sum);
+ }
+ }
+
+ @Test
+ public void rpcPayloadRoundTripAndOriginalTimeFilter() throws Exception {
+ for (ClusterTableOptions options :
+ Arrays.asList(ClusterTableOptions.aCluster(),
ClusterTableOptions.kCluster())) {
+ ISession session = mock(ISession.class);
+ List<RowRecord> stored = new ArrayList<>();
+ doAnswer(
+ call -> {
+ long id = call.getArgument(1);
+ List<Object> values = call.getArgument(4);
+ ClusterTable page = ClusterTableCodec.decode(((Binary)
values.get(0)).getValues());
+ assertEquals(page.rowCount(), values.get(3));
+ long min = Long.MAX_VALUE, max = Long.MIN_VALUE;
+ for (int r = 0; r < page.rowCount(); r++) {
+ min = Math.min(min, page.timestamp(r));
+ max = Math.max(max, page.timestamp(r));
+ }
+ assertEquals(min, values.get(1));
+ assertEquals(max, values.get(2));
+ RowRecord row = new RowRecord(id);
+ Field field = new Field(TSDataType.BLOB);
+ field.setBinaryV((Binary) values.get(0));
+ row.addField(field);
+ stored.add(row);
+ return null;
+ })
+ .when(session)
+ .insertRecord(anyString(), anyLong(), anyList(), anyList(),
anyList());
+ when(session.executeQueryStatement(anyString()))
+ .thenAnswer(
+ call -> {
+ SessionDataSet result = mock(SessionDataSet.class);
+ Iterator<RowRecord> rows = stored.iterator();
+ when(result.hasNext()).thenAnswer(ignore -> rows.hasNext());
+ when(result.next()).thenAnswer(ignore -> rows.next());
+ return result;
+ });
+ ClusterTableSession client = new ClusterTableSession(session,
"root.cluster_test.a", options);
+ ClusterTable table = sample(180);
+ client.createSchema();
+ client.writeTable(1, table);
+ client.writeBlock(2, table.slice(0, 20));
+ Map<String, Integer> expected = new HashMap<>(), actual = new
HashMap<>();
+ records(expected, table);
+ records(expected, table.slice(0, 20));
+ client.readAll(page -> records(actual, page));
+ assertEquals(expected, actual);
+ expected.clear();
+ actual.clear();
+ long start = 1700000000010L, end = 1700000000020L;
+ records(expected, table.timeRange(start, end));
+ records(expected, table.slice(0, 20).timeRange(start, end));
+ client.readTimeRange(start, end, page -> records(actual, page));
+ assertEquals(expected, actual);
+ assertThrows(IllegalArgumentException.class, () -> client.writeBlock(2,
table));
+ assertThrows(
+ IllegalArgumentException.class, () -> client.readTimeRange(end,
start, page -> {}));
+ }
+ }
+
+ @Test
+ public void failedInsertCannotSilentlyAdvance() throws Exception {
+ ISession session = mock(ISession.class);
+ doThrow(new StatementExecutionException("unknown outcome"))
+ .when(session)
+ .insertRecord(anyString(), anyLong(), anyList(), anyList(), anyList());
+ ClusterTableSession client =
+ new ClusterTableSession(session, "root.cluster_test.a",
ClusterTableOptions.aCluster());
+ assertThrows(StatementExecutionException.class, () -> client.writeBlock(1,
sample(2)));
+ assertThrows(IOException.class, () -> client.writeBlock(2, sample(2)));
+ }
+
+ @Test
+ public void rejectsUnsafeDevicePathAndEmptyBlocks() throws Exception {
+ ISession session = mock(ISession.class);
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new ClusterTableSession(
+ session, "root.a; delete timeseries root.**",
ClusterTableOptions.aCluster()));
+ ClusterTableSession client =
+ new ClusterTableSession(session, "root.cluster_test.a",
ClusterTableOptions.aCluster());
+ assertThrows(IllegalArgumentException.class, () -> client.writeBlock(1,
sample(0)));
+ }
+}
diff --git a/pom.xml b/pom.xml
index e2a099fdc85..1e2578fdf48 100644
--- a/pom.xml
+++ b/pom.xml
@@ -167,7 +167,7 @@
<thrift.version>0.14.1</thrift.version>
<xz.version>1.9</xz.version>
<zstd-jni.version>1.5.6-3</zstd-jni.version>
- <tsfile.version>2.1.0-250207-SNAPSHOT</tsfile.version>
+ <tsfile.version>2.1.0-SNAPSHOT</tsfile.version>
</properties>
<!--
if we claim dependencies in dependencyManagement, then we do not claim
