You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/10/09 07:57:38 UTC
(tsfile) branch research/cluster-compress updated: Add joint Cluster compression for aligned numeric series (#1000)
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch research/cluster-compress
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/research/cluster-compress by this push:
new 50eb213f2 Add joint Cluster compression for aligned numeric series (#1000)
50eb213f2 is described below
commit 50eb213f2a31cbf6e266f8497146fa609e4f5d53
Author: SamM755 <19...@users.noreply.github.com>
AuthorDate: Fri Oct 9 15:57:31 2026 +0800
Add joint Cluster compression for aligned numeric series (#1000)
---
java/tsfile/ALIGNED_CLUSTER.md | 155 +++++++
java/tsfile/CLUSTER_TABLE.md | 58 +++
.../tsfile/encoding/decoder/ClusterDecoder.java | 298 ++++++------
.../tsfile/encoding/decoder/ClusterReader.java | 10 +
.../apache/tsfile/encoding/decoder/Decoder.java | 4 +-
.../tsfile/encoding/encoder/AClusterAlgorithm.java | 22 +-
.../tsfile/encoding/encoder/AClusterEncoder.java | 441 +-----------------
.../tsfile/encoding/encoder/ClusterEncoder.java | 303 +++++++++++++
.../tsfile/encoding/encoder/ClusterResult.java | 71 +++
.../tsfile/encoding/encoder/KClusterAlgorithm.java | 96 ++--
.../tsfile/encoding/encoder/KClusterEncoder.java | 423 +----------------
.../tsfile/encoding/encoder/TSEncodingBuilder.java | 22 +-
.../tsfile/encoding/table/AlpColumnCodec.java | 179 ++++++++
.../tsfile/encoding/table/ClusterColumnBuffer.java | 61 +++
.../encoding/table/ClusterColumnSelector.java | 105 +++++
.../encoding/table/ClusterNativePageCodec.java | 261 +++++++++++
.../apache/tsfile/encoding/table/ClusterTable.java | 227 ++++++++++
.../tsfile/encoding/table/ClusterTableCodec.java | 361 +++++++++++++++
.../tsfile/encoding/table/ClusterTableEncoder.java | 61 +++
.../tsfile/encoding/table/ClusterTableOptions.java | 55 +++
.../tsfile/encoding/table/ClusterTableTsFile.java | 119 +++++
.../encoding/table/MultidimensionalCluster.java | 216 +++++++++
.../apache/tsfile/encoding/table/TableBlockIO.java | 119 +++++
.../apache/tsfile/read/reader/page/PageReader.java | 29 +-
.../tsfile/read/reader/page/ValuePageReader.java | 19 +-
.../write/chunk/AlignedChunkGroupWriterImpl.java | 8 +
.../tsfile/write/chunk/AlignedChunkWriterImpl.java | 13 +
.../tsfile/write/chunk/ValueChunkWriter.java | 15 +
.../write/page/ClusterAlignedPageWriter.java | 90 ++++
.../org/apache/tsfile/write/page/PageWriter.java | 103 ++++-
.../apache/tsfile/write/page/TimePageWriter.java | 32 +-
.../apache/tsfile/write/page/ValuePageWriter.java | 72 ++-
.../tsfile/encoding/decoder/ClusterCodecTest.java | 502 +++++++++++++++++++++
.../encoding/table/ClusterAlignedTsFileTest.java | 272 +++++++++++
.../encoding/table/ClusterTableCodecTest.java | 317 +++++++++++++
.../encoding/table/ClusterTableDataCheck.java | 116 +++++
36 files changed, 4165 insertions(+), 1090 deletions(-)
diff --git a/java/tsfile/ALIGNED_CLUSTER.md b/java/tsfile/ALIGNED_CLUSTER.md
new file mode 100644
index 000000000..4f142165e
--- /dev/null
+++ b/java/tsfile/ALIGNED_CLUSTER.md
@@ -0,0 +1,155 @@
+<!--
+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.
+-->
+
+# Native numeric Cluster pages (research)
+
+This contribution targets `research/cluster-compress` in apache/tsfile, with a
+companion contribution to the same research branch in apache/iotdb.
+
+Aligned INT32, INT64, FLOAT and DOUBLE measurements marked ACLUSTER or KCLUSTER
+use joint numeric-table encoding. Within an aligned writer, columns marked with
+the same method form one group. Other encodings keep their ordinary column path.
+This is native typed-column storage: the fields are not BLOB measurements.
+
+The writer collects complete aligned rows, selects columns with the CV-elbow
+selector, and performs one multidimensional ACluster/KCluster operation per block.
+Unselected or unsafe-to-scale columns use ALP/exceptions/RAW. The selection is
+fitted on the first block of this writer/schema and reused across its subsequent
+pages. A new chunk writer or changed schema fits a new selection; this is not a
+durable database-wide trained model. Native defaults are K=100, iterations=10,
+seed=0. Scalar codec properties are not currently exposed as joint-group options.
+
+## SQL through the companion IoTDB research branch
+
+Install this branch's Java common/tsfile artifacts before building IoTDB. All
+nodes opening the new files must use the matching research TsFile reader.
+
+```sql
+CREATE DATABASE root.cluster_demo;
+CREATE ALIGNED TIMESERIES root.cluster_demo.d (
+ c1 DOUBLE ENCODING=KCLUSTER,
+ c2 DOUBLE ENCODING=KCLUSTER,
+ c3 INT64 ENCODING=KCLUSTER
+);
+INSERT INTO root.cluster_demo.d(time,c1,c2,c3)
+ALIGNED VALUES (1,1.2,3.4,5),(2,2.4,6.8,10);
+FLUSH;
+SELECT c1,c2 FROM root.cluster_demo.d WHERE c3 >= 5;
+```
+
+Use ACLUSTER in place of KCLUSTER for the adaptive method. Mark the entire
+candidate numeric group; the selector chooses its clustered subspace internally.
+An ALP companion still has the group's encoding name in schema metadata, because
+the native page decoder dispatches its internal per-column mode.
+
+Single-column CREATE TIMESERIES ... WITH DATATYPE=FLOAT ENCODING=KCluster (or
+ACluster) remains supported. Its PageWriter also uses the timestamp-aware native
+page envelope, while direct scalar Encoder/Decoder APIs retain their existing
+unordered numeric-multiset stream contract.
+
+## Layout and query behavior
+
+ClusterNativePageCodec encodes a full table once, then copies each compressed
+column body into its own independently decodable column page. It does not run
+independent clustering per column. The prototype duplicates necessary counts,
+schema and timestamp keys in these projections; it does not duplicate other
+columns' values. This favors correctness and projection simplicity over metadata
+size. Do not report its file sizes as the benchmark container's compression ratio.
+
+Aligned value pages start with a negative native marker, version and bounded
+block counts. Scalar native pages add a zero time-stream-length prefix, which
+distinguishes them from nonempty legacy scalar pages. Nested CLTB v1 blocks retain
+their checksums; the existing BLOB block format is unchanged. Blocks have at most
+10,000 rows and 2,000,000 cells; a larger native page is split internally without
+changing the native page's record membership. Native pages are limited to 100,000
+rows and 1,024 candidate columns per method group.
+
+Compressed records remain in cluster order. Native readers expose actual
+timestamp keys in increasing order to satisfy IoTDB's scan interface. No original
+row number or inverse permutation is serialized, and the table codec does not
+promise restoration of input physical order. This SQL scan adaptation adds work
+and must be included in future native scan performance measurements.
+
+Timestamp keys are separate from the candidate numeric value columns. They follow
+the same row reordering but are stored as exact fixed-width INT64 keys in the
+table blocks, without ALP encoding. The aligned shared time column retains its
+configured native TsFile time encoding and compression.
+
+Native page/chunk statistics are computed from original typed input, including
+nulls, before compression. Existing filtering, projection and aggregation consume
+normal typed columns after page decoding. Native IoTDB duplicate-timestamp update
+semantics still apply: arbitrary duplicate timestamp tuples are not a bag of
+independent records at the native SQL interface. Equal-valued records at distinct
+timestamp keys remain separate.
+
+## Validation and limits
+
+ClusterAlignedTsFileTest covers row-wise and column-wise aligned writers,
+multiple pages/chunks, timestamp/value associations, single-column projection,
+nulls/all-null columns, mixed ordinary encodings, exact large integers, scalar
+native files, and malformed native envelopes. Existing ClusterTableCodecTest and
+ClusterCodecTest preserve the independent BLOB and direct scalar APIs.
+
+The companion Session module's ClusterAlignedSqlIT issues ordinary SQL only:
+aligned and scalar schema creation, numeric INSERT, FLUSH, typed SELECT,
+projection, time/value predicates, descending scans, count/sum/first/last, and a
+read-only mode for checking persisted data after server restart.
+
+Verified on 2026-10-08 (Asia/Shanghai): 72 distinct TsFile unit/regression cases and
+3 Session cases passed, including the retained BLOB tests. DataNode's 3,266 Java
+sources compiled against the companion TsFile dependency. On an isolated IoTDB
+2.0.2-SNAPSHOT server at the research baseline e5dacc0, using the newly built
+TsFile/common jars, each method passed 1,200 aligned SQL records and 400 scalar
+SQL records. A forced DataNode restart recovered the final 200 unflushed aligned
+records per method from WAL; the read-only checks repeated successfully. The
+separate BLOB API also read back 1,200 complete records per method on this same
+server before and after restart.
+
+The TsFile build ran Checkstyle, Spotless and license checks. Full-repository
+Failsafe integration tests were not run: the offline cache lacked their JUnit
+provider. The final targeted build used `-Dtsfile.it.skip=true`; live SQL checks
+were executed separately. Repeated Maven executions of the same tests are not
+counted as additional test cases.
+
+Example local build commands, after installing upstream dependencies:
+
+```text
+# TsFile root (common and parent artifacts already installed)
+mvn -o -B -P with-java -pl java/tsfile -Dtest=ClusterAlignedTsFileTest,ClusterTableCodecTest,ClusterCodecTest,ValuePageWriterTest,PageWriterTest -Dtsfile.it.skip=true -Dmaven.javadoc.skip=true install
+# IoTDB root
+mvn -o -B -pl iotdb-client/session -Dtest=ClusterTableSessionTest test
+mvn -o -B -pl iotdb-core/datanode -DskipTests compile
+```
+
+The isolated workspace runtime is `runtime/aligned_20261008`, with RPC port 22667;
+it does not use production directories. Logs include `native-verified-build.log`,
+`native-datanode-compile.log`, `native-session-final-test.log`,
+`native-sql-final-write.log`, `native-sql-final-restart-read.log`, and
+`native-blob-coexist-restart-read.log`. Start scripts and logs are in the outer
+research workspace rather than the Apache source tree.
+
+This remains a research prototype. Compaction variants, deletion/overwrite,
+schema evolution, replication, migration of older research files, and full-scale
+compression/throughput/memory benchmarks have not been comprehensively validated.
+The custom page format requires the modified readers. This document does not
+claim compatibility with unmodified released TsFile readers or completion of
+the repositories' entire test suites.
+
+The separate ClusterTableTsFile / ClusterTableSession BLOB APIs remain available;
+see CLUSTER_TABLE.md. Native SQL does not route through those APIs.
diff --git a/java/tsfile/CLUSTER_TABLE.md b/java/tsfile/CLUSTER_TABLE.md
new file mode 100644
index 000000000..a9b3aa42a
--- /dev/null
+++ b/java/tsfile/CLUSTER_TABLE.md
@@ -0,0 +1,58 @@
+<!--
+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 blocks (research)
+
+This opt-in Java API combines CV-elbow column selection, multidimensional ACluster/KCluster,
+and ALP for companion columns. It stores independent, versioned CLTB blocks in standard TsFile
+BLOB measurements. Its CLTB block format and direct scalar codec APIs are retained.
+Native numeric SQL pages are a separate opt-in path described in [ALIGNED_CLUSTER.md](ALIGNED_CLUSTER.md).
+
+Use `org.apache.tsfile.encoding.table.ClusterTable` with explicit timestamps, column names,
+INT32/INT64/FLOAT/DOUBLE types and strictly matching boxed values (or null).
+Use `ClusterTableOptions.aCluster()` or `kCluster()`.
+
+```java
+try (ClusterTableTsFile.Writer writer =
+ new ClusterTableTsFile.Writer(new File("table.tsfile"), options)) {
+ writer.append(table);
+}
+ClusterTableTsFile.read(new File("table.tsfile"), decoded -> {
+ // Consume complete records; original physical order is not promised.
+});
+```
+
+The writer selects from at most the first 10,000 rows and reuses that selection.
+Pages are independent, with at most 10,000 rows and 2,000,000 numeric cells each.
+All columns, timestamps and nulls share the cluster traversal; no inverse permutation is stored.
+Inexact scaling or unsupported ranges use exact ALP exceptions/RAW storage.
+All IEEE bits and all INT64 values are retained, including duplicate records.
+
+This is a block-container API. Standard SQL/ordinary TsFile readers see BLOB blocks,
+not the original numeric fields. The companion IoTDB research branch provides
+`org.apache.iotdb.session.ClusterTableSession` to decode and filter original records.
+
+The ALP implementation ports this project's exponent/factor + FOR + exceptions approach.
+CLTB is not wire-compatible with either the earlier C++ benchmark container or upstream ALP.
+The medoid objective uses the paper's exact bit-width formula (zero residual costs one bit).
+Java and C++ seeded random initialization are not bit-identical.
+
+Run `ClusterTableCodecTest` and the existing `ClusterCodecTest` through the normal Maven build.
+`ClusterTableDataCheck <outer-workspace>` optionally verifies the public-data selector manifest
+and bounded complete-record round trips. It is a functional check, not a performance benchmark.
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterDecoder.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterDecoder.java
index 59a914fdf..c7c2e540b 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterDecoder.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterDecoder.java
@@ -20,206 +20,218 @@
package org.apache.tsfile.encoding.decoder;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.encoding.TsFileDecodingException;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Arrays;
+/** Decodes cluster-grouped table values; input row order is intentionally not restored. */
public class ClusterDecoder extends Decoder {
-
private final TSDataType dataType;
-
- private long[] longValues;
- private double[] doubleValues;
- private int readIndex = 0;
- private int count = 0;
- private boolean hasDecoded = false;
+ private long[] values;
+ private int readIndex;
public ClusterDecoder(TSDataType dataType) {
- super(TSEncoding.ACLUSTER);
- this.dataType = dataType;
+ this(TSEncoding.ACLUSTER, dataType);
}
- /** check has next */
- @Override
- public boolean hasNext(ByteBuffer buffer) throws IOException {
- if (!hasDecoded) {
- decodeInternally(buffer);
+ public ClusterDecoder(TSEncoding encoding, TSDataType dataType) {
+ super(encoding);
+ switch (dataType) {
+ case INT32:
+ case DATE:
+ case INT64:
+ case TIMESTAMP:
+ case FLOAT:
+ case DOUBLE:
+ this.dataType = dataType;
+ break;
+ default:
+ throw new TsFileDecodingException("Unsupported cluster type: " + dataType);
}
- return readIndex < count;
}
- /** reset */
+ @Override
+ public boolean hasNext(ByteBuffer buffer) {
+ return ensureBlock(buffer);
+ }
+
@Override
public void reset() {
- this.hasDecoded = false;
- this.readIndex = 0;
- this.count = 0;
- this.longValues = null;
- this.doubleValues = null;
+ values = null;
+ readIndex = 0;
}
@Override
public int readInt(ByteBuffer buffer) {
- return (int) longValues[readIndex++];
+ require(dataType == TSDataType.INT32 || dataType == TSDataType.DATE, "Wrong read type");
+ return (int) nextValue(buffer);
}
@Override
public long readLong(ByteBuffer buffer) {
- return longValues[readIndex++];
+ require(dataType == TSDataType.INT64 || dataType == TSDataType.TIMESTAMP, "Wrong read type");
+ return nextValue(buffer);
}
@Override
public float readFloat(ByteBuffer buffer) {
- return (float) doubleValues[readIndex++];
+ require(dataType == TSDataType.FLOAT, "Wrong read type");
+ return Float.intBitsToFloat((int) nextValue(buffer));
}
@Override
public double readDouble(ByteBuffer buffer) {
- return doubleValues[readIndex++];
+ require(dataType == TSDataType.DOUBLE, "Wrong read type");
+ return Double.longBitsToDouble(nextValue(buffer));
}
- /** Internal decode method Decode ACluster */
- private void decodeInternally(ByteBuffer buffer) {
- if (hasDecoded) {
- return;
- }
+ private long nextValue(ByteBuffer buffer) {
+ require(ensureBlock(buffer), "No more cluster values");
+ return values[readIndex++];
+ }
- ClusterReader reader = new ClusterReader(buffer);
+ private boolean ensureBlock(ByteBuffer buffer) {
+ if (values != null && readIndex < values.length) {
+ return true;
+ }
+ if (!buffer.hasRemaining()) {
+ return false;
+ }
+ // Only consume the caller's bytes after a whole block has passed validation.
+ ByteBuffer input = buffer.duplicate();
+ try {
+ long[] decoded = decodeBlock(new ClusterReader(input));
+ buffer.position(input.position());
+ values = decoded;
+ readIndex = 0;
+ return true;
+ } catch (ArithmeticException | IllegalArgumentException e) {
+ throw new TsFileDecodingException("Invalid cluster block: " + e.getMessage());
+ }
+ }
- // --- Header ---
- int scalingExponent = (int) reader.read(8);
+ private long[] decodeBlock(ClusterReader reader) {
+ int scale = (int) reader.read(8);
int k = (int) reader.read(16);
- this.count = (int) reader.read(16);
+ int count = (int) reader.read(16);
int packSize = (int) reader.read(16);
-
- // --- Global Minimum Value (minVal) ---
- int minValBit = (int) reader.read(8);
- long minValSign = reader.read(1);
- long absMinVal = reader.read(minValBit); // Read the absolute value part
- long minVal = (minValSign == 1) ? -absMinVal : absMinVal; // Apply the sign
-
- // Allocate memory based on count and dataType
- if (this.count > 0) {
- if (dataType == TSDataType.FLOAT || dataType == TSDataType.DOUBLE) {
- this.doubleValues = new double[this.count];
- } else {
- this.longValues = new long[this.count];
+ require(count > 0 && k <= count && packSize > 0, "Invalid cluster header");
+ int minBits = (int) reader.read(8);
+ if (minBits == 0) {
+ require(scale == 0 && k == 0, "Invalid RAW block header");
+ require(reader.read(8) == 1, "Unsupported cluster RAW version");
+ require(reader.read(8) == (dataType.serialize() & 0xff), "RAW block type mismatch");
+ int width = is32Bit() ? 32 : 64;
+ require(reader.remainingBits() >= (long) count * width, "Truncated RAW block");
+ long[] raw = new long[count];
+ for (int i = 0; i < count; i++) {
+ long bits = reader.read(width);
+ raw[i] = dataType == TSDataType.INT32 || dataType == TSDataType.DATE ? (int) bits : bits;
}
+ return raw;
}
-
- // Handle case where all values are the same (k=0)
+ require(isFloating() || scale == 0, "Integer cluster block has a decimal scale");
+ long min = readSignedMagnitude(reader, minBits);
+ long[] result = new long[count];
if (k == 0) {
- if (this.count > 0) {
- reconstructFromMinValOnly(minVal, scalingExponent);
- }
- this.hasDecoded = true;
- return;
+ Arrays.fill(result, toValueBits(min, scale));
+ return result;
}
- // --- Medoids ---
- long[] medoids = new long[k];
- int minMedoidBit = (int) reader.read(8);
- long minMedoidSign = reader.read(1);
- long absMinMedoid = reader.read(minMedoidBit); // Read the absolute value part
- long minMedoid = (minMedoidSign == 1) ? -absMinMedoid : absMinMedoid; // Apply the sign
-
- int maxMedoidOffsetBits = (int) reader.read(8);
+ int referenceMinBits = (int) reader.read(8);
+ long referenceMin = readSignedMagnitude(reader, referenceMinBits);
+ int referenceBits = (int) reader.read(8);
+ require(referenceBits >= 1 && referenceBits <= 63, "Invalid reference width");
+ require(referenceMin >= 0, "Negative normalized reference");
+ long[] references = new long[k];
for (int i = 0; i < k; i++) {
- long offset = reader.read(maxMedoidOffsetBits);
- medoids[i] = minMedoid + offset;
+ references[i] = Math.addExact(referenceMin, reader.read(referenceBits));
}
-
- // --- Frequencies (Cluster Sizes) ---
- // The encoder wrote deltas (cluster sizes), so we read them and rebuild the cumulative array
- long[] cumulativeFrequencies = new long[k];
- int numFreqBlocks = (int) reader.read(16);
-
- // Metadata pass for frequencies
- int[] freqBlockMaxBits = new int[numFreqBlocks];
- for (int i = 0; i < numFreqBlocks; i++) {
- freqBlockMaxBits[i] = (int) reader.read(8);
+ long[] counts = readPacks(reader, k, packSize, 16, false);
+ long previousCount = 0;
+ long total = 0;
+ for (int i = 0; i < k; i++) {
+ // Stored values are differences between sorted cluster sizes, not prefix positions.
+ counts[i] = Math.addExact(previousCount, counts[i]);
+ require(counts[i] >= 0 && counts[i] <= count, "Invalid cluster frequency");
+ previousCount = counts[i];
+ total = Math.addExact(total, counts[i]);
}
-
- // Data pass for frequencies - Reconstruct cumulative frequencies
- long currentCumulativeFreq = 0;
- int freqIndex = 0;
- for (int i = 0; i < numFreqBlocks; i++) {
- int start = i * packSize;
- int end = Math.min(start + packSize, k);
- int bitsForBlock = freqBlockMaxBits[i];
- for (int j = start; j < end; j++) {
- long delta = reader.read(bitsForBlock); // This delta is the actual cluster size
- currentCumulativeFreq += delta;
- cumulativeFrequencies[freqIndex++] = currentCumulativeFreq;
+ require(total == count, "Cluster frequencies do not sum to the value count");
+ long[] residuals = readPacks(reader, count, packSize, 32, true);
+ int position = 0;
+ for (int i = 0; i < k; i++) {
+ for (int j = 0; j < counts[i]; j++) {
+ long zigzag = residuals[position];
+ long residual = (zigzag >>> 1) ^ -(zigzag & 1);
+ long offset = Math.addExact(references[i], residual);
+ require(offset >= 0, "Negative normalized value");
+ result[position++] = toValueBits(Math.addExact(min, offset), scale);
}
}
+ return result;
+ }
- // --- Residuals ---
- long[] residuals = new long[this.count];
- int numPacks = (int) reader.read(32);
-
- // Metadata pass for residuals
- int[] resPackMaxBits = new int[numPacks];
- for (int i = 0; i < numPacks; i++) {
- resPackMaxBits[i] = (int) reader.read(8);
+ private long[] readPacks(
+ ClusterReader reader, int count, int packSize, int packCountBits, boolean allowZeroWidth) {
+ int expected = (count + packSize - 1) / packSize;
+ require(reader.read(packCountBits) == expected, "Invalid pack count");
+ int[] widths = new int[expected];
+ long requiredBits = 0;
+ for (int i = 0; i < expected; i++) {
+ widths[i] = (int) reader.read(8);
+ require(widths[i] >= (allowZeroWidth ? 0 : 1) && widths[i] <= 64, "Invalid pack width");
+ requiredBits += (long) widths[i] * Math.min(packSize, count - i * packSize);
}
-
- // Data pass for residuals
- int residualIdx = 0;
- for (int i = 0; i < numPacks; i++) {
- int start = i * packSize;
- int end = Math.min(start + packSize, this.count);
- int bitsForPack = resPackMaxBits[i];
- if (bitsForPack > 0) {
- for (int j = start; j < end; j++) {
- residuals[residualIdx++] = reader.read(bitsForPack);
- }
- } else {
- // If bitsForPack is 0, all residuals in this pack are 0.
- // We just need to advance the index, as the array is already initialized to 0.
- residualIdx += (end - start);
- }
+ require(reader.remainingBits() >= requiredBits, "Truncated cluster payload");
+ long[] data = new long[count];
+ for (int i = 0; i < count; i++) {
+ int width = widths[i / packSize];
+ data[i] = width == 0 ? 0 : reader.read(width);
}
-
- // --- Final Data Reconstruction ---
- // Use the correctly reconstructed cumulativeFrequencies array
- reconstructData(medoids, cumulativeFrequencies, residuals, minVal, scalingExponent);
-
- this.hasDecoded = true;
+ return data;
}
- private void reconstructData(
- long[] medoids, long[] frequencies, long[] residuals, long minVal, int scalingExponent) {
- int residualReadPos = 0;
- int dataWritePos = 0;
- double scalingFactor = Math.pow(10, scalingExponent);
+ private long readSignedMagnitude(ClusterReader reader, int width) {
+ require(width >= 1 && width <= 64, "Invalid signed value width");
+ boolean negative = reader.read(1) != 0;
+ long magnitude = reader.read(width);
+ require(
+ magnitude >= 0 || (negative && magnitude == Long.MIN_VALUE),
+ "Signed magnitude out of range");
+ return negative ? -magnitude : magnitude;
+ }
- for (int clusterId = 0; clusterId < medoids.length; clusterId++) {
- long pointsInThisCluster = frequencies[clusterId];
- long medoid = medoids[clusterId];
+ private long toValueBits(long scaled, int scale) {
+ if (isFloating()) {
+ double value = scaled / Math.pow(10, scale);
+ return dataType == TSDataType.FLOAT
+ ? Float.floatToRawIntBits((float) value) & 0xffffffffL
+ : Double.doubleToRawLongBits(value);
+ }
+ if (is32Bit()) {
+ require(
+ scaled >= Integer.MIN_VALUE && scaled <= Integer.MAX_VALUE,
+ "Cluster value exceeds INT32");
+ }
+ return scaled;
+ }
- for (int i = 0; i < pointsInThisCluster; i++) {
- long zigzagResidual = residuals[residualReadPos++];
- long residual = (zigzagResidual >>> 1) ^ -(zigzagResidual & 1);
- long scaledDataPoint = medoid + residual + minVal;
+ private boolean isFloating() {
+ return dataType == TSDataType.FLOAT || dataType == TSDataType.DOUBLE;
+ }
- if (dataType == TSDataType.FLOAT || dataType == TSDataType.DOUBLE) {
- doubleValues[dataWritePos++] = scaledDataPoint / scalingFactor;
- } else {
- longValues[dataWritePos++] = scaledDataPoint;
- }
- }
- }
+ private boolean is32Bit() {
+ return dataType == TSDataType.INT32
+ || dataType == TSDataType.DATE
+ || dataType == TSDataType.FLOAT;
}
- private void reconstructFromMinValOnly(long minVal, int scalingExponent) {
- if (dataType == TSDataType.FLOAT || dataType == TSDataType.DOUBLE) {
- double scalingFactor = Math.pow(10, scalingExponent);
- double finalValue = minVal / scalingFactor;
- for (int i = 0; i < this.count; i++) doubleValues[i] = finalValue;
- } else {
- for (int i = 0; i < this.count; i++) longValues[i] = minVal;
+ private static void require(boolean condition, String message) {
+ if (!condition) {
+ throw new TsFileDecodingException(message);
}
}
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterReader.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterReader.java
index 99ebb660e..f56758445 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterReader.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/ClusterReader.java
@@ -19,6 +19,8 @@
package org.apache.tsfile.encoding.decoder;
+import org.apache.tsfile.exception.encoding.TsFileDecodingException;
+
import java.nio.ByteBuffer;
public class ClusterReader {
@@ -38,6 +40,10 @@ public class ClusterReader {
"Cannot read more than 64 bits or non-positive bits at once.");
}
+ if (remainingBits() < numBits) {
+ throw new TsFileDecodingException("Truncated cluster bit stream");
+ }
+
long result = 0;
for (int i = 0; i < numBits; i++) {
if (bitPosition < 0) {
@@ -52,4 +58,8 @@ public class ClusterReader {
}
return result;
}
+
+ public long remainingBits() {
+ return (long) buffer.remaining() * 8 + bitPosition + 1;
+ }
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/Decoder.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/Decoder.java
index 9f3a72462..15a779f23 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/Decoder.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/decoder/Decoder.java
@@ -181,10 +181,12 @@ public abstract class Decoder {
case KCLUSTER:
switch (dataType) {
case INT32:
+ case DATE:
case INT64:
+ case TIMESTAMP:
case FLOAT:
case DOUBLE:
- return new ClusterDecoder(dataType);
+ return new ClusterDecoder(encoding, dataType);
default:
throw new TsFileDecodingException(String.format(ERROR_MSG, encoding, dataType));
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterAlgorithm.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterAlgorithm.java
index f4e079f5a..83c492ac8 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterAlgorithm.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterAlgorithm.java
@@ -32,18 +32,16 @@ public final class AClusterAlgorithm {
/**
* The main entry point for the ACluster algorithm. It processes a page of data and returns the
- * results as an Object array.
+ * frequency-sorted references, assignments and counts.
*
* @param data The input time series data for a single page, represented as a long array.
- * @return An Object array where: <br>
- * - index 0: long[] medoids (sorted by cluster frequency) <br>
- * - index 1: int[] clusterAssignments (mapped to the sorted medoids) <br>
- * - index 2: long[] clusterFrequencies (sorted)
+ * @return References sorted by frequency, with matching assignments and counts.
*/
- public static Object[] run(long[] data) {
+ public static ClusterResult run(long[] data) {
+ ClusterResult.validateInput(data);
int n = data.length;
if (n == 0) {
- return new Object[] {new long[0], new int[0], new long[0]};
+ return new ClusterResult(new long[0], new int[0], new long[0]);
}
// --- Initialization ---
@@ -84,7 +82,8 @@ public final class AClusterAlgorithm {
}
// --- Step 4: Calculate potential savings ---
- long savingsFromCurrentPoint = minCostToExistingMedoid;
+ long savingsFromCurrentPoint =
+ minCostToExistingMedoid - calculateResidualCost(currentPoint, currentPoint);
long savingsFromReassignment = 0;
Set<Integer> pointsInBestCluster = pointsInClusters.get(bestMedoidIndex);
for (int pointIndexInCluster : pointsInBestCluster) {
@@ -161,9 +160,10 @@ public final class AClusterAlgorithm {
* @param medoids The discovered medoids.
* @param clusterAssignment The assignment map for each data point.
* @param clusterSize The frequency of each cluster.
- * @return A sorted and correctly mapped Object array.
+ * @return Sorted references with matching assignments and counts.
*/
- private static Object[] sortResults(long[] medoids, int[] clusterAssignment, long[] clusterSize) {
+ private static ClusterResult sortResults(
+ long[] medoids, int[] clusterAssignment, long[] clusterSize) {
int k = medoids.length;
List<MedoidSortHelper> sorters = new ArrayList<>();
for (int i = 0; i < k; i++) {
@@ -188,7 +188,7 @@ public final class AClusterAlgorithm {
sortedClusterAssignment[i] = oldToNewIndexMap[oldIndex];
}
- return new Object[] {sortedMedoids, sortedClusterAssignment, sortedClusterSize};
+ return new ClusterResult(sortedMedoids, sortedClusterAssignment, sortedClusterSize);
}
// --- Cost Calculation Functions ---
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterEncoder.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterEncoder.java
index 1cea597af..43e9f941e 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterEncoder.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/AClusterEncoder.java
@@ -20,447 +20,16 @@
package org.apache.tsfile.encoding.encoder;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.exception.encoding.TsFileEncodingException;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.utils.Binary;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * AClusterEncoder is a batch-based compressor for numerical data (INT32, INT64, FLOAT, DOUBLE). It
- * buffers all data points within a page, then on flush(), applies a clustering algorithm to find
- * optimal global reference points and encodes residuals.
- */
-public class AClusterEncoder extends Encoder {
-
- private static final int DEFAULT_PACK_SIZE = 10;
-
- /** A buffer to store all values of a page before flushing. We use long to unify all types. */
- private interface ValueBuffer {
- /** Adds a value to the buffer. */
- void add(int value);
-
- void add(long value);
-
- void add(float value);
-
- void add(double value);
-
- /** Clears the internal buffer. */
- void clear();
-
- /** Checks if the buffer is empty. */
- boolean isEmpty();
-
- /** Gets the current size of the buffer. */
- int size();
-
- ProcessingResult processAndGet();
- }
-
- private static class Int64Buffer implements ValueBuffer {
- private final List<Long> values = new ArrayList<>();
-
- @Override
- public void add(int value) {
- values.add((long) value);
- }
-
- @Override
- public void add(long value) {
- values.add(value);
- }
-
- @Override
- public void add(float value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public void add(double value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public ProcessingResult processAndGet() { // <--- 实现新方法
- long[] data = values.stream().mapToLong(l -> l).toArray();
- return new ProcessingResult(data, 0); // Exponent is 0 for integers
- }
-
- @Override
- public void clear() {
- values.clear();
- }
-
- @Override
- public boolean isEmpty() {
- return values.isEmpty();
- }
-
- @Override
- public int size() {
- return values.size();
- }
- }
-
- /** A buffer for FLOAT and DOUBLE types. It performs scaling in processAndGetLongs(). */
- private static class DoubleBuffer implements ValueBuffer {
- private final List<Double> values = new ArrayList<>();
-
- @Override
- public void add(int value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public void add(long value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public void add(float value) {
- values.add((double) value);
- } // Store as double to unify
-
- @Override
- public void add(double value) {
- values.add(value);
- }
-
- @Override
- public ProcessingResult processAndGet() {
- // --- Edge Case: Handle empty buffer ---
- if (values.isEmpty()) {
- return new ProcessingResult(new long[0], 0);
- }
- int maxDecimalPlaces = 0;
- for (double v : values) {
- String s = BigDecimal.valueOf(v).toPlainString();
- int dotIndex = s.indexOf('.');
- if (dotIndex != -1) {
- int decimalPlaces = s.length() - dotIndex - 1;
- if (decimalPlaces > maxDecimalPlaces) {
- maxDecimalPlaces = decimalPlaces;
- }
- }
- }
-
- double scalingFactor = Math.pow(10, maxDecimalPlaces);
-
- long[] scaledLongs = new long[values.size()];
- for (int i = 0; i < values.size(); i++) {
- scaledLongs[i] = Math.round(values.get(i) * scalingFactor);
- }
-
- return new ProcessingResult(scaledLongs, maxDecimalPlaces);
- }
-
- @Override
- public void clear() {
- values.clear();
- }
-
- @Override
- public boolean isEmpty() {
- return values.isEmpty();
- }
-
- @Override
- public int size() {
- return values.size();
- }
- }
-
- private static class ProcessingResult {
- final long[] scaledLongs;
- final int scalingExponent; // e.g., 3 for a scaling factor of 1000
-
- ProcessingResult(long[] scaledLongs, int scalingExponent) {
- this.scaledLongs = scaledLongs;
- this.scalingExponent = scalingExponent;
- }
-
- long[] getScaledLongs() {
- return this.scaledLongs;
- }
-
- int getScalingExponent() {
- return this.scalingExponent;
- }
- }
-
- private final ValueBuffer buffer;
-
- /**
- * Constructor for AClusterEncoder. It's called by AClusterEncodingBuilder.
- *
- * @param dataType The data type of the time series, used for potential type-specific logic.
- */
+/** Adaptive reference-point encoding for an unordered, single-column table. */
+public class AClusterEncoder extends ClusterEncoder {
public AClusterEncoder(TSDataType dataType) {
- super(TSEncoding.ACLUSTER);
- switch (dataType) {
- case INT32:
- case INT64:
- this.buffer = new Int64Buffer();
- break;
- case FLOAT:
- case DOUBLE:
- this.buffer = new DoubleBuffer();
- break;
- default:
- throw new TsFileEncodingException(
- "AClusterEncoder does not support data type: " + dataType);
- }
- }
-
- @Override
- public void encode(int value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- @Override
- public void encode(long value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- @Override
- public void encode(float value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- @Override
- public void encode(double value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- @Override
- public void flush(ByteArrayOutputStream out) throws IOException {
- if (buffer.isEmpty()) {
- return;
- }
-
- ProcessingResult procResult = buffer.processAndGet();
- long[] originalData = procResult.getScaledLongs();
- int scalingExponent = procResult.getScalingExponent();
- if (originalData.length == 0) return;
- long minVal = findMin(originalData);
- long[] data = new long[originalData.length];
- for (int i = 0; i < data.length; i++) {
- data[i] = originalData[i] - minVal;
- }
-
- Object[] clusterResult = AClusterAlgorithm.run(data);
- long[] sortedMedoids = (long[]) clusterResult[0];
- int[] clusterAssignments = (int[]) clusterResult[1];
- long[] clusterFrequencies = (long[]) clusterResult[2];
-
- long[] sortedZigzagResiduals =
- calculateSortedZigzagResiduals(data, sortedMedoids, clusterAssignments, clusterFrequencies);
-
- encodeResults(
- out, scalingExponent, minVal, sortedMedoids, clusterFrequencies, sortedZigzagResiduals);
-
- buffer.clear();
- }
-
- private static class MedoidFreqPair {
- long medoid;
- long frequency;
- int originalIndex;
-
- MedoidFreqPair(long medoid, long frequency, int originalIndex) {
- this.medoid = medoid;
- this.frequency = frequency;
- this.originalIndex = originalIndex;
- }
- }
-
- /**
- * A direct translation of your `residualCalculationZigzagNoHuff_sorted` logic. It computes
- * residuals, applies zigzag encoding, and groups them by cluster.
- */
- private long[] calculateSortedZigzagResiduals(
- long[] data, long[] medoids, int[] assignments, long[] frequencies) {
- int n = data.length;
- int k = medoids.length;
- if (n == 0) return new long[0];
-
- long[] sortedResiduals = new long[n];
- int[] writePointers = new int[k];
- int cumulativeCount = 0;
- for (int i = 0; i < k; i++) {
- writePointers[i] = cumulativeCount;
- cumulativeCount += (int) frequencies[i];
- }
-
- for (int i = 0; i < n; i++) {
- int clusterId = assignments[i];
- long medoid = medoids[clusterId];
- long residual = data[i] - medoid;
- long zigzagResidual = (residual << 1) ^ (residual >> 63); // Zigzag Encoding
-
- int targetIndex = writePointers[clusterId];
- sortedResiduals[targetIndex] = zigzagResidual;
- writePointers[clusterId]++;
- }
- return sortedResiduals;
- }
-
- private void encodeResults(
- ByteArrayOutputStream out,
- int scalingExponent,
- long minVal,
- long[] medoids,
- long[] frequencies,
- long[] residuals)
- throws IOException {
-
- ClusterSupport writer = new ClusterSupport(out);
- int numPoints = residuals.length;
- int k = medoids.length;
-
- writer.write(scalingExponent, 8);
- writer.write(k, 16);
- writer.write(numPoints, 16);
- writer.write(DEFAULT_PACK_SIZE, 16);
-
- int minValBit = ClusterSupport.bitsRequired(Math.abs(minVal));
- writer.write(minValBit, 8);
- writer.write(minVal >= 0 ? 0 : 1, 1);
- writer.write(minVal, minValBit);
-
- if (k == 0) {
- writer.flush();
- return;
- }
-
- long minMedoid = findMin(medoids);
- long[] medoidOffsets = new long[k];
- int maxMedoidOffsetBits = 0;
- for (int i = 0; i < k; i++) {
- medoidOffsets[i] = medoids[i] - minMedoid;
- maxMedoidOffsetBits =
- Math.max(maxMedoidOffsetBits, ClusterSupport.bitsRequired(medoidOffsets[i]));
- }
- int minMedoidBit = ClusterSupport.bitsRequired(Math.abs(minMedoid));
- writer.write(minMedoidBit, 8);
- writer.write(minMedoid > 0 ? 0 : 1, 1);
- writer.write(minMedoid, minMedoidBit);
-
- writer.write(maxMedoidOffsetBits, 8);
- for (long offset : medoidOffsets) {
- writer.write(offset, maxMedoidOffsetBits);
- }
-
- long[] freqDeltas = new long[k];
- if (k > 0) {
- freqDeltas[0] = frequencies[0];
- for (int i = 1; i < k; i++) {
- freqDeltas[i] = frequencies[i] - frequencies[i - 1];
- }
- }
- int numFreqBlocks = (k + DEFAULT_PACK_SIZE - 1) / DEFAULT_PACK_SIZE;
- writer.write(numFreqBlocks, 16);
-
- int[] freqBlockMaxBits = new int[numFreqBlocks];
- // Metadata pass for frequencies
- for (int i = 0; i < numFreqBlocks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, k);
- long maxDelta = 0;
- for (int j = start; j < end; j++) {
- maxDelta = Math.max(maxDelta, freqDeltas[j]);
- }
- freqBlockMaxBits[i] = ClusterSupport.bitsRequired(maxDelta);
- writer.write(freqBlockMaxBits[i], 8);
- }
- // Data pass for frequencies
- for (int i = 0; i < numFreqBlocks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, k);
- int bitsForBlock = freqBlockMaxBits[i];
- for (int j = start; j < end; j++) {
- writer.write(freqDeltas[j], bitsForBlock);
- }
- }
-
- int numPacks = (numPoints + DEFAULT_PACK_SIZE - 1) / DEFAULT_PACK_SIZE;
- writer.write(numPacks, 32);
-
- int[] resPackMaxBits = new int[numPacks];
- // Metadata pass for residuals
- for (int i = 0; i < numPacks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, numPoints);
- long maxOffset = 0;
- for (int j = start; j < end; j++) {
- maxOffset = Math.max(maxOffset, residuals[j]);
- }
- resPackMaxBits[i] = ClusterSupport.bitsRequired(maxOffset);
- writer.write(resPackMaxBits[i], 8);
- }
- // Data pass for residuals
- for (int i = 0; i < numPacks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, numPoints);
- int bitsForPack = resPackMaxBits[i];
- if (bitsForPack > 0) {
- for (int j = start; j < end; j++) {
- writer.write(residuals[j], bitsForPack);
- }
- }
- }
-
- writer.flush();
- }
-
- private long findMin(long[] data) {
- if (data == null || data.length == 0) {
- throw new IllegalArgumentException("Data array cannot be null or empty.");
- }
- long min = data[0];
- for (int i = 1; i < data.length; i++) {
- if (data[i] < min) {
- min = data[i];
- }
- }
- return min;
- }
-
- @Override
- public int getOneItemMaxSize() {
- return 8;
- }
-
- @Override
- public long getMaxByteSize() {
- if (this.buffer.isEmpty()) {
- return 0;
- }
- return (long) this.buffer.size() * getOneItemMaxSize() * 3 / 2;
- }
-
- @Override
- public void encode(boolean value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support boolean values.");
- }
-
- @Override
- public void encode(short value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support short values.");
- }
-
- @Override
- public void encode(Binary value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support Binary values.");
+ super(TSEncoding.ACLUSTER, dataType);
}
@Override
- public void encode(BigDecimal value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support BigDecimal values.");
+ protected ClusterResult cluster(long[] data) {
+ return AClusterAlgorithm.run(data);
}
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/ClusterEncoder.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/ClusterEncoder.java
new file mode 100644
index 000000000..01a8362c2
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/ClusterEncoder.java
@@ -0,0 +1,303 @@
+/*
+ * 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.tsfile.encoding.encoder;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.encoding.TsFileEncodingException;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.Arrays;
+
+/**
+ * Shared single-column table-data codec. Values may be reordered within each block; only the
+ * multiset of values is preserved. No row-position or cluster-ID stream is stored.
+ */
+public abstract class ClusterEncoder extends Encoder {
+ public static final int MAX_BLOCK_VALUES = 10_000;
+ private static final int PACK_SIZE = 10;
+ private final TSDataType dataType;
+ // Integers or raw IEEE bits, so even NaN payloads and negative zero survive a RAW block.
+ private long[] values = new long[128];
+ private int size;
+
+ protected ClusterEncoder(TSEncoding encoding, TSDataType dataType) {
+ super(encoding);
+ switch (dataType) {
+ case INT32:
+ case DATE:
+ case INT64:
+ case TIMESTAMP:
+ case FLOAT:
+ case DOUBLE:
+ this.dataType = dataType;
+ break;
+ default:
+ throw new TsFileEncodingException(encoding + " does not support " + dataType);
+ }
+ }
+
+ protected abstract ClusterResult cluster(long[] data);
+
+ @Override
+ public void encode(int value, ByteArrayOutputStream out) {
+ requireType(dataType == TSDataType.INT32 || dataType == TSDataType.DATE);
+ append(value, out);
+ }
+
+ @Override
+ public void encode(long value, ByteArrayOutputStream out) {
+ requireType(dataType == TSDataType.INT64 || dataType == TSDataType.TIMESTAMP);
+ append(value, out);
+ }
+
+ @Override
+ public void encode(float value, ByteArrayOutputStream out) {
+ requireType(dataType == TSDataType.FLOAT);
+ append(Float.floatToRawIntBits(value) & 0xffffffffL, out);
+ }
+
+ @Override
+ public void encode(double value, ByteArrayOutputStream out) {
+ requireType(dataType == TSDataType.DOUBLE);
+ append(Double.doubleToRawLongBits(value), out);
+ }
+
+ private void requireType(boolean valid) {
+ if (!valid) {
+ throw new TsFileEncodingException("Value type does not match cluster schema " + dataType);
+ }
+ }
+
+ private void append(long value, ByteArrayOutputStream out) {
+ if (size == MAX_BLOCK_VALUES) {
+ try {
+ flush(out);
+ } catch (IOException e) {
+ throw new TsFileEncodingException("Cannot flush cluster block: " + e.getMessage());
+ }
+ }
+ if (size == values.length) {
+ values = Arrays.copyOf(values, Math.min(MAX_BLOCK_VALUES, values.length * 2));
+ }
+ values[size++] = value;
+ }
+
+ @Override
+ public void flush(ByteArrayOutputStream out) throws IOException {
+ if (size == 0) {
+ return;
+ }
+ long[] scaled = new long[size];
+ int scale = scaleValues(scaled);
+ ByteArrayOutputStream candidate = null;
+ if (scale >= 0) {
+ long min = scaled[0];
+ for (long value : scaled) {
+ min = Math.min(min, value);
+ }
+ boolean safe = min != Long.MIN_VALUE;
+ for (int i = 0; safe && i < size; i++) {
+ try {
+ scaled[i] = Math.subtractExact(scaled[i], min);
+ } catch (ArithmeticException e) {
+ safe = false;
+ }
+ // Bound residual ZigZag and every KCluster initialization distance sum.
+ safe &= scaled[i] >= 0 && scaled[i] <= Long.MAX_VALUE / (2L * size);
+ }
+ if (safe) {
+ candidate = new ByteArrayOutputStream();
+ boolean constant = true;
+ for (long value : scaled) {
+ constant &= value == 0;
+ }
+ writeCluster(candidate, scale, min, scaled, constant ? null : cluster(scaled));
+ }
+ }
+
+ int rawBytes = 10 + size * valueByteWidth();
+ if (candidate != null && candidate.size() <= rawBytes) {
+ candidate.writeTo(out);
+ } else {
+ writeRaw(out);
+ }
+ size = 0;
+ }
+
+ // A negative result selects RAW. A scale is accepted only after exact bitwise round-trip.
+ private int scaleValues(long[] scaled) {
+ if (dataType != TSDataType.FLOAT && dataType != TSDataType.DOUBLE) {
+ System.arraycopy(values, 0, scaled, 0, size);
+ return 0;
+ }
+ int scale = 0;
+ for (int i = 0; i < size; i++) {
+ double value = floatingValue(i);
+ if (!Double.isFinite(value)) {
+ return -1;
+ }
+ scale = Math.max(scale, decimalValue(i).stripTrailingZeros().scale());
+ if (scale > 18) {
+ return -1;
+ }
+ }
+ double factor = Math.pow(10, scale);
+ for (int i = 0; i < size; i++) {
+ try {
+ scaled[i] = decimalValue(i).movePointRight(scale).longValueExact();
+ } catch (ArithmeticException e) {
+ return -1;
+ }
+ double restored = scaled[i] / factor;
+ long bits =
+ dataType == TSDataType.FLOAT
+ ? Float.floatToRawIntBits((float) restored) & 0xffffffffL
+ : Double.doubleToRawLongBits(restored);
+ if (bits != values[i]) {
+ return -1;
+ }
+ }
+ return scale;
+ }
+
+ private double floatingValue(int i) {
+ return dataType == TSDataType.FLOAT
+ ? Float.intBitsToFloat((int) values[i])
+ : Double.longBitsToDouble(values[i]);
+ }
+
+ private BigDecimal decimalValue(int i) {
+ return dataType == TSDataType.FLOAT
+ ? new BigDecimal(Float.toString(Float.intBitsToFloat((int) values[i])))
+ : BigDecimal.valueOf(Double.longBitsToDouble(values[i]));
+ }
+
+ private void writeHeader(ClusterSupport writer, int scale, int k) throws IOException {
+ writer.write(scale, 8);
+ writer.write(k, 16);
+ writer.write(size, 16);
+ writer.write(PACK_SIZE, 16);
+ }
+
+ private void writeRaw(ByteArrayOutputStream out) throws IOException {
+ ClusterSupport writer = new ClusterSupport(out);
+ writeHeader(writer, 0, 0);
+ // A zero min-value width is impossible in the legacy writer. Old readers reject it.
+ writer.write(0, 8);
+ writer.write(1, 8); // RAW extension version
+ writer.write(dataType.serialize(), 8);
+ for (int i = 0; i < size; i++) {
+ writer.write(values[i], valueByteWidth() * 8);
+ }
+ writer.flush();
+ }
+
+ private void writeCluster(
+ ByteArrayOutputStream out, int scale, long min, long[] data, ClusterResult result)
+ throws IOException {
+ ClusterSupport writer = new ClusterSupport(out);
+ int k = result == null ? 0 : result.references.length;
+ writeHeader(writer, scale, k);
+ int minBits = ClusterSupport.bitsRequired(Math.abs(min));
+ writer.write(minBits, 8);
+ writer.write(min < 0 ? 1 : 0, 1);
+ writer.write(Math.abs(min), minBits);
+ if (k == 0) {
+ writer.flush();
+ return;
+ }
+ result.validate(size);
+ long minReference = result.references[0];
+ for (long reference : result.references) {
+ minReference = Math.min(minReference, reference);
+ }
+ writer.write(ClusterSupport.bitsRequired(minReference), 8);
+ writer.write(0, 1);
+ writer.write(minReference, ClusterSupport.bitsRequired(minReference));
+ int referenceBits = 1;
+ for (long reference : result.references) {
+ referenceBits =
+ Math.max(referenceBits, ClusterSupport.bitsRequired(reference - minReference));
+ }
+ writer.write(referenceBits, 8);
+ for (long reference : result.references) {
+ writer.write(reference - minReference, referenceBits);
+ }
+ long[] countDeltas = new long[k];
+ for (int i = 0; i < k; i++) {
+ countDeltas[i] = result.counts[i] - (i == 0 ? 0 : result.counts[i - 1]);
+ }
+ writePacks(writer, countDeltas, 16);
+
+ long[] residuals = new long[size];
+ int[] positions = new int[k];
+ for (int i = 1; i < k; i++) {
+ positions[i] = positions[i - 1] + (int) result.counts[i - 1];
+ }
+ for (int i = 0; i < size; i++) {
+ int id = result.assignments[i];
+ long residual = data[i] - result.references[id];
+ residuals[positions[id]++] = (residual << 1) ^ (residual >> 63);
+ }
+ writePacks(writer, residuals, 32);
+ writer.flush();
+ }
+
+ private void writePacks(ClusterSupport writer, long[] data, int countBits) throws IOException {
+ int packs = (data.length + PACK_SIZE - 1) / PACK_SIZE;
+ writer.write(packs, countBits);
+ int[] widths = new int[packs];
+ for (int i = 0; i < packs; i++) {
+ int width = 1;
+ for (int j = i * PACK_SIZE; j < Math.min(data.length, (i + 1) * PACK_SIZE); j++) {
+ width = Math.max(width, ClusterSupport.bitsRequired(data[j]));
+ }
+ widths[i] = width;
+ writer.write(width, 8);
+ }
+ for (int i = 0; i < packs; i++) {
+ for (int j = i * PACK_SIZE; j < Math.min(data.length, (i + 1) * PACK_SIZE); j++) {
+ writer.write(data[j], widths[i]);
+ }
+ }
+ }
+
+ @Override
+ public int getOneItemMaxSize() {
+ return Long.BYTES;
+ }
+
+ private int valueByteWidth() {
+ return dataType == TSDataType.INT32
+ || dataType == TSDataType.DATE
+ || dataType == TSDataType.FLOAT
+ ? 4
+ : 8;
+ }
+
+ @Override
+ public long getMaxByteSize() {
+ // Current retained memory, including unused array capacity; flush scratch is transient.
+ return 64L + 8L * values.length;
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/ClusterResult.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/ClusterResult.java
new file mode 100644
index 000000000..f89205ea4
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/ClusterResult.java
@@ -0,0 +1,71 @@
+/*
+ * 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.tsfile.encoding.encoder;
+
+import java.util.Arrays;
+
+/** References, original-input assignments and frequency-sorted cluster sizes. */
+public final class ClusterResult {
+ public final long[] references;
+ public final int[] assignments;
+ public final long[] counts;
+
+ public ClusterResult(long[] references, int[] assignments, long[] counts) {
+ this.references = references;
+ this.assignments = assignments;
+ this.counts = counts;
+ }
+
+ public void validate(int pointCount) {
+ if (references.length == 0
+ || references.length > pointCount
+ || counts.length != references.length
+ || assignments.length != pointCount) {
+ throw new IllegalArgumentException("Invalid cluster result dimensions");
+ }
+ long[] actualCounts = new long[counts.length];
+ for (int id : assignments) {
+ if (id < 0 || id >= counts.length) {
+ throw new IllegalArgumentException("Invalid cluster assignment");
+ }
+ actualCounts[id]++;
+ }
+ for (int i = 0; i < counts.length; i++) {
+ if (references[i] < 0 || counts[i] < 0 || (i > 0 && counts[i] < counts[i - 1])) {
+ throw new IllegalArgumentException("Invalid references or unsorted cluster counts");
+ }
+ }
+ if (!Arrays.equals(counts, actualCounts)) {
+ throw new IllegalArgumentException("Cluster counts do not match assignments");
+ }
+ }
+
+ static void validateInput(long[] data) {
+ if (data == null) {
+ throw new IllegalArgumentException("Cluster input cannot be null");
+ }
+ long limit = Long.MAX_VALUE / (2L * Math.max(1, data.length));
+ for (long value : data) {
+ if (value < 0 || value > limit) {
+ throw new IllegalArgumentException("Cluster input exceeds safe normalized range");
+ }
+ }
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterAlgorithm.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterAlgorithm.java
index 128824424..6d55ec1f5 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterAlgorithm.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterAlgorithm.java
@@ -29,22 +29,22 @@ import java.util.Set;
public class KClusterAlgorithm {
- private static final Random rand = new Random();
-
/** Private constructor to prevent instantiation. */
private KClusterAlgorithm() {}
- public static Object[] run(long[] data, int k) {
- if (data == null || data.length == 0) {
- return new Object[] {new long[0], new int[0], new long[0]};
- }
+ public static ClusterResult run(long[] data, int k) {
+ return run(data, k, 2, 0L);
+ }
- long[][] data2D = new long[data.length][1];
- for (int i = 0; i < data.length; i++) {
- data2D[i][0] = data[i];
+ public static ClusterResult run(long[] data, int k, int maxIterations, long seed) {
+ ClusterResult.validateInput(data);
+ if (k <= 0 || maxIterations <= 0) {
+ throw new IllegalArgumentException("k and maxIterations must be positive");
}
-
- return kMedoidLogCost(data, k, 2, 0.01);
+ if (data.length == 0) {
+ return new ClusterResult(new long[0], new int[0], new long[0]);
+ }
+ return kMedoidLogCost(data, k, maxIterations, 0.01, new Random(seed));
}
/** Helper class for sorting medoids based on their cluster size (frequency). */
@@ -66,7 +66,8 @@ public class KClusterAlgorithm {
}
/** The core K-Medoids algorithm, specifically implemented for 1D data. */
- private static Object[] kMedoidLogCost(long[] data, int k, int maxIter, double tol) {
+ private static ClusterResult kMedoidLogCost(
+ long[] data, int k, int maxIter, double tol, Random random) {
int n = data.length;
// Use a HashSet for efficient uniqueness checking on primitive longs.
@@ -79,52 +80,27 @@ public class KClusterAlgorithm {
}
int distinctCount = uniquePoints.size();
if (distinctCount < k) {
- System.err.println(
- "Warning: Distinct data points ("
- + distinctCount
- + ") is less than input k ("
- + k
- + "), setting k to "
- + distinctCount);
k = distinctCount;
}
- if (k <= 0) {
- return new Object[] {new long[0], new int[n], new long[0]};
- }
-
// 1. Initialize medoids using a K-Medoids++ style approach.
- long[] medoids = acceleratedInitialization(data, k);
+ long[] medoids = acceleratedInitialization(data, k, random);
int[] clusterAssignment = new int[n];
long previousTotalCost = Long.MAX_VALUE;
// 2. Main iterative loop (Build and Swap phases).
for (int iteration = 0; iteration < maxIter; iteration++) {
// --- Assignment Step ---
- long totalCostThisRound = 0L;
- for (int i = 0; i < n; i++) {
- long minCost = Long.MAX_VALUE;
- int assignedMedoidIndex = -1;
- for (int m = 0; m < k; m++) {
- long cost = calculateResidualCost(data[i], medoids[m]);
- if (cost < minCost) {
- minCost = cost;
- assignedMedoidIndex = m;
- if (minCost == 0) break; // Optimization: A perfect match is unbeatable.
- }
- }
- clusterAssignment[i] = assignedMedoidIndex;
- totalCostThisRound += minCost;
- }
+ long totalCostThisRound = assign(data, medoids, clusterAssignment);
// --- Convergence Check (Cost) ---
- if (iteration > 0 && (previousTotalCost - totalCostThisRound) < tol) {
+ if (iteration > 0 && Math.abs(previousTotalCost - totalCostThisRound) < tol) {
break;
}
previousTotalCost = totalCostThisRound;
// --- Update Step ---
- long[] newMedoids = updateMedoids(data, clusterAssignment, k);
+ long[] newMedoids = updateMedoids(data, clusterAssignment, medoids);
// --- Convergence Check (Medoids) ---
if (Arrays.equals(medoids, newMedoids)) {
@@ -134,6 +110,7 @@ public class KClusterAlgorithm {
}
// 3. Calculate final cluster sizes and sort results.
+ assign(data, medoids, clusterAssignment);
long[] finalClusterSizes = new long[k];
for (int assignment : clusterAssignment) {
if (assignment != -1) finalClusterSizes[assignment]++;
@@ -142,7 +119,7 @@ public class KClusterAlgorithm {
}
/** K-Medoids++ style initialization for 1D data. */
- private static long[] acceleratedInitialization(long[] data, int k) {
+ private static long[] acceleratedInitialization(long[] data, int k, Random rand) {
long[] medoids = new long[k];
Set<Long> selectedMedoids = new HashSet<>();
@@ -195,8 +172,9 @@ public class KClusterAlgorithm {
* Updates medoids by finding the point within each cluster that minimizes the total intra-cluster
* cost.
*/
- private static long[] updateMedoids(long[] data, int[] clusterAssignment, int k) {
- long[] newMedoids = new long[k];
+ private static long[] updateMedoids(long[] data, int[] clusterAssignment, long[] medoids) {
+ int k = medoids.length;
+ long[] newMedoids = medoids.clone();
List<Long>[] clusterPoints = new ArrayList[k];
for (int i = 0; i < k; i++) {
clusterPoints[i] = new ArrayList<>();
@@ -235,9 +213,10 @@ public class KClusterAlgorithm {
* @param medoids The discovered medoids.
* @param clusterAssignment The assignment map for each data point.
* @param clusterSize The frequency of each cluster.
- * @return A sorted and correctly mapped Object array.
+ * @return Sorted references with matching assignments and counts.
*/
- private static Object[] sortResults(long[] medoids, int[] clusterAssignment, long[] clusterSize) {
+ private static ClusterResult sortResults(
+ long[] medoids, int[] clusterAssignment, long[] clusterSize) {
int k = medoids.length;
List<KClusterAlgorithm.MedoidSortHelper> sorters = new ArrayList<>();
for (int i = 0; i < k; i++) {
@@ -262,11 +241,32 @@ public class KClusterAlgorithm {
sortedClusterAssignment[i] = oldToNewIndexMap[oldIndex];
}
- return new Object[] {sortedMedoids, sortedClusterAssignment, sortedClusterSize};
+ return new ClusterResult(sortedMedoids, sortedClusterAssignment, sortedClusterSize);
}
// --- Cost Calculation Functions ---
+ private static long assign(long[] data, long[] medoids, int[] assignments) {
+ long total = 0;
+ for (int i = 0; i < data.length; i++) {
+ long bestCost = Long.MAX_VALUE;
+ int best = 0;
+ for (int m = 0; m < medoids.length; m++) {
+ long cost = calculateResidualCost(data[i], medoids[m]);
+ if (cost < bestCost || (cost == bestCost && data[i] == medoids[m])) {
+ bestCost = cost;
+ best = m;
+ }
+ if (data[i] == medoids[m]) {
+ break;
+ }
+ }
+ assignments[i] = best;
+ total += bestCost;
+ }
+ return total;
+ }
+
private static long bitLengthCost(long value) {
if (value == 0) return 1;
return 64 - Long.numberOfLeadingZeros(value);
@@ -282,7 +282,7 @@ public class KClusterAlgorithm {
int ans = -1;
while (low <= high) {
int mid = low + (high - low) / 2;
- if (prefixSums[mid] >= value) {
+ if (prefixSums[mid] > value) {
ans = mid;
high = mid - 1;
} else {
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterEncoder.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterEncoder.java
index c8b347bb9..036c7b48d 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterEncoder.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/KClusterEncoder.java
@@ -20,425 +20,30 @@
package org.apache.tsfile.encoding.encoder;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.exception.encoding.TsFileEncodingException;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.utils.Binary;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.util.ArrayList;
-import java.util.List;
-
-public class KClusterEncoder extends Encoder {
-
- private static final int DEFAULT_PACK_SIZE = 10;
+/** Fixed-reference-count encoding for an unordered, single-column table. */
+public class KClusterEncoder extends ClusterEncoder {
private final int k;
- private long[] scaledData;
- private int scalingExponent;
- private final ValueBuffer buffer;
+ private final int maxIterations;
+ private final long seed;
- // MODIFIED: Constructor takes 'k'
public KClusterEncoder(TSDataType dataType, int k) {
- super(TSEncoding.KCLUSTER);
- this.k = k;
- switch (dataType) {
- case INT32:
- case INT64:
- this.buffer = new Int64Buffer();
- break;
- case FLOAT:
- case DOUBLE:
- this.buffer = new DoubleBuffer();
- break;
- default:
- throw new TsFileEncodingException(
- "AClusterEncoder does not support data type: " + dataType);
- }
- }
-
- /** A buffer to store all values of a page before flushing. We use long to unify all types. */
- private interface ValueBuffer {
- /** Adds a value to the buffer. */
- void add(int value);
-
- void add(long value);
-
- void add(float value);
-
- void add(double value);
-
- /** Clears the internal buffer. */
- void clear();
-
- /** Checks if the buffer is empty. */
- boolean isEmpty();
-
- /** Gets the current size of the buffer. */
- int size();
-
- KClusterEncoder.ProcessingResult processAndGet();
- }
-
- private static class Int64Buffer implements KClusterEncoder.ValueBuffer {
- private final List<Long> values = new ArrayList<>();
-
- @Override
- public void add(int value) {
- values.add((long) value);
- }
-
- @Override
- public void add(long value) {
- values.add(value);
- }
-
- @Override
- public void add(float value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public void add(double value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public KClusterEncoder.ProcessingResult processAndGet() { // <--- 实现新方法
- long[] data = values.stream().mapToLong(l -> l).toArray();
- return new KClusterEncoder.ProcessingResult(data, 0); // Exponent is 0 for integers
- }
-
- @Override
- public void clear() {
- values.clear();
- }
-
- @Override
- public boolean isEmpty() {
- return values.isEmpty();
- }
-
- @Override
- public int size() {
- return values.size();
- }
- }
-
- /** A buffer for FLOAT and DOUBLE types. It performs scaling in processAndGetLongs(). */
- private static class DoubleBuffer implements KClusterEncoder.ValueBuffer {
- private final List<Double> values = new ArrayList<>();
-
- @Override
- public void add(int value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public void add(long value) {
- /* Do nothing, type mismatch */
- }
-
- @Override
- public void add(float value) {
- values.add((double) value);
- } // Store as double to unify
-
- @Override
- public void add(double value) {
- values.add(value);
- }
-
- @Override
- public KClusterEncoder.ProcessingResult processAndGet() {
- // --- Edge Case: Handle empty buffer ---
- if (values.isEmpty()) {
- return new KClusterEncoder.ProcessingResult(new long[0], 0);
- }
- int maxDecimalPlaces = 0;
- for (double v : values) {
- String s = BigDecimal.valueOf(v).toPlainString();
- int dotIndex = s.indexOf('.');
- if (dotIndex != -1) {
- int decimalPlaces = s.length() - dotIndex - 1;
- if (decimalPlaces > maxDecimalPlaces) {
- maxDecimalPlaces = decimalPlaces;
- }
- }
- }
-
- double scalingFactor = Math.pow(10, maxDecimalPlaces);
-
- long[] scaledLongs = new long[values.size()];
- for (int i = 0; i < values.size(); i++) {
- scaledLongs[i] = Math.round(values.get(i) * scalingFactor);
- }
-
- return new KClusterEncoder.ProcessingResult(scaledLongs, maxDecimalPlaces);
- }
-
- @Override
- public void clear() {
- values.clear();
- }
-
- @Override
- public boolean isEmpty() {
- return values.isEmpty();
- }
-
- @Override
- public int size() {
- return values.size();
- }
- }
-
- private static class ProcessingResult {
- final long[] scaledLongs;
- final int scalingExponent; // e.g., 3 for a scaling factor of 1000
-
- ProcessingResult(long[] scaledLongs, int scalingExponent) {
- this.scaledLongs = scaledLongs;
- this.scalingExponent = scalingExponent;
- }
-
- long[] getScaledLongs() {
- return this.scaledLongs;
- }
-
- int getScalingExponent() {
- return this.scalingExponent;
- }
+ this(dataType, k, 2, 0L);
}
- @Override
- public void flush(ByteArrayOutputStream out) throws IOException {
- if (buffer.isEmpty()) {
- return;
- }
-
- ProcessingResult procResult = buffer.processAndGet();
- long[] originalData = procResult.getScaledLongs();
- int scalingExponent = procResult.getScalingExponent();
- if (originalData.length == 0) return;
- long minVal = findMin(originalData);
- long[] data = new long[originalData.length];
- for (int i = 0; i < data.length; i++) {
- data[i] = originalData[i] - minVal;
+ public KClusterEncoder(TSDataType dataType, int k, int maxIterations, long seed) {
+ super(TSEncoding.KCLUSTER, dataType);
+ if (k <= 0 || maxIterations <= 0) {
+ throw new IllegalArgumentException("k and maxIterations must be positive");
}
-
- Object[] clusterResult = KClusterAlgorithm.run(data, k);
- long[] sortedMedoids = (long[]) clusterResult[0];
- int[] clusterAssignments = (int[]) clusterResult[1];
- long[] clusterFrequencies = (long[]) clusterResult[2];
-
- long[] sortedZigzagResiduals =
- calculateSortedZigzagResiduals(data, sortedMedoids, clusterAssignments, clusterFrequencies);
-
- encodeResults(
- out, scalingExponent, minVal, sortedMedoids, clusterAssignments, sortedZigzagResiduals);
-
- buffer.clear();
- }
-
- @Override
- public void encode(int value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- @Override
- public void encode(long value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- @Override
- public void encode(float value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- @Override
- public void encode(double value, ByteArrayOutputStream out) {
- buffer.add(value);
- }
-
- private long[] calculateSortedZigzagResiduals(
- long[] data, long[] medoids, int[] assignments, long[] frequencies) {
- int n = data.length;
- int k = medoids.length;
- if (n == 0) return new long[0];
-
- long[] sortedResiduals = new long[n];
- int[] writePointers = new int[k];
- int cumulativeCount = 0;
- for (int i = 0; i < k; i++) {
- writePointers[i] = cumulativeCount;
- cumulativeCount += (int) frequencies[i];
- }
-
- for (int i = 0; i < n; i++) {
- int clusterId = assignments[i];
- long medoid = medoids[clusterId];
- long residual = data[i] - medoid;
- long zigzagResidual = (residual << 1) ^ (residual >> 63); // Zigzag Encoding
-
- int targetIndex = writePointers[clusterId];
- sortedResiduals[targetIndex] = zigzagResidual;
- writePointers[clusterId]++;
- }
- return sortedResiduals;
- }
-
- private void encodeResults(
- ByteArrayOutputStream out,
- int scalingExponent,
- long minVal,
- long[] medoids,
- int[] frequencies,
- long[] residuals)
- throws IOException {
-
- ClusterSupport writer = new ClusterSupport(out);
- int numPoints = frequencies.length;
- int k = medoids.length;
-
- writer.write(scalingExponent, 8);
- writer.write(k, 16);
- writer.write(numPoints, 16);
- writer.write(DEFAULT_PACK_SIZE, 16);
-
- int minValBit = ClusterSupport.bitsRequired(minVal);
- writer.write(minValBit, 8);
- writer.write(minVal >= 0 ? 0 : 1, 1);
- writer.write(minVal, minValBit);
-
- if (k == 0) {
- writer.flush();
- return;
- }
-
- long minMedoid = findMin(medoids);
- long[] medoidOffsets = new long[k];
- int maxMedoidOffsetBits = 0;
- for (int i = 0; i < k; i++) {
- medoidOffsets[i] = medoids[i] - minMedoid;
- maxMedoidOffsetBits =
- Math.max(maxMedoidOffsetBits, ClusterSupport.bitsRequired(medoidOffsets[i]));
- }
- int minMedoidBit = ClusterSupport.bitsRequired(minMedoid);
- writer.write(minMedoidBit, 8);
- writer.write(minMedoid > 0 ? 0 : 1, 1);
- writer.write(minMedoid, minMedoidBit);
-
- writer.write(maxMedoidOffsetBits, 8);
- for (long offset : medoidOffsets) {
- writer.write(offset, maxMedoidOffsetBits);
- }
-
- long[] freqDeltas = new long[k];
- if (k > 0) {
- freqDeltas[0] = frequencies[0];
- for (int i = 1; i < k; i++) {
- freqDeltas[i] = frequencies[i] - frequencies[i - 1];
- }
- }
- int numFreqBlocks = (k + DEFAULT_PACK_SIZE - 1) / DEFAULT_PACK_SIZE;
- writer.write(numFreqBlocks, 16);
-
- int[] freqBlockMaxBits = new int[numFreqBlocks];
- // Metadata pass for frequencies
- for (int i = 0; i < numFreqBlocks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, k);
- long maxDelta = 0;
- for (int j = start; j < end; j++) {
- maxDelta = Math.max(maxDelta, freqDeltas[j]);
- }
- freqBlockMaxBits[i] = ClusterSupport.bitsRequired(maxDelta);
- writer.write(freqBlockMaxBits[i], 8);
- }
- // Data pass for frequencies
- for (int i = 0; i < numFreqBlocks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, k);
- int bitsForBlock = freqBlockMaxBits[i];
- for (int j = start; j < end; j++) {
- writer.write(freqDeltas[j], bitsForBlock);
- }
- }
-
- int numPacks = (numPoints + DEFAULT_PACK_SIZE - 1) / DEFAULT_PACK_SIZE;
- writer.write(numPacks, 32);
-
- int[] resPackMaxBits = new int[numPacks];
- // Metadata pass for residuals
- for (int i = 0; i < numPacks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, numPoints);
- long maxOffset = 0;
- for (int j = start; j < end; j++) {
- maxOffset = Math.max(maxOffset, residuals[j]);
- }
- resPackMaxBits[i] = ClusterSupport.bitsRequired(maxOffset);
- writer.write(resPackMaxBits[i], 8);
- }
- // Data pass for residuals
- for (int i = 0; i < numPacks; i++) {
- int start = i * DEFAULT_PACK_SIZE;
- int end = Math.min(start + DEFAULT_PACK_SIZE, numPoints);
- int bitsForPack = resPackMaxBits[i];
- if (bitsForPack > 0) {
- for (int j = start; j < end; j++) {
- writer.write(residuals[j], bitsForPack);
- }
- }
- }
-
- writer.flush();
- }
-
- private long findMin(long[] data) {
- if (data == null || data.length == 0) {
- throw new IllegalArgumentException("Data array cannot be null or empty.");
- }
- long min = data[0];
- for (int i = 1; i < data.length; i++) {
- if (data[i] < min) {
- min = data[i];
- }
- }
- return min;
- }
-
- @Override
- public int getOneItemMaxSize() {
- return 8;
- }
-
- @Override
- public long getMaxByteSize() {
- if (this.buffer.isEmpty()) {
- return 0;
- }
- return (long) this.buffer.size() * getOneItemMaxSize() * 3 / 2;
- }
-
- @Override
- public void encode(boolean value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support boolean values.");
- }
-
- @Override
- public void encode(short value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support short values.");
- }
-
- @Override
- public void encode(Binary value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support Binary values.");
+ this.k = k;
+ this.maxIterations = maxIterations;
+ this.seed = seed;
}
@Override
- public void encode(BigDecimal value, ByteArrayOutputStream out) {
- throw new TsFileEncodingException("AClusterEncoder does not support BigDecimal values.");
+ protected ClusterResult cluster(long[] data) {
+ return KClusterAlgorithm.run(data, k, maxIterations, seed);
}
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/TSEncodingBuilder.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/TSEncodingBuilder.java
index 5bcb40142..13e141b46 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/TSEncodingBuilder.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/encoder/TSEncodingBuilder.java
@@ -220,6 +220,8 @@ public abstract class TSEncodingBuilder {
public static class KCluster extends TSEncodingBuilder {
private int k;
+ private int maxIterations = 2;
+ private long seed = 0L;
private static final int KCLUSTER_DEFAULT_K = 1000;
private static final String K_KEY = "k";
@@ -231,15 +233,15 @@ public abstract class TSEncodingBuilder {
public Encoder getEncoder(TSDataType type) {
switch (type) {
case INT32:
- return new KClusterEncoder(type, this.k);
+ return new KClusterEncoder(type, this.k, maxIterations, seed);
case DATE:
case INT64:
- return new KClusterEncoder(type, this.k);
+ return new KClusterEncoder(type, this.k, maxIterations, seed);
case TIMESTAMP:
case FLOAT:
- return new KClusterEncoder(type, this.k);
+ return new KClusterEncoder(type, this.k, maxIterations, seed);
case DOUBLE:
- return new KClusterEncoder(type, this.k);
+ return new KClusterEncoder(type, this.k, maxIterations, seed);
default:
throw new UnSupportedDataTypeException("KCLUSTER doesn't support data type: " + type);
}
@@ -261,6 +263,15 @@ public abstract class TSEncodingBuilder {
"KCLUSTER parameter k must be an integer, but was " + kStr);
}
}
+ if (props != null && props.containsKey("max_iterations")) {
+ maxIterations = Integer.parseInt(props.get("max_iterations"));
+ if (maxIterations <= 0) {
+ throw new IllegalArgumentException("KCLUSTER max_iterations must be positive");
+ }
+ }
+ if (props != null && props.containsKey("seed")) {
+ seed = Long.parseLong(props.get("seed"));
+ }
}
}
@@ -274,9 +285,10 @@ public abstract class TSEncodingBuilder {
switch (type) {
case INT32:
case DATE:
+ return new DeltaBinaryEncoder.IntDeltaEncoder();
case INT64:
case TIMESTAMP:
- // return new DeltaBinaryEncoder.LongDeltaEncoder();
+ return new DeltaBinaryEncoder.LongDeltaEncoder();
case FLOAT:
case DOUBLE:
return new FloatEncoder(TSEncoding.TS_2DIFF, type, maxPointNumber);
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/AlpColumnCodec.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/AlpColumnCodec.java
new file mode 100644
index 000000000..593c2e73b
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/AlpColumnCodec.java
@@ -0,0 +1,179 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.enums.TSDataType;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/**
+ * Java port of the repository ALP path: exponent/factor search, integer FOR, exceptions, 1024-value
+ * vectors and RAW fallback. A versioned table-block format, not upstream ALP bytes.
+ */
+final class AlpColumnCodec {
+ private static final int VECTOR = 1024;
+ private static final double[] POW = new double[19];
+ private static final double[] INV = new double[19];
+
+ static {
+ for (int i = 0; i < POW.length; i++) {
+ POW[i] = Double.parseDouble("1e" + i);
+ INV[i] = Double.parseDouble("1e-" + i);
+ }
+ }
+
+ private AlpColumnCodec() {}
+
+ static byte[] encode(long[] raw, TSDataType type) throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ for (int offset = 0; offset < raw.length; offset += VECTOR) {
+ int n = Math.min(VECTOR, raw.length - offset);
+ int wordBytes = type == TSDataType.INT32 || type == TSDataType.FLOAT ? 4 : 8;
+ long bestBytes = 1L + n * wordBytes;
+ int bestE = -1, bestF = 0, bestWidth = 0;
+ long bestBase = 0;
+ long[] integers = new long[n];
+ boolean[] exact = new boolean[n];
+ for (int e = 0; e <= 18; e++) {
+ for (int f = 0; f <= e; f++) {
+ int exceptions = 0;
+ long min = Long.MAX_VALUE, max = Long.MIN_VALUE;
+ for (int i = 0; i < n; i++) {
+ exact[i] = convert(raw[offset + i], type, e, f, integers, i);
+ if (!exact[i]) exceptions++;
+ else {
+ min = Math.min(min, integers[i]);
+ max = Math.max(max, integers[i]);
+ }
+ }
+ if (exceptions == n) continue;
+ int width = TableBlockIO.width(max - min); // unsigned span, possibly 64 bits
+ long size = 14L + exceptions * (2L + wordBytes) + ((long) n * width + 7) / 8;
+ if (size < bestBytes) {
+ bestBytes = size;
+ bestE = e;
+ bestF = f;
+ bestBase = min;
+ bestWidth = width;
+ }
+ }
+ }
+ if (bestE < 0) {
+ out.writeByte(0);
+ for (int i = 0; i < n; i++) writeRaw(out, raw[offset + i], type);
+ continue;
+ }
+ out.writeByte(1);
+ out.writeByte(bestE);
+ out.writeByte(bestF);
+ out.writeByte(bestWidth);
+ out.writeLong(bestBase);
+ int exceptions = 0;
+ for (int i = 0; i < n; i++) {
+ exact[i] = convert(raw[offset + i], type, bestE, bestF, integers, i);
+ if (!exact[i]) exceptions++;
+ }
+ out.writeShort(exceptions);
+ for (int i = 0; i < n; i++) {
+ if (!exact[i]) {
+ out.writeShort(i);
+ writeRaw(out, raw[offset + i], type);
+ integers[i] = bestBase;
+ }
+ integers[i] -= bestBase;
+ }
+ TableBlockIO.pack(out, integers, 0, n, bestWidth);
+ }
+ return bytes.toByteArray();
+ }
+
+ private static boolean convert(long raw, TSDataType type, int e, int f, long[] ints, int i) {
+ // INT64 values outside binary64's exact-integer range are always literal exceptions.
+ if (type == TSDataType.INT64 && (raw < -(1L << 53) || raw > (1L << 53))) return false;
+ double value = ClusterTable.numeric(raw, type);
+ double scaled = value * POW[e] * INV[f];
+ if (!Double.isFinite(scaled) || scaled <= -0x1p63 || scaled >= 0x1p63) return false;
+ // llround-style ties away from zero, without adding .5 to large integral doubles.
+ double absolute = Math.abs(scaled);
+ double floor = Math.floor(absolute);
+ long integer = (long) (floor + (absolute - floor >= .5 ? 1 : 0));
+ if (scaled < 0) integer = -integer;
+ double restored = (double) integer * POW[f] * INV[e];
+ if (type == TSDataType.INT64
+ && (restored < -0x1p53 || restored > 0x1p53 || restored != Math.rint(restored))) {
+ return false;
+ }
+ if (type == TSDataType.INT32
+ && (restored < Integer.MIN_VALUE
+ || restored > Integer.MAX_VALUE
+ || restored != Math.rint(restored))) return false;
+ if (ClusterTable.raw(restored, type) != raw) return false;
+ ints[i] = integer;
+ return true;
+ }
+
+ static long[] decode(DataInputStream in, int rows, TSDataType type) throws IOException {
+ long[] result = new long[rows];
+ for (int offset = 0; offset < rows; offset += VECTOR) {
+ int n = Math.min(VECTOR, rows - offset);
+ int mode = in.readUnsignedByte();
+ if (mode == 0) {
+ for (int i = 0; i < n; i++) result[offset + i] = readRaw(in, type);
+ } else if (mode == 1) {
+ int e = in.readUnsignedByte(), f = in.readUnsignedByte(), width = in.readUnsignedByte();
+ if (e > 18 || f > e || width > 64) throw new IOException("Invalid ALP parameters");
+ long base = in.readLong();
+ int count = in.readUnsignedShort();
+ if (count > n) throw new IOException("Invalid ALP exception count");
+ int[] positions = new int[count];
+ long[] exceptions = new long[count];
+ for (int i = 0; i < count; i++) {
+ positions[i] = in.readUnsignedShort();
+ if (positions[i] >= n || (i > 0 && positions[i] <= positions[i - 1])) {
+ throw new IOException("Invalid ALP exception positions");
+ }
+ exceptions[i] = readRaw(in, type);
+ }
+ long[] values = TableBlockIO.unpack(in, n, width);
+ for (int i = 0; i < n; i++) {
+ long integer = base + values[i]; // defined modular arithmetic for unsigned FOR
+ double value = (double) integer * POW[f] * INV[e];
+ result[offset + i] = ClusterTable.raw(value, type);
+ }
+ for (int i = 0; i < count; i++) result[offset + positions[i]] = exceptions[i];
+ } else throw new IOException("Unknown ALP vector mode");
+ }
+ TableBlockIO.exhausted(in);
+ return result;
+ }
+
+ static void writeRaw(DataOutputStream out, long raw, TSDataType type) throws IOException {
+ if (type == TSDataType.INT32 || type == TSDataType.FLOAT) out.writeInt((int) raw);
+ else out.writeLong(raw);
+ }
+
+ static long readRaw(DataInputStream in, TSDataType type) throws IOException {
+ return type == TSDataType.INT32 || type == TSDataType.FLOAT ? in.readInt() : in.readLong();
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterColumnBuffer.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterColumnBuffer.java
new file mode 100644
index 000000000..500245990
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterColumnBuffer.java
@@ -0,0 +1,61 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import java.util.Arrays;
+import java.util.BitSet;
+
+/** Page-local raw numeric cells awaiting joint encoding by an aligned writer. */
+public final class ClusterColumnBuffer {
+ private long[] bits = new long[128];
+ private final BitSet nulls = new BitSet();
+ private int size;
+
+ public void append(long valueBits, boolean isNull) {
+ if (size >= ClusterNativePageCodec.MAX_PAGE_ROWS) {
+ throw new IllegalArgumentException("Cluster native page exceeds row limit");
+ }
+ if (size == bits.length) bits = Arrays.copyOf(bits, bits.length * 2);
+ bits[size] = valueBits;
+ if (isNull) nulls.set(size);
+ size++;
+ }
+
+ public int size() {
+ return size;
+ }
+
+ public long bits(int row) {
+ return bits[row];
+ }
+
+ public boolean isNull(int row) {
+ return nulls.get(row);
+ }
+
+ public long memoryBytes() {
+ return 8L * bits.length + nulls.size() / 8;
+ }
+
+ public void reset() {
+ size = 0;
+ nulls.clear();
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterColumnSelector.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterColumnSelector.java
new file mode 100644
index 000000000..d3c690869
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterColumnSelector.java
@@ -0,0 +1,105 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Port of select_end_to_end_columns: Pearson, alpha=.5, global CV elbow, 64 rounds. */
+public final class ClusterColumnSelector {
+ private ClusterColumnSelector() {}
+
+ public static int[] select(ClusterTable table) {
+ int dim = table.columnCount();
+ int n = Math.min(10000, table.rowCount());
+ List<Integer> active = new ArrayList<>();
+ for (int c = 0; c < dim; c++) active.add(c);
+ if (dim == 1 || n == 0) return array(active);
+ double[][] centered = new double[dim][n];
+ double[] norm = new double[dim];
+ for (int c = 0; c < dim; c++) {
+ double mean = 0;
+ boolean finite = true;
+ for (int r = 0; r < n; r++) {
+ double x = table.number(r, c);
+ finite &= Double.isFinite(x);
+ mean += x;
+ }
+ mean /= n;
+ if (!finite || !Double.isFinite(mean)) continue;
+ for (int r = 0; r < n; r++) {
+ centered[c][r] = table.number(r, c) - mean;
+ norm[c] += centered[c][r] * centered[c][r];
+ }
+ norm[c] = Math.sqrt(norm[c]);
+ }
+ double[][] corr = new double[dim][dim];
+ for (int c = 0; c < dim; c++) {
+ for (int j = c + 1; j < dim; j++) {
+ double denominator = norm[c] * norm[j];
+ if (denominator <= 1e-18 || !Double.isFinite(denominator)) continue;
+ double dot = 0;
+ for (int r = 0; r < n; r++) dot += centered[c][r] * centered[j][r];
+ double v = Math.abs(dot / denominator);
+ if (Double.isFinite(v)) corr[c][j] = corr[j][c] = v;
+ }
+ }
+ int[] result = array(active);
+ double previous = 0;
+ double bestDrop = Double.NEGATIVE_INFINITY;
+ boolean hasBest = false;
+ for (int step = 0; step < 64 && active.size() > 1; step++) {
+ double[] scores = new double[active.size()];
+ double mean = 0;
+ for (int i = 0; i < active.size(); i++) {
+ for (int j : active) scores[i] += corr[active.get(i)][j];
+ scores[i] /= active.size() - 1;
+ mean += scores[i];
+ }
+ mean /= scores.length;
+ double variance = 0;
+ for (double score : scores) variance += (score - mean) * (score - mean);
+ double sd = Math.sqrt(variance / scores.length);
+ double cv = sd / Math.max(Math.abs(mean), 1e-18);
+ if (step > 0 && previous - cv > bestDrop + 1e-15) {
+ result = array(active);
+ bestDrop = previous - cv;
+ hasBest = true;
+ }
+ previous = cv;
+ List<Integer> next = new ArrayList<>();
+ int weakest = 0;
+ for (int i = 0; i < scores.length; i++) {
+ if (scores[i] < scores[weakest]) weakest = i;
+ if (scores[i] >= mean - .5 * sd) next.add(active.get(i));
+ }
+ // This fallback is part of the retained Python/C++ implementation.
+ if (next.size() == active.size()) next.remove(weakest);
+ active = next;
+ }
+ return hasBest ? result : array(active);
+ }
+
+ private static int[] array(List<Integer> values) {
+ int[] result = new int[values.size()];
+ for (int i = 0; i < result.length; i++) result[i] = values.get(i);
+ return result;
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterNativePageCodec.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterNativePageCodec.java
new file mode 100644
index 000000000..349d0b8d4
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterNativePageCodec.java
@@ -0,0 +1,261 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.ReadWriteForEncodingUtils;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.zip.CRC32;
+
+/**
+ * Native typed-column pages produced by one joint table encoding. Columns carry independent
+ * projections of the joint stream, so projecting a column never requires reading another column.
+ * Timestamp keys are data, not input positions. The SQL adapter visits these keys in time order;
+ * the table codec itself continues to decode in cluster order without restoring input order.
+ */
+public final class ClusterNativePageCodec {
+ public static final int MAGIC = 0xC34C4E50;
+ public static final int MAX_PAGE_ROWS = 100000;
+ private static final int VERSION = 1;
+
+ private ClusterNativePageCodec() {}
+
+ public static final class Selection {
+ private String[] names;
+ private TSDataType[] types;
+ private int[] selected;
+
+ private int[] fit(ClusterTable table) {
+ if (!Arrays.equals(names, table.names) || !Arrays.equals(types, table.types)) {
+ names = table.names.clone();
+ types = table.types.clone();
+ selected = ClusterColumnSelector.select(table);
+ }
+ return selected;
+ }
+ }
+
+ public static byte[][] encode(
+ long[] times,
+ String[] names,
+ TSDataType[] types,
+ ClusterColumnBuffer[] columns,
+ ClusterTableOptions options,
+ Selection selection)
+ throws IOException {
+ ClusterTable.validateSchema(names, types);
+ int n = times.length, d = names.length;
+ if (n < 1 || n > MAX_PAGE_ROWS || columns.length != d) {
+ throw new IOException("Invalid native cluster page shape");
+ }
+ for (int r = 1; r < n; r++) {
+ if (times[r] <= times[r - 1]) throw new IOException("Native time keys must increase");
+ }
+ for (ClusterColumnBuffer column : columns) {
+ if (column.size() != n) throw new IOException("Unequal aligned column lengths");
+ }
+ int blockRows = Math.min(ClusterTableCodec.MAX_ROWS, ClusterTableCodec.MAX_CELLS / d);
+ int blocks = (n + blockRows - 1) / blockRows;
+ ByteArrayOutputStream[] buffers = new ByteArrayOutputStream[d];
+ DataOutputStream[] outputs = new DataOutputStream[d];
+ for (int c = 0; c < d; c++) {
+ buffers[c] = new ByteArrayOutputStream();
+ outputs[c] = new DataOutputStream(buffers[c]);
+ outputs[c].writeInt(MAGIC);
+ outputs[c].writeByte(VERSION);
+ outputs[c].writeInt(n);
+ outputs[c].writeInt(blocks);
+ }
+ for (int start = 0; start < n; start += blockRows) {
+ int end = Math.min(n, start + blockRows), count = end - start;
+ long[][] values = new long[count][d];
+ BitSet[] nulls = new BitSet[d];
+ for (int c = 0; c < d; c++) {
+ nulls[c] = new BitSet(count);
+ for (int r = 0; r < count; r++) {
+ values[r][c] = columns[c].bits(start + r);
+ if (columns[c].isNull(start + r)) nulls[c].set(r);
+ }
+ }
+ ClusterTable table =
+ new ClusterTable(Arrays.copyOfRange(times, start, end), names, types, values, nulls);
+ byte[] joint = ClusterTableCodec.encode(table, options, selection.fit(table)).bytes();
+ byte[][] projected = project(joint);
+ for (int c = 0; c < d; c++) TableBlockIO.payload(outputs[c], projected[c]);
+ }
+ byte[][] result = new byte[d][];
+ for (int c = 0; c < d; c++) {
+ if (buffers[c].size() > ClusterTableCodec.MAX_ENCODED_BYTES) {
+ throw new IOException("Native column page too large");
+ }
+ result[c] = buffers[c].toByteArray();
+ }
+ return result;
+ }
+
+ // Only called on bytes just produced by the checked table encoder. Copy the existing compressed
+ // column bodies verbatim: this does not rerun clustering or recompress columns independently.
+ private static byte[][] project(byte[] joint) throws IOException {
+ DataInputStream in = new DataInputStream(new ByteArrayInputStream(joint));
+ int magic = in.readInt(), version = in.readUnsignedByte(), method = in.readUnsignedByte();
+ int n = in.readInt(), d = in.readUnsignedShort(), requestedK = in.readUnsignedShort();
+ int iterations = in.readUnsignedByte();
+ long seed = in.readLong();
+ boolean[] selected = new boolean[d];
+ int selectedCount = in.readUnsignedShort();
+ for (int i = 0; i < selectedCount; i++) selected[in.readUnsignedShort()] = true;
+ String[] names = new String[d];
+ byte[] types = new byte[d];
+ for (int c = 0; c < d; c++) {
+ names[c] = in.readUTF();
+ types[c] = in.readByte();
+ }
+ int k = in.readUnsignedShort();
+ byte[] counts = new byte[k * 2], times = new byte[n * 8];
+ in.readFully(counts);
+ in.readFully(times);
+ byte[][] result = new byte[d][];
+ for (int c = 0; c < d; c++) {
+ byte[] bitmap = new byte[(n + 7) / 8];
+ in.readFully(bitmap);
+ int mode = in.readUnsignedByte();
+ int length = in.readInt();
+ byte[] payload = new byte[length];
+ in.readFully(payload);
+ ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(buffer);
+ out.writeInt(magic);
+ out.writeByte(version);
+ out.writeByte(method);
+ out.writeInt(n);
+ out.writeShort(1);
+ out.writeShort(requestedK);
+ out.writeByte(iterations);
+ out.writeLong(seed);
+ out.writeShort(selected[c] ? 1 : 0);
+ if (selected[c]) out.writeShort(0);
+ out.writeUTF(names[c]);
+ out.writeByte(types[c]);
+ out.writeShort(mode == 1 ? k : 0);
+ if (mode == 1) out.write(counts);
+ out.write(times);
+ out.write(bitmap);
+ out.writeByte(mode);
+ TableBlockIO.payload(out, payload);
+ CRC32 crc = new CRC32();
+ crc.update(buffer.toByteArray());
+ out.writeInt((int) crc.getValue());
+ result[c] = buffer.toByteArray();
+ }
+ return result;
+ }
+
+ public static boolean isNativePage(ByteBuffer buffer) {
+ return buffer.remaining() >= 4 && buffer.getInt(buffer.position()) == MAGIC;
+ }
+
+ public static final class Decoded {
+ public final long[] times;
+ public final byte[] bitmap;
+ public final ByteBuffer values;
+
+ private Decoded(long[] times, byte[] bitmap, ByteBuffer values) {
+ this.times = times;
+ this.bitmap = bitmap;
+ this.values = values;
+ }
+ }
+
+ /** Native scan adapter: expose actual timestamp keys in the order expected by IoTDB readers. */
+ public static Decoded decode(ByteBuffer source, TSDataType type) throws IOException {
+ if (source.remaining() > ClusterTableCodec.MAX_ENCODED_BYTES)
+ throw new IOException("Native page too large");
+ byte[] bytes = new byte[source.remaining()];
+ source.duplicate().get(bytes);
+ DataInputStream in = new DataInputStream(new ByteArrayInputStream(bytes));
+ if (in.readInt() != MAGIC || in.readUnsignedByte() != VERSION)
+ throw new IOException("Unknown native cluster page");
+ int n = in.readInt(), blocks = in.readInt();
+ if (n < 1 || n > MAX_PAGE_ROWS || blocks < 1 || blocks > n)
+ throw new IOException("Invalid native page counts");
+ TreeMap<Long, Long> cells = new TreeMap<>();
+ int total = 0;
+ String name = null;
+ for (int b = 0; b < blocks; b++) {
+ DataInputStream body = TableBlockIO.payload(in);
+ byte[] encoded = new byte[body.available()];
+ body.readFully(encoded);
+ ClusterTable column = ClusterTableCodec.decode(encoded);
+ if (column.columnCount() != 1
+ || column.types[0] != type
+ || column.rowCount() == 0
+ || (name != null && !name.equals(column.names[0])))
+ throw new IOException("Native column schema mismatch");
+ name = column.names[0];
+ total += column.rowCount();
+ if (total > n) throw new IOException("Native page row count overflow");
+ for (int r = 0; r < column.rowCount(); r++) {
+ long time = column.timestamps[r];
+ if (cells.containsKey(time)) throw new IOException("Duplicate native timestamp key");
+ cells.put(time, column.nulls[0].get(r) ? null : column.bits[r][0]);
+ }
+ }
+ if (total != n) throw new IOException("Native page row count mismatch");
+ TableBlockIO.exhausted(in);
+ long[] times = new long[n];
+ byte[] bitmap = new byte[(n + 7) / 8];
+ ByteArrayOutputStream values = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(values);
+ int r = 0;
+ for (Map.Entry<Long, Long> cell : cells.entrySet()) {
+ times[r] = cell.getKey();
+ if (cell.getValue() != null) {
+ bitmap[r / 8] |= 0x80 >>> (r % 8);
+ long bits = cell.getValue();
+ switch (type) {
+ case INT32:
+ ReadWriteForEncodingUtils.writeVarInt((int) bits, values);
+ break;
+ case FLOAT:
+ out.writeInt((int) bits);
+ break;
+ case INT64:
+ case DOUBLE:
+ out.writeLong(bits);
+ break;
+ default:
+ throw new IOException("Unsupported native cluster type");
+ }
+ }
+ r++;
+ }
+ return new Decoded(times, bitmap, ByteBuffer.wrap(values.toByteArray()));
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTable.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTable.java
new file mode 100644
index 000000000..19559f6c6
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTable.java
@@ -0,0 +1,227 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.enums.TSDataType;
+
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.HashSet;
+import java.util.Set;
+
+/** Numeric records with explicit timestamps. Physical row order is not part of the contract. */
+public final class ClusterTable {
+ final long[] timestamps;
+ final String[] names;
+ final TSDataType[] types;
+ final long[][] bits;
+ final BitSet[] nulls;
+
+ public ClusterTable(long[] timestamps, String[] names, TSDataType[] types, Object[][] rows) {
+ validateSchema(names, types);
+ if (timestamps.length != rows.length) {
+ throw new IllegalArgumentException("Timestamp/row count mismatch");
+ }
+ this.timestamps = timestamps.clone();
+ this.names = names.clone();
+ this.types = types.clone();
+ this.bits = new long[rows.length][names.length];
+ this.nulls = new BitSet[names.length];
+ for (int c = 0; c < names.length; c++) {
+ nulls[c] = new BitSet(rows.length);
+ }
+ for (int r = 0; r < rows.length; r++) {
+ if (rows[r] == null || rows[r].length != names.length) {
+ throw new IllegalArgumentException("Ragged table");
+ }
+ for (int c = 0; c < names.length; c++) {
+ Object value = rows[r][c];
+ if (value == null) {
+ nulls[c].set(r);
+ continue;
+ }
+ // Reject implicit numeric casts: a long must never silently pass through binary64.
+ switch (types[c]) {
+ case INT32:
+ if (!(value instanceof Integer)) throw wrongType(c);
+ bits[r][c] = (Integer) value;
+ break;
+ case INT64:
+ if (!(value instanceof Long)) throw wrongType(c);
+ bits[r][c] = (Long) value;
+ break;
+ case FLOAT:
+ if (!(value instanceof Float)) throw wrongType(c);
+ bits[r][c] = Float.floatToRawIntBits((Float) value);
+ break;
+ case DOUBLE:
+ if (!(value instanceof Double)) throw wrongType(c);
+ bits[r][c] = Double.doubleToRawLongBits((Double) value);
+ break;
+ default:
+ throw wrongType(c);
+ }
+ }
+ }
+ }
+
+ ClusterTable(
+ long[] timestamps, String[] names, TSDataType[] types, long[][] bits, BitSet[] nulls) {
+ validateSchema(names, types);
+ this.timestamps = timestamps;
+ this.names = names;
+ this.types = types;
+ this.bits = bits;
+ this.nulls = nulls;
+ }
+
+ static void validateSchema(String[] names, TSDataType[] types) {
+ if (names == null
+ || types == null
+ || names.length == 0
+ || names.length > ClusterTableCodec.MAX_COLUMNS
+ || names.length != types.length) {
+ throw new IllegalArgumentException("Expected 1..1024 named numeric columns");
+ }
+ Set<String> unique = new HashSet<>();
+ for (int c = 0; c < names.length; c++) {
+ if (names[c] == null
+ || names[c].isEmpty()
+ || names[c].length() > 1024
+ || !unique.add(names[c])) {
+ throw new IllegalArgumentException("Invalid or duplicate column name");
+ }
+ if (types[c] != TSDataType.INT32
+ && types[c] != TSDataType.INT64
+ && types[c] != TSDataType.FLOAT
+ && types[c] != TSDataType.DOUBLE) {
+ throw new IllegalArgumentException("Supported types: INT32, INT64, FLOAT, DOUBLE");
+ }
+ }
+ }
+
+ private static IllegalArgumentException wrongType(int column) {
+ return new IllegalArgumentException("Value does not match declared type at column " + column);
+ }
+
+ public int rowCount() {
+ return timestamps.length;
+ }
+
+ public int columnCount() {
+ return names.length;
+ }
+
+ public String[] columnNames() {
+ return names.clone();
+ }
+
+ public TSDataType[] columnTypes() {
+ return types.clone();
+ }
+
+ public long timestamp(int row) {
+ return timestamps[row];
+ }
+
+ public boolean isNull(int row, int column) {
+ return nulls[column].get(row);
+ }
+
+ public long rawBits(int row, int column) {
+ return bits[row][column];
+ }
+
+ public Number value(int row, int column) {
+ if (isNull(row, column)) return null;
+ switch (types[column]) {
+ case INT32:
+ return (int) bits[row][column];
+ case INT64:
+ return bits[row][column];
+ case FLOAT:
+ return Float.intBitsToFloat((int) bits[row][column]);
+ case DOUBLE:
+ return Double.longBitsToDouble(bits[row][column]);
+ default:
+ throw wrongType(column);
+ }
+ }
+
+ double number(int row, int column) {
+ return isNull(row, column) ? Double.NaN : numeric(bits[row][column], types[column]);
+ }
+
+ static double numeric(long raw, TSDataType type) {
+ switch (type) {
+ case DOUBLE:
+ return Double.longBitsToDouble(raw);
+ case FLOAT:
+ return Float.intBitsToFloat((int) raw);
+ default:
+ return raw;
+ }
+ }
+
+ static long raw(double value, TSDataType type) {
+ switch (type) {
+ case DOUBLE:
+ return Double.doubleToRawLongBits(value);
+ case FLOAT:
+ return Float.floatToRawIntBits((float) value);
+ case INT32:
+ return (int) value;
+ default:
+ return (long) value;
+ }
+ }
+
+ public ClusterTable slice(int from, int to) {
+ if (from < 0 || to < from || to > rowCount()) throw new IndexOutOfBoundsException();
+ int[] indices = new int[to - from];
+ for (int i = 0; i < indices.length; i++) indices[i] = from + i;
+ return selectRows(indices);
+ }
+
+ /** Inclusive bounds on the ORIGINAL timestamps, including duplicate timestamps. */
+ public ClusterTable timeRange(long start, long end) {
+ if (start > end) throw new IllegalArgumentException("Reversed time range");
+ int[] indices = new int[rowCount()];
+ int count = 0;
+ for (int r = 0; r < rowCount(); r++) {
+ if (timestamps[r] >= start && timestamps[r] <= end) indices[count++] = r;
+ }
+ return selectRows(Arrays.copyOf(indices, count));
+ }
+
+ private ClusterTable selectRows(int[] indices) {
+ long[] times = new long[indices.length];
+ long[][] values = new long[indices.length][];
+ BitSet[] missing = new BitSet[columnCount()];
+ for (int c = 0; c < columnCount(); c++) missing[c] = new BitSet(indices.length);
+ for (int i = 0; i < indices.length; i++) {
+ int r = indices[i];
+ times[i] = timestamps[r];
+ values[i] = bits[r].clone();
+ for (int c = 0; c < columnCount(); c++) if (nulls[c].get(r)) missing[c].set(i);
+ }
+ return new ClusterTable(times, names.clone(), types.clone(), values, missing);
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableCodec.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableCodec.java
new file mode 100644
index 000000000..4cb56d5a2
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableCodec.java
@@ -0,0 +1,361 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.enums.TSDataType;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.List;
+import java.util.zip.CRC32;
+
+/**
+ * Complete independent numeric table blocks: CV-elbow selection, joint A/K clustering, ALP
+ * companions and timestamps in the same traversal. No permutation is stored or restored.
+ */
+public final class ClusterTableCodec {
+ public static final int MAX_ROWS = 10000;
+ public static final int MAX_COLUMNS = 1024;
+ public static final int MAX_CELLS = 2000000;
+ public static final int MAX_ENCODED_BYTES = 64 * 1024 * 1024;
+ private static final int MAGIC = 0x434c5442; // CLTB
+ private static final int VERSION = 1;
+
+ private ClusterTableCodec() {}
+
+ public static final class Encoded {
+ private final byte[] bytes;
+ private final int[] selected;
+ private final int[] clustered;
+ public final int referenceCount;
+ public final long clusterBytes;
+ public final long alpBytes;
+
+ private Encoded(
+ byte[] bytes,
+ int[] selected,
+ int[] clustered,
+ int references,
+ long clusterBytes,
+ long alpBytes) {
+ this.bytes = bytes;
+ this.selected = selected.clone();
+ this.clustered = clustered;
+ this.referenceCount = references;
+ this.clusterBytes = clusterBytes;
+ this.alpBytes = alpBytes;
+ }
+
+ public byte[] bytes() {
+ return bytes.clone();
+ }
+
+ public int[] selectedColumns() {
+ return selected.clone();
+ }
+
+ public int[] clusteredColumns() {
+ return clustered.clone();
+ }
+ }
+
+ private static final class Scaled {
+ final int column;
+ final int scale;
+ final long min;
+ final long[] values;
+
+ Scaled(int column, int scale, long min, long[] values) {
+ this.column = column;
+ this.scale = scale;
+ this.min = min;
+ this.values = values;
+ }
+ }
+
+ public static Encoded encode(ClusterTable table, ClusterTableOptions options) throws IOException {
+ return encode(table, options, ClusterColumnSelector.select(table));
+ }
+
+ /**
+ * Reuse the first sample's selection across later blocks; every block records it independently.
+ */
+ public static Encoded encode(ClusterTable table, ClusterTableOptions options, int[] selected)
+ throws IOException {
+ int n = table.rowCount(), d = table.columnCount();
+ shape(n, d);
+ if (options == null) throw new IllegalArgumentException("Missing algorithm options");
+ boolean[] requested = new boolean[d];
+ for (int c : selected) {
+ if (c < 0 || c >= d || requested[c]) throw new IllegalArgumentException("Invalid selection");
+ requested[c] = true;
+ }
+ List<Scaled> scaled = new ArrayList<>();
+ for (int c : selected) {
+ Scaled column = scale(table, c);
+ if (column != null) scaled.add(column);
+ }
+ int[] order = new int[n];
+ for (int r = 0; r < n; r++) order[r] = r;
+ MultidimensionalCluster.Result clusters = null;
+ if (!scaled.isEmpty()) {
+ long[][] points = new long[n][scaled.size()];
+ for (int c = 0; c < scaled.size(); c++) {
+ for (int r = 0; r < n; r++) points[r][c] = scaled.get(c).values[r];
+ }
+ clusters = MultidimensionalCluster.run(points, options);
+ order = clusters.order;
+ }
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeInt(MAGIC);
+ out.writeByte(VERSION);
+ out.writeByte(options.method.ordinal());
+ out.writeInt(n);
+ out.writeShort(d);
+ out.writeShort(options.k);
+ out.writeByte(options.maxIterations);
+ out.writeLong(options.seed);
+ out.writeShort(selected.length);
+ for (int c : selected) out.writeShort(c);
+ for (int c = 0; c < d; c++) {
+ out.writeUTF(table.names[c]);
+ out.writeByte(table.types[c].serialize());
+ }
+ int k = clusters == null ? 0 : clusters.counts.length;
+ out.writeShort(k);
+ if (clusters != null) for (int count : clusters.counts) out.writeShort(count);
+ // Original timestamps travel with entire records; these are not outer block identifiers.
+ for (int r : order) out.writeLong(table.timestamps[r]);
+ long clusterBytes = 0, alpBytes = 0;
+ int[] clustered = new int[scaled.size()];
+ for (int i = 0; i < scaled.size(); i++) clustered[i] = scaled.get(i).column;
+ for (int c = 0; c < d; c++) {
+ byte[] missing = new byte[(n + 7) / 8];
+ for (int i = 0; i < n; i++) if (table.nulls[c].get(order[i])) missing[i / 8] |= 1 << (i % 8);
+ out.write(missing);
+ int index = -1;
+ for (int i = 0; i < scaled.size(); i++) if (scaled.get(i).column == c) index = i;
+ out.writeByte(index < 0 ? 0 : 1);
+ byte[] payload;
+ if (index < 0) {
+ long[] values = new long[n];
+ for (int i = 0; i < n; i++) values[i] = table.bits[order[i]][c];
+ payload = AlpColumnCodec.encode(values, table.types[c]);
+ alpBytes += payload.length;
+ } else {
+ Scaled column = scaled.get(index);
+ ByteArrayOutputStream body = new ByteArrayOutputStream();
+ DataOutputStream data = new DataOutputStream(body);
+ data.writeByte(column.scale);
+ data.writeLong(column.min);
+ for (long[] reference : clusters.references) data.writeLong(reference[index]);
+ long[] residuals = new long[n];
+ int position = 0;
+ for (int id = 0; id < k; id++) {
+ for (int j = 0; j < clusters.counts[id]; j++, position++) {
+ long delta = column.values[order[position]] - clusters.references[id][index];
+ residuals[position] = (delta << 1) ^ (delta >> 63);
+ }
+ }
+ data.write(TableBlockIO.packs(residuals));
+ payload = body.toByteArray();
+ clusterBytes += payload.length;
+ }
+ TableBlockIO.payload(out, payload);
+ }
+ if (bytes.size() + 4 > MAX_ENCODED_BYTES) throw new IOException("Encoded block too large");
+ CRC32 crc = new CRC32();
+ byte[] body = bytes.toByteArray();
+ crc.update(body);
+ out.writeInt((int) crc.getValue());
+ return new Encoded(bytes.toByteArray(), selected, clustered, k, clusterBytes, alpBytes);
+ }
+
+ private static Scaled scale(ClusterTable table, int c) {
+ int n = table.rowCount();
+ if (n == 0 || !table.nulls[c].isEmpty()) return null;
+ TSDataType type = table.types[c];
+ long[] values = new long[n];
+ int scale = 0;
+ BigDecimal[] decimals = null;
+ if (type == TSDataType.DOUBLE || type == TSDataType.FLOAT) {
+ decimals = new BigDecimal[n];
+ for (int r = 0; r < n; r++) {
+ double value = table.number(r, c);
+ if (!Double.isFinite(value) || (value == 0 && table.bits[r][c] != 0)) return null;
+ decimals[r] =
+ type == TSDataType.FLOAT
+ ? new BigDecimal(Float.toString((float) value))
+ : BigDecimal.valueOf(value);
+ scale = Math.max(scale, decimals[r].stripTrailingZeros().scale());
+ }
+ if (scale > 18) return null;
+ }
+ long min = Long.MAX_VALUE, max = Long.MIN_VALUE;
+ try {
+ for (int r = 0; r < n; r++) {
+ values[r] =
+ decimals == null
+ ? table.bits[r][c]
+ : decimals[r].movePointRight(scale).longValueExact();
+ if (decimals != null
+ && ClusterTable.raw(values[r] / Math.pow(10, scale), type) != table.bits[r][c]) {
+ return null;
+ }
+ min = Math.min(min, values[r]);
+ max = Math.max(max, values[r]);
+ }
+ long span = Math.subtractExact(max, min);
+ if (span > Long.MAX_VALUE / 2) return null;
+ for (int r = 0; r < n; r++) values[r] = Math.subtractExact(values[r], min);
+ } catch (ArithmeticException unsafe) {
+ return null;
+ }
+ return new Scaled(c, scale, min, values);
+ }
+
+ public static ClusterTable decode(byte[] bytes) throws IOException {
+ if (bytes == null || bytes.length < 8 || bytes.length > MAX_ENCODED_BYTES) {
+ throw new IOException("Invalid encoded block length");
+ }
+ CRC32 crc = new CRC32();
+ crc.update(bytes, 0, bytes.length - 4);
+ DataInputStream checksum =
+ new DataInputStream(new ByteArrayInputStream(bytes, bytes.length - 4, 4));
+ if ((int) crc.getValue() != checksum.readInt())
+ throw new IOException("Table block checksum mismatch");
+ DataInputStream in = new DataInputStream(new ByteArrayInputStream(bytes, 0, bytes.length - 4));
+ try {
+ if (in.readInt() != MAGIC || in.readUnsignedByte() != VERSION)
+ throw new IOException("Unknown table block format");
+ int method = in.readUnsignedByte();
+ if (method >= ClusterTableOptions.Method.values().length)
+ throw new IOException("Unknown clustering method");
+ int n = in.readInt(), d = in.readUnsignedShort();
+ shape(n, d);
+ new ClusterTableOptions(
+ ClusterTableOptions.Method.values()[method],
+ in.readUnsignedShort(),
+ in.readUnsignedByte(),
+ in.readLong());
+ int selectionCount = in.readUnsignedShort();
+ if (selectionCount > d) throw new IOException("Invalid selected column count");
+ boolean[] selected = new boolean[d];
+ for (int i = 0; i < selectionCount; i++) {
+ int c = in.readUnsignedShort();
+ if (c >= d || selected[c]) throw new IOException("Invalid selected column ID");
+ selected[c] = true;
+ }
+ String[] names = new String[d];
+ TSDataType[] types = new TSDataType[d];
+ for (int c = 0; c < d; c++) {
+ names[c] = in.readUTF();
+ types[c] = TSDataType.deserialize(in.readByte());
+ }
+ ClusterTable.validateSchema(names, types);
+ int k = in.readUnsignedShort();
+ if (k > n) throw new IOException("Invalid reference count");
+ int[] counts = new int[k];
+ int total = 0;
+ for (int id = 0; id < k; id++) {
+ counts[id] = in.readUnsignedShort();
+ if (counts[id] == 0 || (id > 0 && counts[id] < counts[id - 1]))
+ throw new IOException("Invalid cluster count");
+ total += counts[id];
+ }
+ if (k > 0 && total != n)
+ throw new IOException("Cluster frequency sum differs from row count");
+ // Check available timestamp bytes before any row-sized allocation.
+ if (8L * n > in.available()) throw new IOException("Truncated timestamp stream");
+ long[] times = new long[n];
+ for (int r = 0; r < n; r++) times[r] = in.readLong();
+ long[][] values = new long[n][d];
+ BitSet[] nulls = new BitSet[d];
+ int clusterColumns = 0;
+ for (int c = 0; c < d; c++) {
+ byte[] bitmap = new byte[(n + 7) / 8];
+ in.readFully(bitmap);
+ nulls[c] = BitSet.valueOf(bitmap);
+ if (nulls[c].length() > n) throw new IOException("Invalid null bitmap");
+ int mode = in.readUnsignedByte();
+ DataInputStream body = TableBlockIO.payload(in);
+ long[] column;
+ if (mode == 0) column = AlpColumnCodec.decode(body, n, types[c]);
+ else if (mode == 1) {
+ if (!selected[c] || k == 0 || !nulls[c].isEmpty())
+ throw new IOException("Invalid clustered column");
+ clusterColumns++;
+ int scale = body.readUnsignedByte();
+ if (scale > 18
+ || ((types[c] == TSDataType.INT32 || types[c] == TSDataType.INT64) && scale != 0)) {
+ throw new IOException("Invalid scale");
+ }
+ long min = body.readLong();
+ long[] refs = new long[k];
+ for (int id = 0; id < k; id++) {
+ refs[id] = body.readLong();
+ if (refs[id] < 0 || refs[id] > Long.MAX_VALUE / 2)
+ throw new IOException("Invalid reference");
+ }
+ column = TableBlockIO.readPacks(body, n);
+ int position = 0;
+ for (int id = 0; id < k; id++) {
+ for (int j = 0; j < counts[id]; j++, position++) {
+ long residual = (column[position] >>> 1) ^ -(column[position] & 1);
+ long normalized = Math.addExact(refs[id], residual);
+ if (normalized < 0 || normalized > Long.MAX_VALUE / 2)
+ throw new IOException("Invalid residual");
+ long integer = Math.addExact(min, normalized);
+ if (types[c] == TSDataType.INT64) column[position] = integer;
+ else if (types[c] == TSDataType.INT32) {
+ if (integer != (int) integer) throw new IOException("INT32 overflow");
+ column[position] = integer;
+ } else column[position] = ClusterTable.raw(integer / Math.pow(10, scale), types[c]);
+ }
+ }
+ TableBlockIO.exhausted(body);
+ } else throw new IOException("Unknown column codec");
+ for (int r = 0; r < n; r++) values[r][c] = column[r];
+ }
+ if ((k > 0) != (clusterColumns > 0)) throw new IOException("Inconsistent cluster metadata");
+ TableBlockIO.exhausted(in);
+ return new ClusterTable(times, names, types, values, nulls);
+ } catch (IllegalArgumentException | ArithmeticException bad) {
+ throw new IOException("Invalid table block", bad);
+ }
+ }
+
+ private static void shape(int rows, int columns) {
+ if (rows < 0
+ || rows > MAX_ROWS
+ || columns < 1
+ || columns > MAX_COLUMNS
+ || (long) rows * columns > MAX_CELLS) {
+ throw new IllegalArgumentException("Block exceeds row/column/cell limits");
+ }
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableEncoder.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableEncoder.java
new file mode 100644
index 000000000..de6273f7e
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableEncoder.java
@@ -0,0 +1,61 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/** Fits selection once on up to 10,000 leading rows, then encodes independent blocks. */
+public final class ClusterTableEncoder {
+ private final ClusterTableOptions options;
+ private String[] names;
+ private org.apache.tsfile.enums.TSDataType[] types;
+ private int[] selected;
+
+ public ClusterTableEncoder(ClusterTableOptions options) {
+ if (options == null) throw new IllegalArgumentException("Missing options");
+ this.options = options;
+ }
+
+ /** Calling fit again checks the schema; it never silently refits a stream. */
+ public void fit(ClusterTable sample) {
+ if (names == null) {
+ if (sample.rowCount() == 0) return;
+ selected = ClusterColumnSelector.select(sample);
+ names = sample.columnNames();
+ types = sample.columnTypes();
+ } else if (!Arrays.equals(names, sample.names) || !Arrays.equals(types, sample.types)) {
+ throw new IllegalArgumentException("Schema changed within cluster table stream");
+ }
+ }
+
+ public ClusterTableCodec.Encoded encode(ClusterTable page) throws IOException {
+ fit(page);
+ return ClusterTableCodec.encode(page, options, selected == null ? new int[0] : selected);
+ }
+
+ public int[] selectedColumns() {
+ return selected == null ? new int[0] : selected.clone();
+ }
+
+ public static int pageSize(ClusterTable table) {
+ return Math.min(ClusterTableCodec.MAX_ROWS, ClusterTableCodec.MAX_CELLS / table.columnCount());
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableOptions.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableOptions.java
new file mode 100644
index 000000000..aa17b125f
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableOptions.java
@@ -0,0 +1,55 @@
+/*
+ * 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.tsfile.encoding.table;
+
+/** Explicit, reproducible algorithm configuration; no dataset-specific branches. */
+public final class ClusterTableOptions {
+ public enum Method {
+ ACLUSTER,
+ KCLUSTER
+ }
+
+ public final Method method;
+ public final int k;
+ public final int maxIterations;
+ public final long seed;
+
+ public ClusterTableOptions(Method method, int k, int maxIterations, long seed) {
+ if (method == null
+ || k < 1
+ || k > ClusterTableCodec.MAX_ROWS
+ || maxIterations < 1
+ || maxIterations > 100) {
+ throw new IllegalArgumentException("Invalid cluster options");
+ }
+ this.method = method;
+ this.k = k;
+ this.maxIterations = maxIterations;
+ this.seed = seed;
+ }
+
+ public static ClusterTableOptions aCluster() {
+ return new ClusterTableOptions(Method.ACLUSTER, 100, 10, 0);
+ }
+
+ public static ClusterTableOptions kCluster() {
+ return new ClusterTableOptions(Method.KCLUSTER, 100, 10, 0);
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableTsFile.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableTsFile.java
new file mode 100644
index 000000000..e77802349
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/ClusterTableTsFile.java
@@ -0,0 +1,119 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+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.TsFileSequenceReader;
+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.write.TsFileWriter;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+
+/** Standard TsFile BLOB container for independently decodable complete-record blocks. */
+public final class ClusterTableTsFile {
+ public static final String DEVICE = "root.cluster.table";
+ public static final String PAYLOAD = "payload";
+
+ private ClusterTableTsFile() {}
+
+ @FunctionalInterface
+ public interface BlockConsumer {
+ void accept(ClusterTable table) throws IOException;
+ }
+
+ public static final class Writer implements Closeable {
+ private final TsFileWriter writer;
+ private final ClusterTableEncoder encoder;
+ private final IMeasurementSchema schema;
+ private long blockId;
+ private boolean failed;
+
+ public Writer(File file, ClusterTableOptions options) throws IOException {
+ if (file.exists()) throw new IOException("Refusing to overwrite existing TsFile: " + file);
+ encoder = new ClusterTableEncoder(options);
+ schema =
+ new MeasurementSchema(
+ PAYLOAD, TSDataType.BLOB, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED);
+ writer = new TsFileWriter(file);
+ try {
+ writer.registerTimeseries(DEVICE, schema);
+ } catch (WriteProcessException error) {
+ writer.close();
+ throw new IOException(error);
+ }
+ }
+
+ /** Caller may pass a whole numeric table; bounded pages each carry complete metadata. */
+ public void append(ClusterTable table) throws IOException {
+ if (failed) throw new IOException("Writer failed; use a new output file");
+ encoder.fit(table);
+ int pageSize = ClusterTableEncoder.pageSize(table);
+ for (int start = 0; start < table.rowCount(); start += pageSize) {
+ ClusterTable page = table.slice(start, Math.min(table.rowCount(), start + pageSize));
+ byte[] bytes = encoder.encode(page).bytes();
+ Tablet tablet = new Tablet(DEVICE, Collections.singletonList(schema), 1);
+ tablet.addTimestamp(0, blockId);
+ tablet.addValue(0, PAYLOAD, bytes);
+ tablet.setRowSize(1);
+ try {
+ writer.writeTree(tablet);
+ blockId++;
+ } catch (IOException | WriteProcessException error) {
+ failed = true;
+ throw new IOException("Cannot write cluster table block", error);
+ }
+ }
+ }
+
+ public void flush() throws IOException {
+ writer.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ writer.close();
+ }
+ }
+
+ public static void read(File file, BlockConsumer consumer) throws IOException {
+ try (TsFileReader reader = new TsFileReader(new TsFileSequenceReader(file.getPath()))) {
+ QueryDataSet result =
+ reader.query(
+ QueryExpression.create(
+ Collections.singletonList(new Path(DEVICE, PAYLOAD, true)), null));
+ while (result.hasNext()) {
+ byte[] bytes = result.next().getFields().get(0).getBinaryV().getValues();
+ consumer.accept(ClusterTableCodec.decode(bytes));
+ }
+ }
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/MultidimensionalCluster.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/MultidimensionalCluster.java
new file mode 100644
index 000000000..823ea644a
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/MultidimensionalCluster.java
@@ -0,0 +1,216 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/** Multidimensional bit-cost medoids. All coordinates are checked, normalized nonnegative longs. */
+final class MultidimensionalCluster {
+ private MultidimensionalCluster() {}
+
+ static final class Result {
+ final long[][] references;
+ final int[] counts;
+ final int[] order;
+
+ Result(long[][] references, int[] counts, int[] order) {
+ this.references = references;
+ this.counts = counts;
+ this.order = order;
+ }
+ }
+
+ private static final class Key {
+ final long[] row;
+
+ Key(long[] row) {
+ this.row = row;
+ }
+
+ @Override
+ public int hashCode() {
+ return Arrays.hashCode(row);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ return other instanceof Key && Arrays.equals(row, ((Key) other).row);
+ }
+ }
+
+ static long cost(long[] a, long[] b) {
+ long sum = 0;
+ for (int c = 0; c < a.length; c++) sum += 1 + TableBlockIO.width(Math.abs(a[c] - b[c]));
+ return sum;
+ }
+
+ static Result run(long[][] data, ClusterTableOptions options) {
+ List<long[]> refs = new ArrayList<>();
+ int[] assignment = new int[data.length];
+ if (options.method == ClusterTableOptions.Method.ACLUSTER) adaptive(data, refs, assignment);
+ else fixed(data, refs, assignment, options);
+ int[] counts = new int[refs.size()];
+ for (int id : assignment) counts[id]++;
+ List<Integer> ids = new ArrayList<>();
+ for (int i = 0; i < counts.length; i++) if (counts[i] != 0) ids.add(i);
+ ids.sort(Comparator.comparingInt((Integer id) -> counts[id]).thenComparingInt(id -> id));
+ int[] offsets = new int[counts.length];
+ int[] sortedCounts = new int[ids.size()];
+ long[][] sortedRefs = new long[ids.size()][];
+ int position = 0;
+ for (int i = 0; i < ids.size(); i++) {
+ int id = ids.get(i);
+ offsets[id] = position;
+ position += counts[id];
+ sortedCounts[i] = counts[id];
+ sortedRefs[i] = refs.get(id);
+ }
+ int[] order = new int[data.length];
+ for (int r = 0; r < data.length; r++) order[offsets[assignment[r]]++] = r;
+ return new Result(sortedRefs, sortedCounts, order);
+ }
+
+ private static int nearest(long[] point, List<long[]> refs) {
+ int best = 0;
+ long min = Long.MAX_VALUE;
+ for (int i = 0; i < refs.size(); i++) {
+ long score = cost(point, refs.get(i));
+ if (score < min || (score == min && Arrays.equals(point, refs.get(i)))) {
+ best = i;
+ min = score;
+ }
+ }
+ return best;
+ }
+
+ private static void adaptive(long[][] data, List<long[]> refs, int[] assignment) {
+ refs.add(data[0]);
+ List<List<Integer>> members = new ArrayList<>();
+ members.add(new ArrayList<>());
+ members.get(0).add(0);
+ Map<Key, Integer> known = new HashMap<>();
+ known.put(new Key(data[0]), 0);
+ for (int r = 1; r < data.length; r++) {
+ Integer duplicate = known.get(new Key(data[r]));
+ if (duplicate != null) {
+ assignment[r] = duplicate;
+ members.get(duplicate).add(r);
+ continue;
+ }
+ int best = nearest(data[r], refs);
+ long gain = cost(data[r], refs.get(best)) - data[r].length;
+ List<Integer> move = new ArrayList<>();
+ for (int p : members.get(best)) {
+ long saving = cost(data[p], refs.get(best)) - cost(data[p], data[r]);
+ if (saving > 0) {
+ gain += saving;
+ move.add(p);
+ }
+ }
+ long referenceCost = 0;
+ for (long value : data[r]) referenceCost += 1 + TableBlockIO.width(value);
+ if (gain > referenceCost) {
+ int id = refs.size();
+ refs.add(data[r]);
+ known.put(new Key(data[r]), id);
+ for (int p : move) assignment[p] = id;
+ members.get(best).removeIf(p -> assignment[p] == id);
+ move.add(r);
+ members.add(move);
+ assignment[r] = id;
+ } else {
+ assignment[r] = best;
+ members.get(best).add(r);
+ }
+ }
+ }
+
+ private static double distance(long[] a, long[] b) {
+ double sum = 0;
+ for (int c = 0; c < a.length; c++) sum += Math.abs(a[c] - b[c]);
+ return sum;
+ }
+
+ private static void fixed(
+ long[][] data, List<long[]> refs, int[] assignment, ClusterTableOptions options) {
+ Map<Key, long[]> unique = new LinkedHashMap<>();
+ for (long[] row : data) unique.putIfAbsent(new Key(row), row);
+ int k = Math.min(options.k, unique.size());
+ Random random = new Random(options.seed);
+ refs.add(data[random.nextInt(data.length)]);
+ double[] nearestDistance = new double[data.length];
+ Arrays.fill(nearestDistance, Double.POSITIVE_INFINITY);
+ while (refs.size() < k) {
+ double total = 0;
+ long[] newest = refs.get(refs.size() - 1);
+ for (int r = 0; r < data.length; r++) {
+ nearestDistance[r] = Math.min(nearestDistance[r], distance(data[r], newest));
+ total += nearestDistance[r];
+ }
+ double sample = random.nextDouble() * total;
+ int chosen = 0;
+ for (; chosen < data.length - 1; chosen++) {
+ sample -= nearestDistance[chosen];
+ if (sample < 0) break;
+ }
+ // Never select the same value twice, including random/rounding boundary cases.
+ while (nearestDistance[chosen] == 0) chosen = (chosen + 1) % data.length;
+ refs.add(data[chosen]);
+ }
+ for (int iteration = 0; iteration < options.maxIterations; iteration++) {
+ List<Map<Key, Integer>> groups = new ArrayList<>();
+ for (int id = 0; id < k; id++) groups.add(new LinkedHashMap<>());
+ for (int r = 0; r < data.length; r++) {
+ assignment[r] = nearest(data[r], refs);
+ groups.get(assignment[r]).merge(new Key(data[r]), 1, Integer::sum);
+ }
+ boolean changed = false;
+ for (int id = 0; id < k; id++) {
+ Map<Key, Integer> group = groups.get(id);
+ if (group.isEmpty()) continue;
+ long bestCost = Long.MAX_VALUE;
+ long[] best = refs.get(id);
+ for (Key candidate : group.keySet()) {
+ long score = 0;
+ for (Map.Entry<Key, Integer> member : group.entrySet()) {
+ score += cost(candidate.row, member.getKey().row) * member.getValue();
+ if (score >= bestCost) break;
+ }
+ if (score < bestCost) {
+ bestCost = score;
+ best = candidate.row;
+ }
+ }
+ changed |= !Arrays.equals(best, refs.get(id));
+ refs.set(id, best);
+ }
+ if (!changed) break;
+ }
+ // The final update must be followed by assignment against the final medoids.
+ for (int r = 0; r < data.length; r++) assignment[r] = nearest(data[r], refs);
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/TableBlockIO.java b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/TableBlockIO.java
new file mode 100644
index 000000000..2b814c024
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/encoding/table/TableBlockIO.java
@@ -0,0 +1,119 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/** Bounds-checked, byte-aligned unsigned bit packing, including widths zero and 64. */
+final class TableBlockIO {
+ private TableBlockIO() {}
+
+ static int width(long value) {
+ return 64 - Long.numberOfLeadingZeros(value);
+ }
+
+ static void pack(DataOutputStream out, long[] values, int start, int end, int width)
+ throws IOException {
+ int current = 0;
+ int used = 0;
+ for (int i = start; i < end; i++) {
+ for (int b = width - 1; b >= 0; b--) {
+ current = (current << 1) | (int) ((values[i] >>> b) & 1);
+ if (++used == 8) {
+ out.writeByte(current);
+ current = 0;
+ used = 0;
+ }
+ }
+ }
+ if (used != 0) out.writeByte(current << (8 - used));
+ }
+
+ static long[] unpack(DataInputStream in, int count, int width) throws IOException {
+ if (width < 0
+ || width > 64
+ || count < 0
+ || count > ClusterTableCodec.MAX_ROWS
+ || ((long) count * width + 7) / 8 > in.available()) {
+ throw new IOException("Invalid packed column");
+ }
+ long[] result = new long[count];
+ int current = 0;
+ int left = 0;
+ for (int i = 0; i < count; i++) {
+ for (int b = 0; b < width; b++) {
+ if (left == 0) {
+ current = in.readUnsignedByte();
+ left = 8;
+ }
+ result[i] = (result[i] << 1) | ((current >>> --left) & 1);
+ }
+ }
+ if (left != 0 && (current & ((1 << left) - 1)) != 0) {
+ throw new IOException("Nonzero bit padding");
+ }
+ return result;
+ }
+
+ static byte[] packs(long[] values) throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ for (int start = 0; start < values.length; start += 10) {
+ int end = Math.min(values.length, start + 10);
+ long mask = 0;
+ for (int i = start; i < end; i++) mask |= values[i];
+ int width = width(mask);
+ out.writeByte(width);
+ pack(out, values, start, end, width);
+ }
+ return bytes.toByteArray();
+ }
+
+ static long[] readPacks(DataInputStream in, int count) throws IOException {
+ long[] result = new long[count];
+ for (int start = 0; start < count; start += 10) {
+ int n = Math.min(10, count - start);
+ long[] block = unpack(in, n, in.readUnsignedByte());
+ System.arraycopy(block, 0, result, start, n);
+ }
+ return result;
+ }
+
+ static void payload(DataOutputStream out, byte[] bytes) throws IOException {
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
+
+ static DataInputStream payload(DataInputStream in) throws IOException {
+ int size = in.readInt();
+ if (size < 0 || size > in.available()) throw new IOException("Invalid payload length");
+ byte[] bytes = new byte[size];
+ in.readFully(bytes);
+ return new DataInputStream(new ByteArrayInputStream(bytes));
+ }
+
+ static void exhausted(DataInputStream in) throws IOException {
+ if (in.available() != 0) throw new IOException("Trailing data in block");
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/PageReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/PageReader.java
index ee74f815d..c7b5f754b 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/PageReader.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/PageReader.java
@@ -21,7 +21,10 @@ package org.apache.tsfile.read.reader.page;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.encoding.decoder.Decoder;
+import org.apache.tsfile.encoding.decoder.PlainDecoder;
+import org.apache.tsfile.encoding.table.ClusterNativePageCodec;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.encoding.TsFileDecodingException;
import org.apache.tsfile.file.header.PageHeader;
import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.read.common.BatchData;
@@ -55,10 +58,10 @@ public class PageReader implements IPageReader {
private final TSDataType dataType;
/** decoder for value column */
- private final Decoder valueDecoder;
+ private Decoder valueDecoder;
/** decoder for time column */
- private final Decoder timeDecoder;
+ private Decoder timeDecoder;
/** time column in memory */
private ByteBuffer timeBuffer;
@@ -128,6 +131,28 @@ public class PageReader implements IPageReader {
* @param pageData uncompressed bytes size of time column, time column, value column
*/
private void splitDataToTimeStampAndValue(ByteBuffer pageData) {
+ ByteBuffer nativeData = pageData.duplicate();
+ if (nativeData.hasRemaining()
+ && nativeData.get() == 0
+ && ClusterNativePageCodec.isNativePage(nativeData)) {
+ try {
+ ClusterNativePageCodec.Decoded decoded =
+ ClusterNativePageCodec.decode(nativeData, dataType);
+ timeBuffer = ByteBuffer.allocate(decoded.times.length * Long.BYTES);
+ for (int r = 0; r < decoded.times.length; r++) {
+ if ((decoded.bitmap[r / 8] & (0x80 >>> (r % 8))) == 0)
+ throw new IOException("Null cell in non-aligned page");
+ timeBuffer.putLong(decoded.times[r]);
+ }
+ timeBuffer.flip();
+ valueBuffer = decoded.values;
+ timeDecoder = new PlainDecoder();
+ valueDecoder = new PlainDecoder();
+ return;
+ } catch (IOException invalid) {
+ throw new TsFileDecodingException("Invalid native cluster page: " + invalid.getMessage());
+ }
+ }
int timeBufferLength = ReadWriteForEncodingUtils.readUnsignedVarInt(pageData);
timeBuffer = pageData.slice();
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/ValuePageReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/ValuePageReader.java
index 317b493c5..b3e10cd2c 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/ValuePageReader.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/ValuePageReader.java
@@ -21,7 +21,10 @@ package org.apache.tsfile.read.reader.page;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.encoding.decoder.Decoder;
+import org.apache.tsfile.encoding.decoder.PlainDecoder;
+import org.apache.tsfile.encoding.table.ClusterNativePageCodec;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.encoding.TsFileDecodingException;
import org.apache.tsfile.file.header.PageHeader;
import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.read.common.BatchData;
@@ -48,7 +51,7 @@ public class ValuePageReader {
private final TSDataType dataType;
/** decoder for value column */
- private final Decoder valueDecoder;
+ private Decoder valueDecoder;
private byte[] bitmap;
@@ -72,7 +75,6 @@ public class ValuePageReader {
if (pageData != null) {
splitDataToBitmapAndValue(pageData);
}
- this.valueBuffer = pageData;
}
public ValuePageReader(
@@ -90,6 +92,18 @@ public class ValuePageReader {
if (!pageData.hasRemaining()) { // Empty Page
return;
}
+ if (ClusterNativePageCodec.isNativePage(pageData)) {
+ try {
+ ClusterNativePageCodec.Decoded decoded = ClusterNativePageCodec.decode(pageData, dataType);
+ this.size = decoded.times.length;
+ this.bitmap = decoded.bitmap;
+ this.valueBuffer = decoded.values;
+ this.valueDecoder = new PlainDecoder();
+ return;
+ } catch (IOException invalid) {
+ throw new TsFileDecodingException("Invalid native cluster page: " + invalid.getMessage());
+ }
+ }
this.size = ReadWriteIOUtils.readInt(pageData);
this.bitmap = new byte[(size + 7) / 8];
pageData.get(bitmap);
@@ -101,7 +115,6 @@ public class ValuePageReader {
if (lazyLoadPageData != null && valueBuffer == null) {
ByteBuffer pageData = lazyLoadPageData.uncompressPageData(pageHeader);
splitDataToBitmapAndValue(pageData);
- this.valueBuffer = pageData;
lazyLoadPageData = null;
}
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java
index c051a702d..c2f9dc205 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkGroupWriterImpl.java
@@ -32,6 +32,7 @@ import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.DateUtils;
import org.apache.tsfile.write.UnSupportedDataTypeException;
+import org.apache.tsfile.write.page.ClusterAlignedPageWriter;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.record.Tablet.ColumnCategory;
import org.apache.tsfile.write.record.datapoint.DataPoint;
@@ -59,6 +60,7 @@ public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
private final Map<String, ValueChunkWriter> valueChunkWriterMap = new LinkedHashMap<>();
private final TimeChunkWriter timeChunkWriter;
+ private final ClusterAlignedPageWriter clusterPages = new ClusterAlignedPageWriter();
private final EncryptParameter encryprParam;
@@ -115,6 +117,8 @@ public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
measurementSchema.getEncodingType(),
measurementSchema.getValueEncoder());
valueChunkWriterMap.put(measurementName, valueChunkWriter);
+ ClusterAlignedPageWriter.attach(
+ timeChunkWriter.getPageWriter(), valueChunkWriterMap.values());
tryToAddEmptyPageAndData(valueChunkWriter);
}
return valueChunkWriter;
@@ -136,6 +140,8 @@ public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
schema.getEncodingType(),
schema.getValueEncoder());
valueChunkWriterMap.put(measurementName, valueChunkWriter);
+ ClusterAlignedPageWriter.attach(
+ timeChunkWriter.getPageWriter(), valueChunkWriterMap.values());
tryToAddEmptyPageAndData(valueChunkWriter);
}
}
@@ -397,6 +403,7 @@ public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
}
private void writePageToPageBuffer() {
+ clusterPages.prepare(timeChunkWriter.getPageWriter(), valueChunkWriterMap.values());
timeChunkWriter.writePageToPageBuffer();
for (ValueChunkWriter valueChunkWriter : valueChunkWriterMap.values()) {
valueChunkWriter.writePageToPageBuffer();
@@ -404,6 +411,7 @@ public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
}
private void sealAllChunks() {
+ clusterPages.prepare(timeChunkWriter.getPageWriter(), valueChunkWriterMap.values());
timeChunkWriter.sealCurrentPage();
for (ValueChunkWriter valueChunkWriter : valueChunkWriterMap.values()) {
valueChunkWriter.sealCurrentPage();
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java
index a82ca2310..991ef1bda 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java
@@ -32,6 +32,7 @@ import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.TsPrimitiveType;
+import org.apache.tsfile.write.page.ClusterAlignedPageWriter;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.VectorMeasurementSchema;
import org.apache.tsfile.write.writer.TsFileIOWriter;
@@ -46,6 +47,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
protected TimeChunkWriter timeChunkWriter;
protected List<ValueChunkWriter> valueChunkWriterList;
+ private final ClusterAlignedPageWriter clusterPages = new ClusterAlignedPageWriter();
protected int valueIndex;
protected EncryptParameter encryptParam;
@@ -84,6 +86,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
}
this.valueIndex = 0;
+ ClusterAlignedPageWriter.attach(timeChunkWriter.getPageWriter(), valueChunkWriterList);
this.remainingPointsNumber = timeChunkWriter.getRemainingPointNumberForCurrentPage();
}
@@ -115,6 +118,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
}
this.valueIndex = 0;
+ ClusterAlignedPageWriter.attach(timeChunkWriter.getPageWriter(), valueChunkWriterList);
this.remainingPointsNumber = timeChunkWriter.getRemainingPointNumberForCurrentPage();
}
@@ -149,6 +153,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
}
this.valueIndex = 0;
+ ClusterAlignedPageWriter.attach(timeChunkWriter.getPageWriter(), valueChunkWriterList);
this.remainingPointsNumber = timeChunkWriter.getRemainingPointNumberForCurrentPage();
}
@@ -178,6 +183,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
}
this.valueIndex = 0;
+ ClusterAlignedPageWriter.attach(timeChunkWriter.getPageWriter(), valueChunkWriterList);
this.remainingPointsNumber = timeChunkWriter.getRemainingPointNumberForCurrentPage();
}
@@ -216,6 +222,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
this.valueIndex = 0;
+ ClusterAlignedPageWriter.attach(timeChunkWriter.getPageWriter(), valueChunkWriterList);
this.remainingPointsNumber = timeChunkWriter.getRemainingPointNumberForCurrentPage();
}
@@ -248,6 +255,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
this.valueIndex = 0;
+ ClusterAlignedPageWriter.attach(timeChunkWriter.getPageWriter(), valueChunkWriterList);
this.remainingPointsNumber = timeChunkWriter.getRemainingPointNumberForCurrentPage();
}
@@ -450,6 +458,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
}
protected void writePageToPageBuffer() {
+ clusterPages.prepare(timeChunkWriter.getPageWriter(), valueChunkWriterList);
timeChunkWriter.writePageToPageBuffer();
for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
valueChunkWriter.writePageToPageBuffer();
@@ -468,6 +477,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
@Override
public void writeToFileWriter(TsFileIOWriter tsfileWriter) throws IOException {
+ clusterPages.prepare(timeChunkWriter.getPageWriter(), valueChunkWriterList);
timeChunkWriter.writeToFileWriter(tsfileWriter);
for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
valueChunkWriter.writeToFileWriter(tsfileWriter);
@@ -493,6 +503,7 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
@Override
public void sealCurrentPage() {
+ clusterPages.prepare(timeChunkWriter.getPageWriter(), valueChunkWriterList);
timeChunkWriter.sealCurrentPage();
for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
valueChunkWriter.sealCurrentPage();
@@ -500,10 +511,12 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
}
public void sealCurrentTimePage() {
+ clusterPages.prepare(timeChunkWriter.getPageWriter(), valueChunkWriterList);
timeChunkWriter.sealCurrentPage();
}
public void sealCurrentValuePage(int valueIndex) {
+ clusterPages.prepare(timeChunkWriter.getPageWriter(), valueChunkWriterList);
valueChunkWriterList.get(valueIndex).sealCurrentPage();
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java
index 3a3ccfa8c..5d6d97a29 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java
@@ -110,6 +110,13 @@ public class ValueChunkWriter {
this.pageWriter =
new ValuePageWriter(
valueEncoder, ICompressor.getCompressor(compressionType), dataType, this.encryptParam);
+ if ((encodingType == TSEncoding.ACLUSTER || encodingType == TSEncoding.KCLUSTER)
+ && (dataType == TSDataType.INT32
+ || dataType == TSDataType.INT64
+ || dataType == TSDataType.FLOAT
+ || dataType == TSDataType.DOUBLE)) {
+ pageWriter.enableClusterBuffer();
+ }
}
public ValueChunkWriter(
@@ -136,6 +143,13 @@ public class ValueChunkWriter {
this.pageWriter =
new ValuePageWriter(
valueEncoder, ICompressor.getCompressor(compressionType), dataType, this.encryptParam);
+ if ((encodingType == TSEncoding.ACLUSTER || encodingType == TSEncoding.KCLUSTER)
+ && (dataType == TSDataType.INT32
+ || dataType == TSDataType.INT64
+ || dataType == TSDataType.FLOAT
+ || dataType == TSDataType.DOUBLE)) {
+ pageWriter.enableClusterBuffer();
+ }
}
public void write(long time, long value, boolean isNull) {
@@ -226,6 +240,7 @@ public class ValueChunkWriter {
numOfPages++;
this.statistics.mergeStatistics(pageWriter.getStatistics());
} catch (IOException e) {
+ if (pageWriter.getClusterBuffer() != null) throw new java.io.UncheckedIOException(e);
logger.error("meet error in pageWriter.writePageHeaderAndDataIntoBuff,ignore this page:", e);
} finally {
// clear start time stamp for next initializing
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/ClusterAlignedPageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ClusterAlignedPageWriter.java
new file mode 100644
index 000000000..8ac456f02
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ClusterAlignedPageWriter.java
@@ -0,0 +1,90 @@
+/*
+ * 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.tsfile.write.page;
+
+import org.apache.tsfile.encoding.table.ClusterColumnBuffer;
+import org.apache.tsfile.encoding.table.ClusterNativePageCodec;
+import org.apache.tsfile.encoding.table.ClusterTableOptions;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.write.chunk.ValueChunkWriter;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+
+/** Coordinates page-local joint encoding before any of the aligned column pages are sealed. */
+public final class ClusterAlignedPageWriter {
+ private final Map<TSEncoding, ClusterNativePageCodec.Selection> selections =
+ new EnumMap<>(TSEncoding.class);
+
+ public static void attach(TimePageWriter time, Collection<ValueChunkWriter> writers) {
+ for (ValueChunkWriter writer : writers) {
+ if (writer.getPageWriter().getClusterBuffer() != null) {
+ time.enableClusterTimestamps();
+ return;
+ }
+ }
+ }
+
+ public void prepare(TimePageWriter time, Collection<ValueChunkWriter> writers) {
+ if (time == null || time.getPointNumber() == 0) return;
+ for (TSEncoding method : new TSEncoding[] {TSEncoding.ACLUSTER, TSEncoding.KCLUSTER}) {
+ List<ValueChunkWriter> group = new ArrayList<>();
+ for (ValueChunkWriter writer : writers) {
+ ValuePageWriter page = writer.getPageWriter();
+ if (writer.getEncodingType() == method && page != null && page.getClusterBuffer() != null)
+ group.add(writer);
+ }
+ if (group.isEmpty() || group.get(0).getPageWriter().hasClusterPage()) continue;
+ String[] names = new String[group.size()];
+ TSDataType[] types = new TSDataType[group.size()];
+ ClusterColumnBuffer[] columns = new ClusterColumnBuffer[group.size()];
+ for (int c = 0; c < group.size(); c++) {
+ names[c] = group.get(c).getMeasurementId();
+ types[c] = group.get(c).getDataType();
+ columns[c] = group.get(c).getPageWriter().getClusterBuffer();
+ }
+ ClusterNativePageCodec.Selection selection =
+ selections.computeIfAbsent(method, ignored -> new ClusterNativePageCodec.Selection());
+ try {
+ byte[][] pages =
+ ClusterNativePageCodec.encode(
+ time.getClusterTimestamps(),
+ names,
+ types,
+ columns,
+ method == TSEncoding.ACLUSTER
+ ? ClusterTableOptions.aCluster()
+ : ClusterTableOptions.kCluster(),
+ selection);
+ for (int c = 0; c < group.size(); c++)
+ group.get(c).getPageWriter().setClusterPage(pages[c]);
+ } catch (IOException failure) {
+ // Never fall back to independently reordered scalar streams or silently drop a page.
+ throw new UncheckedIOException("Cannot encode joint aligned cluster page", failure);
+ }
+ }
+ }
+}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java
index 877404304..19a8103c2 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java
@@ -19,13 +19,18 @@
package org.apache.tsfile.write.page;
import org.apache.tsfile.compress.ICompressor;
+import org.apache.tsfile.encoding.encoder.ClusterEncoder;
import org.apache.tsfile.encoding.encoder.Encoder;
+import org.apache.tsfile.encoding.table.ClusterColumnBuffer;
+import org.apache.tsfile.encoding.table.ClusterNativePageCodec;
+import org.apache.tsfile.encoding.table.ClusterTableOptions;
import org.apache.tsfile.encrypt.EncryptParameter;
import org.apache.tsfile.encrypt.EncryptUtils;
import org.apache.tsfile.encrypt.IEncryptor;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.EncryptionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.PublicBAOS;
@@ -59,6 +64,10 @@ public class PageWriter {
// value
private Encoder valueEncoder;
private PublicBAOS valueOut;
+ private ClusterColumnBuffer clusterValues;
+ private ClusterColumnBuffer clusterTimes;
+ private final ClusterNativePageCodec.Selection clusterSelection =
+ new ClusterNativePageCodec.Selection();
/**
* statistic of current page. It will be reset after calling {@code
@@ -119,29 +128,37 @@ public class PageWriter {
/** write a time value pair into encoder */
public void write(long time, int value) {
- timeEncoder.encode(time, timeOut);
- valueEncoder.encode(value, valueOut);
+ if (!captureClusterValue(time, value)) {
+ timeEncoder.encode(time, timeOut);
+ valueEncoder.encode(value, valueOut);
+ }
statistics.update(time, value);
}
/** write a time value pair into encoder */
public void write(long time, long value) {
- timeEncoder.encode(time, timeOut);
- valueEncoder.encode(value, valueOut);
+ if (!captureClusterValue(time, value)) {
+ timeEncoder.encode(time, timeOut);
+ valueEncoder.encode(value, valueOut);
+ }
statistics.update(time, value);
}
/** write a time value pair into encoder */
public void write(long time, float value) {
- timeEncoder.encode(time, timeOut);
- valueEncoder.encode(value, valueOut);
+ if (!captureClusterValue(time, Float.floatToRawIntBits(value))) {
+ timeEncoder.encode(time, timeOut);
+ valueEncoder.encode(value, valueOut);
+ }
statistics.update(time, value);
}
/** write a time value pair into encoder */
public void write(long time, double value) {
- timeEncoder.encode(time, timeOut);
- valueEncoder.encode(value, valueOut);
+ if (!captureClusterValue(time, Double.doubleToRawLongBits(value))) {
+ timeEncoder.encode(time, timeOut);
+ valueEncoder.encode(value, valueOut);
+ }
statistics.update(time, value);
}
@@ -164,8 +181,10 @@ public class PageWriter {
/** write time series into encoder */
public void write(long[] timestamps, int[] values, int batchSize) {
for (int i = 0; i < batchSize; i++) {
- timeEncoder.encode(timestamps[i], timeOut);
- valueEncoder.encode(values[i], valueOut);
+ if (!captureClusterValue(timestamps[i], values[i])) {
+ timeEncoder.encode(timestamps[i], timeOut);
+ valueEncoder.encode(values[i], valueOut);
+ }
}
statistics.update(timestamps, values, batchSize);
}
@@ -173,8 +192,10 @@ public class PageWriter {
/** write time series into encoder */
public void write(long[] timestamps, long[] values, int batchSize) {
for (int i = 0; i < batchSize; i++) {
- timeEncoder.encode(timestamps[i], timeOut);
- valueEncoder.encode(values[i], valueOut);
+ if (!captureClusterValue(timestamps[i], values[i])) {
+ timeEncoder.encode(timestamps[i], timeOut);
+ valueEncoder.encode(values[i], valueOut);
+ }
}
statistics.update(timestamps, values, batchSize);
}
@@ -182,8 +203,10 @@ public class PageWriter {
/** write time series into encoder */
public void write(long[] timestamps, float[] values, int batchSize) {
for (int i = 0; i < batchSize; i++) {
- timeEncoder.encode(timestamps[i], timeOut);
- valueEncoder.encode(values[i], valueOut);
+ if (!captureClusterValue(timestamps[i], Float.floatToRawIntBits(values[i]))) {
+ timeEncoder.encode(timestamps[i], timeOut);
+ valueEncoder.encode(values[i], valueOut);
+ }
}
statistics.update(timestamps, values, batchSize);
}
@@ -191,8 +214,10 @@ public class PageWriter {
/** write time series into encoder */
public void write(long[] timestamps, double[] values, int batchSize) {
for (int i = 0; i < batchSize; i++) {
- timeEncoder.encode(timestamps[i], timeOut);
- valueEncoder.encode(values[i], valueOut);
+ if (!captureClusterValue(timestamps[i], Double.doubleToRawLongBits(values[i]))) {
+ timeEncoder.encode(timestamps[i], timeOut);
+ valueEncoder.encode(values[i], valueOut);
+ }
}
statistics.update(timestamps, values, batchSize);
}
@@ -206,6 +231,22 @@ public class PageWriter {
statistics.update(timestamps, values, batchSize);
}
+ private boolean captureClusterValue(long time, long bits) {
+ TSDataType type = statistics.getType();
+ if (!(valueEncoder instanceof ClusterEncoder)
+ || (type != TSDataType.INT32
+ && type != TSDataType.INT64
+ && type != TSDataType.FLOAT
+ && type != TSDataType.DOUBLE)) return false;
+ if (clusterValues == null) {
+ clusterValues = new ClusterColumnBuffer();
+ clusterTimes = new ClusterColumnBuffer();
+ }
+ clusterValues.append(bits, false);
+ clusterTimes.append(time, false);
+ return true;
+ }
+
/** flush all data remained in encoders. */
private void prepareEndWriteOnePage() throws IOException {
timeEncoder.flush(timeOut);
@@ -219,6 +260,29 @@ public class PageWriter {
* @return a new readable ByteBuffer whose position is 0.
*/
public ByteBuffer getUncompressedBytes() throws IOException {
+ if (clusterValues != null && clusterValues.size() != 0) {
+ long[] times = new long[clusterTimes.size()];
+ for (int r = 0; r < times.length; r++) times[r] = clusterTimes.bits(r);
+ byte[][] pages;
+ try {
+ pages =
+ ClusterNativePageCodec.encode(
+ times,
+ new String[] {"value"},
+ new TSDataType[] {statistics.getType()},
+ new ClusterColumnBuffer[] {clusterValues},
+ valueEncoder.getType() == TSEncoding.ACLUSTER
+ ? ClusterTableOptions.aCluster()
+ : ClusterTableOptions.kCluster(),
+ clusterSelection);
+ } catch (IOException failure) {
+ throw new java.io.UncheckedIOException("Cannot encode native scalar cluster page", failure);
+ }
+ // Zero time-stream length cannot occur in a nonempty legacy scalar page.
+ ByteBuffer nativePage = ByteBuffer.allocate(1 + pages[0].length);
+ nativePage.put((byte) 0).put(pages[0]).flip();
+ return nativePage;
+ }
prepareEndWriteOnePage();
ByteBuffer buffer = ByteBuffer.allocate(timeOut.size() + valueOut.size() + 4);
ReadWriteForEncodingUtils.writeUnsignedVarInt(timeOut.size(), buffer);
@@ -309,11 +373,16 @@ public class PageWriter {
return timeOut.size()
+ valueOut.size()
+ timeEncoder.getMaxByteSize()
- + valueEncoder.getMaxByteSize();
+ + valueEncoder.getMaxByteSize()
+ + (clusterValues == null ? 0 : clusterValues.memoryBytes() + clusterTimes.memoryBytes());
}
/** reset this page */
public void reset(IMeasurementSchema measurementSchema) {
+ if (clusterValues != null) {
+ clusterValues.reset();
+ clusterTimes.reset();
+ }
timeOut.reset();
valueOut.reset();
statistics = Statistics.getStatsByType(measurementSchema.getType());
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java
index b83743ef9..bb806077e 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java
@@ -36,6 +36,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
+import java.util.Arrays;
/**
* This writer is used to write time into a page. It consists of a time encoder and respective
@@ -59,6 +60,9 @@ public class TimePageWriter {
*/
private TimeStatistics statistics;
+ private long[] clusterTimestamps;
+ private int clusterTimeCount;
+
public TimePageWriter(Encoder timeEncoder, ICompressor compressor) {
this.timeOut = new PublicBAOS();
this.timeEncoder = timeEncoder;
@@ -78,6 +82,7 @@ public class TimePageWriter {
/** write a time into encoder */
public void write(long time) {
+ captureClusterTime(time);
timeEncoder.encode(time, timeOut);
statistics.update(time);
}
@@ -85,6 +90,7 @@ public class TimePageWriter {
/** write time series into encoder */
public void write(long[] timestamps, int batchSize, int arrayOffset) {
for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ captureClusterTime(timestamps[i]);
timeEncoder.encode(timestamps[i], timeOut);
}
if (batchSize != 0) {
@@ -190,15 +196,39 @@ public class TimePageWriter {
* @return allocated size in time, value and outputStream
*/
public long estimateMaxMemSize() {
- return timeOut.size() + timeEncoder.getMaxByteSize();
+ return timeOut.size()
+ + timeEncoder.getMaxByteSize()
+ + (clusterTimestamps == null ? 0 : 8L * clusterTimestamps.length);
}
/** reset this page */
public void reset() {
+ clusterTimeCount = 0;
timeOut.reset();
statistics = new TimeStatistics();
}
+ public void enableClusterTimestamps() {
+ if (clusterTimestamps == null) {
+ if (statistics.getCount() != 0)
+ throw new IllegalStateException("Register cluster columns before writing rows");
+ clusterTimestamps = new long[128];
+ }
+ }
+
+ private void captureClusterTime(long time) {
+ if (clusterTimestamps == null) return;
+ if (clusterTimeCount == clusterTimestamps.length)
+ clusterTimestamps = Arrays.copyOf(clusterTimestamps, clusterTimestamps.length * 2);
+ clusterTimestamps[clusterTimeCount++] = time;
+ }
+
+ public long[] getClusterTimestamps() {
+ if (clusterTimestamps == null || clusterTimeCount != statistics.getCount())
+ throw new IllegalStateException("Missing native cluster time keys");
+ return Arrays.copyOf(clusterTimestamps, clusterTimeCount);
+ }
+
public void setTimeEncoder(Encoder encoder) {
this.timeEncoder = encoder;
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java
index 6199171a0..3f505441b 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java
@@ -20,6 +20,7 @@ package org.apache.tsfile.write.page;
import org.apache.tsfile.compress.ICompressor;
import org.apache.tsfile.encoding.encoder.Encoder;
+import org.apache.tsfile.encoding.table.ClusterColumnBuffer;
import org.apache.tsfile.encrypt.EncryptParameter;
import org.apache.tsfile.encrypt.EncryptUtils;
import org.apache.tsfile.encrypt.IEncryptor;
@@ -61,6 +62,9 @@ public class ValuePageWriter {
*/
private Statistics<? extends Serializable> statistics;
+ private ClusterColumnBuffer clusterBuffer;
+ private byte[] clusterPage;
+
private byte bitmap;
private int size;
@@ -99,7 +103,7 @@ public class ValuePageWriter {
public void write(long time, boolean value, boolean isNull) {
setBit(isNull);
if (!isNull) {
- valueEncoder.encode(value, valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(value, valueOut);
statistics.update(time, value);
}
}
@@ -108,43 +112,47 @@ public class ValuePageWriter {
public void write(long time, short value, boolean isNull) {
setBit(isNull);
if (!isNull) {
- valueEncoder.encode(value, valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(value, valueOut);
statistics.update(time, value);
}
}
/** write a time value pair into encoder */
public void write(long time, int value, boolean isNull) {
+ if (clusterBuffer != null) clusterBuffer.append(value, isNull);
setBit(isNull);
if (!isNull) {
- valueEncoder.encode(value, valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(value, valueOut);
statistics.update(time, value);
}
}
/** write a time value pair into encoder */
public void write(long time, long value, boolean isNull) {
+ if (clusterBuffer != null) clusterBuffer.append(value, isNull);
setBit(isNull);
if (!isNull) {
- valueEncoder.encode(value, valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(value, valueOut);
statistics.update(time, value);
}
}
/** write a time value pair into encoder */
public void write(long time, float value, boolean isNull) {
+ if (clusterBuffer != null) clusterBuffer.append(Float.floatToRawIntBits(value), isNull);
setBit(isNull);
if (!isNull) {
- valueEncoder.encode(value, valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(value, valueOut);
statistics.update(time, value);
}
}
/** write a time value pair into encoder */
public void write(long time, double value, boolean isNull) {
+ if (clusterBuffer != null) clusterBuffer.append(Double.doubleToRawLongBits(value), isNull);
setBit(isNull);
if (!isNull) {
- valueEncoder.encode(value, valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(value, valueOut);
statistics.update(time, value);
}
}
@@ -153,7 +161,7 @@ public class ValuePageWriter {
public void write(long time, Binary value, boolean isNull) {
setBit(isNull);
if (!isNull) {
- valueEncoder.encode(value, valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(value, valueOut);
statistics.update(time, value);
}
}
@@ -175,7 +183,7 @@ public class ValuePageWriter {
for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
setBit(isNull[i]);
if (!isNull[i]) {
- valueEncoder.encode(values[i], valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(values[i], valueOut);
statistics.update(timestamps[i], values[i]);
}
}
@@ -185,9 +193,10 @@ public class ValuePageWriter {
public void write(
long[] timestamps, int[] values, boolean[] isNull, int batchSize, int arrayOffset) {
for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ if (clusterBuffer != null) clusterBuffer.append(values[i], isNull[i]);
setBit(isNull[i]);
if (!isNull[i]) {
- valueEncoder.encode(values[i], valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(values[i], valueOut);
statistics.update(timestamps[i], values[i]);
}
}
@@ -197,9 +206,10 @@ public class ValuePageWriter {
public void write(
long[] timestamps, long[] values, boolean[] isNull, int batchSize, int arrayOffset) {
for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ if (clusterBuffer != null) clusterBuffer.append(values[i], isNull[i]);
setBit(isNull[i]);
if (!isNull[i]) {
- valueEncoder.encode(values[i], valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(values[i], valueOut);
statistics.update(timestamps[i], values[i]);
}
}
@@ -209,9 +219,11 @@ public class ValuePageWriter {
public void write(
long[] timestamps, float[] values, boolean[] isNull, int batchSize, int arrayOffset) {
for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ if (clusterBuffer != null)
+ clusterBuffer.append(Float.floatToRawIntBits(values[i]), isNull[i]);
setBit(isNull[i]);
if (!isNull[i]) {
- valueEncoder.encode(values[i], valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(values[i], valueOut);
statistics.update(timestamps[i], values[i]);
}
}
@@ -221,9 +233,11 @@ public class ValuePageWriter {
public void write(
long[] timestamps, double[] values, boolean[] isNull, int batchSize, int arrayOffset) {
for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
+ if (clusterBuffer != null)
+ clusterBuffer.append(Double.doubleToRawLongBits(values[i]), isNull[i]);
setBit(isNull[i]);
if (!isNull[i]) {
- valueEncoder.encode(values[i], valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(values[i], valueOut);
statistics.update(timestamps[i], values[i]);
}
}
@@ -235,7 +249,7 @@ public class ValuePageWriter {
for (int i = arrayOffset; i < batchSize + arrayOffset; i++) {
setBit(isNull[i]);
if (!isNull[i]) {
- valueEncoder.encode(values[i], valueOut);
+ if (clusterBuffer == null) valueEncoder.encode(values[i], valueOut);
statistics.update(timestamps[i], values[i]);
}
}
@@ -256,6 +270,11 @@ public class ValuePageWriter {
* @return a new readable ByteBuffer whose position is 0.
*/
public ByteBuffer getUncompressedBytes() throws IOException {
+ if (clusterBuffer != null) {
+ if (clusterPage == null)
+ throw new IllegalStateException("Joint cluster page was not prepared");
+ return ByteBuffer.wrap(clusterPage);
+ }
prepareEndWriteOnePage();
ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES + bitmapOut.size() + valueOut.size());
buffer.putInt(size);
@@ -348,11 +367,19 @@ public class ValuePageWriter {
* @return allocated size in time, value and outputStream
*/
public long estimateMaxMemSize() {
- return Integer.BYTES + bitmapOut.size() + 1 + valueOut.size() + valueEncoder.getMaxByteSize();
+ return Integer.BYTES
+ + bitmapOut.size()
+ + 1
+ + valueOut.size()
+ + valueEncoder.getMaxByteSize()
+ + (clusterBuffer == null ? 0 : clusterBuffer.memoryBytes())
+ + (clusterPage == null ? 0 : clusterPage.length);
}
/** reset this page */
public void reset(TSDataType dataType) {
+ if (clusterBuffer != null) clusterBuffer.reset();
+ clusterPage = null;
bitmapOut.reset();
size = 0;
bitmap = 0;
@@ -360,6 +387,23 @@ public class ValuePageWriter {
statistics = Statistics.getStatsByType(dataType);
}
+ public void enableClusterBuffer() {
+ if (size != 0) throw new IllegalStateException("Cannot change a populated page codec");
+ clusterBuffer = new ClusterColumnBuffer();
+ }
+
+ public ClusterColumnBuffer getClusterBuffer() {
+ return clusterBuffer;
+ }
+
+ public boolean hasClusterPage() {
+ return clusterPage != null;
+ }
+
+ public void setClusterPage(byte[] page) {
+ clusterPage = page;
+ }
+
public void setValueEncoder(Encoder encoder) {
this.valueEncoder = encoder;
}
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/encoding/decoder/ClusterCodecTest.java b/java/tsfile/src/test/java/org/apache/tsfile/encoding/decoder/ClusterCodecTest.java
new file mode 100644
index 000000000..c88b93523
--- /dev/null
+++ b/java/tsfile/src/test/java/org/apache/tsfile/encoding/decoder/ClusterCodecTest.java
@@ -0,0 +1,502 @@
+/*
+ * 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.tsfile.encoding.decoder;
+
+import org.apache.tsfile.compress.ICompressor;
+import org.apache.tsfile.encoding.encoder.AClusterAlgorithm;
+import org.apache.tsfile.encoding.encoder.AClusterEncoder;
+import org.apache.tsfile.encoding.encoder.ClusterEncoder;
+import org.apache.tsfile.encoding.encoder.ClusterResult;
+import org.apache.tsfile.encoding.encoder.Encoder;
+import org.apache.tsfile.encoding.encoder.KClusterAlgorithm;
+import org.apache.tsfile.encoding.encoder.KClusterEncoder;
+import org.apache.tsfile.encoding.encoder.TSEncodingBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.encoding.TsFileDecodingException;
+import org.apache.tsfile.exception.encoding.TsFileEncodingException;
+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.TsFileSequenceReader;
+import org.apache.tsfile.read.common.BatchData;
+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.reader.page.PageReader;
+import org.apache.tsfile.read.reader.page.ValuePageReader;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.page.PageWriter;
+import org.apache.tsfile.write.page.ValuePageWriter;
+import org.apache.tsfile.write.record.TSRecord;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(Parameterized.class)
+public class ClusterCodecTest {
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<Object[]> encodings() {
+ return Arrays.asList(new Object[][] {{TSEncoding.ACLUSTER}, {TSEncoding.KCLUSTER}});
+ }
+
+ private final TSEncoding encoding;
+
+ public ClusterCodecTest(TSEncoding encoding) {
+ this.encoding = encoding;
+ }
+
+ private Encoder encoder(TSDataType type) {
+ return encoding == TSEncoding.ACLUSTER
+ ? new AClusterEncoder(type)
+ : new KClusterEncoder(type, 4, 3, 42);
+ }
+
+ private void encode(Encoder encoder, TSDataType type, long value, ByteArrayOutputStream out) {
+ switch (type) {
+ case INT32:
+ case DATE:
+ encoder.encode((int) value, out);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ encoder.encode(value, out);
+ break;
+ case FLOAT:
+ encoder.encode(Float.intBitsToFloat((int) value), out);
+ break;
+ case DOUBLE:
+ encoder.encode(Double.longBitsToDouble(value), out);
+ break;
+ default:
+ throw new AssertionError(type);
+ }
+ }
+
+ private long read(Decoder decoder, TSDataType type, ByteBuffer buffer) {
+ switch (type) {
+ case INT32:
+ case DATE:
+ return decoder.readInt(buffer);
+ case INT64:
+ case TIMESTAMP:
+ return decoder.readLong(buffer);
+ case FLOAT:
+ return Float.floatToRawIntBits(decoder.readFloat(buffer)) & 0xffffffffL;
+ case DOUBLE:
+ return Double.doubleToRawLongBits(decoder.readDouble(buffer));
+ default:
+ throw new AssertionError(type);
+ }
+ }
+
+ private byte[] roundTrip(TSDataType type, long[] input) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ Encoder writer = encoder(type);
+ for (long value : input) {
+ encode(writer, type, value, out);
+ }
+ writer.flush(out);
+ int bytes = out.size();
+ writer.flush(out);
+ assertEquals(bytes, out.size());
+ Decoder reader = Decoder.getDecoderByType(encoding, type);
+ assertEquals(encoding, reader.getType());
+ ByteBuffer buffer = ByteBuffer.wrap(out.toByteArray());
+ long[] decoded = new long[input.length];
+ // Real TsFile page readers call readXXX directly without first calling hasNext.
+ for (int i = 0; i < input.length; i++) {
+ if (i > 0) {
+ assertTrue(reader.hasNext(buffer));
+ assertTrue(reader.hasNext(buffer));
+ }
+ decoded[i] = read(reader, type, buffer);
+ }
+ assertFalse(reader.hasNext(buffer));
+ assertEquals(0, buffer.remaining());
+ assertEquals(frequencies(input), frequencies(decoded));
+ return out.toByteArray();
+ }
+
+ private Map<Long, Integer> frequencies(long[] values) {
+ Map<Long, Integer> result = new HashMap<>();
+ for (long value : values) {
+ result.put(value, result.getOrDefault(value, 0) + 1);
+ }
+ return result;
+ }
+
+ @Test
+ public void testGroupedIntegerMultisets() throws IOException {
+ long[] values = new long[600];
+ for (int i = 0; i < values.length; i++) {
+ values[i] = new long[] {-5, 100, -4, 10000, -5, 10000}[i % 6];
+ }
+ for (TSDataType type :
+ new TSDataType[] {
+ TSDataType.INT32, TSDataType.INT64, TSDataType.DATE, TSDataType.TIMESTAMP
+ }) {
+ byte[] encoded = roundTrip(type, values);
+ assertTrue("Test must exercise cluster storage, not RAW", encoded[7] != 0);
+ }
+ }
+
+ @Test
+ public void testEmptyConstantAndShortBlocks() throws IOException {
+ roundTrip(TSDataType.INT64, new long[0]);
+ roundTrip(TSDataType.INT64, new long[] {-5});
+ roundTrip(TSDataType.INT32, new long[] {0, 0, 0});
+ roundTrip(TSDataType.INT64, new long[] {Long.MIN_VALUE});
+ long[] same = new long[200];
+ Arrays.fill(same, -5);
+ assertTrue(roundTrip(TSDataType.INT64, same).length < 32);
+ }
+
+ @Test
+ public void testIntegerExtremes() throws IOException {
+ roundTrip(
+ TSDataType.INT32,
+ new long[] {Integer.MIN_VALUE, Integer.MAX_VALUE, -1, 0, Integer.MIN_VALUE});
+ byte[] raw =
+ roundTrip(
+ TSDataType.INT64,
+ new long[] {Long.MIN_VALUE, Long.MAX_VALUE, 0, -1, 1, Long.MIN_VALUE});
+ assertEquals(0, raw[7]);
+ roundTrip(
+ TSDataType.INT64, new long[] {Long.MAX_VALUE, Long.MAX_VALUE - 1, Long.MAX_VALUE - 3});
+ }
+
+ @Test
+ public void testExactDecimalScaling() throws IOException {
+ for (TSDataType type : new TSDataType[] {TSDataType.FLOAT, TSDataType.DOUBLE}) {
+ long[] values = new long[500];
+ for (int i = 0; i < values.length; i++) {
+ double value = new double[] {-5.25, 0.1, 100.125, 7.75}[i % 4];
+ values[i] =
+ type == TSDataType.FLOAT
+ ? Float.floatToRawIntBits((float) value) & 0xffffffffL
+ : Double.doubleToRawLongBits(value);
+ }
+ assertTrue(roundTrip(type, values)[7] != 0);
+ }
+ }
+
+ @Test
+ public void testRawFloatingBits() throws IOException {
+ long[] doubles = {
+ 0L,
+ Long.MIN_VALUE,
+ 0x7ff0000000000000L,
+ 0xfff0000000000000L,
+ 0x7ff8000000000001L,
+ 0x7ff8000000000002L,
+ 1L,
+ 0x7fefffffffffffffL,
+ Double.doubleToRawLongBits(0.12345678901234567)
+ };
+ assertEquals(0, roundTrip(TSDataType.DOUBLE, doubles)[7]);
+ long[] floats = {
+ 0, 0x80000000L, 0x7f800000L, 0xff800000L, 0x7fc00001L, 0x7fc00002L, 1, 0x7f7fffffL
+ };
+ assertEquals(0, roundTrip(TSDataType.FLOAT, floats)[7]);
+ // Finite values alone can also require RAW, without NaN or infinity.
+ assertEquals(
+ 0,
+ roundTrip(
+ TSDataType.DOUBLE,
+ new long[] {Double.doubleToRawLongBits(1e300), Double.doubleToRawLongBits(1e-300)})[7]);
+ }
+
+ @Test
+ public void testSeededRandomRoundTrips() throws IOException {
+ Random random = new Random(123);
+ for (int trial = 0; trial < 12; trial++) {
+ long[] integers = new long[1 + random.nextInt(150)];
+ long[] doubles = new long[integers.length];
+ for (int i = 0; i < integers.length; i++) {
+ integers[i] = random.nextInt(1000) - 500;
+ doubles[i] = Double.doubleToRawLongBits(integers[i] / 100.0);
+ }
+ roundTrip(TSDataType.INT64, integers);
+ roundTrip(TSDataType.DOUBLE, doubles);
+ }
+ }
+
+ @Test
+ public void testBoundedBlocksBeyondUnsignedShortCount() throws IOException {
+ long[] values = new long[65537];
+ Arrays.fill(values, -7);
+ assertTrue(roundTrip(TSDataType.INT64, values).length < 256);
+ Encoder writer = encoder(TSDataType.INT64);
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ for (int i = 0; i < 3 * ClusterEncoder.MAX_BLOCK_VALUES; i++) {
+ writer.encode(7L, out);
+ }
+ assertTrue(writer.getMaxByteSize() <= 64L + 8L * ClusterEncoder.MAX_BLOCK_VALUES);
+ writer.flush(out);
+ }
+
+ @Test
+ public void testReuseAndReset() throws IOException {
+ Encoder writer = encoder(TSDataType.INT64);
+ Decoder reader = Decoder.getDecoderByType(encoding, TSDataType.INT64);
+ for (long value : new long[] {-5, 0, 100}) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ writer.encode(value, out);
+ writer.flush(out);
+ reader.reset();
+ ByteBuffer buffer = ByteBuffer.wrap(out.toByteArray());
+ assertEquals(value, reader.readLong(buffer));
+ assertFalse(reader.hasNext(buffer));
+ }
+ }
+
+ @Test
+ public void testMalformedAndTruncatedBlocks() throws IOException {
+ byte[] raw = roundTrip(TSDataType.INT64, new long[] {Long.MIN_VALUE, Long.MAX_VALUE, 3});
+ for (int end = 1; end < raw.length; end++) {
+ expectInvalid(Arrays.copyOf(raw, end));
+ }
+ byte[] wrongVersion = raw.clone();
+ wrongVersion[8] = 2;
+ expectInvalid(wrongVersion);
+ byte[] wrongType = raw.clone();
+ wrongType[9] = TSDataType.FLOAT.serialize();
+ expectInvalid(wrongType);
+ byte[] zeroPack = raw.clone();
+ zeroPack[5] = zeroPack[6] = 0;
+ expectInvalid(zeroPack);
+ long[] repeated = new long[100];
+ for (int i = 0; i < repeated.length; i++) {
+ repeated[i] = i % 2 == 0 ? -3 : 1000;
+ }
+ byte[] clustered = roundTrip(TSDataType.INT64, repeated);
+ clustered[4]++; // Header no longer matches the encoded group counts.
+ expectInvalid(clustered);
+ }
+
+ private void expectInvalid(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ try {
+ Decoder.getDecoderByType(encoding, TSDataType.INT64).readLong(buffer);
+ fail("Malformed block should fail");
+ } catch (TsFileDecodingException expected) {
+ assertEquals("Failed parsing must not consume caller bytes", 0, buffer.position());
+ }
+ }
+
+ @Test
+ public void testTypeMismatchIsRejected() {
+ try {
+ encoder(TSDataType.INT64).encode(1.0, new ByteArrayOutputStream());
+ fail("Mismatched writes must not be silently dropped");
+ } catch (TsFileEncodingException expected) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testRealPageReaderDirectReads() throws IOException {
+ PageWriter writer =
+ new PageWriter(
+ new MeasurementSchema("v", TSDataType.INT64, encoding, CompressionType.UNCOMPRESSED));
+ long[] input = new long[100];
+ for (int i = 0; i < input.length; i++) {
+ input[i] = i % 3 == 0 ? -5 : 100;
+ writer.write(i, input[i]);
+ }
+ PageReader reader =
+ new PageReader(
+ writer.getUncompressedBytes(),
+ TSDataType.INT64,
+ Decoder.getDecoderByType(encoding, TSDataType.INT64),
+ Decoder.getDecoderByType(TSEncoding.TS_2DIFF, TSDataType.INT64));
+ assertEquals(frequencies(input), batchFrequencies(reader.getAllSatisfiedPageData(true)));
+ }
+
+ @Test
+ public void testRealValuePageReaderWithNulls() throws IOException {
+ ValuePageWriter writer =
+ new ValuePageWriter(
+ encoder(TSDataType.INT64),
+ ICompressor.getCompressor(CompressionType.UNCOMPRESSED),
+ TSDataType.INT64);
+ long[] times = new long[120];
+ long[] nonNull = new long[80];
+ int index = 0;
+ for (int i = 0; i < times.length; i++) {
+ times[i] = i;
+ long value = i % 2 == 0 ? -5 : 100;
+ boolean isNull = i % 3 == 0;
+ writer.write(i, value, isNull);
+ if (!isNull) {
+ nonNull[index++] = value;
+ }
+ }
+ ValuePageReader reader =
+ new ValuePageReader(
+ null,
+ writer.getUncompressedBytes(),
+ TSDataType.INT64,
+ Decoder.getDecoderByType(encoding, TSDataType.INT64));
+ assertEquals(frequencies(nonNull), batchFrequencies(reader.nextBatch(times, true, null)));
+ }
+
+ private Map<Long, Integer> batchFrequencies(BatchData batch) {
+ Map<Long, Integer> result = new HashMap<>();
+ while (batch.hasCurrent()) {
+ long value = ((Number) batch.currentValue()).longValue();
+ result.put(value, result.getOrDefault(value, 0) + 1);
+ batch.next();
+ }
+ return result;
+ }
+
+ @Test
+ public void testLegacyClusterFixture() throws IOException {
+ // Legacy layout: one reference at offset 0, minimum 5, frequency 3, zero residuals.
+ byte[] legacy = {
+ 0, 0, 1, 0, 3, 0, 10, 3, 80, 24, 4, 0, 2, 5, (byte) 128, 0, 0, 0, (byte) 128, (byte) 128
+ };
+ Decoder reader = Decoder.getDecoderByType(encoding, TSDataType.INT64);
+ ByteBuffer input = ByteBuffer.wrap(legacy);
+ for (int i = 0; i < 3; i++) {
+ assertEquals(5L, reader.readLong(input));
+ }
+ assertFalse(reader.hasNext(input));
+ }
+
+ @Test
+ public void testAlgorithmResultContracts() {
+ if (encoding == TSEncoding.ACLUSTER) {
+ // The new point still pays its self-residual cost; opening a second reference loses.
+ assertEquals(1, AClusterAlgorithm.run(new long[] {0, 2, 3}).references.length);
+ return;
+ }
+ Random random = new Random(12);
+ for (int trial = 0; trial < 20; trial++) {
+ long[] data = new long[45];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = random.nextInt(500);
+ }
+ ClusterResult result = KClusterAlgorithm.run(data, 5, 1, trial);
+ result.validate(data.length);
+ ClusterResult again = KClusterAlgorithm.run(data, 5, 1, trial);
+ assertArrayEquals(result.references, again.references);
+ assertArrayEquals(result.assignments, again.assignments);
+ for (int i = 0; i < data.length; i++) {
+ long assignedDistance = Math.abs(data[i] - result.references[result.assignments[i]]);
+ int assignedBits = Math.max(1, 64 - Long.numberOfLeadingZeros(assignedDistance));
+ for (long reference : result.references) {
+ long distance = Math.abs(data[i] - reference);
+ int bits = Math.max(1, 64 - Long.numberOfLeadingZeros(distance));
+ assertTrue("Assignments must match the final references", assignedBits <= bits);
+ if (data[i] == reference) {
+ assertEquals(reference, result.references[result.assignments[i]]);
+ }
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testBuilderPropertiesAndAliases() throws IOException {
+ Map<String, String> props = new HashMap<>();
+ props.put("k", "2");
+ props.put("max_iterations", "1");
+ props.put("seed", "123");
+ TSEncodingBuilder builder = TSEncodingBuilder.getEncodingBuilder(encoding);
+ builder.initFromProps(props);
+ for (TSDataType type : new TSDataType[] {TSDataType.DATE, TSDataType.TIMESTAMP}) {
+ Encoder writer = builder.getEncoder(type);
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ encode(writer, type, -5, out);
+ writer.flush(out);
+ assertEquals(
+ -5L,
+ read(Decoder.getDecoderByType(encoding, type), type, ByteBuffer.wrap(out.toByteArray())));
+ }
+ if (encoding == TSEncoding.KCLUSTER) {
+ props.put("max_iterations", "0");
+ try {
+ builder.initFromProps(props);
+ fail("Invalid iteration limit must be rejected");
+ } catch (IllegalArgumentException expected) {
+ // expected
+ }
+ }
+ }
+
+ @Test
+ public void testFileCloseReopenAcrossChunks() throws Exception {
+ File file = new File(temporaryFolder.getRoot(), "cluster.tsfile");
+ long[] input = new long[1200];
+ try (TsFileWriter writer = new TsFileWriter(file)) {
+ writer.registerTimeseries(
+ "root.table",
+ new MeasurementSchema("v", TSDataType.INT64, encoding, CompressionType.UNCOMPRESSED));
+ for (int i = 0; i < input.length; i++) {
+ input[i] =
+ i < 600
+ ? new long[] {-5, 100, 10000}[i % 3]
+ : new long[] {Long.MIN_VALUE, Long.MAX_VALUE, 0}[i % 3];
+ writer.writeRecord(new TSRecord("root.table", i).addPoint("v", input[i]));
+ if ((i + 1) % 200 == 0) {
+ writer.flush();
+ }
+ }
+ }
+ Map<Long, Integer> result = new HashMap<>();
+ try (TsFileSequenceReader sequence = new TsFileSequenceReader(file.getAbsolutePath());
+ TsFileReader reader = new TsFileReader(sequence)) {
+ QueryDataSet rows =
+ reader.query(
+ QueryExpression.create(
+ Collections.singletonList(new Path("root.table", "v", true)), null));
+ while (rows.hasNext()) {
+ long value = rows.next().getFields().get(0).getLongV();
+ result.put(value, result.getOrDefault(value, 0) + 1);
+ }
+ }
+ assertEquals(frequencies(input), result);
+ }
+}
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterAlignedTsFileTest.java b/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterAlignedTsFileTest.java
new file mode 100644
index 000000000..84bb7a9f5
--- /dev/null
+++ b/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterAlignedTsFileTest.java
@@ -0,0 +1,272 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.common.conf.TSFileDescriptor;
+import org.apache.tsfile.encoding.decoder.PlainDecoder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+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.TsFileSequenceReader;
+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.write.TsFileWriter;
+import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl;
+import org.apache.tsfile.write.record.TSRecord;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.apache.tsfile.write.writer.TsFileIOWriter;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+
+@RunWith(Parameterized.class)
+public class ClusterAlignedTsFileTest {
+ @Parameterized.Parameters(name = "{0}")
+ public static Object[] encodings() {
+ return new Object[] {TSEncoding.ACLUSTER, TSEncoding.KCLUSTER};
+ }
+
+ private final TSEncoding encoding;
+ @Rule public TemporaryFolder temp = new TemporaryFolder();
+
+ public ClusterAlignedTsFileTest(TSEncoding encoding) {
+ this.encoding = encoding;
+ }
+
+ private List<IMeasurementSchema> schemas() {
+ List<IMeasurementSchema> schemas = new ArrayList<>();
+ schemas.add(new MeasurementSchema("a", TSDataType.DOUBLE, encoding, CompressionType.LZ4));
+ schemas.add(new MeasurementSchema("b", TSDataType.INT64, encoding, CompressionType.LZ4));
+ schemas.add(new MeasurementSchema("c", TSDataType.FLOAT, encoding, CompressionType.LZ4));
+ schemas.add(new MeasurementSchema("empty", TSDataType.INT32, encoding, CompressionType.LZ4));
+ schemas.add(new MeasurementSchema("plain", TSDataType.INT32, TSEncoding.PLAIN));
+ return schemas;
+ }
+
+ private static double a(int r) {
+ return r % 31 == 0 ? -0.0 : ((r * 17) % 37) / 10.0;
+ }
+
+ private static long b(int r) {
+ return r % 11 == 0 ? Long.MAX_VALUE - r : (r * 13) % 41;
+ }
+
+ private static float c(int r) {
+ return (float) (((r * 3) % 19) / 10.0);
+ }
+
+ @Test
+ public void alignedFilePreservesTimestampValuePairsAndProjection() throws Exception {
+ File file = new File(temp.getRoot(), "aligned.tsfile");
+ int old = TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
+ TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(127);
+ try {
+ try (TsFileWriter writer = new TsFileWriter(file)) {
+ writer.registerAlignedTimeseries(new Path("root.native.d"), schemas());
+ for (int r = 0; r < 641; r++) {
+ TSRecord row = new TSRecord("root.native.d", 1000 + r);
+ row.addPoint("a", a(r)).addPoint("b", b(r)).addPoint("plain", r);
+ if (r % 7 != 0) row.addPoint("c", c(r));
+ writer.writeRecord(row);
+ if (r == 300) writer.flush();
+ }
+ }
+ verify(file, 641);
+ } finally {
+ TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(old);
+ }
+ }
+
+ @Test
+ public void columnWiseMemtableWriterPath() throws Exception {
+ File file = new File(temp.getRoot(), "column-wise.tsfile");
+ try (TsFileIOWriter fileWriter = new TsFileIOWriter(file)) {
+ fileWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.native.d"));
+ AlignedChunkWriterImpl writer = new AlignedChunkWriterImpl(schemas());
+ for (int start = 0; start < 641; start += 127) {
+ int end = Math.min(641, start + 127);
+ for (int r = start; r < end; r++) writer.writeByColumn(1000 + r, a(r), false);
+ writer.nextColumn();
+ for (int r = start; r < end; r++) writer.writeByColumn(1000 + r, b(r), false);
+ writer.nextColumn();
+ for (int r = start; r < end; r++) writer.writeByColumn(1000 + r, c(r), r % 7 == 0);
+ writer.nextColumn();
+ for (int r = start; r < end; r++) writer.writeByColumn(1000 + r, 0, true);
+ writer.nextColumn();
+ for (int r = start; r < end; r++) writer.writeByColumn(1000 + r, r, false);
+ writer.nextColumn();
+ long[] times = new long[end - start];
+ for (int r = 0; r < times.length; r++) times[r] = 1000 + start + r;
+ writer.write(times, times.length, 0);
+ writer.sealCurrentPage();
+ }
+ writer.writeToFileWriter(fileWriter);
+ fileWriter.endChunkGroup();
+ fileWriter.endFile();
+ }
+ verify(file, 641);
+ }
+
+ private void verify(File file, int count) throws Exception {
+ try (TsFileSequenceReader sequence = new TsFileSequenceReader(file.getAbsolutePath());
+ TsFileReader reader = new TsFileReader(sequence)) {
+ List<Path> fields =
+ Arrays.asList(
+ new Path("root.native.d", "a", true),
+ new Path("root.native.d", "b", true),
+ new Path("root.native.d", "c", true),
+ new Path("root.native.d", "plain", true));
+ QueryDataSet data = reader.query(QueryExpression.create(fields, null));
+ int r = 0;
+ while (data.hasNext()) {
+ RowRecord row = data.next();
+ assertEquals(1000 + r, row.getTimestamp());
+ assertEquals(
+ Double.doubleToRawLongBits(a(r)),
+ Double.doubleToRawLongBits(row.getFields().get(0).getDoubleV()));
+ assertEquals(b(r), row.getFields().get(1).getLongV());
+ if (r % 7 == 0) assertNull(row.getFields().get(2));
+ else
+ assertEquals(
+ Float.floatToRawIntBits(c(r)),
+ Float.floatToRawIntBits(row.getFields().get(2).getFloatV()));
+ assertEquals(r, row.getFields().get(3).getIntV());
+ r++;
+ }
+ assertEquals(count, r);
+ QueryDataSet projected =
+ reader.query(QueryExpression.create(Collections.singletonList(fields.get(1)), null));
+ r = 0;
+ while (projected.hasNext()) {
+ RowRecord row = projected.next();
+ assertEquals(1000 + r, row.getTimestamp());
+ assertEquals(b(r++), row.getFields().get(0).getLongV());
+ }
+ assertEquals(count, r);
+ }
+ }
+
+ @Test
+ public void scalarSqlPageKeepsActualTimestampKeys() throws Exception {
+ File file = new File(temp.getRoot(), "scalar-native.tsfile");
+ try (TsFileWriter writer = new TsFileWriter(file)) {
+ writer.registerTimeseries(
+ "root.scalar.d", new MeasurementSchema("v", TSDataType.FLOAT, encoding));
+ for (int r = 0; r < 701; r++) {
+ writer.writeRecord(new TSRecord("root.scalar.d", 1000 + r).addPoint("v", c(r)));
+ if (r == 300) writer.flush();
+ }
+ }
+ try (TsFileSequenceReader sequence = new TsFileSequenceReader(file.getAbsolutePath());
+ TsFileReader reader = new TsFileReader(sequence)) {
+ QueryDataSet rows =
+ reader.query(
+ QueryExpression.create(
+ Collections.singletonList(new Path("root.scalar.d", "v", true)), null));
+ int r = 0;
+ while (rows.hasNext()) {
+ RowRecord row = rows.next();
+ assertEquals(1000 + r, row.getTimestamp());
+ assertEquals(
+ Float.floatToRawIntBits(c(r++)),
+ Float.floatToRawIntBits(row.getFields().get(0).getFloatV()));
+ }
+ assertEquals(701, r);
+ }
+ }
+
+ @Test
+ public void jointSelectionAndCorruptNativePage() throws Exception {
+ int n = 301;
+ String[] names = {"x", "y", "nullable"};
+ TSDataType[] types = {TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.INT64};
+ ClusterColumnBuffer[] columns = {
+ new ClusterColumnBuffer(), new ClusterColumnBuffer(), new ClusterColumnBuffer()
+ };
+ long[] times = new long[n];
+ for (int r = 0; r < n; r++) {
+ times[r] = 1000 + r;
+ double x = (r * 23 % 37) / 10.0;
+ columns[0].append(Double.doubleToRawLongBits(x), false);
+ columns[1].append(Double.doubleToRawLongBits(x * 2), false);
+ columns[2].append(Long.MAX_VALUE - r, r % 7 == 0);
+ }
+ ClusterTableOptions options =
+ encoding == TSEncoding.ACLUSTER
+ ? ClusterTableOptions.aCluster()
+ : ClusterTableOptions.kCluster();
+ byte[][] pages =
+ ClusterNativePageCodec.encode(
+ times, names, types, columns, options, new ClusterNativePageCodec.Selection());
+ for (int col = 0; col < 3; col++) {
+ ClusterNativePageCodec.Decoded decoded =
+ ClusterNativePageCodec.decode(ByteBuffer.wrap(pages[col]), types[col]);
+ assertArrayEquals(times, decoded.times);
+ PlainDecoder decoder = new PlainDecoder();
+ for (int r = 0; r < n; r++) {
+ boolean missing = columns[col].isNull(r);
+ assertEquals(!missing, (decoded.bitmap[r / 8] & (0x80 >>> (r % 8))) != 0);
+ if (!missing)
+ assertEquals(
+ columns[col].bits(r),
+ col == 2
+ ? decoder.readLong(decoded.values)
+ : Double.doubleToRawLongBits(decoder.readDouble(decoded.values)));
+ }
+ assertFalse(decoded.values.hasRemaining());
+ }
+ byte[] bad = pages[0].clone();
+ bad[bad.length - 1] ^= 1;
+ assertThrows(
+ IOException.class,
+ () -> ClusterNativePageCodec.decode(ByteBuffer.wrap(bad), TSDataType.DOUBLE));
+ assertThrows(
+ IOException.class,
+ () -> ClusterNativePageCodec.decode(ByteBuffer.wrap(pages[0]), TSDataType.INT64));
+ times[1] = times[0];
+ assertThrows(
+ IOException.class,
+ () ->
+ ClusterNativePageCodec.encode(
+ times, names, types, columns, options, new ClusterNativePageCodec.Selection()));
+ }
+}
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterTableCodecTest.java b/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterTableCodecTest.java
new file mode 100644
index 000000000..b9cef50d8
--- /dev/null
+++ b/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterTableCodecTest.java
@@ -0,0 +1,317 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.enums.TSDataType;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+import java.util.zip.CRC32;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class ClusterTableCodecTest {
+ @Rule public TemporaryFolder temporary = new TemporaryFolder();
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<Object[]> methods() {
+ return Arrays.asList(
+ new Object[][] {
+ {ClusterTableOptions.Method.ACLUSTER}, {ClusterTableOptions.Method.KCLUSTER}
+ });
+ }
+
+ private final ClusterTableOptions options;
+
+ public ClusterTableCodecTest(ClusterTableOptions.Method method) {
+ options = new ClusterTableOptions(method, 4, 3, 42);
+ }
+
+ static ClusterTable data(int n) {
+ Object[][] rows = new Object[n][4];
+ long[] times = new long[n];
+ Random random = new Random(19);
+ for (int r = 0; r < n; r++) {
+ int v = (r * 17) % 31;
+ times[r] = 1700000000000L + (r / 2); // duplicates are intentional
+ rows[r][0] = (double) v;
+ rows[r][1] = (double) (v * 2 + 1);
+ rows[r][2] = random.nextGaussian();
+ rows[r][3] = r % 13 == 0 ? null : (double) (r % 7) / 10;
+ }
+ return new ClusterTable(
+ times,
+ new String[] {"x", "y", "noise", "nullable"},
+ new TSDataType[] {
+ TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.DOUBLE, TSDataType.DOUBLE
+ },
+ rows);
+ }
+
+ static Map<String, Integer> records(ClusterTable table) {
+ Map<String, Integer> result = new HashMap<>();
+ addRecords(result, table);
+ return result;
+ }
+
+ static void addRecords(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 jointColumnsAndAlpKeepWholeRecords() throws Exception {
+ ClusterTable input = data(320);
+ ClusterTableCodec.Encoded encoded = ClusterTableCodec.encode(input, options, new int[] {0, 1});
+ assertArrayEquals(new int[] {0, 1}, encoded.clusteredColumns());
+ assertTrue(encoded.referenceCount > 0);
+ assertTrue(encoded.clusterBytes > 0);
+ assertTrue(encoded.alpBytes > 0);
+ ClusterTable output = ClusterTableCodec.decode(encoded.bytes());
+ assertArrayEquals(input.columnNames(), output.columnNames());
+ assertArrayEquals(input.columnTypes(), output.columnTypes());
+ assertEquals(records(input), records(output));
+ assertEquals(
+ records(input.timeRange(1700000000020L, 1700000000040L)),
+ records(output.timeRange(1700000000020L, 1700000000040L)));
+ assertArrayEquals(
+ encoded.bytes(), ClusterTableCodec.encode(input, options, new int[] {0, 1}).bytes());
+ }
+
+ @Test
+ public void automaticSelectorAndSchemaFreeze() throws Exception {
+ ClusterTable input = data(256);
+ assertArrayEquals(new int[] {0, 1}, ClusterColumnSelector.select(input));
+ ClusterTableEncoder stream = new ClusterTableEncoder(options);
+ stream.fit(input);
+ ClusterTable page = input.slice(30, 80);
+ assertArrayEquals(new int[] {0, 1}, stream.encode(page).selectedColumns());
+ ClusterTable other =
+ new ClusterTable(
+ new long[] {0},
+ new String[] {"other"},
+ new TSDataType[] {TSDataType.DOUBLE},
+ new Object[][] {{1.0}});
+ assertThrows(IllegalArgumentException.class, () -> stream.encode(other));
+ }
+
+ @Test
+ public void exactTypesNullsAndUnsafeNumbers() throws Exception {
+ long nan = 0x7ff8000000000042L;
+ Object[][] rows = {
+ {Integer.MIN_VALUE, Long.MIN_VALUE, -0.0f, -0.0},
+ {
+ Integer.MAX_VALUE,
+ Long.MAX_VALUE,
+ Float.intBitsToFloat(0x7fc00042),
+ Double.longBitsToDouble(nan)
+ },
+ {0, 9007199254740993L, Float.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY},
+ {null, null, null, null},
+ {7, -9007199254740993L, Float.MIN_VALUE, Double.MIN_VALUE},
+ {-7, -1L, 1.25f, 1.25}
+ };
+ ClusterTable input =
+ new ClusterTable(
+ new long[] {4, 3, 2, 1, 1, Long.MIN_VALUE},
+ new String[] {"i", "l", "f", "d"},
+ new TSDataType[] {
+ TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.DOUBLE
+ },
+ rows);
+ ClusterTableCodec.Encoded encoded =
+ ClusterTableCodec.encode(input, options, new int[] {0, 1, 2, 3});
+ assertEquals(0, encoded.clusteredColumns().length);
+ assertEquals(records(input), records(ClusterTableCodec.decode(encoded.bytes())));
+ }
+
+ @Test
+ public void longValuesNeverPassThroughDouble() throws Exception {
+ ClusterTable input =
+ new ClusterTable(
+ new long[] {8, 8, 9, 10},
+ new String[] {"large", "float"},
+ new TSDataType[] {TSDataType.INT64, TSDataType.FLOAT},
+ new Object[][] {
+ {Long.MAX_VALUE, 1.2f},
+ {Long.MAX_VALUE - 1, 1.3f},
+ {Long.MAX_VALUE - 2, 1.4f},
+ {Long.MAX_VALUE, 1.2f}
+ });
+ ClusterTableCodec.Encoded encoded = ClusterTableCodec.encode(input, options, new int[] {0, 1});
+ assertEquals(2, encoded.clusteredColumns().length);
+ assertEquals(records(input), records(ClusterTableCodec.decode(encoded.bytes())));
+ }
+
+ @Test
+ public void emptyAndConstantTables() throws Exception {
+ ClusterTable empty = data(0);
+ assertEquals(
+ 0, ClusterTableCodec.decode(ClusterTableCodec.encode(empty, options).bytes()).rowCount());
+ Object[][] rows = new Object[64][2];
+ for (Object[] row : rows) {
+ row[0] = 0.0;
+ row[1] = -12.5;
+ }
+ ClusterTable constant =
+ new ClusterTable(
+ new long[64],
+ new String[] {"a", "b"},
+ new TSDataType[] {TSDataType.DOUBLE, TSDataType.DOUBLE},
+ rows);
+ ClusterTableCodec.Encoded encoded =
+ ClusterTableCodec.encode(constant, options, new int[] {0, 1});
+ assertEquals(1, encoded.referenceCount);
+ assertEquals(records(constant), records(ClusterTableCodec.decode(encoded.bytes())));
+ }
+
+ @Test
+ public void serializedBlockIsIndependentAndRejectsDamage() throws Exception {
+ byte[] original = ClusterTableCodec.encode(data(70), options).bytes();
+ for (int cut : new int[] {0, 1, 8, original.length - 1}) {
+ assertThrows(
+ java.io.IOException.class, () -> ClusterTableCodec.decode(Arrays.copyOf(original, cut)));
+ }
+ byte[] bad = original.clone();
+ bad[bad.length / 2] ^= 1;
+ assertThrows(java.io.IOException.class, () -> ClusterTableCodec.decode(bad));
+ byte[] shape = original.clone();
+ ByteBuffer.wrap(shape).putInt(6, Integer.MAX_VALUE);
+ CRC32 crc = new CRC32();
+ crc.update(shape, 0, shape.length - 4);
+ ByteBuffer.wrap(shape).putInt(shape.length - 4, (int) crc.getValue());
+ assertThrows(java.io.IOException.class, () -> ClusterTableCodec.decode(shape));
+ assertEquals(records(data(70)), records(ClusterTableCodec.decode(original)));
+ }
+
+ @Test
+ public void badInputRejectedWithoutSilentConversion() throws Exception {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new ClusterTable(
+ new long[] {0},
+ new String[] {"v"},
+ new TSDataType[] {TSDataType.INT64},
+ new Object[][] {{9007199254740993.0}}));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> ClusterTableCodec.encode(data(2), options, new int[] {0, 0}));
+ assertThrows(
+ IllegalArgumentException.class, () -> ClusterTableCodec.encode(data(10001), options));
+ }
+
+ @Test
+ public void alpCompressedAndRawVectorsAreExact() throws Exception {
+ long[] raw = new long[2051];
+ for (int i = 0; i < raw.length; i++) raw[i] = Double.doubleToRawLongBits((i % 31) / 10.0);
+ raw[12] = Double.doubleToRawLongBits(-0.0);
+ raw[19] = 0x7ff800000000007aL;
+ byte[] encoded = AlpColumnCodec.encode(raw, TSDataType.DOUBLE);
+ assertEquals(1, encoded[0]);
+ assertTrue(encoded.length < raw.length * 8);
+ assertArrayEquals(
+ raw,
+ AlpColumnCodec.decode(
+ new DataInputStream(new ByteArrayInputStream(encoded)), raw.length, TSDataType.DOUBLE));
+ long[] integers = {Long.MIN_VALUE, Long.MAX_VALUE, 9007199254740993L};
+ byte[] fallback = AlpColumnCodec.encode(integers, TSDataType.INT64);
+ assertEquals(0, fallback[0]);
+ assertArrayEquals(
+ integers,
+ AlpColumnCodec.decode(
+ new DataInputStream(new ByteArrayInputStream(fallback)),
+ integers.length,
+ TSDataType.INT64));
+ }
+
+ @Test
+ public void actualTsFileCloseReopenAndMultiplePages() throws Exception {
+ ClusterTable input = data(10007);
+ File file = new File(temporary.getRoot(), "records.tsfile");
+ try (ClusterTableTsFile.Writer writer = new ClusterTableTsFile.Writer(file, options)) {
+ writer.append(input);
+ writer.flush();
+ writer.append(input.slice(0, 9));
+ }
+ Map<String, Integer> expected = records(input);
+ addRecords(expected, input.slice(0, 9));
+ Map<String, Integer> actual = new HashMap<>();
+ int[] pages = {0};
+ ClusterTableTsFile.read(
+ file,
+ page -> {
+ pages[0]++;
+ addRecords(actual, page);
+ });
+ assertEquals(3, pages[0]);
+ assertEquals(expected, actual);
+ assertThrows(java.io.IOException.class, () -> new ClusterTableTsFile.Writer(file, options));
+ }
+
+ @Test
+ public void randomizedRecordsAndAlpCompanions() throws Exception {
+ Random random = new Random(38);
+ for (int trial = 0; trial < 12; trial++) {
+ int n = 1 + random.nextInt(100);
+ Object[][] rows = new Object[n][3];
+ long[] times = new long[n];
+ for (int r = 0; r < n; r++) {
+ times[r] = random.nextLong();
+ rows[r][0] = random.nextInt(31);
+ rows[r][1] = random.nextLong();
+ rows[r][2] = Double.longBitsToDouble(random.nextLong());
+ }
+ ClusterTable table =
+ new ClusterTable(
+ times,
+ new String[] {"i", "l", "d"},
+ new TSDataType[] {TSDataType.INT32, TSDataType.INT64, TSDataType.DOUBLE},
+ rows);
+ assertEquals(
+ records(table),
+ records(
+ ClusterTableCodec.decode(
+ ClusterTableCodec.encode(table, options, new int[] {0}).bytes())));
+ }
+ }
+}
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterTableDataCheck.java b/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterTableDataCheck.java
new file mode 100644
index 000000000..6c4b2ed75
--- /dev/null
+++ b/java/tsfile/src/test/java/org/apache/tsfile/encoding/table/ClusterTableDataCheck.java
@@ -0,0 +1,116 @@
+/*
+ * 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.tsfile.encoding.table;
+
+import org.apache.tsfile.enums.TSDataType;
+
+import java.io.BufferedReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/** Optional bounded real-data check; no benchmarks or modifications to the C++ repository. */
+public final class ClusterTableDataCheck {
+ private ClusterTableDataCheck() {}
+
+ public static void main(String[] args) throws Exception {
+ Path workspace = Paths.get(args[0]);
+ List<String> manifest =
+ Files.readAllLines(
+ workspace.resolve("data/selected_columns_manifest.csv"), StandardCharsets.UTF_8);
+ System.out.println(
+ "dataset,method,selection_rows,tested_rows,selected_dim,clustered_dim,references,bytes,status");
+ for (String dataset : Arrays.asList("CT", "Crop", "Gas", "BT", "Musk")) {
+ String entry =
+ manifest.stream().filter(line -> line.startsWith(dataset + ",")).findFirst().get();
+ String[] fields = entry.split(",");
+ int dimensions = Integer.parseInt(fields[7]);
+ int[] expected = Arrays.stream(fields[9].split(";")).mapToInt(Integer::parseInt).toArray();
+ List<Object[]> rows = new ArrayList<>();
+ try (BufferedReader reader =
+ Files.newBufferedReader(
+ workspace.resolve("data/all/" + dataset + ".csv"), StandardCharsets.UTF_8)) {
+ String line;
+ while (rows.size() < 10000 && (line = reader.readLine()) != null) {
+ String[] tokens = line.split(",");
+ if (tokens.length != dimensions) throw new AssertionError("Unexpected CSV width");
+ Object[] row = new Object[dimensions];
+ for (int c = 0; c < dimensions; c++) row[c] = Double.parseDouble(tokens[c]);
+ rows.add(row);
+ }
+ }
+ long[] times = new long[rows.size()];
+ for (int r = 0; r < times.length; r++) times[r] = r;
+ String[] names = new String[dimensions];
+ TSDataType[] types = new TSDataType[dimensions];
+ for (int c = 0; c < dimensions; c++) {
+ names[c] = "c" + c;
+ types[c] = TSDataType.DOUBLE;
+ }
+ ClusterTable sample = new ClusterTable(times, names, types, rows.toArray(new Object[0][]));
+ int[] selected = ClusterColumnSelector.select(sample);
+ if (!Arrays.equals(expected, selected)) {
+ throw new AssertionError(dataset + " selector differs: " + Arrays.toString(selected));
+ }
+ for (ClusterTableOptions.Method method : ClusterTableOptions.Method.values()) {
+ ClusterTableEncoder encoder =
+ new ClusterTableEncoder(new ClusterTableOptions(method, 100, 10, 0));
+ encoder.fit(sample);
+ long bytes = 0, refs = 0;
+ int tested = 0, clustered = Integer.MAX_VALUE;
+ for (int from : new int[] {0, 256, rows.size() - 127}) {
+ int to = Math.min(rows.size(), from + 256);
+ ClusterTable page = sample.slice(from, to);
+ ClusterTableCodec.Encoded result = encoder.encode(page);
+ ClusterTable restored = ClusterTableCodec.decode(result.bytes());
+ if (!ClusterTableCodecTest.records(page)
+ .equals(ClusterTableCodecTest.records(restored))) {
+ throw new AssertionError(dataset + " complete-record round trip failed");
+ }
+ clustered = Math.min(clustered, result.clusteredColumns().length);
+ bytes += result.bytes().length;
+ refs += result.referenceCount;
+ tested += page.rowCount();
+ }
+ System.out.println(
+ dataset
+ + ","
+ + method
+ + ","
+ + rows.size()
+ + ","
+ + tested
+ + ","
+ + selected.length
+ + ","
+ + clustered
+ + ","
+ + refs
+ + ","
+ + bytes
+ + ",PASS");
+ }
+ }
+ }
+}