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:53 UTC
(tsfile) branch ty/fix-vector-schema-stream-deserialization created (now 614941278)
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a change to branch ty/fix-vector-schema-stream-deserialization
in repository https://gitbox.apache.org/repos/asf/tsfile.git
at 614941278 Fix VectorMeasurementSchema stream deserialization
This branch includes the following new commits:
new 614941278 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: 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
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 6149412785fa9ffa118cd079ab74adabcf4301ea
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Fri Sep 4 15:39:28 2026 +0800
Fix VectorMeasurementSchema stream deserialization
---
.../write/schema/VectorMeasurementSchema.java | 10 +-
.../writer/MeasurementSchemaSerializeTest.java | 107 +++++++++++++++++++++
2 files changed, 114 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 36bed75e7..6cee451e9 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
@@ -30,6 +30,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;
@@ -420,9 +422,11 @@ 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(Messages.get("error.write.schema_vector_unexpected_end_of_stream"));
+ try {
+ new DataInputStream(inputStream).readFully(compressors);
+ } catch (EOFException e) {
+ throw new IOException(
+ Messages.get("error.write.schema_vector_unexpected_end_of_stream"), 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..97167803b 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.getMeasurementName(), actual.getMeasurementName());
+ 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));
+ }
+ }
}