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:02 UTC
(tsfile) 02/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 9d2e028b0e63ee2f1892f031c7f8d99197581d59
Author: Tian Jiang <jt...@163.com>
AuthorDate: Wed Sep 16 17:52:51 2026 +0800
tmp
---
.../org/apache/tsfile/tools/ArrowSourceReader.java | 123 +++++++++++++--------
.../apache/tsfile/tools/ParquetSourceReader.java | 91 +++++++--------
.../org/apache/tsfile/tools/TabletBuilder.java | 11 +-
.../write/chunk/AlignedChunkGroupWriterImpl.java | 15 ++-
.../chunk/NonAlignedChunkGroupWriterImpl.java | 41 +++++--
.../tsfile/write/chunk/AlignedNullWriteTest.java | 57 ++++++++++
6 files changed, 222 insertions(+), 116 deletions(-)
diff --git a/java/tools/src/main/java/org/apache/tsfile/tools/ArrowSourceReader.java b/java/tools/src/main/java/org/apache/tsfile/tools/ArrowSourceReader.java
index e23222d80..3f439224f 100644
--- a/java/tools/src/main/java/org/apache/tsfile/tools/ArrowSourceReader.java
+++ b/java/tools/src/main/java/org/apache/tsfile/tools/ArrowSourceReader.java
@@ -182,23 +182,15 @@ public class ArrowSourceReader implements SourceReader {
}
int numCols = schemaColumnNames.size();
- List<Object[]> rows = new ArrayList<>(rowCount);
-
- for (int r = 0; r < rowCount; r++) {
- Object[] row = new Object[numCols];
- for (int c = 0; c < numCols; c++) {
- String colName = schemaColumnNames.get(c);
- FieldVector vec = vectorMap.get(colName);
- if (vec == null || vec.isNull(r)) {
- row[c] = null;
- } else {
- row[c] = extractValue(vec, r);
- }
+ Object[][] columns = new Object[numCols][rowCount];
+ // Arrow and SourceBatch are columnar: resolve each vector once and avoid a row transpose.
+ for (int c = 0; c < numCols; c++) {
+ FieldVector vec = vectorMap.get(schemaColumnNames.get(c));
+ if (vec != null) {
+ readColumn(vec, columns[c]);
}
- rows.add(row);
}
-
- return SourceBatch.fromRows(schemaColumnNames, rows);
+ return new SourceBatch(schemaColumnNames.toArray(new String[0]), columns, rowCount);
} catch (IOException e) {
LOGGER.error(Messages.format("log.tools.arrow_read_error", sourceFile.getAbsolutePath()), e);
exhausted = true;
@@ -293,40 +285,75 @@ public class ArrowSourceReader implements SourceReader {
return names;
}
- private Object extractValue(FieldVector vec, int row) {
- // Date / Timestamp checks must come BEFORE the BigIntVector/IntVector branches: although
- // they hold int/long underneath, DateDayVector / TimeStampVector do NOT extend
- // IntVector / BigIntVector, so without these branches Date columns fall through to the
- // generic getObject().toString() path and produce strings that don't match TSDataType.DATE.
- if (vec instanceof DateDayVector) {
- // Days since 1970-01-01. ValueConverter.toLocalDate handles Integer → LocalDate.
- return ((DateDayVector) vec).get(row);
- } else if (vec instanceof DateMilliVector) {
- // Millis since 1970-01-01; collapse to date.
- long millis = ((DateMilliVector) vec).get(row);
- return LocalDate.ofEpochDay(Math.floorDiv(millis, 86_400_000L));
- } else if (vec instanceof TimeStampVector) {
- // Long in the vector's native precision; matches the precision detected by
- // detectTimestampPrecision() and stored on the schema.
- return ((TimeStampVector) vec).get(row);
- } else if (vec instanceof BigIntVector) {
- return ((BigIntVector) vec).get(row);
- } else if (vec instanceof IntVector) {
- return ((IntVector) vec).get(row);
- } else if (vec instanceof Float4Vector) {
- return ((Float4Vector) vec).get(row);
- } else if (vec instanceof Float8Vector) {
- return ((Float8Vector) vec).get(row);
- } else if (vec instanceof BitVector) {
- return ((BitVector) vec).get(row) != 0;
- } else if (vec instanceof VarCharVector) {
- byte[] bytes = ((VarCharVector) vec).get(row);
- return new String(bytes, StandardCharsets.UTF_8);
- } else if (vec instanceof VarBinaryVector) {
- return ((VarBinaryVector) vec).get(row);
+ /** Select the physical vector type once for the entire column. */
+ private void readColumn(FieldVector vec, Object[] output) {
+ if (vec instanceof DateDayVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row);
+ }
+ }
+ } else if (vec instanceof DateMilliVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = LocalDate.ofEpochDay(Math.floorDiv(v.get(row), 86_400_000L));
+ }
+ }
+ } else if (vec instanceof TimeStampVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row);
+ }
+ }
+ } else if (vec instanceof BigIntVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row);
+ }
+ }
+ } else if (vec instanceof IntVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row);
+ }
+ }
+ } else if (vec instanceof Float4Vector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row);
+ }
+ }
+ } else if (vec instanceof Float8Vector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row);
+ }
+ }
+ } else if (vec instanceof BitVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row) != 0;
+ }
+ }
+ } else if (vec instanceof VarCharVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = new String(v.get(row), StandardCharsets.UTF_8);
+ }
+ }
+ } else if (vec instanceof VarBinaryVector v) {
+ for (int row = 0; row < output.length; row++) {
+ if (!v.isNull(row)) {
+ output[row] = v.get(row);
+ }
+ }
} else {
- Object obj = vec.getObject(row);
- return obj != null ? obj.toString() : null;
+ for (int row = 0; row < output.length; row++) {
+ if (!vec.isNull(row)) {
+ Object value = vec.getObject(row);
+ output[row] = value == null ? null : value.toString();
+ }
+ }
}
}
diff --git a/java/tools/src/main/java/org/apache/tsfile/tools/ParquetSourceReader.java b/java/tools/src/main/java/org/apache/tsfile/tools/ParquetSourceReader.java
index 207ef1b60..afe53131c 100644
--- a/java/tools/src/main/java/org/apache/tsfile/tools/ParquetSourceReader.java
+++ b/java/tools/src/main/java/org/apache/tsfile/tools/ParquetSourceReader.java
@@ -46,6 +46,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.Function;
public class ParquetSourceReader implements SourceReader {
@@ -164,34 +165,29 @@ public class ParquetSourceReader implements SourceReader {
Map<String, Integer> parquetColIndex = buildParquetColumnIndex();
int numCols = schemaColumnNames.size();
- List<Object[]> rows = new ArrayList<>((int) rowCount);
-
- for (long r = 0; r < rowCount; r++) {
+ int count = Math.toIntExact(rowCount);
+ Object[][] columns = new Object[numCols][count];
+ List<Function<Group, Object>> extractors = new ArrayList<>(numCols);
+ for (String name : schemaColumnNames) {
+ Integer index = parquetColIndex.get(name);
+ extractors.add(index == null ? null : valueExtractor(index));
+ }
+ // The record reader is row-oriented, but its output can go directly into SourceBatch columns.
+ for (int row = 0; row < count; row++) {
Group group = recordReader.read();
- Object[] row = new Object[numCols];
-
- for (int c = 0; c < numCols; c++) {
- String colName = schemaColumnNames.get(c);
- Integer pIdx = parquetColIndex.get(colName);
- if (pIdx == null) {
- row[c] = null;
- continue;
- }
-
- try {
- if (group.getFieldRepetitionCount(pIdx) == 0) {
- row[c] = null;
- } else {
- row[c] = extractValue(group, pIdx);
+ for (int col = 0; col < numCols; col++) {
+ Function<Group, Object> extractor = extractors.get(col);
+ if (extractor != null) {
+ try {
+ columns[col][row] = extractor.apply(group);
+ } catch (RuntimeException e) {
+ // Preserve the existing treatment of missing or malformed field values as null.
+ columns[col][row] = null;
}
- } catch (RuntimeException e) {
- row[c] = null;
}
}
- rows.add(row);
}
-
- return SourceBatch.fromRows(schemaColumnNames, rows);
+ return new SourceBatch(schemaColumnNames.toArray(new String[0]), columns, count);
} catch (IOException e) {
LOGGER.error(
Messages.format("log.tools.parquet_read_error", sourceFile.getAbsolutePath()), e);
@@ -281,37 +277,30 @@ public class ParquetSourceReader implements SourceReader {
return index;
}
- private Object extractValue(Group group, int fieldIndex) {
+ private Function<Group, Object> valueExtractor(int fieldIndex) {
Type fieldType = parquetSchema.getType(fieldIndex);
+ Function<Group, Object> extractor;
if (!fieldType.isPrimitive()) {
- return group.getGroup(fieldIndex, 0).toString();
- }
-
- PrimitiveType pt = fieldType.asPrimitiveType();
- switch (pt.getPrimitiveTypeName()) {
- case BOOLEAN:
- return group.getBoolean(fieldIndex, 0);
- case INT32:
- return group.getInteger(fieldIndex, 0);
- case INT64:
- return group.getLong(fieldIndex, 0);
- case FLOAT:
- return group.getFloat(fieldIndex, 0);
- case DOUBLE:
- return group.getDouble(fieldIndex, 0);
- case BINARY:
- case FIXED_LEN_BYTE_ARRAY:
- LogicalTypeAnnotation logical = pt.getLogicalTypeAnnotation();
- if (logical instanceof LogicalTypeAnnotation.StringLogicalTypeAnnotation) {
- return group.getString(fieldIndex, 0);
- }
- return group.getBinary(fieldIndex, 0).getBytes();
- case INT96:
- // Use getInt96 — INT96 values are Int96Value, not BinaryValue, so getBinary throws CCE.
- return int96ToEpochNanos(group.getInt96(fieldIndex, 0).getBytes());
- default:
- return group.getValueToString(fieldIndex, 0);
+ extractor = group -> group.getGroup(fieldIndex, 0).toString();
+ } else {
+ PrimitiveType pt = fieldType.asPrimitiveType();
+ extractor =
+ switch (pt.getPrimitiveTypeName()) {
+ case BOOLEAN -> group -> group.getBoolean(fieldIndex, 0);
+ case INT32 -> group -> group.getInteger(fieldIndex, 0);
+ case INT64 -> group -> group.getLong(fieldIndex, 0);
+ case FLOAT -> group -> group.getFloat(fieldIndex, 0);
+ case DOUBLE -> group -> group.getDouble(fieldIndex, 0);
+ case BINARY, FIXED_LEN_BYTE_ARRAY ->
+ pt.getLogicalTypeAnnotation()
+ instanceof LogicalTypeAnnotation.StringLogicalTypeAnnotation
+ ? group -> group.getString(fieldIndex, 0)
+ : group -> group.getBinary(fieldIndex, 0).getBytes();
+ // INT96 values require getInt96 rather than getBinary.
+ case INT96 -> group -> int96ToEpochNanos(group.getInt96(fieldIndex, 0).getBytes());
+ };
}
+ return group -> group.getFieldRepetitionCount(fieldIndex) == 0 ? null : extractor.apply(group);
}
/**
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 e55019627..43ff5ece1 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
@@ -22,6 +22,8 @@ import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.TableSchema;
import org.apache.tsfile.i18n.Messages;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -78,11 +80,18 @@ public class TabletBuilder {
for (int col = 0; col < tableSchema.getColumnSchemas().size(); col++) {
IMeasurementSchema colSchema = tableSchema.getColumnSchemas().get(col);
String colName = colSchema.getMeasurementName();
+ Type type = Type.fromTsDataType(colSchema.getType());
+ Object targetValues = tablet.getValues()[col];
+ // addTimestamp initialized the API bitmap; resolve it and the column type only once.
+ BitMap nulls = rowCount == 0 ? null : tablet.getBitMaps()[col];
if (tagDefaults.containsKey(colName)) {
Object defaultValue = tagDefaults.get(colName);
for (int i = 0; i < rowCount; i++) {
- tablet.addValue(colName, i, defaultValue);
+ type.addValue(i, defaultValue, targetValues);
+ if (defaultValue != null) {
+ nulls.unmark(i);
+ }
}
continue;
}
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 6b66792ef..00e97cb71 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
@@ -207,6 +207,9 @@ public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
emptyValueChunkWriters.add(entry.getValue());
}
}
+ // Resolve lazily on the first row to preserve timestamp validation and late-column catch-up.
+ ValueChunkWriter[] writers = new ValueChunkWriter[measurementSchemas.size()];
+ Type[] types = new Type[measurementSchemas.size()];
// TODO: changing to a column-first style by calculating the remaining page space of each
// column firsts
for (int row = startRowIndex; row < endRowIndex; row++) {
@@ -223,10 +226,14 @@ public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
&& tablet.getBitMaps()[columnIndex] != null
&& tablet.getBitMaps()[columnIndex].isMarked(row);
// check isNull by bitMap in tablet
- ValueChunkWriter valueChunkWriter =
- tryToAddSeriesWriterInternal(measurementSchemas.get(columnIndex));
- Type.fromTsDataType(measurementSchemas.get(columnIndex).getType())
- .write(valueChunkWriter, time, tablet.getValues()[columnIndex], row, isNull);
+ ValueChunkWriter valueChunkWriter = writers[columnIndex];
+ if (valueChunkWriter == null) {
+ valueChunkWriter = tryToAddSeriesWriterInternal(measurementSchemas.get(columnIndex));
+ writers[columnIndex] = valueChunkWriter;
+ types[columnIndex] = Type.fromTsDataType(measurementSchemas.get(columnIndex).getType());
+ }
+ types[columnIndex].write(
+ valueChunkWriter, time, tablet.getValues()[columnIndex], row, isNull);
}
// TODO: we can write the null columns after whole insertion, according to the point number
// in the time chunk before and after, no need to do it in a row-by-row manner
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java
index ea73884d2..64f4b6297 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/NonAlignedChunkGroupWriterImpl.java
@@ -26,6 +26,8 @@ import org.apache.tsfile.exception.write.WriteProcessException;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.i18n.Messages;
import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.TypeServices;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.record.datapoint.DataPoint;
import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -116,20 +118,35 @@ public class NonAlignedChunkGroupWriterImpl implements IChunkGroupWriter {
String measurementId = timeseries.get(column).getMeasurementName();
Type type = Type.fromTsDataType(timeseries.get(column).getType());
ChunkWriterImpl chunkWriter = chunkWriters.get(measurementId);
- pointCount = 0;
- for (int row = startRowIndex; row < endRowIndex; row++) {
- // check isNull in tablet
- if (tablet.getBitMaps() != null
- && tablet.getBitMaps()[column] != null
- && tablet.getBitMaps()[column].isMarked(row)) {
- continue;
+ BitMap nulls = tablet.getBitMaps() == null ? null : tablet.getBitMaps()[column];
+ int firstRow = startRowIndex;
+ while (firstRow < endRowIndex && nulls != null && nulls.isMarked(firstRow)) {
+ firstRow++;
+ }
+ // An empty column must not inspect its value array or invoke a value writer.
+ if (firstRow >= endRowIndex) {
+ continue;
+ }
+ TabletWriteContext context =
+ new TabletWriteContext(deviceId, measurementId, lastTimeMap.get(measurementId));
+ try {
+ TypeServices.WRITE_TABLET_COLUMN_SERVICE
+ .call(type)
+ .write(
+ chunkWriter,
+ tablet.getTimestamps(),
+ tablet.getValues()[column],
+ nulls,
+ firstRow,
+ endRowIndex,
+ context);
+ } finally {
+ // Commit the successfully written prefix even when a later row is out of order.
+ if (context.getPointCount() != 0) {
+ lastTimeMap.put(measurementId, context.getLastTime());
}
- long time = tablet.getTimestamps()[row];
- checkIsHistoryData(measurementId, time);
- pointCount++;
- type.write(chunkWriter, time, tablet.getValues()[column], row);
- lastTimeMap.put(measurementId, time);
}
+ pointCount = context.getPointCount();
maxPointCount = Math.max(pointCount, maxPointCount);
}
return maxPointCount;
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/chunk/AlignedNullWriteTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/chunk/AlignedNullWriteTest.java
index 8e6eebe36..4448ae31e 100644
--- a/java/tsfile/src/test/java/org/apache/tsfile/write/chunk/AlignedNullWriteTest.java
+++ b/java/tsfile/src/test/java/org/apache/tsfile/write/chunk/AlignedNullWriteTest.java
@@ -22,20 +22,77 @@ import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.encoding.decoder.PlainDecoder;
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.write.record.Tablet;
import org.apache.tsfile.write.record.datapoint.IntDataPoint;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.Test;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import java.util.Collections;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
public class AlignedNullWriteTest {
+ @Test
+ public void tabletWriterCacheKeepsLateColumnPagesAndFailureOrder() throws Exception {
+ TSFileConfig config = TSFileDescriptor.getInstance().getConfig();
+ int oldLimit = config.getMaxNumberOfPointsInPage();
+ try {
+ config.setMaxNumberOfPointsInPage(8);
+ AlignedChunkGroupWriterImpl scalar =
+ new AlignedChunkGroupWriterImpl(IDeviceID.Factory.DEFAULT_FACTORY.create("root.tablet"));
+ AlignedChunkGroupWriterImpl batch =
+ new AlignedChunkGroupWriterImpl(IDeviceID.Factory.DEFAULT_FACTORY.create("root.tablet"));
+ MeasurementSchema missing =
+ new MeasurementSchema(
+ "missing", TSDataType.INT32, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED);
+ scalar.tryToAddSeriesWriter(missing);
+ batch.tryToAddSeriesWriter(missing);
+ for (int row = 0; row < 11; row++) {
+ scalar.write(row, Collections.emptyList());
+ batch.write(row, Collections.emptyList());
+ }
+ MeasurementSchema late =
+ new MeasurementSchema(
+ "late", TSDataType.INT32, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED);
+ scalar.tryToAddSeriesWriter(late);
+ Tablet tablet = new Tablet("root.tablet", Collections.singletonList(late), 24);
+ for (int row = 0; row < 24; row++) {
+ tablet.addTimestamp(row, row + 11);
+ tablet.addValue("late", row, row);
+ scalar.write(row + 11, Collections.singletonList(new IntDataPoint("late", row)));
+ }
+ batch.write(tablet);
+ for (String name : Arrays.asList("missing", "late")) {
+ ValueChunkWriter expected = scalar.valueChunkWriterMap.get(name);
+ ValueChunkWriter actual = batch.valueChunkWriterMap.get(name);
+ expected.sealCurrentPage();
+ actual.sealCurrentPage();
+ assertEquals(expected.getByteBuffer(), actual.getByteBuffer());
+ assertEquals(expected.getNumOfPages(), actual.getNumOfPages());
+ assertEquals(expected.getStatistics(), actual.getStatistics());
+ }
+ Tablet rejected =
+ new Tablet(
+ "root.tablet",
+ Collections.singletonList(new MeasurementSchema("rejected", TSDataType.INT32)),
+ 1);
+ rejected.addTimestamp(0, 0);
+ rejected.addValue("rejected", 0, 42);
+ assertThrows(WriteProcessException.class, () -> batch.write(rejected));
+ assertEquals(2, batch.valueChunkWriterMap.size());
+ } finally {
+ config.setMaxNumberOfPointsInPage(oldLimit);
+ }
+ }
+
@Test
public void lateColumnsAndMissingRowsKeepPageBoundaries() throws Exception {
TSFileConfig config = TSFileDescriptor.getInstance().getConfig();