You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by ja...@apache.org on 2026/09/04 07:39:56 UTC
(tsfile) branch ty/fix-vector-schema-stream-deserialization-1.1 created (now fc04c3fbb)
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a change to branch ty/fix-vector-schema-stream-deserialization-1.1
in repository https://gitbox.apache.org/repos/asf/tsfile.git
at fc04c3fbb [To dev/1.1] Fix VectorMeasurementSchema stream deserialization
This branch includes the following new commits:
new fc04c3fbb [To dev/1.1] Fix VectorMeasurementSchema stream deserialization
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tsfile) 01/01: [To dev/1.1] Fix VectorMeasurementSchema stream deserialization
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch ty/fix-vector-schema-stream-deserialization-1.1
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit fc04c3fbb18df9b0e05a1f22f1b37f5e9499e9b8
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Fri Sep 4 15:39:28 2026 +0800
[To dev/1.1] Fix VectorMeasurementSchema stream deserialization
---
.../write/schema/VectorMeasurementSchema.java | 9 +-
.../writer/MeasurementSchemaSerializeTest.java | 107 +++++++++++++++++++++
2 files changed, 113 insertions(+), 3 deletions(-)
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/schema/VectorMeasurementSchema.java b/java/tsfile/src/main/java/org/apache/tsfile/write/schema/VectorMeasurementSchema.java
index 33c21479c..1ee206c84 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/schema/VectorMeasurementSchema.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/schema/VectorMeasurementSchema.java
@@ -29,6 +29,8 @@ import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.apache.tsfile.utils.StringContainer;
+import java.io.DataInputStream;
+import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -415,9 +417,10 @@ public class VectorMeasurementSchema
vectorMeasurementSchema.unifiedCompressor = ReadWriteIOUtils.readByte(inputStream);
if (vectorMeasurementSchema.unifiedCompressor == NO_UNIFIED_COMPRESSOR) {
byte[] compressors = new byte[measurementSize + 1];
- int read = inputStream.read(compressors);
- if (read != measurementSize) {
- throw new IOException("Unexpected end of stream when reading compressors");
+ try {
+ new DataInputStream(inputStream).readFully(compressors);
+ } catch (EOFException e) {
+ throw new IOException("Unexpected end of stream when reading compressors", e);
}
vectorMeasurementSchema.compressors = compressors;
}
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/MeasurementSchemaSerializeTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/MeasurementSchemaSerializeTest.java
index e031b0d49..9a06e733b 100644
--- a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/MeasurementSchemaSerializeTest.java
+++ b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/MeasurementSchemaSerializeTest.java
@@ -19,17 +19,23 @@
package org.apache.tsfile.write.writer;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.apache.tsfile.write.schema.VectorMeasurementSchema;
import org.junit.Test;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
public class MeasurementSchemaSerializeTest {
@@ -54,4 +60,105 @@ public class MeasurementSchemaSerializeTest {
MeasurementSchema measurementSchema = MeasurementSchema.deserializeFrom(inputStream);
assertEquals(standard, measurementSchema);
}
+
+ @Test
+ public void deserializeVectorWithUnifiedCompressorFromInputStreamTest() throws IOException {
+ VectorMeasurementSchema standard = createVectorSchema(CompressionType.LZ4);
+ ByteArrayInputStream inputStream = serializeToInputStream(standard);
+
+ VectorMeasurementSchema measurementSchema =
+ VectorMeasurementSchema.deserializeFrom(inputStream);
+
+ assertVectorSchemaEquals(standard, measurementSchema);
+ assertEquals(-1, inputStream.read());
+ }
+
+ @Test
+ public void deserializeVectorWithPerColumnCompressorsFromInputStreamTest() throws IOException {
+ VectorMeasurementSchema standard = createVectorSchemaWithPerColumnCompressors();
+ ByteArrayInputStream inputStream = serializeToInputStream(standard);
+
+ VectorMeasurementSchema measurementSchema =
+ VectorMeasurementSchema.deserializeFrom(inputStream);
+
+ assertVectorSchemaEquals(standard, measurementSchema);
+ assertEquals(-1, inputStream.read());
+ }
+
+ @Test
+ public void deserializeVectorWithPerColumnCompressorsFromShortReadInputStreamTest()
+ throws IOException {
+ VectorMeasurementSchema standard = createVectorSchemaWithPerColumnCompressors();
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ standard.serializeTo(outputStream);
+ byte[] serialized = outputStream.toByteArray();
+ int compressorListStart = serialized.length - standard.getSubMeasurementsCount() - 1;
+ int shortReadBoundary = compressorListStart + 1;
+ InputStream inputStream =
+ new SequenceInputStream(
+ new ByteArrayInputStream(Arrays.copyOf(serialized, shortReadBoundary)),
+ new ByteArrayInputStream(
+ Arrays.copyOfRange(serialized, shortReadBoundary, serialized.length)));
+
+ VectorMeasurementSchema measurementSchema =
+ VectorMeasurementSchema.deserializeFrom(inputStream);
+
+ assertVectorSchemaEquals(standard, measurementSchema);
+ assertEquals(-1, inputStream.read());
+ }
+
+ @Test
+ public void deserializeVectorRejectsTruncatedPerColumnCompressors() throws IOException {
+ VectorMeasurementSchema standard = createVectorSchemaWithPerColumnCompressors();
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ standard.serializeTo(outputStream);
+ byte[] serialized = outputStream.toByteArray();
+ ByteArrayInputStream inputStream =
+ new ByteArrayInputStream(Arrays.copyOf(serialized, serialized.length - 1));
+
+ assertThrows(IOException.class, () -> VectorMeasurementSchema.deserializeFrom(inputStream));
+ }
+
+ private static VectorMeasurementSchema createVectorSchema(CompressionType compressionType) {
+ return new VectorMeasurementSchema(
+ "device",
+ new String[] {"sensor_1", "sensor_2"},
+ new TSDataType[] {TSDataType.FLOAT, TSDataType.INT64},
+ new TSEncoding[] {TSEncoding.RLE, TSEncoding.TS_2DIFF},
+ compressionType);
+ }
+
+ private static VectorMeasurementSchema createVectorSchemaWithPerColumnCompressors() {
+ return new VectorMeasurementSchema(
+ "device",
+ new String[] {"sensor_1", "sensor_2"},
+ new TSDataType[] {TSDataType.FLOAT, TSDataType.INT64},
+ new TSEncoding[] {TSEncoding.RLE, TSEncoding.TS_2DIFF},
+ new byte[] {
+ CompressionType.LZ4.serialize(),
+ CompressionType.SNAPPY.serialize(),
+ CompressionType.GZIP.serialize()
+ });
+ }
+
+ private static ByteArrayInputStream serializeToInputStream(VectorMeasurementSchema schema)
+ throws IOException {
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ schema.serializeTo(outputStream);
+ return new ByteArrayInputStream(outputStream.toByteArray());
+ }
+
+ private static void assertVectorSchemaEquals(
+ VectorMeasurementSchema expected, VectorMeasurementSchema actual) {
+ assertEquals(expected.getMeasurementId(), actual.getMeasurementId());
+ assertEquals(expected.getSubMeasurementsList(), actual.getSubMeasurementsList());
+ assertEquals(
+ expected.getSubMeasurementsTSDataTypeList(), actual.getSubMeasurementsTSDataTypeList());
+ assertEquals(
+ expected.getSubMeasurementsTSEncodingList(), actual.getSubMeasurementsTSEncodingList());
+ assertEquals(expected.getTimeCompressor(), actual.getTimeCompressor());
+ for (int i = 0; i < expected.getSubMeasurementsCount(); i++) {
+ assertEquals(expected.getValueCompressor(i), actual.getValueCompressor(i));
+ }
+ }
}