Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,10 @@
import java.util.ArrayList;
import java.util.List;

/**
* Column roles in table-model files. TAG columns identify entities, FIELD columns contain
* measurements, ATTRIBUTE columns contain descriptive values, and TIME is the timestamp column.
*/
public enum ColumnCategory {
TAG,
FIELD,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,11 @@

package org.apache.tsfile.encoding.encoder;

/**
* Implements Swinging Door Trending lossy compression. compDeviation bounds value error, while
* compMinTime and compMaxTime control the time window. The pending boundary point is emitted when
* the door closes or flush is called.
*/
public class SDTEncoder {

// the last read time and value if upperDoor >= lowerDoor meaning out of compDeviation range, will
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,10 @@ public static int estimateMaxPageHeaderSizeWithoutStatistics() {
return 2 * (Integer.BYTES + 1);
}

/**
* Deserializes a page header from the input. This overload reads statistics only when
* hasStatistic is true; an empty page may have null statistics.
*/
public static PageHeader deserializeFrom(
InputStream inputStream, TSDataType dataType, boolean hasStatistic) throws IOException {
int uncompressedSize = ReadWriteForEncodingUtils.readUnsignedVarInt(inputStream);
Expand Down Expand Up @@ -173,7 +177,10 @@ public void setModified(boolean modified) {
this.modified |= modified;
}

/** max page header size without statistics. */
/**
* Returns the serialized size of this page header and its page body, including statistics when
* statistics are present.
*/
public int getSerializedPageSize() {
if (uncompressedSize == 0) { // Empty page
return ReadWriteForEncodingUtils.uVarIntSize(uncompressedSize);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,14 @@ public static MetadataIndexNode deserializeFrom(
return new MetadataIndexNode(children, offset, nodeType);
}

/**
* Finds the child entry for key. Exact search requires an equal key; non-exact search returns the
* floor entry. The selected entry's end offset is the exclusive upper bound of the child scan
* range.
*
* @param key -the sorted search key
* @param exactSearch -whether an exact key match is required
*/
public Pair<IMetadataIndexEntry, Long> getChildIndexEntry(Comparable key, boolean exactSearch) {
int index = binarySearchInChildren(key, exactSearch);
if (index == -1) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,10 @@
import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfCharArray;
import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfObjectArray;

/**
* Metadata for one time series. Chunk metadata may be loaded lazily from an in-memory buffer or
* temporary file; callers must initialize the loader before requesting deferred chunk metadata.
*/
public class TimeseriesMetadata implements ITimeSeriesMetadata {

private static final int INSTANCE_SIZE =
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,9 @@ public static TsFileMetadata deserializeWithoutCacheTableSchemaMap(
* deserialize data from the buffer.
*
* @param buffer -buffer use to deserialize
* @param context -reader context used for version and encryption compatibility
* @param needTableSchemaMap -whether table schemas should be materialized and cached; disabling
* it reduces memory usage.
* @return -an instance of TsFileMetaData
*/
public static TsFileMetadata deserializeFrom(
Expand Down Expand Up @@ -293,6 +296,10 @@ public Map<String, MetadataIndexNode> getTableMetadataIndexNodeMap() {
return tableMetadataIndexNodeMap;
}

/**
* Returns the metadata index for tableName. If no table-specific node exists, the default root
* keyed by the empty table name is used.
*/
public MetadataIndexNode getTableMetadataIndexNode(String tableName) {
MetadataIndexNode metadataIndexNode = tableMetadataIndexNodeMap.get(tableName);
if (metadataIndexNode == null) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,13 @@ public TsFileSequenceReader(String file) throws IOException {
this(file, true, null);
}

/**
* Scans the file and counts chunks in each chunk group. The current file position is used during
* scanning and the returned result may be incomplete when a malformed tail is encountered.
*
* @return the number of chunks for each scanned chunk group
* @throws IOException if the file cannot be read
*/
public Map<IDeviceID, Integer> countChunksPerChunkGroup() throws IOException {
Map<IDeviceID, Integer> result = new LinkedHashMap<>();

Expand Down
10 changes: 10 additions & 0 deletions java/tsfile/src/main/java/org/apache/tsfile/read/common/Chunk.java
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,12 @@ public void setDeleteIntervalList(List<TimeRange> list) {
this.deleteIntervalList = list;
}

/**
* Appends the pages of another compatible chunk. When two single-page chunks are merged, the
* resulting chunk must be marked as multi-page and the original chunk statistics must be written
* into their page headers. Chunks must have compatible data types, codecs, and measurement
* semantics.
*/
public void mergeChunkByAppendPage(Chunk chunk) throws IOException {
int dataSize = 0;
// from where the page data of the merged chunk starts, if -1, it means the merged chunk has
Expand Down Expand Up @@ -216,6 +222,10 @@ public long getRetainedSizeInBytes() {
return INSTANCE_SIZE + sizeOfByteArray(chunkData.capacity());
}

/**
* Re-encodes this chunk using the target data type and schema. Null values are represented using
* the target type's null sentinel, and statistics are recomputed from the rewritten records.
*/
public Chunk rewrite(TSDataType newType, Chunk timeChunk) throws IOException {
if (newType == null || newType == chunkHeader.getDataType()) {
return this;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,10 @@
import java.io.IOException;
import java.util.List;

/**
* Provides page readers for one chunk and exposes chunk-level skip and deletion decisions.
* Implementations may defer decompression and decryption until a page is read.
*/
public interface IChunkReader {

boolean hasNextSatisfiedPage() throws IOException;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,10 @@
import java.util.List;
import java.util.function.LongConsumer;

/**
* Reads decoded records from one page. Implementations may consume decoder state; filtering,
* deletion handling, pagination, and empty-page behavior are defined by the methods below.
*/
public interface IPageReader extends IMetadata {

default BatchData getAllSatisfiedPageData() throws IOException {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,10 @@ public ChunkReader(Chunk chunk, long readStopTime) {
this(chunk, readStopTime, null, null);
}

/**
* Parses all page headers and creates page readers. For a single-page chunk, the chunk statistic
* may be reused; for multiple pages, each page follows its own statistic and data boundaries.
*/
private void initAllPageReaders(Statistics<? extends Serializable> chunkStatistic) {
// construct next satisfied page header
while (chunkDataBuffer.remaining() > 0) {
Expand All @@ -105,6 +109,7 @@ private void initAllPageReaders(Statistics<? extends Serializable> chunkStatisti
}
}

/** Determines whether a page can be skipped by time/statistic filters or deletion intervals. */
private boolean pageCanSkip(PageHeader pageHeader) {
if (queryFilter != null
&& !queryFilter.satisfyStartEndTime(pageHeader.getStartTime(), pageHeader.getEndTime())) {
Expand All @@ -116,6 +121,10 @@ private boolean pageCanSkip(PageHeader pageHeader) {
return false;
}

/**
* A fully deleted page is skipped; a partially deleted page remains readable and is marked for
* record-level filtering.
*/
protected boolean pageDeleted(PageHeader pageHeader) {
if (readStopTime > pageHeader.getEndTime()) {
// used for chunk reader by timestamp
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,10 @@ public void initTsBlockBuilder(List<TSDataType> dataTypes) {
// do nothing
}

/**
* Tests whether timestamp is covered by the current deletion intervals. Intervals are inclusive
* and expected to be sorted; the deletion cursor advances monotonically for ascending timestamps.
*/
protected boolean isDeleted(long timestamp) {
while (deleteIntervalList != null && deleteCursor < deleteIntervalList.size()) {
if (deleteIntervalList.get(deleteCursor).contains(timestamp)) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,11 @@
import java.util.Map;
import java.util.Optional;

/**
* Reads table-model data for one device. Table and column names are normalized according to the
* reader's case rules; tagFilter applies only to tag columns. close() releases reader resources and
* may suppress close-time I/O failures.
*/
public class DeviceTableModelReader implements ITsFileReader {

protected TsFileSequenceReader fileReader;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -502,7 +502,11 @@ private void checkIsAllMeasurementsInGroup(
}
}

/** Check whether all measurements of dataPoints list are in the measurementGroup. */
/**
* Validates measurement schemas against the registered group schema. For non-aligned writes,
* unknown measurements are filtered from a private copy; for aligned writes, an unknown
* measurement is rejected. The caller's list is not modified.
*/
private List<IMeasurementSchema> checkIsAllMeasurementsInGroup(
List<DataPoint> dataPoints, MeasurementGroup measurementGroup, boolean isAligned)
throws NoMeasurementException {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,11 @@
import java.util.Set;
import java.util.stream.Collectors;

/**
* Writes an aligned chunk group consisting of one shared time chunk and one value chunk per
* measurement. Time and value pages remain synchronized, and bitmaps represent missing values at
* aligned row positions.
*/
public class AlignedChunkGroupWriterImpl implements IChunkGroupWriter {
protected static final Logger LOG = LoggerFactory.getLogger(AlignedChunkGroupWriterImpl.class);

Expand Down Expand Up @@ -278,6 +283,10 @@ public long getCurrentChunkGroupSize() {
return size;
}

/**
* Adds empty value pages and null bitmap entries when a value measurement is introduced after
* time pages already exist, preserving page boundaries and row alignment across all value chunks.
*/
public void tryToAddEmptyPageAndData(ValueChunkWriter valueChunkWriter) throws IOException {
// add empty page
for (int i = 0; i < timeChunkWriter.getNumOfPages(); i++) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,11 @@
import java.nio.channels.WritableByteChannel;
import java.util.function.Function;

/**
* Writes one measurement chunk by aggregating page data, serializing a chunk header and all page
* bodies, and producing chunk metadata. After the chunk is written, the writer resets its page
* state and can be reused.
*/
public class ChunkWriterImpl implements IChunkWriter {

private static final Logger logger = LoggerFactory.getLogger(ChunkWriterImpl.class);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,12 @@ public interface IChunkGroupWriter {
*/
int write(Tablet tablet) throws WriteProcessException, IOException;

/**
* Writes rows in the half-open range [startRowIndex, endRowIndex).
*
* @param startRowIndex inclusive first row
* @param endRowIndex exclusive end row
*/
int write(Tablet table, int startRowIndex, int endRowIndex)
throws WriteProcessException, IOException;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,10 @@
import java.util.Map;
import java.util.stream.Collectors;

/**
* Describes one measurement's name, data type, encoding, compression, statistic, and optional
* properties. Implementations must preserve these fields during schema serialization and copying.
*/
public interface IMeasurementSchema extends Accountable {

MeasurementSchemaType getSchemaType();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,10 @@ public MeasurementSchema(String measurementName, TSDataType dataType) {
null);
}

/** set properties as an empty Map. */
/**
* Creates a measurement schema. The properties argument may be null; callers must not assume that
* an empty mutable map is created automatically.
*/
public MeasurementSchema(String measurementName, TSDataType dataType, TSEncoding encoding) {
this(
measurementName,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,10 @@

import java.io.IOException;

/**
* Public v4 TsFile writing contract. Implementations must document schema registration,
* record/tablet lifecycle, flush behavior, close requirements, and all checked exceptions.
*/
public interface ITsFileWriter extends AutoCloseable {

@TsFileApi
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,10 @@
import java.io.File;
import java.io.IOException;

/**
* Builder for v4 TsFile writers. build() validates the output path, schema, table-model settings,
* and memory thresholds before creating a writer.
*/
public class TsFileWriterBuilder {

private static final long defaultMemoryThresholdInByte = 32 * 1024 * 1024;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -331,8 +331,10 @@ public boolean isWritingChunkGroup() {
* @param measurementId - measurementId of this time series
* @param compressionCodecName - compression name of this time series
* @param tsDataType - data type
* @param encodingType - the encoding used by the chunk pages
* @param statistics - Chunk statistics
* @param dataSize - the serialized size of all pages
* @param numOfPages - the number of serialized pages
* @param mask - 0x80 for time chunk, 0x40 for value chunk, 0x00 for common chunk
* @throws IOException if I/O error occurs
*/
Expand Down
Loading