You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by ji...@apache.org on 2026/09/17 04:21:03 UTC
(tsfile) 03/04: tmp
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch optimize_loops_with_datatype
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit bf46f2c52e97b7e34c5b8dbb592bce64ef399b20
Author: Tian Jiang <jt...@163.com>
AuthorDate: Wed Sep 16 18:15:38 2026 +0800
tmp
---
.../org/apache/tsfile/tools/TabletBuilder.java | 5 +-
.../org/apache/tsfile/tools/TabletBuilderTest.java | 73 +++++++++
.../templates/TabletColumnWriterTemplate.ftl | 72 +++++++++
.../java/org/apache/tsfile/utils/TypeServices.java | 39 +++++
.../tsfile/write/chunk/TabletWriteContext.java | 68 +++++++++
.../tsfile/write/chunk/TabletColumnWriteTest.java | 164 +++++++++++++++++++++
6 files changed, 420 insertions(+), 1 deletion(-)
diff --git a/java/tools/src/main/java/org/apache/tsfile/tools/TabletBuilder.java b/java/tools/src/main/java/org/apache/tsfile/tools/TabletBuilder.java
index 43ff5ece1..61d4457d7 100644
--- a/java/tools/src/main/java/org/apache/tsfile/tools/TabletBuilder.java
+++ b/java/tools/src/main/java/org/apache/tsfile/tools/TabletBuilder.java
@@ -117,7 +117,10 @@ public class TabletBuilder {
colSchema.getType(), isMeasurement, importSchema.getTimePrecision());
}
Object converted = converter.apply(rawValue);
- tablet.addValue(colName, i, converted);
+ type.addValue(i, converted, targetValues);
+ if (converted != null) {
+ nulls.unmark(i);
+ }
}
}
diff --git a/java/tools/src/test/java/org/apache/tsfile/tools/TabletBuilderTest.java b/java/tools/src/test/java/org/apache/tsfile/tools/TabletBuilderTest.java
index 5dcfbb3ef..3e1c1ae00 100644
--- a/java/tools/src/test/java/org/apache/tsfile/tools/TabletBuilderTest.java
+++ b/java/tools/src/test/java/org/apache/tsfile/tools/TabletBuilderTest.java
@@ -21,6 +21,7 @@ package org.apache.tsfile.tools;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.junit.Test;
@@ -35,6 +36,78 @@ import static org.junit.Assert.assertTrue;
public class TabletBuilderTest {
+ @Test
+ public void testColumnInsertionMatchesPublicTabletApi() {
+ TSDataType[] types = {
+ TSDataType.BOOLEAN,
+ TSDataType.INT32,
+ TSDataType.INT64,
+ TSDataType.FLOAT,
+ TSDataType.DOUBLE,
+ TSDataType.DATE,
+ TSDataType.TIMESTAMP,
+ TSDataType.TEXT,
+ TSDataType.STRING,
+ TSDataType.BLOB,
+ TSDataType.OBJECT
+ };
+ List<ImportSchema.SourceColumn> columns = new ArrayList<>();
+ columns.add(new ImportSchema.SourceColumn("time", TSDataType.INT64));
+ String[] names = new String[types.length + 1];
+ names[0] = "time";
+ for (int col = 0; col < types.length; col++) {
+ names[col + 1] = "s" + col;
+ columns.add(new ImportSchema.SourceColumn(names[col + 1], types[col]));
+ }
+ ImportSchema schema =
+ buildSchema(
+ "test",
+ "time",
+ Collections.singletonList(new ImportSchema.TagColumn("tag", "default")),
+ columns.toArray(new ImportSchema.SourceColumn[0]));
+ schema.setNullFormat("NULL");
+ TabletBuilder builder = new TabletBuilder(schema, new TimeConverter("ms"));
+ for (int count : new int[] {0, 1, 35}) {
+ Object[][] values = new Object[names.length][count];
+ for (int row = 0; row < count; row++) {
+ values[0][row] = (long) (count - row);
+ for (int col = 0; col < types.length; col++) {
+ values[col + 1][row] =
+ row % 4 == 0
+ ? "NULL"
+ : switch (types[col]) {
+ case BOOLEAN -> "true";
+ case DATE -> "2024-02-29";
+ case INT32, INT64, TIMESTAMP, FLOAT, DOUBLE -> row;
+ case TEXT, STRING, BLOB, OBJECT -> "v" + row;
+ case VECTOR, UNKNOWN -> throw new AssertionError();
+ };
+ }
+ }
+ Tablet actual = builder.build(new SourceBatch(names, values, count));
+ List<IMeasurementSchema> schemas = builder.getTableSchema().getColumnSchemas();
+ Tablet expected =
+ new Tablet(
+ "test",
+ IMeasurementSchema.getMeasurementNameList(schemas),
+ IMeasurementSchema.getDataTypeList(schemas),
+ builder.getTableSchema().getColumnTypes(),
+ count);
+ for (int row = 0; row < count; row++) {
+ expected.addTimestamp(row, row + 1);
+ expected.addValue("tag", row, "default");
+ for (int col = 0; col < types.length; col++) {
+ Object raw = values[col + 1][count - row - 1];
+ if (!"NULL".equals(raw)) {
+ expected.addValue(names[col + 1], row, ValueConverter.convert(raw, types[col], true));
+ }
+ }
+ }
+ expected.setRowSize(count);
+ assertEquals(expected, actual);
+ }
+ }
+
@Test
public void testColumnConversionKeepsSortedRowsAndNulls() {
ImportSchema schema =
diff --git a/java/tsfile/src/main/codegen/templates/TabletColumnWriterTemplate.ftl b/java/tsfile/src/main/codegen/templates/TabletColumnWriterTemplate.ftl
new file mode 100644
index 000000000..ae6a3c083
--- /dev/null
+++ b/java/tsfile/src/main/codegen/templates/TabletColumnWriterTemplate.ftl
@@ -0,0 +1,72 @@
+<@pp.dropOutputFile />
+<#list [["Boolean", "boolean"], ["Int", "int"], ["Long", "long"],
+ ["Float", "float"], ["Double", "double"], ["Binary", "Binary"],
+ ["Date", "LocalDate"]] as t>
+<#assign name=t[0] primitive=t[1]>
+<@pp.changeOutputFile name="/org/apache/tsfile/read/common/type/service/${name}TabletColumnWriter.java" />
+/*
+ * 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.read.common.type.service;
+
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.utils.BitMap;
+<#if name == "Binary">
+import org.apache.tsfile.utils.Binary;
+</#if>
+<#if name == "Date">
+import org.apache.tsfile.utils.DateUtils;
+import java.time.LocalDate;
+</#if>
+import org.apache.tsfile.utils.TypeServices.TabletColumnWriter;
+import org.apache.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.tsfile.write.chunk.TabletWriteContext;
+
+/** Generated typed loop; retain scalar encoding to preserve page boundaries and SDT state. */
+public final class ${name}TabletColumnWriter implements TabletColumnWriter {
+ public static final ${name}TabletColumnWriter INSTANCE = new ${name}TabletColumnWriter();
+
+ private ${name}TabletColumnWriter() {}
+
+ @Override
+ public void write(ChunkWriterImpl writer, long[] times, Object values, BitMap nulls,
+ int start, int end, TabletWriteContext context) throws WriteProcessException {
+<#if name == "Date">
+ if (values instanceof int[]) {
+ IntTabletColumnWriter.INSTANCE.write(writer, times, values, nulls, start, end, context);
+ return;
+ }
+</#if>
+ ${primitive}[] column = (${primitive}[]) values;
+ for (int row = start; row < end; row++) {
+ if (nulls != null && nulls.isMarked(row)) {
+ continue;
+ }
+ long time = times[row];
+ context.checkTime(time);
+<#if name == "Date">
+ writer.write(time, DateUtils.parseDateExpressionToInt(column[row]));
+<#else>
+ writer.write(time, column[row]);
+</#if>
+ context.recordTime(time);
+ }
+ }
+}
+</#list>
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/utils/TypeServices.java b/java/tsfile/src/main/java/org/apache/tsfile/utils/TypeServices.java
index 01763e1af..0afe8b77a 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/utils/TypeServices.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/utils/TypeServices.java
@@ -22,21 +22,31 @@ package org.apache.tsfile.utils;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.encoding.decoder.Decoder;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
import org.apache.tsfile.i18n.Messages;
import org.apache.tsfile.read.common.BatchData;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder;
import org.apache.tsfile.read.common.type.service.BinaryPageDataReader;
+import org.apache.tsfile.read.common.type.service.BinaryTabletColumnWriter;
import org.apache.tsfile.read.common.type.service.BooleanPageDataReader;
+import org.apache.tsfile.read.common.type.service.BooleanTabletColumnWriter;
import org.apache.tsfile.read.common.type.service.DatePageDataReader;
+import org.apache.tsfile.read.common.type.service.DateTabletColumnWriter;
import org.apache.tsfile.read.common.type.service.DoublePageDataReader;
+import org.apache.tsfile.read.common.type.service.DoubleTabletColumnWriter;
import org.apache.tsfile.read.common.type.service.FloatPageDataReader;
+import org.apache.tsfile.read.common.type.service.FloatTabletColumnWriter;
import org.apache.tsfile.read.common.type.service.IntPageDataReader;
+import org.apache.tsfile.read.common.type.service.IntTabletColumnWriter;
import org.apache.tsfile.read.common.type.service.LongPageDataReader;
+import org.apache.tsfile.read.common.type.service.LongTabletColumnWriter;
import org.apache.tsfile.read.common.type.service.TypeService;
import org.apache.tsfile.read.filter.basic.Filter;
import org.apache.tsfile.read.reader.series.PaginationController;
import org.apache.tsfile.write.UnSupportedDataTypeException;
+import org.apache.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.tsfile.write.chunk.TabletWriteContext;
import org.apache.tsfile.write.chunk.ValueChunkWriter;
import java.io.IOException;
@@ -61,7 +71,36 @@ public final class TypeServices {
.setChecked(true);
};
+ /** Resolve the Tablet column type before entering its row loop. */
+ public static final TypeService<TabletColumnWriter> WRITE_TABLET_COLUMN_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN -> BooleanTabletColumnWriter.INSTANCE;
+ case INT32 -> IntTabletColumnWriter.INSTANCE;
+ case DATE -> DateTabletColumnWriter.INSTANCE;
+ case INT64, TIMESTAMP -> LongTabletColumnWriter.INSTANCE;
+ case FLOAT -> FloatTabletColumnWriter.INSTANCE;
+ case DOUBLE -> DoubleTabletColumnWriter.INSTANCE;
+ case TEXT, BLOB, STRING, OBJECT -> BinaryTabletColumnWriter.INSTANCE;
+ case ROW, UNKNOWN, VECTOR ->
+ throw new UnSupportedDataTypeException(String.valueOf(type.getTypeEnum()))
+ .setChecked(true);
+ };
+
+ public interface TabletColumnWriter {
+ void write(
+ ChunkWriterImpl writer,
+ long[] times,
+ Object values,
+ BitMap nulls,
+ int start,
+ int end,
+ TabletWriteContext context)
+ throws WriteProcessException;
+ }
+
static {
+ WRITE_TABLET_COLUMN_SERVICE.check();
READ_PAGE_BATCH_SERVICE.check();
}
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TabletWriteContext.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TabletWriteContext.java
new file mode 100644
index 000000000..3a97619dc
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TabletWriteContext.java
@@ -0,0 +1,68 @@
+/*
+ * 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.chunk;
+
+import org.apache.tsfile.common.constant.TsFileConstant;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.i18n.Messages;
+
+/** Per-column state shared by the type-specialized Tablet write loops. */
+public final class TabletWriteContext {
+ private final IDeviceID deviceId;
+ private final String measurementId;
+ private boolean hasLastTime;
+ private long lastTime;
+ private int pointCount;
+
+ public TabletWriteContext(IDeviceID deviceId, String measurementId, Long lastTime) {
+ this.deviceId = deviceId;
+ this.measurementId = measurementId;
+ this.hasLastTime = lastTime != null;
+ this.lastTime = lastTime == null ? 0 : lastTime;
+ }
+
+ public void checkTime(long time) throws WriteProcessException {
+ if (hasLastTime && time <= lastTime) {
+ throw new WriteProcessException(
+ Messages.format(
+ "error.write.chunk_group_non_aligned_out_of_order",
+ deviceId,
+ TsFileConstant.PATH_SEPARATOR,
+ measurementId,
+ lastTime));
+ }
+ }
+
+ /** Advance only after a successful write, so failures retain the last accepted timestamp. */
+ public void recordTime(long time) {
+ lastTime = time;
+ hasLastTime = true;
+ pointCount++;
+ }
+
+ public long getLastTime() {
+ return lastTime;
+ }
+
+ public int getPointCount() {
+ return pointCount;
+ }
+}
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/chunk/TabletColumnWriteTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/chunk/TabletColumnWriteTest.java
new file mode 100644
index 000000000..af65a99c8
--- /dev/null
+++ b/java/tsfile/src/test/java/org/apache/tsfile/write/chunk/TabletColumnWriteTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.chunk;
+
+import org.apache.tsfile.common.conf.TSFileDescriptor;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+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.common.type.Type;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.TypeServices;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import org.junit.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDate;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+public class TabletColumnWriteTest {
+ @Test
+ public void testEncodedPagesMatchScalarForAllTypes() throws Exception {
+ int oldLimit = TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
+ try {
+ // Force multiple pages and exercise both bitmap boundaries and a nonzero source offset.
+ TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(7);
+ for (TSDataType dataType : TSDataType.values()) {
+ if (dataType == TSDataType.UNKNOWN || dataType == TSDataType.VECTOR) {
+ continue;
+ }
+ Type type = Type.fromTsDataType(dataType);
+ for (boolean sparse : new boolean[] {false, true}) {
+ Object values = type.createArray(35);
+ long[] times = new long[35];
+ BitMap nulls = sparse ? new BitMap(35) : null;
+ for (int row = 0; row < 35; row++) {
+ times[row] = row - 20;
+ type.addValue(row, value(dataType, row), values);
+ if (sparse && row % 3 == 0) {
+ nulls.mark(row);
+ }
+ }
+ compare(type, dataType, times, values, nulls);
+ if (dataType == TSDataType.DATE) {
+ int[] intDates = new int[35];
+ for (int row = 0; row < intDates.length; row++) {
+ intDates[row] = 20240101 + row % 28;
+ }
+ compare(type, dataType, times, intDates, nulls);
+ }
+ }
+ }
+ } finally {
+ TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(oldLimit);
+ }
+ }
+
+ private void compare(Type type, TSDataType dataType, long[] times, Object values, BitMap nulls)
+ throws Exception {
+ MeasurementSchema schema =
+ new MeasurementSchema("s", dataType, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED);
+ compareWriters(type, dataType, times, values, nulls, schema);
+ if (dataType == TSDataType.INT32
+ || dataType == TSDataType.INT64
+ || dataType == TSDataType.FLOAT
+ || dataType == TSDataType.DOUBLE) {
+ schema.setProps(Map.of("loss", "sdt", "compdev", "0.1", "compmaxtime", "5"));
+ compareWriters(type, dataType, times, values, nulls, schema);
+ }
+ }
+
+ private void compareWriters(
+ Type type,
+ TSDataType dataType,
+ long[] times,
+ Object values,
+ BitMap nulls,
+ MeasurementSchema schema)
+ throws Exception {
+ ChunkWriterImpl scalar = new ChunkWriterImpl(schema);
+ ChunkWriterImpl batch = new ChunkWriterImpl(schema);
+ int count = 0;
+ for (int row = 2; row < 34; row++) {
+ if (nulls == null || !nulls.isMarked(row)) {
+ type.write(scalar, times[row], values, row);
+ count++;
+ }
+ }
+ TabletWriteContext context =
+ new TabletWriteContext(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d"), "s", null);
+ TypeServices.WRITE_TABLET_COLUMN_SERVICE
+ .call(type)
+ .write(batch, times, values, nulls, 2, 34, context);
+ scalar.sealCurrentPage();
+ batch.sealCurrentPage();
+ assertEquals(dataType.toString(), scalar.getByteBuffer(), batch.getByteBuffer());
+ assertEquals(scalar.getStatistics(), batch.getStatistics());
+ assertEquals(scalar.getNumOfPages(), batch.getNumOfPages());
+ assertEquals(count, context.getPointCount());
+ }
+
+ @Test
+ public void testOutOfOrderKeepsWrittenPrefixAndSkipsNulls() throws Exception {
+ NonAlignedChunkGroupWriterImpl writer =
+ new NonAlignedChunkGroupWriterImpl(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d"));
+ writer.tryToAddSeriesWriter(new MeasurementSchema("s", TSDataType.INT32));
+ Tablet tablet =
+ new Tablet(
+ "root.d", Collections.singletonList(new MeasurementSchema("s", TSDataType.INT32)), 5);
+ long[] times = {Long.MIN_VALUE, 10, 3, 9, 20};
+ for (int row = 0; row < times.length; row++) {
+ tablet.addTimestamp(row, times[row]);
+ if (row != 2) {
+ tablet.addValue("s", row, row);
+ }
+ }
+ assertThrows(WriteProcessException.class, () -> writer.write(tablet, 0, 5));
+ assertEquals(Long.valueOf(10), writer.getLastTimeMap().get("s"));
+ assertEquals(1, writer.write(tablet, 4, 5));
+ assertEquals(Long.valueOf(20), writer.getLastTimeMap().get("s"));
+ assertEquals(0, writer.write(tablet, 2, 3));
+ assertEquals(Long.valueOf(20), writer.getLastTimeMap().get("s"));
+ tablet.getValues()[0] = null;
+ assertEquals(0, writer.write(tablet, 2, 3));
+ assertEquals(0, writer.write(tablet, 0, 0));
+ }
+
+ private Object value(TSDataType type, int row) {
+ return switch (type) {
+ case BOOLEAN -> row % 2 == 0;
+ case INT32 -> row;
+ case DATE -> LocalDate.of(2024, 1, 1 + row % 28);
+ case INT64, TIMESTAMP -> (long) row;
+ case FLOAT -> row * 1.5f;
+ case DOUBLE -> row * 1.5;
+ case TEXT, STRING, BLOB, OBJECT -> new Binary("v" + row, StandardCharsets.UTF_8);
+ case VECTOR, UNKNOWN -> throw new AssertionError(type);
+ };
+ }
+}