This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d2cb37a40a1abeab98d400b97ce292491c974a5b Author: Tian Jiang <[email protected]> AuthorDate: Sat Oct 10 20:00:21 2026 +0800 tmp --- .gitignore | 2 + .../apache/iotdb/session/SessionConnection.java | 18 +- .../iotdb/session/SessionConnectionTest.java | 82 ++++++++ .../consensus/iot/IoTConsensusServerImpl.java | 26 ++- .../consensus/iot/IoTConsensusServerMetrics.java | 96 +++++++++ .../execution/operator/source/FileLoaderUtils.java | 56 ++++- .../operator/source/FlatDevicePageReaderQueue.java | 14 +- .../iotdb/db/service/metrics/WritingMetrics.java | 129 +++++++++++- .../dataregion/memtable/TsFileProcessor.java | 48 ++++- .../read/reader/chunk/AlignedChunkPageReader.java | 112 ++++++++++ .../read/reader/chunk/DiskAlignedChunkLoader.java | 38 +++- .../source/FlatDevicePageReaderQueueTest.java | 197 ++++++++++++++++++ .../dataregion/memtable/TsFileProcessorTest.java | 43 ++-- .../reader/chunk/AlignedChunkPageReaderTest.java | 230 +++++++++++++++++++++ 14 files changed, 1042 insertions(+), 49 deletions(-) diff --git a/.gitignore b/.gitignore index 76e774d9286..5a57e52c2cb 100644 --- a/.gitignore +++ b/.gitignore @@ -139,3 +139,5 @@ iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/gra .claude/todos/ .claude/worktrees/ .claude/scheduled_tasks.json + +.codex diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java index b498784ad2c..aae1c18b5d3 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java @@ -942,7 +942,8 @@ public class SessionConnection { return new RetryResult<>(status, lastTException, i); } - private RetryResult<TSStatus> callWithRetryAndReconnect(TFunction<TSStatus> rpc) { + private RetryResult<TSStatus> callWithRetryAndReconnect(TFunction<TSStatus> rpc) + throws IoTDBConnectionException { return callWithRetryAndReconnect( rpc, status -> status.isSetNeedRetry() && status.isNeedRetry(), @@ -950,7 +951,7 @@ public class SessionConnection { } private <T> RetryResult<T> callWithRetryAndReconnect( - TFunction<T> rpc, Function<T, TSStatus> statusGetter) { + TFunction<T> rpc, Function<T, TSStatus> statusGetter) throws IoTDBConnectionException { return callWithRetryAndReconnect( rpc, t -> { @@ -964,7 +965,8 @@ public class SessionConnection { /** reconnect if the remote datanode is unreachable retry if the status is set to needRetry */ private <T> RetryResult<T> callWithRetryAndReconnect( - TFunction<T> rpc, Predicate<T> shouldRetry, Predicate<T> forceReconnect) { + TFunction<T> rpc, Predicate<T> shouldRetry, Predicate<T> forceReconnect) + throws IoTDBConnectionException { TException lastTException = null; T result = null; int retryAttempt; @@ -1012,7 +1014,15 @@ public class SessionConnection { } } - return new RetryResult<>(result, lastTException, retryAttempt); + // Callers need a response to verify its status. Preserve the RPC failure instead of + // letting a null response hide it behind a NullPointerException. + if (lastTException != null) { + throw new IoTDBConnectionException(lastTException); + } + if (result == null) { + throw new IoTDBConnectionException(logForReconnectionFailure()); + } + return new RetryResult<>(result, null, retryAttempt); } private TSStatus deleteDataInternal(TSDeleteDataReq request) throws TException { diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java index fea78466a78..655a5147965 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java @@ -56,6 +56,7 @@ import org.apache.iotdb.service.rpc.thrift.TSUnsetSchemaTemplateReq; import org.apache.thrift.TException; import org.apache.thrift.transport.TTransport; +import org.apache.thrift.transport.TTransportException; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -258,6 +259,87 @@ public class SessionConnectionTest { sessionConnection.executeQueryStatement("show version", 500l); } + @Test + public void testQueryRetryExhaustionPreservesTransportException() throws Exception { + TTransportException firstFailure = new TTransportException("first read timeout"); + TTransportException lastFailure = new TTransportException("last read timeout"); + Mockito.when(client.executeQueryStatementV2(any())).thenThrow(firstFailure, lastFailure); + // No live transport: exercise failed reconnects without opening network connections. + Whitebox.setInternalState(sessionConnection, "transport", (TTransport) null); + Whitebox.setInternalState(sessionConnection, "retryIntervalInMs", 0L); + try { + IoTDBConnectionException failure = + Assert.assertThrows( + IoTDBConnectionException.class, + () -> sessionConnection.executeQueryStatement("show version", 500)); + Assert.assertSame(lastFailure, failure.getCause()); + Mockito.verify(client, Mockito.times(11)).executeQueryStatementV2(any()); + } finally { + Whitebox.setInternalState(sessionConnection, "transport", transport); + } + } + + @Test + public void testQueryInterruptedRetryPreservesTransportException() throws Exception { + TTransportException rpcFailure = new TTransportException("read timeout"); + Mockito.when(client.executeQueryStatementV2(any())).thenThrow(rpcFailure); + Whitebox.setInternalState(sessionConnection, "transport", (TTransport) null); + try { + Thread.currentThread().interrupt(); + IoTDBConnectionException failure = + Assert.assertThrows( + IoTDBConnectionException.class, + () -> sessionConnection.executeQueryStatement("show version", 500)); + Assert.assertSame(rpcFailure, failure.getCause()); + Assert.assertTrue(Thread.currentThread().isInterrupted()); + Mockito.verify(client).executeQueryStatementV2(any()); + } finally { + Thread.interrupted(); + Whitebox.setInternalState(sessionConnection, "transport", transport); + } + } + + @Test + public void testQueryRetryCanRecover() throws Exception { + TSExecuteStatementResp response = + new TSExecuteStatementResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + response.setColumns(Collections.emptyList()); + response.setDataTypeList(Collections.emptyList()); + response.setQueryResult(Collections.emptyList()); + Mockito.when(client.executeQueryStatementV2(any())) + .thenThrow(new TTransportException("read timeout")) + .thenReturn(response); + Whitebox.setInternalState(sessionConnection, "transport", (TTransport) null); + Whitebox.setInternalState(sessionConnection, "retryIntervalInMs", 0L); + try { + Assert.assertNotNull(sessionConnection.executeQueryStatement("show version", 500)); + Mockito.verify(client, Mockito.times(2)).executeQueryStatementV2(any()); + } finally { + Whitebox.setInternalState(sessionConnection, "transport", transport); + } + } + + @Test + public void testQueryRetryExhaustionPreservesServerStatus() throws Exception { + TSStatus status = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); + status.setNeedRetry(true); + status.setMessage("query still unavailable"); + Mockito.when(client.executeQueryStatementV2(any())) + .thenReturn(new TSExecuteStatementResp(status)); + Whitebox.setInternalState(sessionConnection, "transport", (TTransport) null); + Whitebox.setInternalState(sessionConnection, "retryIntervalInMs", 0L); + try { + StatementExecutionException failure = + Assert.assertThrows( + StatementExecutionException.class, + () -> sessionConnection.executeQueryStatement("show version", 500)); + Assert.assertEquals(status.getCode(), failure.getStatusCode()); + Mockito.verify(client, Mockito.times(11)).executeQueryStatementV2(any()); + } finally { + Whitebox.setInternalState(sessionConnection, "transport", transport); + } + } + @Test public void testExecuteNonQueryStatement() throws IoTDBConnectionException, StatementExecutionException { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index 66d917bebc2..216dd29398d 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -252,8 +252,8 @@ public class IoTConsensusServerImpl { public TSStatus write(IConsensusRequest request) { long consensusWriteStartTime = System.nanoTime(); stateMachineLock.lock(); + long getStateMachineLockTime = System.nanoTime(); try { - long getStateMachineLockTime = System.nanoTime(); // statistic the time of acquiring stateMachine lock ioTConsensusServerMetrics.recordGetStateMachineLockTime( getStateMachineLockTime - consensusWriteStartTime); @@ -281,10 +281,13 @@ public class IoTConsensusServerImpl { // statistic the time of checking write block ioTConsensusServerMetrics.recordCheckingBeforeWriteTime( writeToStateMachineStartTime - getStateMachineLockTime); + long buildIndexedRequestStartTime = System.nanoTime(); IndexedConsensusRequest indexedConsensusRequest = buildIndexedConsensusRequestForLocalRequest(request); indexedConsensusRequest.setRoutingEpoch(currentRoutingEpoch); lastConsensusRequest = indexedConsensusRequest; + ioTConsensusServerMetrics.recordBuildIndexedRequestTime( + System.nanoTime() - buildIndexedRequestStartTime); if (indexedConsensusRequest.getSearchIndex() % 100000 == 0) { logger.info( IoTConsensusMessages.DATA_REGION_INDEX_AFTER_BUILD, @@ -293,26 +296,35 @@ public class IoTConsensusServerImpl { indexedConsensusRequest.getSearchIndex(), lastConsensusRequest.getSerializedRequests()); } + long deserializeRequestStartTime = System.nanoTime(); IConsensusRequest planNode = stateMachine.deserializeRequest(indexedConsensusRequest); + ioTConsensusServerMetrics.recordDeserializeRequestTime( + System.nanoTime() - deserializeRequestStartTime); long startWriteTime = System.nanoTime(); TSStatus result = stateMachine.write(planNode); - PERFORMANCE_OVERVIEW_METRICS.recordEngineCost(System.nanoTime() - startWriteTime); + long stateMachineWriteCost = System.nanoTime() - startWriteTime; + PERFORMANCE_OVERVIEW_METRICS.recordEngineCost(stateMachineWriteCost); + ioTConsensusServerMetrics.recordStateMachineWriteInnerTime(stateMachineWriteCost); long writeToStateMachineEndTime = System.nanoTime(); // statistic the time of writing request into stateMachine ioTConsensusServerMetrics.recordWriteStateMachineTime( writeToStateMachineEndTime - writeToStateMachineStartTime); if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + long recordAppliedProgressStartTime = System.nanoTime(); writerSafeFrontierTracker.recordAppliedProgress( indexedConsensusRequest.getPhysicalTime(), thisNode.getNodeId(), indexedConsensusRequest.getLocalSeq()); + ioTConsensusServerMetrics.recordAppliedProgressTime( + System.nanoTime() - recordAppliedProgressStartTime); // The index is used when constructing batch in LogDispatcher. If its value // increases but the corresponding request does not exist or is not put into // the queue, the dispatcher will try to find the request in WAL. This behavior // is not expected and will slow down the preparation speed for batch. // So we need to use the lock to ensure the `offer()` and `incrementAndGet()` are // in one transaction. + long offerToQueuesStartTime = System.nanoTime(); synchronized (searchIndex) { logDispatcher.offer(indexedConsensusRequest); // Deliver to subscription queues for real-time in-memory consumption. @@ -333,9 +345,14 @@ public class IoTConsensusServerImpl { } searchIndex.incrementAndGet(); } + ioTConsensusServerMetrics.recordOfferToQueuesTime( + System.nanoTime() - offerToQueuesStartTime); + long updateWriterMetaStartTime = System.nanoTime(); latestWriterMeta = new WriterMeta( indexedConsensusRequest.getLocalSeq(), indexedConsensusRequest.getPhysicalTime()); + ioTConsensusServerMetrics.recordUpdateWriterMetaTime( + System.nanoTime() - updateWriterMetaStartTime); // statistic the time of offering request into queue ioTConsensusServerMetrics.recordOfferRequestToQueueTime( System.nanoTime() - writeToStateMachineEndTime); @@ -354,8 +371,13 @@ public class IoTConsensusServerImpl { System.nanoTime() - consensusWriteStartTime); return result; } finally { + long stateMachineLockHoldTime = System.nanoTime() - getStateMachineLockTime; stateMachineLock.unlock(); + ioTConsensusServerMetrics.recordStateMachineLockHoldTime(stateMachineLockHoldTime); + long persistWriterMetaStartTime = System.nanoTime(); persistLatestWriterMeta(false); + ioTConsensusServerMetrics.recordPersistWriterMetaTime( + System.nanoTime() - persistWriterMetaStartTime); } } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerMetrics.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerMetrics.java index 117f9712635..06b7ac1e814 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerMetrics.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerMetrics.java @@ -36,6 +36,22 @@ public class IoTConsensusServerMetrics implements IMetricSet { private Timer writeStateMachineTimer = DoNothingMetricManager.DO_NOTHING_TIMER; private Timer offerRequestToQueueTimer = DoNothingMetricManager.DO_NOTHING_TIMER; private Timer consensusWriteTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer stateMachineLockHoldTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer buildIndexedRequestTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer deserializeRequestTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer stateMachineWriteInnerTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer recordAppliedProgressTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer offerToQueuesTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer updateWriterMetaTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private Timer persistWriterMetaTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + private static final String STATE_MACHINE_LOCK_HOLD = "stateMachineLockHold"; + private static final String BUILD_INDEXED_REQUEST = "buildIndexedRequest"; + private static final String DESERIALIZE_REQUEST = "deserializeRequest"; + private static final String STATE_MACHINE_WRITE = "stateMachineWrite"; + private static final String RECORD_APPLIED_PROGRESS = "recordAppliedProgress"; + private static final String OFFER_TO_QUEUES = "offerToQueues"; + private static final String UPDATE_WRITER_META = "updateWriterMeta"; + private static final String PERSIST_WRITER_META = "persistWriterMeta"; private static final String IOT_RECEIVE_LOG = Metric.IOT_RECEIVE_LOG.toString(); private Timer deserializeTimer = DoNothingMetricManager.DO_NOTHING_TIMER; private Timer sortTimer = DoNothingMetricManager.DO_NOTHING_TIMER; @@ -186,6 +202,26 @@ public class IoTConsensusServerMetrics implements IMetricSet { "consensusWrite", Tag.REGION.toString(), impl.getConsensusGroupId()); + stateMachineLockHoldTimer = createWriteStageTimer(metricService, STATE_MACHINE_LOCK_HOLD); + buildIndexedRequestTimer = createWriteStageTimer(metricService, BUILD_INDEXED_REQUEST); + deserializeRequestTimer = createWriteStageTimer(metricService, DESERIALIZE_REQUEST); + stateMachineWriteInnerTimer = createWriteStageTimer(metricService, STATE_MACHINE_WRITE); + recordAppliedProgressTimer = createWriteStageTimer(metricService, RECORD_APPLIED_PROGRESS); + offerToQueuesTimer = createWriteStageTimer(metricService, OFFER_TO_QUEUES); + updateWriterMetaTimer = createWriteStageTimer(metricService, UPDATE_WRITER_META); + persistWriterMetaTimer = createWriteStageTimer(metricService, PERSIST_WRITER_META); + } + + private Timer createWriteStageTimer(AbstractMetricService metricService, String stage) { + return metricService.getOrCreateTimer( + Metric.STAGE.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + Metric.IOT_CONSENSUS.toString(), + Tag.TYPE.toString(), + stage, + Tag.REGION.toString(), + impl.getConsensusGroupId()); } private void bindSyncLogTimer(AbstractMetricService metricService) { @@ -270,6 +306,14 @@ public class IoTConsensusServerMetrics implements IMetricSet { writeStateMachineTimer = DoNothingMetricManager.DO_NOTHING_TIMER; offerRequestToQueueTimer = DoNothingMetricManager.DO_NOTHING_TIMER; consensusWriteTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + stateMachineLockHoldTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + buildIndexedRequestTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + deserializeRequestTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + stateMachineWriteInnerTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + recordAppliedProgressTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + offerToQueuesTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + updateWriterMetaTimer = DoNothingMetricManager.DO_NOTHING_TIMER; + persistWriterMetaTimer = DoNothingMetricManager.DO_NOTHING_TIMER; metricService.remove( MetricType.TIMER, Metric.STAGE.toString(), @@ -315,6 +359,26 @@ public class IoTConsensusServerMetrics implements IMetricSet { "consensusWrite", Tag.REGION.toString(), impl.getConsensusGroupId()); + removeWriteStageTimer(metricService, STATE_MACHINE_LOCK_HOLD); + removeWriteStageTimer(metricService, BUILD_INDEXED_REQUEST); + removeWriteStageTimer(metricService, DESERIALIZE_REQUEST); + removeWriteStageTimer(metricService, STATE_MACHINE_WRITE); + removeWriteStageTimer(metricService, RECORD_APPLIED_PROGRESS); + removeWriteStageTimer(metricService, OFFER_TO_QUEUES); + removeWriteStageTimer(metricService, UPDATE_WRITER_META); + removeWriteStageTimer(metricService, PERSIST_WRITER_META); + } + + private void removeWriteStageTimer(AbstractMetricService metricService, String stage) { + metricService.remove( + MetricType.TIMER, + Metric.STAGE.toString(), + Tag.NAME.toString(), + Metric.IOT_CONSENSUS.toString(), + Tag.TYPE.toString(), + stage, + Tag.REGION.toString(), + impl.getConsensusGroupId()); } private void unbindSyncLogTimer(AbstractMetricService metricService) { @@ -361,4 +425,36 @@ public class IoTConsensusServerMetrics implements IMetricSet { public void recordConsensusWriteTime(long costTimeInNanos) { consensusWriteTimer.updateNanos(costTimeInNanos); } + + public void recordStateMachineLockHoldTime(long costTimeInNanos) { + stateMachineLockHoldTimer.updateNanos(costTimeInNanos); + } + + public void recordBuildIndexedRequestTime(long costTimeInNanos) { + buildIndexedRequestTimer.updateNanos(costTimeInNanos); + } + + public void recordDeserializeRequestTime(long costTimeInNanos) { + deserializeRequestTimer.updateNanos(costTimeInNanos); + } + + public void recordStateMachineWriteInnerTime(long costTimeInNanos) { + stateMachineWriteInnerTimer.updateNanos(costTimeInNanos); + } + + public void recordAppliedProgressTime(long costTimeInNanos) { + recordAppliedProgressTimer.updateNanos(costTimeInNanos); + } + + public void recordOfferToQueuesTime(long costTimeInNanos) { + offerToQueuesTimer.updateNanos(costTimeInNanos); + } + + public void recordUpdateWriterMetaTime(long costTimeInNanos) { + updateWriterMetaTimer.updateNanos(costTimeInNanos); + } + + public void recordPersistWriterMetaTime(long costTimeInNanos) { + persistWriterMetaTimer.updateNanos(costTimeInNanos); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java index 023dd69fd57..8c256c4fbbd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java @@ -61,6 +61,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Objects; import java.util.Set; import java.util.stream.Collectors; @@ -511,20 +512,65 @@ public class FileLoaderUtils { List<TSDataType> targetDataTypeList, FragmentInstanceContext context) throws IOException { + return loadPageReaderList(chunkMetaData, globalTimeFilter, context, -1, null); + } + + /** + * Loads disk aligned pages and appends their zero-based physical indexes to physicalPageIndexes. + * The returned list excludes skipped pages, so its indexes cannot be used for random page loads. + */ + public static List<IPageReader> loadPageReaderList( + IChunkMetadata chunkMetaData, + Filter globalTimeFilter, + List<TSDataType> targetDataTypeList, + FragmentInstanceContext context, + List<Integer> physicalPageIndexes) + throws IOException { + return loadPageReaderList( + chunkMetaData, globalTimeFilter, context, -1, Objects.requireNonNull(physicalPageIndexes)); + } + + /** + * Loads only the requested physical page of a disk aligned chunk. Returns null if the page is + * absent or skipped by the time filter or deletions. ChunkCache still reads complete chunks. + */ + public static IPageReader loadPageReader( + IChunkMetadata chunkMetaData, + Filter globalTimeFilter, + FragmentInstanceContext context, + int pageIndex) + throws IOException { + Objects.checkIndex(pageIndex, Integer.MAX_VALUE); + List<IPageReader> pages = + loadPageReaderList(chunkMetaData, globalTimeFilter, context, pageIndex, null); + return pages.isEmpty() ? null : pages.get(0); + } + + private static List<IPageReader> loadPageReaderList( + IChunkMetadata chunkMetaData, + Filter globalTimeFilter, + FragmentInstanceContext context, + int pageIndex, + List<Integer> physicalPageIndexes) + throws IOException { checkArgument( chunkMetaData != null, DataNodeQueryMessages.EXCEPTION_CAN_QUOTE_T_INIT_NULL_CHUNKMETA_15C12BEE); IChunkLoader chunkLoader = chunkMetaData.getChunkLoader(); File tsFile = null; - if (chunkLoader instanceof DiskChunkLoader) { - tsFile = ((DiskChunkLoader) chunkLoader).getTsFile(); - } else if (chunkLoader instanceof DiskAlignedChunkLoader) { - tsFile = ((DiskAlignedChunkLoader) chunkLoader).getTsFile(); + if (chunkLoader instanceof DiskChunkLoader diskLoader) { + tsFile = diskLoader.getTsFile(); + } else if (chunkLoader instanceof DiskAlignedChunkLoader alignedLoader) { + tsFile = alignedLoader.getTsFile(); } final IChunkReader chunkReader; try { - chunkReader = chunkLoader.getChunkReader(chunkMetaData, globalTimeFilter); + chunkReader = + pageIndex >= 0 || physicalPageIndexes != null + ? ((DiskAlignedChunkLoader) chunkLoader) + .getChunkReader(chunkMetaData, globalTimeFilter, pageIndex, physicalPageIndexes) + : chunkLoader.getChunkReader(chunkMetaData, globalTimeFilter); } catch (ChunkTypeInconsistentException e) { // if the chunk in tsfile is a value chunk of aligned series but registered series is // non-aligned, we should skip all data of this chunk. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java index 0ad96dabc48..fcc645c2954 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java @@ -80,8 +80,10 @@ final class FlatDevicePageReaderQueue { keyChunk.setChunkLoader(metadata.getChunkLoader()); List<TSDataType> keyTypes = keyIndices.stream().map(dataTypes::get).collect(Collectors.toList()); + List<Integer> physicalPageIndexes = new ArrayList<>(); List<IPageReader> keyPages = - FileLoaderUtils.loadPageReaderList(keyChunk, timeFilter, keyTypes, context); + FileLoaderUtils.loadPageReaderList( + keyChunk, timeFilter, keyTypes, context, physicalPageIndexes); ChunkPages chunkPages = new ChunkPages(metadata, deletionFilter); // TsFile statistics contain no compound TAG/time bounds. Decode only TAG and time columns // to build the pending-page queue, retaining each page's first key in scan order. @@ -103,7 +105,7 @@ final class FlatDevicePageReaderQueue { new PendingPage( new TimeValuePair(first.getTimestamp(), new TsPrimitiveType.TsVector(projected)), chunkPages, - pageIndex)); + physicalPageIndexes.get(pageIndex))); } } } @@ -134,11 +136,11 @@ final class FlatDevicePageReaderQueue { this.deletionFilter = deletionFilter; } - private void open(int index) throws IOException { - if (readers == null) { - readers = FileLoaderUtils.loadPageReaderList(metadata, timeFilter, dataTypes, context); + private void open(int pageIndex) throws IOException { + IPageReader page = FileLoaderUtils.loadPageReader(metadata, timeFilter, context, pageIndex); + if (page == null) { + return; } - IPageReader page = readers.set(index, null); page.initTsBlockBuilder(dataTypes); if (deletionFilter != null) { page.addRecordFilter(deletionFilter); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java index 88197d03cd6..128bdc7a29b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java @@ -47,11 +47,136 @@ import java.util.List; import java.util.Objects; public class WritingMetrics implements IMetricSet { + private static final FlatDeviceWriteStage[] FLAT_DEVICE_WRITE_STAGES = + FlatDeviceWriteStage.values(); private static final WritingMetrics INSTANCE = new WritingMetrics(); private static final WALManager WAL_MANAGER = WALManager.getInstance(); private WritingMetrics() { - // empty constructor + Arrays.fill(flatDeviceWriteTimers, DoNothingMetricManager.DO_NOTHING_TIMER); + } + + public enum FlatDeviceWriteStage { + RAM_COST_SNAPSHOT, + MEMTABLE_INSERT, + RESOURCE_TIME_UPDATE, + RAM_COST_RECONCILE, + MEMTABLE_NULL_COUNT, + MEMTABLE_SCHEMA_PREPARE, + MEMTABLE_DEVICE_SPLIT, + MEMTABLE_CHUNK_LOOKUP, + MEMTABLE_COLUMN_REORDER, + MEMTABLE_TVLIST_APPEND, + MEMTABLE_DEVICE_INDEX, + MEMTABLE_FIELD_APPEND, + MEMTABLE_APPEND_FINISH, + MEMTABLE_TVLIST_HANDOVER, + MEMTABLE_STATISTICS + } + + // These timers describe TsFileProcessor tablet stages, including replica apply. They are + // nested in the consensus state-machine timer, not additional end-to-end write latency. + private final Timer[] flatDeviceWriteTimers = new Timer[FlatDeviceWriteStage.values().length]; + private volatile boolean flatDeviceWriteMetricsEnabled; + private final ThreadLocal<FlatDeviceWriteCostTracker> flatDeviceWriteCostTracker = + new ThreadLocal<>(); + + /** + * Reused by a writing thread to aggregate logical-device work into one sample per MemTable insert + * range. TVLIST_APPEND includes DEVICE_INDEX, FIELD_APPEND and APPEND_FINISH, as well as waiting + * for the TVList monitor; its children must not be added to it again. + */ + public static final class FlatDeviceWriteCostTracker { + private final long[] costs = new long[FLAT_DEVICE_WRITE_STAGES.length]; + private final boolean[] recorded = new boolean[FLAT_DEVICE_WRITE_STAGES.length]; + private boolean active; + + private FlatDeviceWriteCostTracker() {} + + public void recordCost(FlatDeviceWriteStage stage, long costTimeInNanos) { + costs[stage.ordinal()] += costTimeInNanos; + recorded[stage.ordinal()] = true; + } + } + + public FlatDeviceWriteCostTracker startFlatDeviceMemTableInsert() { + if (!flatDeviceWriteMetricsEnabled) { + return null; + } + FlatDeviceWriteCostTracker tracker = flatDeviceWriteCostTracker.get(); + if (tracker == null) { + tracker = new FlatDeviceWriteCostTracker(); + flatDeviceWriteCostTracker.set(tracker); + } else if (tracker.active) { + // A nested caller borrows the active scope; only its owner resets and publishes it. + return null; + } + Arrays.fill(tracker.costs, 0); + Arrays.fill(tracker.recorded, false); + tracker.active = true; + return tracker; + } + + public FlatDeviceWriteCostTracker getFlatDeviceWriteCostTracker() { + if (!flatDeviceWriteMetricsEnabled) { + return null; + } + FlatDeviceWriteCostTracker tracker = flatDeviceWriteCostTracker.get(); + return tracker != null && tracker.active ? tracker : null; + } + + public void finishFlatDeviceMemTableInsert(FlatDeviceWriteCostTracker tracker) { + if (tracker == null) { + return; + } + // Clear the scope before publishing, including when the insert or a metric update fails. + // TsFileProcessor records MEMTABLE_INSERT first, excluding these Timer updates from it. + tracker.active = false; + for (FlatDeviceWriteStage stage : FLAT_DEVICE_WRITE_STAGES) { + if (tracker.recorded[stage.ordinal()]) { + recordFlatDeviceWriteCost(stage, tracker.costs[stage.ordinal()]); + } + } + } + + private void bindFlatDeviceWriteMetrics(AbstractMetricService metricService) { + // All STAGE timers must share the consensus timers' label keys, or registration silently + // returns a no-op timer. "all" identifies these node-wide, rather than per-region, samples. + for (FlatDeviceWriteStage stage : FlatDeviceWriteStage.values()) { + flatDeviceWriteTimers[stage.ordinal()] = + metricService.getOrCreateTimer( + Metric.STAGE.toString(), + MetricLevel.IMPORTANT, + Tag.NAME.toString(), + "flat_device_write", + Tag.TYPE.toString(), + stage.name(), + Tag.REGION.toString(), + "all"); + } + flatDeviceWriteMetricsEnabled = + flatDeviceWriteTimers[FlatDeviceWriteStage.MEMTABLE_INSERT.ordinal()] + != DoNothingMetricManager.DO_NOTHING_TIMER; + } + + private void unbindFlatDeviceWriteMetrics(AbstractMetricService metricService) { + flatDeviceWriteMetricsEnabled = false; + for (FlatDeviceWriteStage stage : FlatDeviceWriteStage.values()) { + flatDeviceWriteTimers[stage.ordinal()] = DoNothingMetricManager.DO_NOTHING_TIMER; + metricService.remove( + MetricType.TIMER, + Metric.STAGE.toString(), + Tag.NAME.toString(), + "flat_device_write", + Tag.TYPE.toString(), + stage.name(), + Tag.REGION.toString(), + "all"); + } + } + + public void recordFlatDeviceWriteCost(FlatDeviceWriteStage stage, long costTimeInNanos) { + flatDeviceWriteTimers[stage.ordinal()].updateNanos(costTimeInNanos); } // region flush overview metrics @@ -1026,6 +1151,7 @@ public class WritingMetrics implements IMetricSet { @Override public void bindTo(AbstractMetricService metricService) { + bindFlatDeviceWriteMetrics(metricService); bindFlushMetrics(metricService); bindFlushSubTaskMetrics(metricService); bindWALMetrics(metricService); @@ -1035,6 +1161,7 @@ public class WritingMetrics implements IMetricSet { @Override public void unbindFrom(AbstractMetricService metricService) { + unbindFlatDeviceWriteMetrics(metricService); unbindFlushMetrics(metricService); unbindFlushSubTaskMetrics(metricService); unbindWALMetrics(metricService); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 8570124ac3b..808e4a87131 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -52,6 +52,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDe import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.schemaengine.schemaregion.utils.ResourceByPathUtils; import org.apache.iotdb.db.service.metrics.WritingMetrics; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteCostTracker; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteStage; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo; import org.apache.iotdb.db.storageengine.dataregion.flush.CloseFileListener; @@ -675,8 +677,18 @@ public class TsFileProcessor { ensureMemTable(infoForMetrics); workMemTable.checkDataType(insertTabletNode); - AlignedTVListRamCostSnapshot alignedRamCostSnapshot = - takeAlignedTVListRamCostSnapshot(workMemTable, insertTabletNode, rangeList); + // Memory snapshots scan existing TVList state; time them separately from appending the + // incoming tablet so a growing device map is not mistaken for slow value insertion. + final boolean measureFlatDeviceWrite = insertTabletNode.isFlatDevice(); + long flatDeviceStageStart = measureFlatDeviceWrite ? System.nanoTime() : 0; + AlignedTVListRamCostSnapshot alignedRamCostSnapshot; + try { + alignedRamCostSnapshot = + takeAlignedTVListRamCostSnapshot(workMemTable, insertTabletNode, rangeList); + } finally { + recordFlatDeviceWriteCost( + measureFlatDeviceWrite, FlatDeviceWriteStage.RAM_COST_SNAPSHOT, flatDeviceStageStart); + } long[] memIncrements = scheduleMemoryBlock(insertTabletNode, rangeList, results, infoForMetrics); @@ -729,6 +741,11 @@ public class TsFileProcessor { final int[] rangePair = rangeList.get(rangeIndex); int start = rangePair[0]; int end = rangePair[1]; + final FlatDeviceWriteCostTracker flatDeviceWriteCostTracker = + measureFlatDeviceWrite + ? WritingMetrics.getInstance().startFlatDeviceMemTableInsert() + : null; + flatDeviceStageStart = measureFlatDeviceWrite ? System.nanoTime() : 0; try { if (insertTabletNode.isAligned()) { pointInserted += @@ -748,7 +765,15 @@ public class TsFileProcessor { } } throw e; + } finally { + try { + recordFlatDeviceWriteCost( + measureFlatDeviceWrite, FlatDeviceWriteStage.MEMTABLE_INSERT, flatDeviceStageStart); + } finally { + WritingMetrics.getInstance().finishFlatDeviceMemTableInsert(flatDeviceWriteCostTracker); + } } + flatDeviceStageStart = measureFlatDeviceWrite ? System.nanoTime() : 0; for (int i = start; i < end; i++) { if (results[i] == null || results[i].getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { @@ -768,9 +793,19 @@ public class TsFileProcessor { deviceEndOffset.right - deviceStartOffset); deviceStartOffset = deviceEndOffset.right; } + recordFlatDeviceWriteCost( + measureFlatDeviceWrite, + FlatDeviceWriteStage.RESOURCE_TIME_UPDATE, + flatDeviceStageStart); } } finally { - reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); + flatDeviceStageStart = measureFlatDeviceWrite ? System.nanoTime() : 0; + try { + reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); + } finally { + recordFlatDeviceWriteCost( + measureFlatDeviceWrite, FlatDeviceWriteStage.RAM_COST_RECONCILE, flatDeviceStageStart); + } } tsFileResource.updateProgressIndex(insertTabletNode.getProgressIndex()); @@ -1458,6 +1493,13 @@ public class TsFileProcessor { || includeTagColumns && columnCategories[index] == TsTableColumnCategory.TAG)); } + private static void recordFlatDeviceWriteCost( + boolean enabled, FlatDeviceWriteStage stage, long startNanos) { + if (enabled) { + WritingMetrics.getInstance().recordFlatDeviceWriteCost(stage, System.nanoTime() - startNanos); + } + } + private void reconcileAlignedTVListRamCost( AlignedTVListRamCostSnapshot snapshot, long estimatedMemTableIncrement) { if (snapshot == null) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/AlignedChunkPageReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/AlignedChunkPageReader.java new file mode 100644 index 00000000000..de3459ef23d --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/AlignedChunkPageReader.java @@ -0,0 +1,112 @@ +/* + * 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.iotdb.db.storageengine.dataregion.read.reader.chunk; + +import org.apache.tsfile.file.header.PageHeader; +import org.apache.tsfile.read.common.Chunk; +import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.read.reader.IChunkReader; +import org.apache.tsfile.read.reader.IPageReader; +import org.apache.tsfile.read.reader.chunk.AbstractChunkReader; +import org.apache.tsfile.read.reader.chunk.AlignedChunkReader; +import org.apache.tsfile.read.reader.chunk.TableChunkReader; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.function.LongConsumer; + +/** Selects compressed page windows before constructing TsFile readers. */ +class AlignedChunkPageReader extends AbstractChunkReader { + + AlignedChunkPageReader( + Chunk timeChunk, + List<Chunk> valueChunks, + Filter timeFilter, + LongConsumer filterRowsRecorder, + boolean ignoreAllNullRows, + int requestedPageIndex, + List<Integer> physicalPageIndexes) + throws IOException { + super(Long.MIN_VALUE, timeFilter, filterRowsRecorder); + ByteBuffer timeBuffer = timeChunk.getData().duplicate(); + List<ByteBuffer> valueBuffers = new ArrayList<>(valueChunks.size()); + for (Chunk valueChunk : valueChunks) { + valueBuffers.add(valueChunk == null ? null : valueChunk.getData().duplicate()); + } + for (int pageIndex = 0; timeBuffer.hasRemaining(); pageIndex++) { + boolean selected = requestedPageIndex < 0 || pageIndex == requestedPageIndex; + Chunk timePage = nextPage(timeChunk, timeBuffer, selected); + List<Chunk> valuePages = selected ? new ArrayList<>(valueChunks.size()) : null; + for (int columnIndex = 0; columnIndex < valueChunks.size(); columnIndex++) { + Chunk valuePage = + nextPage(valueChunks.get(columnIndex), valueBuffers.get(columnIndex), selected); + if (selected) { + valuePages.add(valuePage); + } + } + if (!selected) { + continue; + } + // Reuse TsFile's handling of encryption, null columns, filters, deletions and decoders. + // Each view contains exactly one encoded page, retaining its original header format. + IChunkReader reader = + ignoreAllNullRows + ? new AlignedChunkReader(timePage, valuePages, timeFilter, filterRowsRecorder) + : new TableChunkReader(timePage, valuePages, timeFilter, filterRowsRecorder); + for (IPageReader page : reader.loadPageReaderList()) { + pageReaderList.add(page); + if (physicalPageIndexes != null) { + physicalPageIndexes.add(pageIndex); + } + } + if (requestedPageIndex >= 0) { + break; + } + } + } + + private static Chunk nextPage(Chunk chunk, ByteBuffer buffer, boolean selected) { + if (chunk == null) { + return null; + } + int start = buffer.position(); + PageHeader pageHeader = + chunk.isSinglePageChunk() + ? PageHeader.deserializeFrom(buffer, chunk.getChunkStatistic()) + : PageHeader.deserializeFrom(buffer, chunk.getHeader().getDataType()); + buffer.position(buffer.position() + pageHeader.getCompressedSize()); + if (!selected) { + return null; + } + // Use duplicate rather than slice: LazyLoadPageData uses positions relative to array(), + // without adding arrayOffset(). No compressed data is copied or decompressed here. + ByteBuffer pageBuffer = buffer.duplicate(); + pageBuffer.limit(buffer.position()); + pageBuffer.position(start); + return new Chunk( + chunk.getHeader(), + pageBuffer, + chunk.getDeleteIntervalList(), + chunk.getChunkStatistic(), + chunk.getEncryptParam()); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java index ceee05ba807..d154ab4b312 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java @@ -78,6 +78,19 @@ public class DiskAlignedChunkLoader implements IChunkLoader { @Override public IChunkReader getChunkReader(IChunkMetadata chunkMetaData, Filter globalTimeFilter) throws IOException { + return getChunkReader(chunkMetaData, globalTimeFilter, -1, null); + } + + /** + * Loads one physical page, or all pages with their physical indexes when pageIndex is -1. Indexes + * are independent of page filtering and deletion. + */ + public IChunkReader getChunkReader( + IChunkMetadata chunkMetaData, + Filter globalTimeFilter, + int pageIndex, + List<Integer> physicalPageIndexes) + throws IOException { long t1 = System.nanoTime(); try { AbstractAlignedChunkMetadata alignedChunkMetadata = @@ -123,12 +136,25 @@ public class DiskAlignedChunkLoader implements IChunkLoader { LongConsumer filterRowsRecorder = this.context.getQueryStatistics()::addFilteredRowsOfPageLevel; long t2 = System.nanoTime(); - IChunkReader chunkReader = - ignoreAllNullRows - ? new AlignedChunkReader( - timeChunk, valueChunkList, globalTimeFilter, filterRowsRecorder) - : new TableChunkReader( - timeChunk, valueChunkList, globalTimeFilter, filterRowsRecorder); + IChunkReader chunkReader; + if (pageIndex >= 0 || physicalPageIndexes != null) { + chunkReader = + new AlignedChunkPageReader( + timeChunk, + valueChunkList, + globalTimeFilter, + filterRowsRecorder, + ignoreAllNullRows, + pageIndex, + physicalPageIndexes); + } else { + chunkReader = + ignoreAllNullRows + ? new AlignedChunkReader( + timeChunk, valueChunkList, globalTimeFilter, filterRowsRecorder) + : new TableChunkReader( + timeChunk, valueChunkList, globalTimeFilter, filterRowsRecorder); + } SERIES_SCAN_COST_METRIC_SET.recordSeriesScanCost( INIT_CHUNK_READER_ALIGNED_DISK, System.nanoTime() - t2); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueueTest.java new file mode 100644 index 00000000000..0a33559c312 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueueTest.java @@ -0,0 +1,197 @@ +/* + * 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.iotdb.db.queryengine.execution.operator.source; + +import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; +import org.apache.iotdb.db.storageengine.buffer.ChunkCache; +import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.DiskAlignedChunkLoader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.FlatDeviceDescPriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.FlatDevicePriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.PriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; +import org.apache.iotdb.db.utils.EncryptDBUtils; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TableDeviceChunkMetadata; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.filter.factory.TimeFilterApi; +import org.apache.tsfile.read.reader.IPageReader; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +public class FlatDevicePageReaderQueueTest { + private static final List<TSDataType> TYPES = List.of(TSDataType.STRING, TSDataType.INT64); + private static final IDeviceID DEVICE = + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"flat_table"}); + @Rule public TemporaryFolder folder = new TemporaryFolder(); + private TsFileResource resource; + private TableDeviceChunkMetadata metadata; + private FragmentInstanceContext context; + + @Before + public void setUp() throws Exception { + context = FragmentInstanceContext.createFragmentInstanceContextForCompaction(0); + File file = new File(folder.newFolder("sequence", "root.test", "0", "0"), "1-1-0-0.tsfile"); + resource = new TsFileResource(file); + FlatDeviceAlignedChunkWriter chunkWriter = + new FlatDeviceAlignedChunkWriter( + List.of( + new MeasurementSchema("tag", TSDataType.STRING), + new MeasurementSchema("v", TSDataType.INT64)), + EncryptDBUtils.getDefaultFirstEncryptParam()); + long[][] pages = {{0, 1}, {100, 101}, {10, 11}}; + try (TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer.startChunkGroup(DEVICE); + for (int pageIndex = 0; pageIndex < pages.length; pageIndex++) { + for (long time : pages[pageIndex]) { + chunkWriter.write( + time, + new Binary(String.valueOf((char) ('a' + pageIndex)), TSFileConfig.STRING_CHARSET), + false); + chunkWriter.write(time, time * 10, false); + chunkWriter.write(time); + } + chunkWriter.sealCurrentPage(); + } + chunkWriter.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.endFile(); + } + resource.setStatusForTest(TsFileResourceStatus.NORMAL); + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath())) { + AbstractAlignedChunkMetadata original = + (AbstractAlignedChunkMetadata) + reader.getIChunkMetadataList(DEVICE, Set.of("tag", "v"), null).get(0).get(0); + metadata = + new TableDeviceChunkMetadata( + original.getTimeChunkMetadata(), original.getValueChunkMetadataList()); + } + metadata.setChunkLoader(new DiskAlignedChunkLoader(context, resource, false)); + } + + @After + public void tearDown() throws Exception { + if (resource != null) { + FileReaderManager.getInstance().closeFileAndRemoveReader(resource.getTsFileID()); + } + ChunkCache.getInstance().clear(); + } + + @Test + public void testPhysicalPageLoadAfterSkippedPages() throws Exception { + metadata.getTimeChunkMetadata().insertIntoSortedDeletions(new TimeRange(100, 101)); + metadata.setDataTypeModifiedAndCannotUseStatistics(true); + List<Integer> indexes = new ArrayList<>(); + List<IPageReader> pages = + FileLoaderUtils.loadPageReaderList( + metadata, TimeFilterApi.gtEq(5), TYPES, context, indexes); + assertEquals(List.of(2), indexes); + assertEquals(1, pages.size()); + assertNull(FileLoaderUtils.loadPageReader(metadata, TimeFilterApi.gtEq(5), context, 0)); + assertNull(FileLoaderUtils.loadPageReader(metadata, TimeFilterApi.gtEq(5), context, 1)); + assertNull(FileLoaderUtils.loadPageReader(metadata, null, context, 3)); + IPageReader page = + FileLoaderUtils.loadPageReader(metadata, TimeFilterApi.gtEq(5), context, indexes.get(0)); + assertTrue(page.isModified()); + page.initTsBlockBuilder(TYPES); + TsBlock block = page.getAllSatisfiedData(); + assertEquals(2, block.getPositionCount()); + assertEquals("c", block.getColumn(0).getBinary(0).toString()); + assertEquals(100, block.getColumn(1).getLong(0)); + } + + @Test + public void testQueueTagTimeOrderBothDirections() throws Exception { + for (boolean ascending : new boolean[] {true, false}) { + PriorityMergeReader merge; + if (ascending) { + merge = new FlatDevicePriorityMergeReader(TYPES, List.of(0)); + } else { + FlatDeviceDescPriorityMergeReader descending = new FlatDeviceDescPriorityMergeReader(); + descending.configureRowKeyColumns(TYPES, List.of(0)); + merge = descending; + } + try { + FlatDevicePageReaderQueue queue = + new FlatDevicePageReaderQueue(context, TYPES, List.of(0), ascending, null, merge); + queue.addChunk(metadata, null); + List<String> rows = new ArrayList<>(); + while (queue.hasNext()) { + TimeValuePair row = merge.nextTimeValuePair(); + rows.add(row.getValue().getVector()[0].getBinary().toString() + ":" + row.getTimestamp()); + assertEquals(row.getTimestamp() * 10, row.getValue().getVector()[1].getLong()); + } + assertEquals( + ascending + ? List.of("a:0", "a:1", "b:100", "b:101", "c:10", "c:11") + : List.of("c:11", "c:10", "b:101", "b:100", "a:1", "a:0"), + rows); + } finally { + merge.close(); + } + } + } + + @Test + public void testQueueFilteringAndDeletion() throws Exception { + metadata.getTimeChunkMetadata().insertIntoSortedDeletions(new TimeRange(100, 101)); + metadata.getValueChunkMetadataList().get(1).insertIntoSortedDeletions(new TimeRange(11, 11)); + PriorityMergeReader merge = new FlatDevicePriorityMergeReader(TYPES, List.of(0)); + try { + FlatDevicePageReaderQueue queue = + new FlatDevicePageReaderQueue( + context, TYPES, List.of(0), true, TimeFilterApi.gtEq(5), merge); + queue.addChunk(metadata, TimeFilterApi.gtEq(11)); + assertTrue(queue.hasNext()); + TimeValuePair row = merge.nextTimeValuePair(); + assertEquals(11, row.getTimestamp()); + assertEquals("c", row.getValue().getVector()[0].getBinary().toString()); + assertNull(row.getValue().getVector()[1]); + assertFalse(queue.hasNext()); + } finally { + merge.close(); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 7dd344a2b22..598a67d83da 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -1488,11 +1488,13 @@ public class TsFileProcessorTest { assertTrue(processor.getTsFileResource().isClosed()); } + // All insertion APIs, flush, and reload retain only table-level bounds for flat sequence files. @Test public void testFlatDeviceSequenceResourceTimes() throws Exception { assertFlatDeviceResourceTimes(true); } + // The same table-only index must support unsequence file filtering, including null TAGs. @Test public void testFlatDeviceUnsequenceResourceTimes() throws Exception { assertFlatDeviceResourceTimes(false); @@ -1522,23 +1524,25 @@ public class TsFileProcessorTest { table, Arrays.asList(schemas), List.of(ColumnCategory.TAG, ColumnCategory.FIELD)), true); - // Different logical devices share one MemTable group, but keep independent resource ranges - // for all three insertion APIs. Tablet ranges exclude the first and last input rows. + // Logical devices share one table-level resource range across all three insertion APIs. + // Tablet ranges exclude the first and last input rows. RelationalInsertRowNode first = flatDeviceRow(10, "d1", schemas); RelationalInsertRowNode second = flatDeviceRow(30, "d2", schemas); processor.insert(first, new long[5]); processor.insert(second, new long[5]); TsFileResource resource = processor.getTsFileResource(); - assertResourceTimes(resource, first.getDeviceID(), 10, 10); - assertResourceTimes(resource, second.getDeviceID(), 30, 30); + IDeviceID physicalDevice = first.getMemTableDeviceID(first.getDeviceID()); + assertEquals(Set.of(physicalDevice), resource.getDevices()); + assertResourceTimes(resource, physicalDevice, 10, 30); + assertFalse(resource.getStartTime(first.getDeviceID()).isPresent()); + assertFalse(resource.getStartTime(second.getDeviceID()).isPresent()); processor.insertRows( new RelationalInsertRowsNode( new PlanNodeId("rows"), List.of(0, 1), List.of(flatDeviceRow(20, "d1", schemas), flatDeviceRow(40, "d2", schemas))), new long[5]); - assertResourceTimes(resource, first.getDeviceID(), 10, 20); - assertResourceTimes(resource, second.getDeviceID(), 30, 40); + assertResourceTimes(resource, physicalDevice, 10, 40); RelationalInsertTabletNode tablet = new RelationalInsertTabletNode( new PlanNodeId("tablet"), @@ -1560,26 +1564,21 @@ public class TsFileProcessorTest { tablet.setFlatDevice(true); processor.insertTablet( tablet, List.of(new int[] {1, 3}, new int[] {3, 5}), new TSStatus[6], false, new long[5]); - assertEquals(Set.of(first.getDeviceID(), second.getDeviceID()), resource.getDevices()); - assertResourceTimes(resource, first.getDeviceID(), 10, 60); - assertResourceTimes(resource, second.getDeviceID(), 30, 80); + assertEquals(Set.of(physicalDevice), resource.getDevices()); + assertResourceTimes(resource, physicalDevice, 10, 80); assertEquals(10L, resource.getTableTime("flat_table").get().left.longValue()); assertEquals(80L, resource.getTableTime("flat_table").get().right.longValue()); - IDeviceID physicalDevice = first.getMemTableDeviceID(first.getDeviceID()); assertEquals(Set.of(physicalDevice), processor.getWorkMemTable().getMemTableMap().keySet()); assertTrue(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(70), sequence, false)); - // Flush and close the file once. An all-null logical ID equals the physical table-name ID, - // but must retain its own range instead of the entire physical group's range. + // An all-null logical ID equals the table-name ID. It must extend the table range, + // never narrow it to the null-TAG device alone. processor.insert(flatDeviceRow(90, null, schemas), new long[5]); processor.insert(flatDeviceRow(100, "d2", schemas), new long[5]); processor.asyncClose().get(30, TimeUnit.SECONDS); assertTrue(resource.isClosed()); - assertEquals( - Set.of(first.getDeviceID(), second.getDeviceID(), physicalDevice), resource.getDevices()); - assertResourceTimes(resource, first.getDeviceID(), 10, 60); - assertResourceTimes(resource, second.getDeviceID(), 30, 100); - assertResourceTimes(resource, physicalDevice, 90, 90); + assertEquals(Set.of(physicalDevice), resource.getDevices()); + assertResourceTimes(resource, physicalDevice, 10, 100); assertEquals(100L, flushedTimes.get(physicalDevice).longValue()); assertTrue(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(70), sequence, false)); assertFalse(resource.isSatisfied(physicalDevice, TimeFilterApi.eq(200), sequence, false)); @@ -1589,12 +1588,12 @@ public class TsFileProcessorTest { TsFileResource restored = new TsFileResource(new File(filePath)); restored.deserialize(); assertEquals(resource.getDevices(), restored.getDevices()); - assertResourceTimes(restored, first.getDeviceID(), 10, 60); - assertResourceTimes(restored, second.getDeviceID(), 30, 100); - assertResourceTimes(restored, physicalDevice, 90, 90); + assertResourceTimes(restored, physicalDevice, 10, 100); + assertFalse(restored.getStartTime(first.getDeviceID()).isPresent()); + assertFalse(restored.getStartTime(second.getDeviceID()).isPresent()); - // Read through the query scanner, which uses the physical ID even though the resource has - // logical ranges. The null-TAG device's [90, 90] range must not prune d2's row at time 70. + // Read through the query scanner using the table-level index. The null-TAG row at 90 + // must not prune d2's row at time 70. SeriesScanOptions.Builder scanOptions = new SeriesScanOptions.Builder(); scanOptions.withAllSensors(Set.of("tag1", "s1")); scanOptions.withGlobalTimeFilter(TimeFilterApi.eq(70)); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/AlignedChunkPageReaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/AlignedChunkPageReaderTest.java new file mode 100644 index 00000000000..c8137c04e5b --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/AlignedChunkPageReaderTest.java @@ -0,0 +1,230 @@ +/* + * 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.iotdb.db.storageengine.dataregion.read.reader.chunk; + +import org.apache.tsfile.encoding.encoder.PlainEncoder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.header.ChunkHeader; +import org.apache.tsfile.file.header.PageHeader; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.common.Chunk; +import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.read.filter.factory.TimeFilterApi; +import org.apache.tsfile.read.reader.IPageReader; +import org.apache.tsfile.read.reader.chunk.TableChunkReader; +import org.apache.tsfile.write.chunk.TimeChunkWriter; +import org.apache.tsfile.write.chunk.ValueChunkWriter; +import org.junit.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class AlignedChunkPageReaderTest { + private static final List<TSDataType> TYPES = List.of(TSDataType.INT64); + + @Test + public void testFirstMiddleLastPageAndUnchangedBuffers() throws IOException { + for (CompressionType compression : + List.of(CompressionType.UNCOMPRESSED, CompressionType.LZ4, CompressionType.SNAPPY)) { + Chunk[] chunks = chunks(compression, new long[][] {{10, 11}, {20, 21}, {1, 2}}); + for (int pageIndex : new int[] {2, 0, 1, 2}) { + long first = new long[] {10, 20, 1}[pageIndex]; + TsBlock block = read(chunks, null, pageIndex).getAllSatisfiedData(); + assertEquals(2, block.getPositionCount()); + assertEquals(first, block.getTimeByIndex(0)); + assertEquals(first * 10, block.getColumn(0).getLong(0)); + assertEquals(first + 1, block.getTimeByIndex(1)); + assertEquals((first + 1) * 10, block.getColumn(0).getLong(1)); + } + for (Chunk chunk : chunks) { + assertEquals(0, chunk.getData().position()); + assertEquals(chunk.getData().capacity(), chunk.getData().limit()); + } + } + } + + @Test + public void testPhysicalIndexesAfterFilteringAndDeletion() throws IOException { + Chunk[] chunks = chunks(CompressionType.LZ4, new long[][] {{0, 1}, {10, 11}, {20, 21}}); + chunks[0].setDeleteIntervalList(List.of(new TimeRange(10, 11))); + List<Integer> indexes = new ArrayList<>(); + List<IPageReader> pages = + reader(chunks, TimeFilterApi.gtEq(5), -1, indexes).loadPageReaderList(); + assertEquals(List.of(2), indexes); + assertEquals(1, pages.size()); + assertTrue(reader(chunks, TimeFilterApi.gtEq(5), 0, null).loadPageReaderList().isEmpty()); + assertTrue(reader(chunks, TimeFilterApi.gtEq(5), 1, null).loadPageReaderList().isEmpty()); + TsBlock block = read(chunks, TimeFilterApi.gtEq(5), indexes.get(0)).getAllSatisfiedData(); + assertEquals(20, block.getTimeByIndex(0)); + assertEquals(200, block.getColumn(0).getLong(0)); + } + + @Test + public void testPartialTimeAndValueDeletionAndRecordFilter() throws IOException { + Chunk[] chunks = chunks(CompressionType.LZ4, new long[][] {{0, 1}, {10, 11, 12, 13}}); + chunks[0].setDeleteIntervalList(List.of(new TimeRange(10, 10))); + chunks[1].setDeleteIntervalList(List.of(new TimeRange(11, 11))); + IPageReader page = read(chunks, TimeFilterApi.ltEq(12), 1); + assertTrue(page.isModified()); + page.addRecordFilter(TimeFilterApi.gtEq(11)); + TsBlock block = page.getAllSatisfiedData(); + assertEquals(2, block.getPositionCount()); + assertEquals(11, block.getTimeByIndex(0)); + assertTrue(block.getColumn(0).isNull(0)); + assertEquals(120, block.getColumn(0).getLong(1)); + } + + @Test + public void testSinglePageAndMissingPage() throws IOException { + Chunk[] chunks = chunks(CompressionType.LZ4, new long[][] {{1, 2, 3}}); + List<Integer> indexes = new ArrayList<>(); + assertEquals(1, reader(chunks, null, -1, indexes).loadPageReaderList().size()); + assertEquals(List.of(0), indexes); + TsBlock block = read(chunks, TimeFilterApi.gt(1), 0).getAllSatisfiedData(); + assertEquals(2, block.getPositionCount()); + assertEquals(2, block.getTimeByIndex(0)); + assertTrue(reader(chunks, null, 1, null).loadPageReaderList().isEmpty()); + } + + @Test + public void testMissingColumnEmptyPageAndAllNullTableRows() throws IOException { + Chunk[] chunks = chunks(CompressionType.LZ4, new long[][] {{1, 2}, {3, 4}}, true); + for (int pageIndex = 0; pageIndex < 2; pageIndex++) { + AlignedChunkPageReader reader = + new AlignedChunkPageReader( + chunks[0], Arrays.asList(chunks[1], null), null, null, false, pageIndex, null); + IPageReader page = reader.loadPageReaderList().get(0); + page.initTsBlockBuilder(List.of(TSDataType.INT64, TSDataType.INT64)); + TsBlock block = page.getAllSatisfiedData(); + assertEquals(2, block.getPositionCount()); + assertTrue(block.getColumn(1).isNull(0)); + assertEquals(pageIndex == 0, block.getColumn(0).isNull(0)); + } + } + + @Test + public void testDoesNotDecompressOtherPages() throws IOException { + Chunk[] chunks = chunks(CompressionType.LZ4, new long[][] {{1, 2}, {3, 4}, {5, 6}}); + // Keep headers intact but make both surrounding time-page bodies invalid LZ4 streams. + // Full-list construction must fail; loading the middle page must not inspect either body. + ByteBuffer buffer = chunks[0].getData().duplicate(); + for (int pageIndex = 0; pageIndex < 3; pageIndex++) { + PageHeader header = PageHeader.deserializeFrom(buffer, TSDataType.VECTOR); + if (pageIndex != 1) { + Arrays.fill( + buffer.array(), + buffer.position(), + buffer.position() + header.getCompressedSize(), + (byte) 0xFF); + } + buffer.position(buffer.position() + header.getCompressedSize()); + } + TsBlock block = read(chunks, null, 1).getAllSatisfiedData(); + assertEquals(3, block.getTimeByIndex(0)); + assertEquals(40, block.getColumn(0).getLong(1)); + assertThrows(Exception.class, () -> new TableChunkReader(chunks[0], List.of(chunks[1]), null)); + } + + @Test + public void testDataTypeModification() throws IOException { + Chunk[] chunks = chunks(CompressionType.LZ4, new long[][] {{1, 2}, {3, 4}}); + AlignedChunkPageReader reader = reader(chunks, null, 1, null); + reader.markDataTypeModifiedAndCannotUseStatistics(); + assertTrue(reader.loadPageReaderList().get(0).isModified()); + } + + private static IPageReader read(Chunk[] chunks, Filter filter, int pageIndex) throws IOException { + List<IPageReader> pages = reader(chunks, filter, pageIndex, null).loadPageReaderList(); + assertEquals(1, pages.size()); + IPageReader page = pages.get(0); + page.initTsBlockBuilder(TYPES); + return page; + } + + private static AlignedChunkPageReader reader( + Chunk[] chunks, Filter filter, int pageIndex, List<Integer> indexes) throws IOException { + return new AlignedChunkPageReader( + chunks[0], List.of(chunks[1]), filter, null, false, pageIndex, indexes); + } + + private static Chunk[] chunks(CompressionType compression, long[][] pages) throws IOException { + return chunks(compression, pages, false); + } + + private static Chunk[] chunks(CompressionType compression, long[][] pages, boolean firstPageNull) + throws IOException { + TimeChunkWriter times = + new TimeChunkWriter( + "", compression, TSEncoding.PLAIN, new PlainEncoder(TSDataType.INT64, 0)); + ValueChunkWriter values = + new ValueChunkWriter( + "v", + compression, + TSDataType.INT64, + TSEncoding.PLAIN, + new PlainEncoder(TSDataType.INT64, 0)); + for (int pageIndex = 0; pageIndex < pages.length; pageIndex++) { + for (long time : pages[pageIndex]) { + times.write(time); + values.write(time, time * 10, firstPageNull && pageIndex == 0); + } + times.sealCurrentPage(); + values.sealCurrentPage(); + } + byte[] timeBytes = times.getPageBuffer().toByteArray(); + byte[] valueBytes = values.getPageBuffer().toByteArray(); + return new Chunk[] { + new Chunk( + new ChunkHeader( + "", + timeBytes.length, + TSDataType.VECTOR, + compression, + TSEncoding.PLAIN, + pages.length, + 0x80), + ByteBuffer.wrap(timeBytes), + null, + times.getStatistics()), + new Chunk( + new ChunkHeader( + "v", + valueBytes.length, + TSDataType.INT64, + compression, + TSEncoding.PLAIN, + pages.length, + 0x40), + ByteBuffer.wrap(valueBytes), + null, + values.getStatistics()) + }; + } +}
