> clientFiles = MetaUtil.getClientFiles();
+ assertEquals(1, clientFiles.size());
+ assertEquals(1, clientFiles.get(0).size());
+ assertTrue(clientFiles.get(0).get(0).endsWith("data_0.tsfile"));
+
+ DataReader reader = DataReader.getInstance(clientFiles.get(0));
+ assertTrue(reader instanceof TsFileDataReader);
+ int rows = 0;
+ while (reader.hasNextBatch()) {
+ IBatch b = reader.nextBatch();
+ rows += b.getRecords().size();
+ }
+ assertEquals(4, rows);
+ }
+
+ /**
+ * Regression: when client count exceeds device count, every schema/data client id must still get
+ * a (possibly empty) device list, not null — otherwise SchemaClient NPEs during registration.
+ */
+ @Test
+ public void everyClientGetsNonNullBucketWhenClientCountExceedsDevices() throws Exception {
+ // fixture has 2 devices (d_0, d_1); use 20 clients so ids 2..19 receive no device
+ File data = new File(folder.getRoot(), "data_0.tsfile");
+ TsFileTestFixtures.writeSampleTable(data);
+ config.setREAL_DATASET_FORMAT(RealDatasetFormat.TSFILE);
+ config.setFILE_PATH(folder.getRoot().getAbsolutePath());
+ config.setBENCHMARK_WORK_MODE(BenchmarkMode.VERIFICATION_WRITE);
+ config.setDATA_CLIENT_NUMBER(20);
+ config.setSCHEMA_CLIENT_NUMBER(20);
+ Files.write(
+ Paths.get(folder.getRoot().getAbsolutePath(), Constants.INFO_PATH),
+ config.toInfoText().getBytes(StandardCharsets.UTF_8));
+
+ RealMetaDataSchema schema = new RealMetaDataSchema();
+ assertTrue(schema.createMetaDataSchema());
+
+ for (int id = 0; id < 20; id++) {
+ assertNotNull("schema client " + id, schema.getDeviceSchemaBySchemaClientId(id));
+ assertNotNull("data client " + id, schema.getDeviceSchemaByDataClientId(id));
+ }
+ }
+}
diff --git a/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/RealDatasetTsFileEndToEndTest.java b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/RealDatasetTsFileEndToEndTest.java
new file mode 100644
index 000000000..2007e3d5e
--- /dev/null
+++ b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/RealDatasetTsFileEndToEndTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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 cn.edu.tsinghua.iot.benchmark.source;
+
+import cn.edu.tsinghua.iot.benchmark.BenchmarkTestBase;
+import cn.edu.tsinghua.iot.benchmark.conf.Config;
+import cn.edu.tsinghua.iot.benchmark.conf.ConfigDescriptor;
+import cn.edu.tsinghua.iot.benchmark.conf.RealDatasetFormat;
+import cn.edu.tsinghua.iot.benchmark.entity.Batch.Batch;
+import cn.edu.tsinghua.iot.benchmark.entity.Batch.IBatch;
+import cn.edu.tsinghua.iot.benchmark.entity.Record;
+import cn.edu.tsinghua.iot.benchmark.entity.Sensor;
+import cn.edu.tsinghua.iot.benchmark.entity.enums.SensorType;
+import cn.edu.tsinghua.iot.benchmark.extern.TsFileDataWriter;
+import cn.edu.tsinghua.iot.benchmark.extern.TsFileSchemaWriter;
+import cn.edu.tsinghua.iot.benchmark.mode.enums.BenchmarkMode;
+import cn.edu.tsinghua.iot.benchmark.schema.MetaUtil;
+import cn.edu.tsinghua.iot.benchmark.schema.schemaImpl.DeviceSchema;
+import cn.edu.tsinghua.iot.benchmark.schema.schemaImpl.RealMetaDataSchema;
+import org.junit.After;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * End-to-end: drive the production TsFile real-dataset generate path ({@link
+ * TsFileSchemaWriter} + {@link TsFileDataWriter}) to produce a small dataset, then read it back
+ * through the production verification path ({@link RealMetaDataSchema} + {@link
+ * TsFileDataReader}) and assert the values round-trip.
+ *
+ * The write phase runs under {@link BenchmarkMode#GENERATE_DATA}. This matters because
+ * constructing any {@code SchemaWriter} eagerly forces {@code MetaDataSchema.getInstance()} (a
+ * once-per-JVM singleton). Under {@code GENERATE_DATA} that resolves to {@code
+ * GenerateMetaDataSchema}, which synthesizes device schemas from config; under a verification mode
+ * it would instead build a {@code RealMetaDataSchema} whose constructor demands an
+ * already-generated dataset and calls {@code System.exit} when one is absent — exactly the
+ * generate-then-read ordering this test exercises. The read-back below does not rely on that
+ * singleton: it instantiates {@code RealMetaDataSchema} directly.
+ */
+public class RealDatasetTsFileEndToEndTest extends BenchmarkTestBase {
+
+ private static final Config config = ConfigDescriptor.getInstance().getConfig();
+
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ /**
+ * {@code config} is a process-wide singleton, and {@code DataReader.getInstance} short-circuits
+ * to {@link TsFileDataReader} whenever {@code REAL_DATASET_FORMAT == TSFILE}. Restore the default
+ * so a later CSV/copy test in the same fork still gets its expected reader.
+ */
+ @After
+ public void restoreDefaultFormat() {
+ config.setREAL_DATASET_FORMAT(RealDatasetFormat.CSV);
+ }
+
+ @Test
+ public void generatedTsFileReadsBackIdentically() throws Exception {
+ config.setREAL_DATASET_FORMAT(RealDatasetFormat.TSFILE);
+ config.setFILE_PATH(folder.getRoot().getAbsolutePath());
+ config.setBENCHMARK_WORK_MODE(BenchmarkMode.GENERATE_DATA);
+ config.setDATA_CLIENT_NUMBER(1);
+ config.setSCHEMA_CLIENT_NUMBER(1);
+ config.setBATCH_SIZE_PER_WRITE(100);
+
+ List sensors =
+ Arrays.asList(new Sensor("s_0", SensorType.INT32), new Sensor("s_1", SensorType.DOUBLE));
+ DeviceSchema d0 = new DeviceSchema("g_0", "t_0", "d_0", sensors, new HashMap<>());
+
+ // 1) write dataset metadata (clears dir + info.txt) then data
+ new TsFileSchemaWriter().writeSchema(Collections.singletonList(d0));
+ TsFileDataWriter writer = new TsFileDataWriter(0);
+ List recs = new ArrayList<>();
+ recs.add(new Record(10L, Arrays.asList((Object) 1, 0.1)));
+ recs.add(new Record(20L, Arrays.asList((Object) 2, 0.2)));
+ assertTrue(writer.writeBatch(new Batch(d0, recs), 0L));
+ writer.close();
+
+ // 2) read back through the production schema + reader path.
+ // createMetaDataSchema() is protected in RealMetaDataSchema (a different package), so it is
+ // invoked via reflection — same approach as RealMetaDataSchemaTest.
+ RealMetaDataSchema schema = new RealMetaDataSchema();
+ Method createMetaDataSchema =
+ RealMetaDataSchema.class.getDeclaredMethod("createMetaDataSchema");
+ createMetaDataSchema.setAccessible(true);
+ assertTrue((boolean) createMetaDataSchema.invoke(schema));
+ List> clientFiles = MetaUtil.getClientFiles();
+
+ List readBack = new ArrayList<>();
+ TsFileDataReader reader = new TsFileDataReader(clientFiles.get(0));
+ while (reader.hasNextBatch()) {
+ IBatch b = reader.nextBatch();
+ assertEquals("d_0", b.getDeviceSchema().getDevice());
+ readBack.addAll(b.getRecords());
+ }
+
+ assertEquals(2, readBack.size());
+ assertEquals(10L, readBack.get(0).getTimestamp());
+ assertEquals(1, readBack.get(0).getRecordDataValue().get(0));
+ assertEquals(0.1, (double) readBack.get(0).getRecordDataValue().get(1), 1e-9);
+ assertEquals(20L, readBack.get(1).getTimestamp());
+ assertEquals(2, readBack.get(1).getRecordDataValue().get(0));
+ }
+}
diff --git a/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/SchemaReaderCheckDataSetTest.java b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/SchemaReaderCheckDataSetTest.java
new file mode 100644
index 000000000..f715cdb7f
--- /dev/null
+++ b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/SchemaReaderCheckDataSetTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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 cn.edu.tsinghua.iot.benchmark.source;
+
+import cn.edu.tsinghua.iot.benchmark.BenchmarkTestBase;
+import cn.edu.tsinghua.iot.benchmark.conf.Config;
+import cn.edu.tsinghua.iot.benchmark.conf.ConfigDescriptor;
+import cn.edu.tsinghua.iot.benchmark.conf.Constants;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SchemaReaderCheckDataSetTest extends BenchmarkTestBase {
+
+ private static final Config config = ConfigDescriptor.getInstance().getConfig();
+
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ /**
+ * A shorter info.txt (e.g. an older dataset with fewer config lines) must return false, not
+ * throw.
+ */
+ @Test
+ public void shortInfoTxtReturnsFalseInsteadOfThrowing() throws Exception {
+ config.setFILE_PATH(folder.getRoot().getAbsolutePath());
+ Files.write(
+ Paths.get(folder.getRoot().getAbsolutePath(), Constants.INFO_PATH),
+ "LOOP=1".getBytes(StandardCharsets.UTF_8));
+
+ assertFalse(new CSVSchemaReader().checkDataSet());
+ }
+
+ @Test
+ public void matchingInfoTxtReturnsTrue() throws Exception {
+ config.setFILE_PATH(folder.getRoot().getAbsolutePath());
+ Files.write(
+ Paths.get(folder.getRoot().getAbsolutePath(), Constants.INFO_PATH),
+ config.toInfoText().getBytes(StandardCharsets.UTF_8));
+
+ assertTrue(new CSVSchemaReader().checkDataSet());
+ }
+}
diff --git a/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileDataReaderTest.java b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileDataReaderTest.java
new file mode 100644
index 000000000..f66e0d4ac
--- /dev/null
+++ b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileDataReaderTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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 cn.edu.tsinghua.iot.benchmark.source;
+
+import cn.edu.tsinghua.iot.benchmark.BenchmarkTestBase;
+import cn.edu.tsinghua.iot.benchmark.conf.Config;
+import cn.edu.tsinghua.iot.benchmark.conf.ConfigDescriptor;
+import cn.edu.tsinghua.iot.benchmark.entity.Batch.IBatch;
+import cn.edu.tsinghua.iot.benchmark.entity.Record;
+import cn.edu.tsinghua.iot.benchmark.schema.schemaImpl.DeviceSchema;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TsFileDataReaderTest extends BenchmarkTestBase {
+
+ private static final Config config = ConfigDescriptor.getInstance().getConfig();
+
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ @Test
+ public void streamsBatchesGroupedByDevice() throws Exception {
+ File data = new File(folder.getRoot(), "data_0.tsfile");
+ TsFileTestFixtures.writeSampleTable(data);
+ config.setFILE_PATH(folder.getRoot().getAbsolutePath());
+ config.setBATCH_SIZE_PER_WRITE(100);
+ // No MetaDataSchema registration needed: the reader derives sensors from the TsFile itself.
+
+ TsFileDataReader reader =
+ new TsFileDataReader(Collections.singletonList(data.getAbsolutePath()));
+
+ Map> timesByDevice = new HashMap<>();
+ int batchCount = 0;
+ while (reader.hasNextBatch()) {
+ IBatch batch = reader.nextBatch();
+ batchCount++;
+ DeviceSchema ds = batch.getDeviceSchema();
+ List times = timesByDevice.computeIfAbsent(ds.getDevice(), k -> new ArrayList<>());
+ for (Record r : batch.getRecords()) {
+ times.add(r.getTimestamp());
+ assertEquals(3, r.getRecordDataValue().size());
+ }
+ }
+ assertFalse(reader.hasNextBatch());
+ assertEquals(2, timesByDevice.size());
+ assertEquals(java.util.Arrays.asList(50L, 60L), timesByDevice.get("d_0"));
+ assertEquals(java.util.Arrays.asList(100L, 200L), timesByDevice.get("d_1"));
+ assertTrue(batchCount >= 2);
+ }
+}
diff --git a/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileRoundTripRawTest.java b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileRoundTripRawTest.java
new file mode 100644
index 000000000..1cf3eb85d
--- /dev/null
+++ b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileRoundTripRawTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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 cn.edu.tsinghua.iot.benchmark.source;
+
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.DeviceTableModelReader;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TsFileRoundTripRawTest {
+
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ @Test
+ public void writesAndReadsBackGroupedByDevice() throws Exception {
+ File f = new File(folder.getRoot(), "sample.tsfile");
+ TsFileTestFixtures.writeSampleTable(f);
+ assertTrue(f.length() > 0);
+
+ try (DeviceTableModelReader reader = new DeviceTableModelReader(f)) {
+ List tables = reader.getAllTableSchema();
+ assertEquals(1, tables.size());
+ TableSchema ts = tables.get(0);
+ assertEquals("t_0", ts.getTableName());
+ assertEquals(1, ts.getTagColumnCnt());
+ assertEquals("device_id", ts.getColumnSchemas().get(0).getMeasurementName());
+ assertEquals(ColumnCategory.TAG, ts.getColumnTypes().get(0));
+
+ ResultSet rs =
+ reader.query(
+ ts.getTableName(),
+ Arrays.asList("device_id", "s_0", "s_1", "s_2"),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ List seen = new ArrayList<>();
+ List times = new ArrayList<>();
+ while (rs.next()) {
+ times.add(rs.getLong(1)); // column 1 == Time
+ seen.add(rs.getString("device_id"));
+ }
+ rs.close();
+ // grouped by device, time-ascending within device
+ assertEquals(Arrays.asList("d_0", "d_0", "d_1", "d_1"), seen);
+ assertEquals(Arrays.asList(50L, 60L, 100L, 200L), times);
+ }
+ }
+}
diff --git a/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileSchemaReaderTest.java b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileSchemaReaderTest.java
new file mode 100644
index 000000000..c9ffff9ef
--- /dev/null
+++ b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileSchemaReaderTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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 cn.edu.tsinghua.iot.benchmark.source;
+
+import cn.edu.tsinghua.iot.benchmark.BenchmarkTestBase;
+import cn.edu.tsinghua.iot.benchmark.conf.Config;
+import cn.edu.tsinghua.iot.benchmark.conf.ConfigDescriptor;
+import cn.edu.tsinghua.iot.benchmark.entity.Sensor;
+import cn.edu.tsinghua.iot.benchmark.entity.enums.SensorType;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TsFileSchemaReaderTest extends BenchmarkTestBase {
+
+ private static final Config config = ConfigDescriptor.getInstance().getConfig();
+
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ @Test
+ public void enumeratesDevicesAndFieldSensors() throws Exception {
+ File data = new File(folder.getRoot(), "data_0.tsfile");
+ TsFileTestFixtures.writeSampleTable(data);
+ config.setFILE_PATH(folder.getRoot().getAbsolutePath());
+
+ Map> result = new TsFileSchemaReader().getDeviceSchemaList();
+
+ assertEquals(2, result.size());
+ assertTrue(result.containsKey("d_0"));
+ assertTrue(result.containsKey("d_1"));
+ List sensors = result.get("d_0");
+ assertEquals(3, sensors.size());
+ assertEquals("s_0", sensors.get(0).getName());
+ assertEquals(SensorType.INT32, sensors.get(0).getSensorType());
+ assertEquals(SensorType.DOUBLE, sensors.get(1).getSensorType());
+ assertEquals(SensorType.BOOLEAN, sensors.get(2).getSensorType());
+ }
+
+ /** A table with no TAG/device_id column must be skipped (no devices), not crash with an NPE. */
+ @Test
+ public void tableWithoutDeviceIdYieldsNoDevicesWithoutNpe() throws Exception {
+ File data = new File(folder.getRoot(), "notag.tsfile");
+ TsFileTestFixtures.writeTableWithoutDeviceId(data);
+ config.setFILE_PATH(folder.getRoot().getAbsolutePath());
+
+ Map> result = new TsFileSchemaReader().getDeviceSchemaList();
+
+ assertTrue(result.isEmpty());
+ }
+}
diff --git a/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileTableModelMappingTest.java b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileTableModelMappingTest.java
new file mode 100644
index 000000000..c820fd33a
--- /dev/null
+++ b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileTableModelMappingTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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 cn.edu.tsinghua.iot.benchmark.source;
+
+import cn.edu.tsinghua.iot.benchmark.entity.enums.SensorType;
+import org.apache.tsfile.enums.TSDataType;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class TsFileTableModelMappingTest {
+
+ @Test
+ public void mapsTsDataTypeToSensorTypeByName() {
+ assertEquals(SensorType.INT32, TsFileTableModelMapping.toSensorType(TSDataType.INT32));
+ assertEquals(SensorType.DOUBLE, TsFileTableModelMapping.toSensorType(TSDataType.DOUBLE));
+ assertEquals(SensorType.BOOLEAN, TsFileTableModelMapping.toSensorType(TSDataType.BOOLEAN));
+ assertEquals(SensorType.STRING, TsFileTableModelMapping.toSensorType(TSDataType.STRING));
+ assertEquals(SensorType.TIMESTAMP, TsFileTableModelMapping.toSensorType(TSDataType.TIMESTAMP));
+ assertEquals(SensorType.DATE, TsFileTableModelMapping.toSensorType(TSDataType.DATE));
+ }
+
+ @Test
+ public void mapsSensorTypeToTsDataTypeByName() {
+ assertEquals(TSDataType.INT64, TsFileTableModelMapping.toTsDataType(SensorType.INT64));
+ assertEquals(TSDataType.FLOAT, TsFileTableModelMapping.toTsDataType(SensorType.FLOAT));
+ assertEquals(TSDataType.TEXT, TsFileTableModelMapping.toTsDataType(SensorType.TEXT));
+ }
+
+ @Test
+ public void unknownTsDataTypeFallsBackToText() {
+ assertEquals(SensorType.TEXT, TsFileTableModelMapping.toSensorType(TSDataType.VECTOR));
+ }
+}
diff --git a/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileTestFixtures.java b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileTestFixtures.java
new file mode 100644
index 000000000..d6496fbe4
--- /dev/null
+++ b/core/src/test/java/cn/edu/tsinghua/iot/benchmark/source/TsFileTestFixtures.java
@@ -0,0 +1,107 @@
+/*
+ * 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 cn.edu.tsinghua.iot.benchmark.source;
+
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.apache.tsfile.write.v4.DeviceTableModelWriter;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/** Test-only helper: writes small table-model TsFiles for reader tests. */
+public class TsFileTestFixtures {
+
+ public static final List FIELD_NAMES = Arrays.asList("s_0", "s_1", "s_2");
+ public static final List FIELD_TYPES =
+ Arrays.asList(TSDataType.INT32, TSDataType.DOUBLE, TSDataType.BOOLEAN);
+
+ /**
+ * Writes table t_0 with device_id TAG + 3 FIELD columns, 2 devices x 2 rows, into {@code file}.
+ */
+ public static void writeSampleTable(File file) throws Exception {
+ List cols = new ArrayList<>();
+ cols.add(new MeasurementSchema("device_id", TSDataType.STRING));
+ for (int i = 0; i < FIELD_NAMES.size(); i++) {
+ cols.add(new MeasurementSchema(FIELD_NAMES.get(i), FIELD_TYPES.get(i)));
+ }
+ List cats =
+ Arrays.asList(
+ ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD, ColumnCategory.FIELD);
+ TableSchema schema = new TableSchema("t_0", cols, cats);
+
+ List names = Arrays.asList("device_id", "s_0", "s_1", "s_2");
+ List types =
+ Arrays.asList(TSDataType.STRING, TSDataType.INT32, TSDataType.DOUBLE, TSDataType.BOOLEAN);
+
+ try (DeviceTableModelWriter w = new DeviceTableModelWriter(file, schema, 10 * 1024 * 1024L)) {
+ Tablet t = new Tablet("t_0", names, types, cats, 4);
+ // interleave d_1 before d_0 to prove the reader regroups by device
+ addRow(t, 0, 100L, "d_1", 11, 1.1, true);
+ addRow(t, 1, 50L, "d_0", 1, 0.1, false);
+ addRow(t, 2, 200L, "d_1", 12, 1.2, false);
+ addRow(t, 3, 60L, "d_0", 2, 0.2, true);
+ t.setRowSize(4);
+ w.write(t);
+ }
+ }
+
+ private static void addRow(
+ Tablet t, int row, long time, String dev, int s0, double s1, boolean s2) {
+ t.addTimestamp(row, time);
+ t.addValue(row, "device_id", dev);
+ t.addValue(row, "s_0", s0);
+ t.addValue(row, "s_1", s1);
+ t.addValue(row, "s_2", s2);
+ }
+
+ /**
+ * Writes a malformed-for-benchmark table with only FIELD columns and NO TAG/device_id column, so
+ * rows have no usable device identifier. Used to verify the reader skips them instead of NPEing.
+ */
+ public static void writeTableWithoutDeviceId(File file) throws Exception {
+ List cols = new ArrayList<>();
+ cols.add(new MeasurementSchema("s_0", TSDataType.INT32));
+ cols.add(new MeasurementSchema("s_1", TSDataType.DOUBLE));
+ List cats = Arrays.asList(ColumnCategory.FIELD, ColumnCategory.FIELD);
+ TableSchema schema = new TableSchema("t_notag", cols, cats);
+
+ List names = Arrays.asList("s_0", "s_1");
+ List types = Arrays.asList(TSDataType.INT32, TSDataType.DOUBLE);
+
+ try (DeviceTableModelWriter w = new DeviceTableModelWriter(file, schema, 10 * 1024 * 1024L)) {
+ Tablet t = new Tablet("t_notag", names, types, cats, 2);
+ t.addTimestamp(0, 1L);
+ t.addValue(0, "s_0", 1);
+ t.addValue(0, "s_1", 0.1);
+ t.addTimestamp(1, 2L);
+ t.addValue(1, "s_0", 2);
+ t.addValue(1, "s_1", 0.2);
+ t.setRowSize(2);
+ w.write(t);
+ }
+ }
+}