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/01/28 04:06:33 UTC
(tsfile) 01/09: add void writeToFileWriter(TsFileIOWriter tsfileWriter, Function measurementNameRemapper) throws IOException;
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch sevo_support
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 24205774c45917a6ac7d64fb3faf96b0ba405156
Author: Tian Jiang <jt...@163.com>
AuthorDate: Sun Jan 4 16:04:41 2026 +0800
add void writeToFileWriter(TsFileIOWriter tsfileWriter, Function<String, String> measurementNameRemapper) throws IOException;
---
.../tsfile/write/chunk/AlignedChunkWriterImpl.java | 10 ++++++++++
.../apache/tsfile/write/chunk/ChunkWriterImpl.java | 19 ++++++++++++++-----
.../org/apache/tsfile/write/chunk/IChunkWriter.java | 6 ++++++
.../apache/tsfile/write/chunk/ValueChunkWriter.java | 16 ++++++++++++----
4 files changed, 42 insertions(+), 9 deletions(-)
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java
index 27761e67..08ae5753 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/AlignedChunkWriterImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.tsfile.write.chunk;
+import java.util.function.Function;
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.encoding.encoder.Encoder;
@@ -477,6 +478,15 @@ public class AlignedChunkWriterImpl implements IChunkWriter {
}
}
+ @Override
+ public void writeToFileWriter(TsFileIOWriter tsfileWriter,
+ Function<String, String> measurementNameRemapper) throws IOException {
+ timeChunkWriter.writeToFileWriter(tsfileWriter);
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ valueChunkWriter.writeToFileWriter(tsfileWriter, measurementNameRemapper);
+ }
+ }
+
@Override
public long estimateMaxSeriesMemSize() {
long estimateMaxSeriesMemSize = timeChunkWriter.estimateMaxSeriesMemSize();
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java
index 25e0410b..d450b373 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.tsfile.write.chunk;
+import java.util.function.Function;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.compress.ICompressor;
import org.apache.tsfile.encoding.encoder.SDTEncoder;
@@ -346,8 +347,14 @@ public class ChunkWriterImpl implements IChunkWriter {
@Override
public void writeToFileWriter(TsFileIOWriter tsfileWriter) throws IOException {
+ writeToFileWriter(tsfileWriter, null);
+ }
+
+ @Override
+ public void writeToFileWriter(TsFileIOWriter tsfileWriter,
+ Function<String, String> measurementNameRemapper) throws IOException {
sealCurrentPage();
- writeAllPagesOfChunkToTsFile(tsfileWriter, statistics);
+ writeAllPagesOfChunkToTsFile(tsfileWriter, statistics, measurementNameRemapper);
// reinit this chunk writer
pageBuffer.reset();
@@ -472,19 +479,21 @@ public class ChunkWriterImpl implements IChunkWriter {
/**
* write the page to specified IOWriter.
*
- * @param writer the specified IOWriter
- * @param statistics the chunk statistics
+ * @param writer the specified IOWriter
+ * @param statistics the chunk statistics
+ * @param measurementNameRemapper
* @throws IOException exception in IO
*/
private void writeAllPagesOfChunkToTsFile(
- TsFileIOWriter writer, Statistics<? extends Serializable> statistics) throws IOException {
+ TsFileIOWriter writer, Statistics<? extends Serializable> statistics,
+ Function<String, String> measurementNameRemapper) throws IOException {
if (statistics.getCount() == 0) {
return;
}
// start to write this column chunk
writer.startFlushChunk(
- measurementSchema.getMeasurementName(),
+ measurementNameRemapper == null ? measurementSchema.getMeasurementName() : measurementNameRemapper.apply(measurementSchema.getMeasurementName()),
compressor.getType(),
measurementSchema.getType(),
measurementSchema.getEncodingType(),
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/IChunkWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/IChunkWriter.java
index 85b9828a..11304073 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/IChunkWriter.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/IChunkWriter.java
@@ -18,6 +18,7 @@
*/
package org.apache.tsfile.write.chunk;
+import java.util.function.Function;
import org.apache.tsfile.write.writer.TsFileIOWriter;
import java.io.IOException;
@@ -28,6 +29,11 @@ public interface IChunkWriter {
/** flush data to TsFileIOWriter. */
void writeToFileWriter(TsFileIOWriter tsfileWriter) throws IOException;
+ /**
+ * flush data to TsFileIOWriter, may rename the measurement in the file according to the remapper.
+ */
+ void writeToFileWriter(TsFileIOWriter tsfileWriter, Function<String, String> measurementNameRemapper) throws IOException;
+
/** estimate memory usage of this series. */
long estimateMaxSeriesMemSize();
diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java
index d011e125..a778f741 100644
--- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java
+++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java
@@ -18,6 +18,7 @@
*/
package org.apache.tsfile.write.chunk;
+import java.util.function.Function;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.compress.ICompressor;
@@ -295,8 +296,13 @@ public class ValueChunkWriter {
}
public void writeToFileWriter(TsFileIOWriter tsfileWriter) throws IOException {
+ writeToFileWriter(tsfileWriter, null);
+ }
+
+ public void writeToFileWriter(TsFileIOWriter tsfileWriter,
+ Function<String, String> measurementNameRemapper) throws IOException {
sealCurrentPage();
- writeAllPagesOfChunkToTsFile(tsfileWriter);
+ writeAllPagesOfChunkToTsFile(tsfileWriter, measurementNameRemapper);
// reinit this chunk writer
pageBuffer.reset();
@@ -385,7 +391,9 @@ public class ValueChunkWriter {
* @param writer the specified IOWriter
* @throws IOException exception in IO
*/
- public void writeAllPagesOfChunkToTsFile(TsFileIOWriter writer) throws IOException {
+ public void writeAllPagesOfChunkToTsFile(TsFileIOWriter writer,
+ Function<String, String> measurementNameRemapper) throws IOException {
+ String finalMeasurementId = measurementNameRemapper == null ? measurementId : measurementNameRemapper.apply(measurementId);
if (statistics.getCount() == 0) {
if (pageBuffer.size() == 0) {
return;
@@ -395,7 +403,7 @@ public class ValueChunkWriter {
// chunkGroup during compaction. To save the disk space, we only serialize chunkHeader for the
// empty valueChunk, whose dataSize is 0.
writer.startFlushChunk(
- measurementId,
+ finalMeasurementId,
compressionType,
dataType,
encodingType,
@@ -409,7 +417,7 @@ public class ValueChunkWriter {
// start to write this column chunk
writer.startFlushChunk(
- measurementId,
+ finalMeasurementId,
compressionType,
dataType,
encodingType,