This is an automated email from the ASF dual-hosted git repository. Wei-hao-Li pushed a commit to branch deviceEntrySpill-dev in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 701f7b07da27c793c9cdca030a318ba11c8a98bf Author: Weihao Li <[email protected]> AuthorDate: Wed Aug 12 09:26:19 2026 +0800 extracted model-1 Signed-off-by: Weihao Li <[email protected]> --- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 ++ .../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 12 ++ .../db/queryengine/common/MPPQueryContext.java | 35 ++++ .../plan/relational/metadata/DeviceEntry.java | 13 ++ .../plan/relational/metadata/Metadata.java | 8 +- .../relational/metadata/TableMetadataImpl.java | 62 +----- .../metadata/fetcher/TableDeviceSchemaFetcher.java | 211 +++++++++++++++++++-- .../spill/AbstractDeviceEntryMaterializer.java | 169 +++++++++++++++++ .../metadata/spill/DeviceEntryDataSet.java | 46 +++++ .../metadata/spill/DeviceEntryDataSetResult.java | 46 +++++ .../metadata/spill/DeviceEntryDiskSpiller.java | 132 +++++++++++++ .../spill/DeviceEntryFileSpillerReader.java | 143 ++++++++++++++ .../metadata/spill/DeviceEntryIOContext.java | 60 ++++++ .../metadata/spill/DeviceEntryMaterializer.java | 132 +++++++++++++ .../metadata/spill/DeviceEntryReader.java | 34 ++++ .../metadata/spill/DeviceEntrySpillManager.java | 189 ++++++++++++++++++ .../metadata/spill/InMemoryDeviceEntryDataSet.java | 76 ++++++++ .../metadata/spill/SpilledDeviceEntryDataSet.java | 85 +++++++++ .../optimizations/PushPredicateIntoTableScan.java | 26 ++- .../FragmentInstanceStatisticsDrawer.java | 19 ++ .../FragmentInstanceStatisticsJsonDrawer.java | 6 + .../statistics/QueryPlanStatistics.java | 37 ++++ .../spill/DeviceEntryMaterializerTest.java | 153 +++++++++++++++ .../FragmentInstanceStatisticsJsonDrawerTest.java | 8 + .../conf/iotdb-system.properties.template | 4 + 25 files changed, 1644 insertions(+), 73 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 1d726366ddd..19bfa1bf9e6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -256,6 +256,9 @@ public class IoTDBConfig { private String queryDir = IoTDBConstant.DN_DEFAULT_DATA_DIR + File.separator + IoTDBConstant.QUERY_FOLDER_NAME; + /** Maximum DeviceEntry bytes kept in memory before a table-query spill. */ + private long tableQueryDeviceEntryBatchSizeInBytes; + /** External lib directory, stores user-uploaded JAR files */ private String extDir = IoTDBConstant.EXT_FOLDER_NAME; @@ -1789,6 +1792,14 @@ public class IoTDBConfig { this.queryDir = queryDir; } + public long getTableQueryDeviceEntryBatchSizeInBytes() { + return tableQueryDeviceEntryBatchSizeInBytes; + } + + public void setTableQueryDeviceEntryBatchSizeInBytes(long tableQueryDeviceEntryBatchSizeInBytes) { + this.tableQueryDeviceEntryBatchSizeInBytes = tableQueryDeviceEntryBatchSizeInBytes; + } + public String getRatisDataRegionSnapshotDir() { return ratisDataRegionSnapshotDir; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 15d2b7d003d..ed46c49e411 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -352,6 +352,18 @@ public class IoTDBDescriptor { conf.setQueryDir( FilePathUtils.regularizePath(conf.getSystemDir() + IoTDBConstant.QUERY_FOLDER_NAME)); + long deviceEntryBatchSize = + Long.parseLong( + properties.getProperty( + "table_query_device_entry_batch_size_in_bytes", + Long.toString(conf.getTableQueryDeviceEntryBatchSizeInBytes()))); + if (deviceEntryBatchSize <= 0) { + deviceEntryBatchSize = + memoryConfig.getOperatorsMemoryManager().getTotalMemorySizeInBytes() + / memoryConfig.getQueryThreadCount() + / 4; + } + conf.setTableQueryDeviceEntryBatchSizeInBytes(deviceEntryBatchSize); String[] defaultTierDirs = new String[conf.getTierDataDirs().length]; for (int i = 0; i < defaultTierDirs.length; ++i) { defaultTierDirs[i] = String.join(",", conf.getTierDataDirs()[i]); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java index 495ebc42b65..f7ff41469ad 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java @@ -528,6 +528,41 @@ public class MPPQueryContext implements IAuditEntity { return queryPlanStatistics.getDispatchCost(); } + public void recordDeviceEntryDiskIO(long bytes, long timeCost) { + getOrCreateQueryPlanStatistics().recordDeviceEntryDiskIO(bytes, timeCost); + } + + public void recordDeviceEntrySegment() { + getOrCreateQueryPlanStatistics().recordDeviceEntrySegment(); + } + + public void recordDeviceEntrySortedRun() { + getOrCreateQueryPlanStatistics().recordDeviceEntrySortedRun(); + } + + public long getDiskIOSizeForDeviceEntry() { + return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDiskIOSizeForDeviceEntry(); + } + + public long getDiskIOTimeCostForDeviceEntry() { + return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDiskIOTimeCostForDeviceEntry(); + } + + public long getDeviceEntrySegmentCount() { + return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntrySegmentCount(); + } + + public long getDeviceEntrySortedRunCount() { + return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntrySortedRunCount(); + } + + private QueryPlanStatistics getOrCreateQueryPlanStatistics() { + if (queryPlanStatistics == null) { + queryPlanStatistics = new QueryPlanStatistics(); + } + return queryPlanStatistics; + } + public void setAnalyzeCost(long analyzeCost) { if (queryPlanStatistics == null) { queryPlanStatistics = new QueryPlanStatistics(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java index 4701e0e9c62..1b5e6ab693b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java @@ -29,6 +29,7 @@ import org.apache.tsfile.utils.Accountable; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.RamUsageEstimator; +import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; @@ -98,6 +99,14 @@ public abstract class DeviceEntry implements Accountable { stream); } + public byte[] serializeToBytes() throws IOException { + final ByteArrayOutputStream byteStream = new ByteArrayOutputStream(); + try (DataOutputStream output = new DataOutputStream(byteStream)) { + serialize(output); + } + return byteStream.toByteArray(); + } + public static DeviceEntry deserialize(final ByteBuffer byteBuffer) { final IDeviceID iDeviceID = StringArrayDeviceID.deserialize(byteBuffer); int size = readInt(byteBuffer); @@ -109,6 +118,10 @@ public abstract class DeviceEntry implements Accountable { return constructDeviceEntry(iDeviceID, attributeColumnValues, readInt(byteBuffer)); } + public static DeviceEntry deserialize(final byte[] bytes) { + return deserialize(ByteBuffer.wrap(bytes)); + } + public static void serializeBinary(final ByteBuffer byteBuffer, final Binary binary) { if (binary == null) { write(-1, byteBuffer); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java index 8b7f0f32656..5c219440c07 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.partition.DataPartition; import org.apache.iotdb.commons.partition.DataPartitionQueryParam; import org.apache.iotdb.commons.partition.SchemaPartition; import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.relational.function.ITableFunctionFactory; import org.apache.iotdb.commons.queryengine.plan.relational.function.OperatorType; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; @@ -37,13 +38,13 @@ import org.apache.iotdb.commons.udf.builtin.relational.TableBuiltinWindowFunctio import org.apache.iotdb.db.exception.load.LoadAnalyzeTableColumnDisorderException; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.common.type.Type; import java.util.List; -import java.util.Map; import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; @@ -86,11 +87,12 @@ public interface Metadata extends ITypeMetadata, ITableFunctionFactory { * index scanning * @param attributeColumns attribute column names */ - Map<String, List<DeviceEntry>> indexScan( + DeviceEntryDataSetResult indexScan( final QualifiedObjectName tableName, final List<Expression> expressionList, final List<String> attributeColumns, - final MPPQueryContext context); + final MPPQueryContext context, + final PlanNodeId planNodeId); /** * This method is used for table column validation and should be invoked before device validation. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java index 9e736a35c58..30ea4a63645 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.partition.DataPartition; import org.apache.iotdb.commons.partition.DataPartitionQueryParam; import org.apache.iotdb.commons.partition.SchemaPartition; import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.relational.function.OperatorType; import org.apache.iotdb.commons.queryengine.plan.relational.function.TableFunctionFactory; import org.apache.iotdb.commons.queryengine.plan.relational.function.arithmetic.AdditionResolver; @@ -57,6 +58,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.function.DataNodeTableBui import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableDeviceSchemaFetcher; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableDeviceSchemaValidator; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.schemaengine.table.ITableCache; @@ -75,7 +77,6 @@ import org.apache.tsfile.read.common.type.TypeFactory; import java.util.Collections; import java.util.List; import java.util.Locale; -import java.util.Map; import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -1439,24 +1440,6 @@ public class TableMetadataImpl implements Metadata { functionName)); } break; - case SqlConstant.IRATE: - validateRateFunctionArguments( - functionName, - argumentTypes, - 2, - DataNodeQueryMessages - .EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_2_ARGUMENTS_VALUE_TIME_E2F55C08); - break; - case SqlConstant.RATE: - case SqlConstant.INCREASE: - case SqlConstant.DELTA: - validateRateFunctionArguments( - functionName, - argumentTypes, - 4, - DataNodeQueryMessages - .EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_4_ARGUMENTS_VALUE_TIME_WINDOW_START_WINDOW_END_FBEC794B); - break; case SqlConstant.COUNT: break; default: @@ -1498,10 +1481,6 @@ public class TableMetadataImpl implements Metadata { case SqlConstant.REGR_INTERCEPT: case SqlConstant.SKEWNESS: case SqlConstant.KURTOSIS: - case SqlConstant.RATE: - case SqlConstant.INCREASE: - case SqlConstant.IRATE: - case SqlConstant.DELTA: return DOUBLE; case SqlConstant.APPROX_MOST_FREQUENT: return STRING; @@ -1631,33 +1610,6 @@ public class TableMetadataImpl implements Metadata { throw new SemanticException(DataNodeQueryMessages.UNKNOWN_FUNCTION + functionName); } - private static void validateRateFunctionArguments( - String functionName, - List<? extends Type> argumentTypes, - int expectedArgumentCount, - String argumentCountError) { - if (argumentTypes.size() != expectedArgumentCount) { - throw new SemanticException(String.format(argumentCountError, functionName)); - } - if (!CommonMetadataUtils.isSupportedMathNumericType(argumentTypes.get(0))) { - throw new SemanticException( - String.format( - DataNodeQueryMessages - .EXCEPTION_AGGREGATE_FUNCTION_ARG_ONLY_SUPPORTS_INT32_INT64_FLOAT_AND_DOUBLE_AS_THE_FIRST_ARGUMENT_8D201434, - functionName)); - } - for (int i = 1; i < argumentTypes.size(); i++) { - Type argumentType = argumentTypes.get(i); - if (!INT64.equals(argumentType) && !TIMESTAMP.equals(argumentType)) { - throw new SemanticException( - String.format( - DataNodeQueryMessages - .EXCEPTION_THE_TIME_ARGUMENTS_OF_AGGREGATE_FUNCTION_ARG_SHOULD_BE_TIMESTAMP_OR_INT64_TYPE_9C736DE3, - functionName)); - } - } - } - @Override public boolean isAggregationFunction( final SessionInfo session, final String functionName, final AccessControl accessControl) { @@ -1677,18 +1629,20 @@ public class TableMetadataImpl implements Metadata { } @Override - public Map<String, List<DeviceEntry>> indexScan( + public DeviceEntryDataSetResult indexScan( final QualifiedObjectName tableName, final List<Expression> expressionList, final List<String> attributeColumns, - final MPPQueryContext context) { + final MPPQueryContext context, + final PlanNodeId planNodeId) { return TableDeviceSchemaFetcher.getInstance() - .fetchDeviceSchemaForDataQuery( + .fetchDeviceSchemaForDataQueryAsDataSet( tableName.getDatabaseName(), tableName.getObjectName(), expressionList, attributeColumns, - context); + context, + planNodeId); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java index 024b9ef44ac..57f20ba6d45 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java @@ -22,6 +22,8 @@ package org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher; import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.QueryTimeoutException; +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression; import org.apache.iotdb.commons.schema.column.ColumnHeader; import org.apache.iotdb.commons.schema.filter.SchemaFilter; @@ -48,6 +50,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.De import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.IDeviceSchema; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceNormalSchema; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryMaterializer; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AbstractTraverseDevice; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FetchDevice; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDevice; @@ -64,6 +69,8 @@ import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.Pair; +import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -238,7 +245,14 @@ public class TableDeviceSchemaFetcher { mayContainDuplicateDevice, false)) { fetchMissingDeviceSchemaForQuery( - database, tableInstance, attributeColumns, statement, deviceEntryMap, queryContext); + database, + tableInstance, + attributeColumns, + statement, + deviceEntryMap, + null, + queryContext, + null); } // TODO table metadata: implement deduplicate during schemaRegion execution @@ -253,6 +267,132 @@ public class TableDeviceSchemaFetcher { : deviceEntryMap; } + public DeviceEntryDataSetResult fetchDeviceSchemaForDataQueryAsDataSet( + final String database, + final String table, + final List<Expression> expressionList, + final List<String> attributeColumns, + final MPPQueryContext queryContext, + final PlanNodeId planNodeId) { + final TsTable tableInstance = DataNodeTableCache.getInstance().getTable(database, table); + if (TreeViewSchema.isTreeViewTable(tableInstance)) { + final Map<String, List<DeviceEntry>> deviceEntryMap = new HashMap<>(); + final AtomicBoolean mayContainDuplicateDevice = new AtomicBoolean(false); + // pass by reference + final AtomicBoolean containsNonAlignedDevice = new AtomicBoolean(false); + final ShowDevice statement = new ShowDevice(database, table); + try (DeviceEntryMaterializer materializer = + new DeviceEntryMaterializer( + queryContext.getQueryId().getId(), + planNodeId, + CONFIG.getTableQueryDeviceEntryBatchSizeInBytes(), + true, + queryContext)) { + final boolean needRemoteFetch = + parseFilter4TraverseDevice( + tableInstance, + expressionList, + statement, + deviceEntryMap, + attributeColumns, + queryContext, + mayContainDuplicateDevice, + false); + for (List<DeviceEntry> entries : deviceEntryMap.values()) { + for (DeviceEntry entry : entries) { + appendToMaterializer(materializer, entry, queryContext, true); + if (entry instanceof NonAlignedDeviceEntry) { + containsNonAlignedDevice.set(true); + } + } + entries.clear(); + } + if (needRemoteFetch) { + fetchMissingDeviceSchemaForQuery( + database, + tableInstance, + attributeColumns, + statement, + deviceEntryMap, + materializer, + queryContext, + containsNonAlignedDevice); + } + if (deviceEntryMap.size() > 1) { + throw new SemanticException( + DataNodeQueryMessages.TREE_DEVICE_VIEW_WITH_MULTIPLE_DATABASES + + deviceEntryMap.keySet() + + DataNodeQueryMessages.IS_UNSUPPORTED_YET); + } + final String resultDatabase = + deviceEntryMap.isEmpty() ? null : deviceEntryMap.keySet().iterator().next(); + return new DeviceEntryDataSetResult( + resultDatabase, materializer.finish(), containsNonAlignedDevice.get()); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + final Map<String, List<DeviceEntry>> cachedEntries = new HashMap<>(); + cachedEntries.put(database, new ArrayList<>()); + final AtomicBoolean mayContainDuplicateDevice = new AtomicBoolean(false); + final ShowDevice statement = new ShowDevice(database, table); + final boolean needRemoteFetch = + parseFilter4TraverseDevice( + tableInstance, + expressionList, + statement, + cachedEntries, + attributeColumns, + queryContext, + mayContainDuplicateDevice, + false); + + if (mayContainDuplicateDevice.get()) { + if (needRemoteFetch) { + fetchMissingDeviceSchemaForQuery( + database, + tableInstance, + attributeColumns, + statement, + cachedEntries, + null, + queryContext, + null); + } + cachedEntries.put( + database, new ArrayList<>(new LinkedHashSet<>(cachedEntries.get(database)))); + } + + try (DeviceEntryMaterializer materializer = + new DeviceEntryMaterializer( + queryContext.getQueryId().getId(), + planNodeId, + CONFIG.getTableQueryDeviceEntryBatchSizeInBytes(), + true, + queryContext)) { + for (DeviceEntry entry : cachedEntries.get(database)) { + appendToMaterializer(materializer, entry, queryContext, true); + } + cachedEntries.get(database).clear(); + if (needRemoteFetch && !mayContainDuplicateDevice.get()) { + fetchMissingDeviceSchemaForQuery( + database, + tableInstance, + attributeColumns, + statement, + cachedEntries, + materializer, + queryContext, + null); + } + final DeviceEntryDataSet dataSet = materializer.finish(); + return new DeviceEntryDataSetResult(database, dataSet, false); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + // Used by show/count device and update device. // Update / Delete device will not access cache public boolean parseFilter4TraverseDevice( @@ -495,7 +635,9 @@ public class TableDeviceSchemaFetcher { final List<String> attributeColumns, final ShowDevice statement, final Map<String, List<DeviceEntry>> deviceEntryMap, - final MPPQueryContext mppQueryContext) { + final DeviceEntryMaterializer materializer, + final MPPQueryContext mppQueryContext, + final AtomicBoolean containsNonAlignedDevice) { Throwable t = null; final long queryId = SessionManager.getInstance().requestQueryId(); @@ -567,10 +709,17 @@ public class TableDeviceSchemaFetcher { statement, mppQueryContext, attributeColumns, - deviceEntryMap.get(database)); + deviceEntryMap.get(database), + materializer); } else { constructTreeResults( - tsBlock.get(), columnHeaderList, tableInstance, mppQueryContext, deviceEntryMap); + tsBlock.get(), + columnHeaderList, + tableInstance, + mppQueryContext, + deviceEntryMap, + materializer, + containsNonAlignedDevice); } } } else { @@ -602,7 +751,8 @@ public class TableDeviceSchemaFetcher { final ShowDevice statement, final MPPQueryContext mppQueryContext, final List<String> attributeColumns, - final List<DeviceEntry> deviceEntryList) { + final List<DeviceEntry> deviceEntryList, + final DeviceEntryMaterializer materializer) { final Column[] columns = tsBlock.getValueColumns(); for (int i = 0; i < tsBlock.getPositionCount(); i++) { final String[] nodes = new String[tableInstance.getTagNum() + 1]; @@ -619,22 +769,48 @@ public class TableDeviceSchemaFetcher { final AlignedDeviceEntry deviceEntry = new AlignedDeviceEntry( deviceID, attributeColumns.stream().map(attributeMap::get).toArray(Binary[]::new)); - mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed()); - deviceEntryList.add(deviceEntry); + if (materializer == null) { + mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed()); + deviceEntryList.add(deviceEntry); + } else { + appendToMaterializer(materializer, deviceEntry, mppQueryContext, false); + } // Only cache those exact device query - // Fetch paths is null iff there are fuzzy queries related to tag columns + // Fetch paths is null iff there are fuzzy queries related to id columns if (Objects.nonNull(statement.getPartitionKeyList())) { cache.putAttributes(statement.getDatabase(), deviceID, attributeMap); } } } + private static void appendToMaterializer( + DeviceEntryMaterializer materializer, + DeviceEntry deviceEntry, + MPPQueryContext queryContext, + boolean memoryAlreadyReserved) { + try { + long releasedRamBytes = materializer.appendWithMemoryControl(deviceEntry); + if (releasedRamBytes > 0) { + queryContext.releaseMemoryReservedForFrontEnd(releasedRamBytes); + } + if (memoryAlreadyReserved && materializer.isSpilled()) { + queryContext.releaseMemoryReservedForFrontEnd(deviceEntry.ramBytesUsed()); + } else if (!memoryAlreadyReserved && !materializer.isSpilled()) { + queryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed()); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + private void constructTreeResults( final TsBlock tsBlock, final List<ColumnHeader> columnHeaderList, final TsTable tableInstance, final MPPQueryContext mppQueryContext, - final Map<String, List<DeviceEntry>> deviceEntryMap) { + final Map<String, List<DeviceEntry>> deviceEntryMap, + final DeviceEntryMaterializer materializer, + final AtomicBoolean containsNonAlignedDevice) { final Column[] columns = tsBlock.getValueColumns(); for (int i = 0; i < tsBlock.getPositionCount(); i++) { final String[] nodes = new String[tableInstance.getTagNum()]; @@ -646,12 +822,19 @@ public class TableDeviceSchemaFetcher { columns[columns.length - 2].getBoolean(i) ? new AlignedDeviceEntry(deviceID, new Binary[0]) : new NonAlignedDeviceEntry(deviceID, new Binary[0]); - mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed()); - deviceEntryMap - .computeIfAbsent( + final List<DeviceEntry> deviceEntries = + deviceEntryMap.computeIfAbsent( columns[columns.length - 1].getBinary(i).getStringValue(TSFileConfig.STRING_CHARSET), - k -> new ArrayList<>()) - .add(deviceEntry); + k -> new ArrayList<>()); + if (materializer == null) { + mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed()); + deviceEntries.add(deviceEntry); + } else { + appendToMaterializer(materializer, deviceEntry, mppQueryContext, false); + if (deviceEntry instanceof NonAlignedDeviceEntry) { + containsNonAlignedDevice.set(true); + } + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java new file mode 100644 index 00000000000..6959387a4aa --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java @@ -0,0 +1,169 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; + +import org.apache.tsfile.external.commons.io.FileUtils; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +public abstract class AbstractDeviceEntryMaterializer implements AutoCloseable { + + private final String queryId; + private final PlanNodeId planNodeId; + private final long thresholdInBytes; + private final List<DeviceEntry> bufferedEntries = new ArrayList<>(); + + private long entryCount; + private Path ownerDirectory; + private boolean ownerRegistered; + private boolean finished; + private DeviceEntryIOContext ioContext; + + protected AbstractDeviceEntryMaterializer( + String queryId, PlanNodeId planNodeId, long thresholdInBytes) { + if (thresholdInBytes <= 0) { + throw new IllegalArgumentException(); + } + this.queryId = queryId; + this.planNodeId = planNodeId; + this.thresholdInBytes = thresholdInBytes; + } + + /** + * Appends a DeviceEntry to this materializer's in-memory buffer. Spill decisions are external. + */ + public abstract void append(DeviceEntry entry) throws IOException; + + public abstract DeviceEntryDataSet finish() throws IOException; + + protected final String queryId() { + return queryId; + } + + protected final long thresholdInBytes() { + return thresholdInBytes; + } + + protected final void appendToBuffer(DeviceEntry entry) { + bufferedEntries.add(entry); + entryCount++; + } + + protected final void incrementEntryCount() { + entryCount++; + } + + protected final Iterable<DeviceEntry> bufferedEntries() { + return bufferedEntries; + } + + protected final boolean isBufferEmpty() { + return bufferedEntries.isEmpty(); + } + + protected final List<DeviceEntry> copyBufferedEntries() { + return new ArrayList<>(bufferedEntries); + } + + protected final void sortBufferedEntries(Comparator<DeviceEntry> comparator) { + bufferedEntries.sort(comparator); + } + + public abstract void forceSpill() throws IOException; + + public final void setQueryContext(MPPQueryContext queryContext) { + ioContext = new DeviceEntryIOContext(queryContext); + } + + protected final DeviceEntryIOContext ioContext() { + return ioContext; + } + + protected final long entryCount() { + return entryCount; + } + + protected final void clearBuffer() { + bufferedEntries.clear(); + } + + protected final Path ownerDirectory() { + return ownerDirectory; + } + + protected final Path ensureOwnerDirectory() throws IOException { + if (ownerDirectory == null) { + ownerDirectory = DeviceEntrySpillManager.getInstance().register(queryId, planNodeId); + ownerRegistered = true; + } + return ownerDirectory; + } + + protected final Path ensureUnregisteredOwnerDirectory(Path rootDirectory) throws IOException { + if (ownerDirectory == null) { + Path normalizedRoot = rootDirectory.normalize(); + Path directory = normalizedRoot.resolve(queryId).resolve(planNodeId.getId()).normalize(); + if (!directory.startsWith(normalizedRoot)) { + throw new IllegalArgumentException(); + } + Files.createDirectories(directory); + ownerDirectory = directory; + } + return ownerDirectory; + } + + protected final void checkNotFinished() { + if (finished) { + throw new IllegalStateException(); + } + } + + protected final void markFinished() { + finished = true; + } + + protected final void cleanupOwnerDirectory() throws IOException { + if (ownerDirectory != null) { + if (ownerRegistered) { + DeviceEntrySpillManager.getInstance().deregisterOwner(queryId, ownerDirectory); + } else { + FileUtils.deleteDirectory(ownerDirectory.toFile()); + } + ownerDirectory = null; + ownerRegistered = false; + } + } + + @Override + public void close() throws IOException { + if (!finished) { + cleanupOwnerDirectory(); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSet.java new file mode 100644 index 00000000000..dabff4daff1 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSet.java @@ -0,0 +1,46 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; + +import java.io.IOException; +import java.util.List; + +public interface DeviceEntryDataSet extends AutoCloseable { + + public long getEntryCount(); + + public boolean isSpilled(); + + public DeviceEntryReader openReader() throws IOException; + + public default DeviceEntryReader openConsumingReader() throws IOException { + throw new UnsupportedOperationException("Open consuming reader is not supported"); + } + + public default List<DeviceEntry> getInlineEntries() { + throw new UnsupportedOperationException( + "Only InMemoryDeviceEntryDataSet supports get inline device entries"); + } + + @Override + public void close() throws IOException; +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSetResult.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSetResult.java new file mode 100644 index 00000000000..7faed3c5a93 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSetResult.java @@ -0,0 +1,46 @@ +/* + * 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.plan.relational.metadata.spill; + +public final class DeviceEntryDataSetResult { + + private final String database; + private final DeviceEntryDataSet dataSet; + private final boolean containsNonAlignedDevice; + + public DeviceEntryDataSetResult( + String database, DeviceEntryDataSet dataSet, boolean containsNonAlignedDevice) { + this.database = database; + this.dataSet = dataSet; + this.containsNonAlignedDevice = containsNonAlignedDevice; + } + + public String getDatabase() { + return database; + } + + public DeviceEntryDataSet getDataSet() { + return dataSet; + } + + public boolean containsNonAlignedDevice() { + return containsNonAlignedDevice; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java new file mode 100644 index 00000000000..50c3965dfee --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java @@ -0,0 +1,132 @@ +/* + * 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.plan.relational.metadata.spill; + +import java.io.BufferedOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.ArrayList; +import java.util.List; + +public final class DeviceEntryDiskSpiller implements AutoCloseable { + + private final Path directory; + private final long targetSegmentBytes; + private final DeviceEntryIOContext ioContext; + private final List<Path> sealedSegments = new ArrayList<>(); + + private DataOutputStream output; + private Path temporaryFile; + private long currentBytes; + private int nextSegmentId; + + public DeviceEntryDiskSpiller(Path directory, long targetSegmentBytes) throws IOException { + this(directory, targetSegmentBytes, null); + } + + public DeviceEntryDiskSpiller( + Path directory, long targetSegmentBytes, DeviceEntryIOContext ioContext) throws IOException { + this.directory = directory; + this.targetSegmentBytes = targetSegmentBytes; + this.ioContext = ioContext; + Files.createDirectories(directory); + } + + public void append(byte[] serializedEntry) throws IOException { + checkTimeout(); + long startNanos = System.nanoTime(); + int recordBytes = Integer.BYTES + serializedEntry.length; + if (currentBytes > 0 && currentBytes + recordBytes > targetSegmentBytes) { + sealCurrentSegment(); + } + ensureOutput(); + output.writeInt(serializedEntry.length); + output.write(serializedEntry); + currentBytes += recordBytes; + recordDiskIO(recordBytes, startNanos); + } + + public List<Path> finish() throws IOException { + sealCurrentSegment(); + return List.copyOf(sealedSegments); + } + + private void ensureOutput() throws IOException { + if (output != null) { + return; + } + temporaryFile = directory.resolve(String.format("segment-%06d.tmp", nextSegmentId)); + output = new DataOutputStream(new BufferedOutputStream(Files.newOutputStream(temporaryFile))); + currentBytes = 0; + } + + private void sealCurrentSegment() throws IOException { + if (output == null) { + return; + } + checkTimeout(); + long startNanos = System.nanoTime(); + output.close(); + output = null; + Path sealedFile = directory.resolve(String.format("segment-%06d.bin", nextSegmentId++)); + try { + Files.move( + temporaryFile, + sealedFile, + StandardCopyOption.ATOMIC_MOVE, + StandardCopyOption.REPLACE_EXISTING); + } catch (IOException e) { + Files.move(temporaryFile, sealedFile, StandardCopyOption.REPLACE_EXISTING); + } + sealedSegments.add(sealedFile); + if (ioContext != null) { + ioContext.recordDiskIO(0, startNanos); + ioContext.recordSegment(); + } + temporaryFile = null; + currentBytes = 0; + } + + private void checkTimeout() { + if (ioContext != null) { + ioContext.checkTimeout(); + } + } + + private void recordDiskIO(long bytes, long startNanos) { + if (ioContext != null) { + ioContext.recordDiskIO(bytes, startNanos); + } + } + + @Override + public void close() throws IOException { + if (output != null) { + output.close(); + output = null; + } + if (temporaryFile != null) { + Files.deleteIfExists(temporaryFile); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryFileSpillerReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryFileSpillerReader.java new file mode 100644 index 00000000000..935c9b51b59 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryFileSpillerReader.java @@ -0,0 +1,143 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; + +import java.io.BufferedInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.NoSuchElementException; + +public final class DeviceEntryFileSpillerReader implements DeviceEntryReader { + + private final List<Path> segments; + private final boolean deleteSegmentAfterRead; + private final DeviceEntryIOContext ioContext; + private int segmentIndex; + private DataInputStream input; + private Path currentSegment; + private DeviceEntry next; + + public DeviceEntryFileSpillerReader(List<Path> segments) { + this(segments, false, null); + } + + public DeviceEntryFileSpillerReader(List<Path> segments, boolean deleteSegmentAfterRead) { + this(segments, deleteSegmentAfterRead, null); + } + + public DeviceEntryFileSpillerReader( + List<Path> segments, boolean deleteSegmentAfterRead, DeviceEntryIOContext ioContext) { + this.segments = segments; + this.deleteSegmentAfterRead = deleteSegmentAfterRead; + this.ioContext = ioContext; + } + + @Override + public boolean hasNext() throws IOException { + if (next != null) { + return true; + } + while (true) { + if (input == null && !openNextSegment()) { + return false; + } + checkTimeout(); + long startNanos = System.nanoTime(); + Integer length = readRecordLength(); + if (length == null) { + closeCurrentSegment(true); + continue; + } + byte[] bytes = new byte[length]; + input.readFully(bytes); + if (ioContext != null) { + ioContext.recordDiskIO(Integer.BYTES + length, startNanos); + } + next = DeviceEntry.deserialize(bytes); + return true; + } + } + + private void checkTimeout() { + if (ioContext != null) { + ioContext.checkTimeout(); + } + } + + @Override + public DeviceEntry next() throws IOException { + if (!hasNext()) { + throw new NoSuchElementException(); + } + DeviceEntry result = next; + next = null; + return result; + } + + private boolean openNextSegment() throws IOException { + if (segmentIndex >= segments.size()) { + return false; + } + currentSegment = segments.get(segmentIndex++); + input = new DataInputStream(new BufferedInputStream(Files.newInputStream(currentSegment))); + return true; + } + + private void closeCurrentSegment(boolean fullyConsumed) throws IOException { + input.close(); + input = null; + if (fullyConsumed && deleteSegmentAfterRead) { + try { + Files.deleteIfExists(currentSegment); + } catch (IOException ignored) { + // Query cleanup will retry deleting a segment that could not be deleted eagerly. + } + } + currentSegment = null; + } + + private Integer readRecordLength() throws IOException { + int firstByte = input.read(); + if (firstByte < 0) { + return null; + } + int length = + (firstByte << 24) + | (input.readUnsignedByte() << 16) + | (input.readUnsignedByte() << 8) + | input.readUnsignedByte(); + if (length < 0) { + throw new IOException(); + } + return length; + } + + @Override + public void close() throws IOException { + if (input != null) { + closeCurrentSegment(false); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java new file mode 100644 index 00000000000..c434e0fece7 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java @@ -0,0 +1,60 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; + +import java.util.concurrent.TimeUnit; + +public final class DeviceEntryIOContext { + + private final MPPQueryContext queryContext; + private final long timeoutStartNanos; + private final long remainingTimeoutNanos; + + public DeviceEntryIOContext(MPPQueryContext queryContext) { + this.queryContext = queryContext; + this.timeoutStartNanos = System.nanoTime(); + long elapsedMillis = Math.max(0, System.currentTimeMillis() - queryContext.getStartTime()); + long remainingTimeoutMillis = Math.max(0, queryContext.getTimeOut() - elapsedMillis); + this.remainingTimeoutNanos = TimeUnit.MILLISECONDS.toNanos(remainingTimeoutMillis); + } + + public void checkTimeout() { + if (System.nanoTime() - timeoutStartNanos >= remainingTimeoutNanos) { + throw new QueryTimeoutRuntimeException( + queryContext.getStartTime(), System.currentTimeMillis(), queryContext.getTimeOut()); + } + } + + public void recordDiskIO(long bytes, long startNanos) { + queryContext.recordDeviceEntryDiskIO(bytes, System.nanoTime() - startNanos); + checkTimeout(); + } + + public void recordSegment() { + queryContext.recordDeviceEntrySegment(); + } + + public void recordSortedRun() { + queryContext.recordDeviceEntrySortedRun(); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java new file mode 100644 index 00000000000..92d7cf64d48 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java @@ -0,0 +1,132 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; + +import java.io.IOException; +import java.nio.file.Path; + +public final class DeviceEntryMaterializer extends AbstractDeviceEntryMaterializer { + + private final boolean rawSegment; + private DeviceEntryDiskSpiller spiller; + // Only be used in fetchDeviceSchema, manages memory itself + private long rawBufferedRamBytes; + + public DeviceEntryMaterializer( + String queryId, PlanNodeId planNodeId, long thresholdInBytes, boolean rawSegment) { + super(queryId, planNodeId, thresholdInBytes); + this.rawSegment = rawSegment; + } + + public DeviceEntryMaterializer( + String queryId, + PlanNodeId planNodeId, + long thresholdInBytes, + boolean rawSegment, + MPPQueryContext queryContext) { + this(queryId, planNodeId, thresholdInBytes, rawSegment); + setQueryContext(queryContext); + } + + @Override + public void append(DeviceEntry entry) throws IOException { + checkNotFinished(); + appendToBuffer(entry); + } + + /** Returns the RAM bytes released when Coordinator Raw Fetch switches to spill mode. */ + public long appendWithMemoryControl(DeviceEntry entry) throws IOException { + checkNotFinished(); + long ramBytesUsed = entry.ramBytesUsed(); + if (spiller == null && rawBufferedRamBytes + ramBytesUsed <= thresholdInBytes()) { + appendToBuffer(entry); + rawBufferedRamBytes += ramBytesUsed; + return 0; + } + long releasedRamBytes = rawBufferedRamBytes; + ensureSpiller(); + rawBufferedRamBytes = 0; + spiller.append(entry.serializeToBytes()); + incrementEntryCount(); + return releasedRamBytes; + } + + @Override + public void forceSpill() throws IOException { + checkNotFinished(); + if (spiller == null && !isBufferEmpty()) { + ensureSpiller(); + } + rawBufferedRamBytes = 0; + } + + public boolean isSpilled() { + return spiller != null; + } + + @Override + public DeviceEntryDataSet finish() throws IOException { + checkNotFinished(); + DeviceEntryDataSet dataSet; + if (spiller == null) { + dataSet = new InMemoryDeviceEntryDataSet(copyBufferedEntries()); + } else { + dataSet = + new SpilledDeviceEntryDataSet( + queryId(), ownerDirectory(), spiller.finish(), entryCount(), !rawSegment); + } + markFinished(); + return dataSet; + } + + private void ensureSpiller() throws IOException { + if (spiller != null) { + return; + } + Path ownerDirectory = + rawSegment + ? ensureUnregisteredOwnerDirectory( + Path.of(IoTDBDescriptor.getInstance().getConfig().getQueryDir(), "device-entry")) + : ensureOwnerDirectory(); + spiller = + new DeviceEntryDiskSpiller( + ownerDirectory.resolve(rawSegment ? "raw" : "fi"), thresholdInBytes(), ioContext()); + for (DeviceEntry entry : bufferedEntries()) { + spiller.append(entry.serializeToBytes()); + } + clearBuffer(); + } + + @Override + public void close() throws IOException { + try { + if (spiller != null) { + spiller.close(); + } + } finally { + super.close(); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryReader.java new file mode 100644 index 00000000000..5fa2e044974 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryReader.java @@ -0,0 +1,34 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; + +import java.io.IOException; + +public interface DeviceEntryReader extends AutoCloseable { + + public boolean hasNext() throws IOException; + + public DeviceEntry next() throws IOException; + + @Override + public void close() throws IOException; +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntrySpillManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntrySpillManager.java new file mode 100644 index 00000000000..162cae055dc --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntrySpillManager.java @@ -0,0 +1,189 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.conf.IoTDBDescriptor; + +import org.apache.tsfile.external.commons.io.FileUtils; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Comparator; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +public final class DeviceEntrySpillManager { + + private final ConcurrentHashMap<String, Set<Path>> queryDirectories = new ConcurrentHashMap<>(); + + private DeviceEntrySpillManager() {} + + public static DeviceEntrySpillManager getInstance() { + return DeviceEntrySpillManagerHolder.INSTANCE; + } + + public Path register(String queryId, PlanNodeId planNodeId) throws IOException { + Path ownerDirectory = rootDirectory().resolve(queryId).resolve(planNodeId.getId()); + Files.createDirectories(ownerDirectory); + queryDirectories + .computeIfAbsent(queryId, ignored -> ConcurrentHashMap.newKeySet()) + .add(ownerDirectory); + return ownerDirectory; + } + + public void deregisterOwner(String queryId, Path ownerDirectory) throws IOException { + Set<Path> owners = queryDirectories.get(queryId); + if (owners != null) { + owners.remove(ownerDirectory); + if (owners.isEmpty()) { + queryDirectories.remove(queryId, owners); + } + } + FileUtils.deleteDirectory(ownerDirectory.toFile()); + } + + public void deregisterQuery(String queryId) throws IOException { + queryDirectories.remove(queryId); + FileUtils.deleteDirectory(rootDirectory().resolve(queryId).toFile()); + } + + @TestOnly + public List<Path> listSegments(String queryId, String planNodeId) throws IOException { + Path dataSetDirectory = resolveRegisteredDataSetDirectory(queryId, planNodeId); + try (java.util.stream.Stream<Path> stream = Files.list(dataSetDirectory)) { + return stream + .filter(path -> path.getFileName().toString().matches("segment-[0-9]{6,}\\.bin")) + .sorted( + Comparator.comparingInt((Path path) -> path.getFileName().toString().length()) + .thenComparing(path -> path.getFileName().toString())) + .toList(); + } + } + + public byte[] readSegment(String queryId, String dataSetId, int segmentId) throws IOException { + return Files.readAllBytes(resolveSegment(queryId, dataSetId, segmentId)); + } + + public Path resolveSegment(String queryId, String dataSetId, int segmentId) throws IOException { + Path segment = getRegisteredSegmentPath(queryId, dataSetId, segmentId); + if (!Files.isRegularFile(segment)) { + throw new java.nio.file.NoSuchFileException(segment.toString()); + } + return segment; + } + + public Path resolveSegment(String queryId, PlanNodeId planNodeId, int segmentId) + throws IOException { + return resolveSegment(queryId, planNodeId.getId(), segmentId); + } + + public void deleteSegment(String queryId, String dataSetId, int segmentId) throws IOException { + Files.deleteIfExists(getRegisteredSegmentPath(queryId, dataSetId, segmentId)); + } + + public void deleteSegment(String queryId, PlanNodeId planNodeId, int segmentId) + throws IOException { + deleteSegment(queryId, planNodeId.getId(), segmentId); + } + + public void finishSegmentDataSet(String queryId, String planNodeId) throws IOException { + deregisterOwner(queryId, rootDirectory().resolve(queryId).resolve(planNodeId)); + } + + public void deregisterFragment(String queryId, String fragmentInstanceId) throws IOException { + FileUtils.deleteDirectory( + resolveUnderRoot(fragmentRootDirectory(), queryId, fragmentInstanceId).toFile()); + } + + public void clearStaleFragmentData() throws IOException { + FileUtils.deleteDirectory(fragmentRootDirectory().toFile()); + Files.createDirectories(fragmentRootDirectory()); + } + + private Path resolveRegisteredDataSetDirectory(String queryId, String dataSetId) + throws IOException { + Path relativeDataSetPath = Path.of(dataSetId); + if (relativeDataSetPath.isAbsolute() + || java.util.stream.StreamSupport.stream(relativeDataSetPath.spliterator(), false) + .anyMatch(path -> path.toString().equals("..") || path.toString().equals("."))) { + throw new IllegalArgumentException(); + } + Path queryDirectory = rootDirectory().resolve(queryId).normalize(); + Path dataSetDirectory = queryDirectory.resolve(relativeDataSetPath).resolve("fi").normalize(); + if (!dataSetDirectory.startsWith(queryDirectory)) { + throw new IllegalArgumentException(); + } + Set<Path> owners = queryDirectories.get(queryId); + boolean registered = + owners != null + && owners.stream() + .map(Path::normalize) + .anyMatch(owner -> dataSetDirectory.startsWith(owner) && Files.isDirectory(owner)); + if (!registered || !Files.isDirectory(dataSetDirectory)) { + throw new java.nio.file.NoSuchFileException(dataSetDirectory.toString()); + } + return dataSetDirectory; + } + + private Path getRegisteredSegmentPath(String queryId, String dataSetId, int segmentId) + throws IOException { + if (segmentId < 0) { + throw new IllegalArgumentException(); + } + return resolveRegisteredDataSetDirectory(queryId, dataSetId) + .resolve(String.format("segment-%06d.bin", segmentId)); + } + + public void clearStaleData() throws IOException { + FileUtils.deleteDirectory(rootDirectory().toFile()); + Files.createDirectories(rootDirectory()); + queryDirectories.clear(); + } + + private Path rootDirectory() { + return Path.of(IoTDBDescriptor.getInstance().getConfig().getQueryDir(), "device-entry"); + } + + private Path fragmentRootDirectory() { + return rootDirectory().resolve("fragment"); + } + + private Path resolveUnderRoot(Path root, String... children) { + Path result = root; + for (String child : children) { + result = result.resolve(child); + } + result = result.normalize(); + if (!result.startsWith(root.normalize())) { + throw new IllegalArgumentException(); + } + return result; + } + + private static class DeviceEntrySpillManagerHolder { + private static final DeviceEntrySpillManager INSTANCE = new DeviceEntrySpillManager(); + + private DeviceEntrySpillManagerHolder() {} + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/InMemoryDeviceEntryDataSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/InMemoryDeviceEntryDataSet.java new file mode 100644 index 00000000000..831cbe351f3 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/InMemoryDeviceEntryDataSet.java @@ -0,0 +1,76 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; + +import java.util.Collections; +import java.util.Iterator; +import java.util.List; + +public final class InMemoryDeviceEntryDataSet implements DeviceEntryDataSet { + + private final List<DeviceEntry> entries; + + public InMemoryDeviceEntryDataSet(List<DeviceEntry> entries) { + this.entries = Collections.unmodifiableList(entries); + } + + @Override + public long getEntryCount() { + return entries.size(); + } + + @Override + public boolean isSpilled() { + return false; + } + + @Override + public DeviceEntryReader openReader() { + Iterator<DeviceEntry> iterator = entries.iterator(); + return new DeviceEntryReader() { + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public DeviceEntry next() { + return iterator.next(); + } + + @Override + public void close() { + // No resource to release. + } + }; + } + + @Override + public List<DeviceEntry> getInlineEntries() { + return entries; + } + + @Override + public void close() { + // The query context owns memory accounting for inline entries. + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/SpilledDeviceEntryDataSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/SpilledDeviceEntryDataSet.java new file mode 100644 index 00000000000..0380f210f34 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/SpilledDeviceEntryDataSet.java @@ -0,0 +1,85 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.tsfile.external.commons.io.FileUtils; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; + +public final class SpilledDeviceEntryDataSet implements DeviceEntryDataSet { + + private final String queryId; + private final Path ownerDirectory; + private final List<Path> segments; + private final long entryCount; + private final boolean managedBySpillManager; + + public SpilledDeviceEntryDataSet( + String queryId, + Path ownerDirectory, + List<Path> segments, + long entryCount, + boolean managedBySpillManager) { + this.queryId = queryId; + this.ownerDirectory = ownerDirectory; + this.segments = segments; + this.entryCount = entryCount; + this.managedBySpillManager = managedBySpillManager; + } + + @Override + public long getEntryCount() { + return entryCount; + } + + @Override + public boolean isSpilled() { + return true; + } + + public Path getOwnerDirectory() { + return ownerDirectory; + } + + public List<Path> getSegments() { + return segments; + } + + @Override + public DeviceEntryReader openReader() { + return new DeviceEntryFileSpillerReader(segments); + } + + @Override + public DeviceEntryReader openConsumingReader() { + return new DeviceEntryFileSpillerReader(segments, true); + } + + @Override + public void close() throws IOException { + if (managedBySpillManager) { + DeviceEntrySpillManager.getInstance().deregisterOwner(queryId, ownerDirectory); + } else { + FileUtils.deleteDirectory(ownerDirectory.toFile()); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java index 1ceb4c86f6c..e74b6bb3a0f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java @@ -70,6 +70,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.analyzer.predicate.schema import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; import org.apache.iotdb.db.queryengine.plan.relational.metadata.NonAlignedDeviceEntry; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryReader; import org.apache.iotdb.db.queryengine.plan.relational.planner.EqualityInference; import org.apache.iotdb.db.queryengine.plan.relational.planner.IrExpressionInterpreter; import org.apache.iotdb.db.queryengine.plan.relational.planner.IrTypeAnalyzer; @@ -91,6 +94,8 @@ import org.apache.tsfile.utils.Pair; import javax.annotation.Nullable; +import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -746,7 +751,7 @@ public class PushPredicateIntoTableScan implements PlanOptimizer { } long startTime = System.nanoTime(); - final Map<String, List<DeviceEntry>> deviceEntriesMap = + final DeviceEntryDataSetResult deviceEntryDataSetResult = metadata.indexScan( tableScanNode.getQualifiedObjectName(), metadataExpressions.stream() @@ -756,7 +761,10 @@ public class PushPredicateIntoTableScan implements PlanOptimizer { expression, tableScanNode.getAssignments())) .collect(Collectors.toList()), attributeColumns, - queryContext); + queryContext, + tableScanNode.getPlanNodeId()); + final Map<String, List<DeviceEntry>> deviceEntriesMap = + readDeviceEntries(deviceEntryDataSetResult); if (deviceEntriesMap.size() > 1) { throw new SemanticException( DataNodeQueryMessages.TREE_DEVICE_VIEW_WITH_MULTIPLE_DATABASES @@ -837,6 +845,20 @@ public class PushPredicateIntoTableScan implements PlanOptimizer { } } + private Map<String, List<DeviceEntry>> readDeviceEntries( + final DeviceEntryDataSetResult result) { + final List<DeviceEntry> entries = new ArrayList<>(); + try (DeviceEntryDataSet dataSet = result.getDataSet(); + DeviceEntryReader reader = dataSet.openReader()) { + while (reader.hasNext()) { + entries.add(reader.next()); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + return Collections.singletonMap(result.getDatabase(), entries); + } + @Override public PlanNode visitJoin(JoinNode node, RewriteContext context) { Expression inheritedPredicate = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java index 02999d57b01..da6f415af8d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java @@ -53,6 +53,25 @@ public class FragmentInstanceStatisticsDrawer { 0, String.format( "Fetch Schema Cost: %.3f ms", context.getFetchSchemaCost() * NS_TO_MS_FACTOR)); + addLine( + planHeader, + 1, + String.format( + "Disk IO Size for DeviceEntry: %d bytes", context.getDiskIOSizeForDeviceEntry())); + addLine( + planHeader, + 1, + String.format( + "Disk IO Time Cost for DeviceEntry: %.3f ms", + context.getDiskIOTimeCostForDeviceEntry() * NS_TO_MS_FACTOR)); + addLine( + planHeader, + 1, + String.format("DeviceEntry Segment Count: %d", context.getDeviceEntrySegmentCount())); + addLine( + planHeader, + 1, + String.format("DeviceEntry Sorted Run Count: %d", context.getDeviceEntrySortedRunCount())); addLine( planHeader, 0, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java index 74571ad152f..9e727cd7517 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java @@ -55,6 +55,12 @@ public class FragmentInstanceStatisticsJsonDrawer { "fetchPartitionCostMs", formatMs(context.getFetchPartitionCost() * NS_TO_MS_FACTOR)); planStatistics.addProperty( "fetchSchemaCostMs", formatMs(context.getFetchSchemaCost() * NS_TO_MS_FACTOR)); + planStatistics.addProperty("diskIOSizeForDeviceEntry", context.getDiskIOSizeForDeviceEntry()); + planStatistics.addProperty( + "diskIOTimeCostForDeviceEntryMs", + formatMs(context.getDiskIOTimeCostForDeviceEntry() * NS_TO_MS_FACTOR)); + planStatistics.addProperty("deviceEntrySegmentCount", context.getDeviceEntrySegmentCount()); + planStatistics.addProperty("deviceEntrySortedRunCount", context.getDeviceEntrySortedRunCount()); planStatistics.addProperty( "logicalPlanCostMs", formatMs(context.getLogicalPlanCost() * NS_TO_MS_FACTOR)); planStatistics.addProperty( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java index edb13217db2..592d45d2e63 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java @@ -19,6 +19,8 @@ package org.apache.iotdb.db.queryengine.statistics; +import java.util.concurrent.atomic.AtomicLong; + public class QueryPlanStatistics { private long analyzeCost; private long fetchPartitionCost; @@ -27,6 +29,12 @@ public class QueryPlanStatistics { private long logicalOptimizationCost; private long distributionPlanCost; private long dispatchCost = 0; + // DeviceEntry materialization may involve multiple Region materializers. Use the same + // lock-free accumulation style as execution.fragment.QueryStatistics. + private final AtomicLong diskIOSizeForDeviceEntry = new AtomicLong(); + private final AtomicLong diskIOTimeCostForDeviceEntry = new AtomicLong(); + private final AtomicLong deviceEntrySegmentCount = new AtomicLong(); + private final AtomicLong deviceEntrySortedRunCount = new AtomicLong(); public void setAnalyzeCost(long analyzeCost) { this.analyzeCost = analyzeCost; @@ -83,4 +91,33 @@ public class QueryPlanStatistics { public long getDispatchCost() { return dispatchCost; } + + public void recordDeviceEntryDiskIO(long bytes, long timeCost) { + diskIOSizeForDeviceEntry.addAndGet(bytes); + diskIOTimeCostForDeviceEntry.addAndGet(timeCost); + } + + public void recordDeviceEntrySegment() { + deviceEntrySegmentCount.incrementAndGet(); + } + + public void recordDeviceEntrySortedRun() { + deviceEntrySortedRunCount.incrementAndGet(); + } + + public long getDiskIOSizeForDeviceEntry() { + return diskIOSizeForDeviceEntry.get(); + } + + public long getDiskIOTimeCostForDeviceEntry() { + return diskIOTimeCostForDeviceEntry.get(); + } + + public long getDeviceEntrySegmentCount() { + return deviceEntrySegmentCount.get(); + } + + public long getDeviceEntrySortedRunCount() { + return deviceEntrySortedRunCount.get(); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializerTest.java new file mode 100644 index 00000000000..9f13e8bd3c7 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializerTest.java @@ -0,0 +1,153 @@ +/* + * 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.plan.relational.metadata.spill; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.Binary; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class DeviceEntryMaterializerTest { + + private Path queryDirectory; + + @Before + public void setUp() throws Exception { + queryDirectory = Files.createTempDirectory("device-entry-spill-test"); + IoTDBDescriptor.getInstance().getConfig().setQueryDir(queryDirectory.toString()); + } + + @After + public void tearDown() throws Exception { + DeviceEntrySpillManager.getInstance().clearStaleData(); + Files.deleteIfExists(queryDirectory.resolve("device-entry")); + Files.deleteIfExists(queryDirectory); + } + + @Test + public void testKeepSmallDataSetInline() throws Exception { + List<DeviceEntry> expected = createEntries(3); + try (DeviceEntryMaterializer materializer = + new DeviceEntryMaterializer("q-inline", new PlanNodeId("scan-0"), Long.MAX_VALUE, true)) { + for (DeviceEntry entry : expected) { + materializer.append(entry); + } + try (DeviceEntryDataSet dataSet = materializer.finish()) { + assertFalse(dataSet.isSpilled()); + assertEquals(expected, dataSet.getInlineEntries()); + } + } + } + + @Test + public void testSpillAndReadMultipleSegments() throws Exception { + List<DeviceEntry> expected = createEntries(20); + DeviceEntryDataSet dataSet; + try (DeviceEntryMaterializer materializer = + new DeviceEntryMaterializer("q-spill", new PlanNodeId("scan-0"), 128, true)) { + for (DeviceEntry entry : expected) { + materializer.append(entry); + } + dataSet = materializer.finish(); + } + + assertTrue(dataSet.isSpilled()); + assertEquals(expected.size(), dataSet.getEntryCount()); + List<DeviceEntry> actual = new ArrayList<>(); + try (DeviceEntryReader reader = dataSet.openReader()) { + while (reader.hasNext()) { + actual.add(reader.next()); + } + } + assertEquals(expected, actual); + + dataSet.close(); + assertFalse(Files.exists(queryDirectory.resolve("device-entry/q-spill/scan-0"))); + } + + @Test + public void testControlledSegmentAccess() throws Exception { + DeviceEntryDataSet dataSet; + try (DeviceEntryMaterializer materializer = + new DeviceEntryMaterializer("q-segment", new PlanNodeId("scan-0"), 128, true)) { + for (DeviceEntry entry : createEntries(20)) { + materializer.append(entry); + } + dataSet = materializer.finish(); + } + + DeviceEntrySpillManager manager = DeviceEntrySpillManager.getInstance(); + Path rawDirectory = queryDirectory.resolve("device-entry/q-segment/scan-0/raw"); + List<Path> segments; + try (java.util.stream.Stream<Path> stream = Files.list(rawDirectory)) { + segments = + stream.filter(path -> path.getFileName().toString().endsWith(".bin")).sorted().toList(); + } + assertTrue(segments.size() > 1); + assertTrue(Files.size(segments.get(0)) > 0); + dataSet.close(); + } + + @Test + public void testSpillFileUsesLengthPrefixedRecordsWithoutHeaderOrCrc() throws Exception { + DeviceEntry entry = createEntries(1).get(0); + DeviceEntryDataSet dataSet; + try (DeviceEntryMaterializer materializer = + new DeviceEntryMaterializer("q-format", new PlanNodeId("scan-0"), 1, true)) { + materializer.append(entry); + dataSet = materializer.finish(); + } + + Path segment = + DeviceEntrySpillManager.getInstance().listSegments("q-format", "scan-0/raw").get(0); + byte[] fileBytes = Files.readAllBytes(segment); + byte[] payload = entry.serializeToBytes(); + assertEquals(Integer.BYTES + payload.length, fileBytes.length); + assertEquals(payload.length, ByteBuffer.wrap(fileBytes).getInt()); + dataSet.close(); + } + + private static List<DeviceEntry> createEntries(int count) { + List<DeviceEntry> entries = new ArrayList<>(count); + for (int i = 0; i < count; i++) { + entries.add( + new AlignedDeviceEntry( + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table", "device" + i}), + new Binary[] {new Binary(("attribute" + i).getBytes(TSFileConfig.STRING_CHARSET))})); + } + return entries; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java index b83501a277c..ae76d47b3fe 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java @@ -60,6 +60,10 @@ public class FragmentInstanceStatisticsJsonDrawerTest { context.setLogicalPlanCost(4000000L); // 4ms context.setLogicalOptimizationCost(5000000L); // 5ms context.setDistributionPlanCost(6000000L); // 6ms + context.recordDeviceEntryDiskIO(8192L, 7000000L); // 8 KiB, 7ms + context.recordDeviceEntrySegment(); + context.recordDeviceEntrySegment(); + context.recordDeviceEntrySortedRun(); drawer.renderPlanStatistics(context); @@ -78,6 +82,10 @@ public class FragmentInstanceStatisticsJsonDrawerTest { assertEquals(4.0, planStats.get("logicalPlanCostMs").getAsDouble(), 0.01); assertEquals(5.0, planStats.get("logicalOptimizationCostMs").getAsDouble(), 0.01); assertEquals(6.0, planStats.get("distributionPlanCostMs").getAsDouble(), 0.01); + assertEquals(8192L, planStats.get("diskIOSizeForDeviceEntry").getAsLong()); + assertEquals(7.0, planStats.get("diskIOTimeCostForDeviceEntryMs").getAsDouble(), 0.01); + assertEquals(2L, planStats.get("deviceEntrySegmentCount").getAsLong()); + assertEquals(1L, planStats.get("deviceEntrySortedRunCount").getAsLong()); } @Test diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index c4a06b68b14..81573e47f54 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -2445,3 +2445,7 @@ enable_retry_for_unknown_error=false # effectiveMode: hot_reload # Datatype: Boolean include_null_value_in_write_throughput_metric=false +# Maximum DeviceEntry bytes retained in memory before spilling for a table query. +# <= 0 uses query execution memory / query_thread_count / 4. +# Datatype: long; effectiveMode: hot_reload; unit: byte. +# table_query_device_entry_batch_size_in_bytes=0
