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 1d726366ddd2..19bfa1bf9e67 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 void setQueryDir(String queryDir) { 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 15d2b7d003d5..ed46c49e4114 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 void loadProperties(TrimProperties properties) throws BadNodeUrlException 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 495ebc42b65c..0f274d9c06e2 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 @@ -48,6 +48,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner; import org.apache.iotdb.db.queryengine.plan.planner.memory.NotThreadSafeMemoryReservationManager; import org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ExternalTsFileQueryResource; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryIOContext; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainOutputFormat; import org.apache.iotdb.db.queryengine.statistics.QueryPlanStatistics; @@ -132,6 +133,8 @@ public enum ExplainType { private QueryPlanStatistics queryPlanStatistics = null; + private DeviceEntryIOContext deviceEntryIOContext; + // To avoid query front-end from consuming too much memory, it needs to reserve memory when // constructing some Expression and PlanNode. private final MemoryReservationManager memoryReservationManager; @@ -403,6 +406,13 @@ public void setStartTime(long startTime) { this.startTime = startTime; } + public DeviceEntryIOContext getOrCreateDeviceEntryIOContext(boolean duringFetchSchema) { + if (deviceEntryIOContext == null) { + deviceEntryIOContext = new DeviceEntryIOContext(this, duringFetchSchema); + } + return deviceEntryIOContext; + } + public void addFailedEndPoint(TEndPoint endPoint) { this.endPointBlackList.add(endPoint); } @@ -528,6 +538,37 @@ public long getDispatchCost() { return queryPlanStatistics.getDispatchCost(); } + public void recordDeviceEntryDiskIODuringFetchSchema(long bytes, long timeCost) { + getOrCreateQueryPlanStatistics().recordDeviceEntryDiskIODuringFetchSchema(bytes, timeCost); + } + + public void recordDeviceEntryCount(long count) { + getOrCreateQueryPlanStatistics().recordDeviceEntryCount(count); + } + + public long getDiskIOSizeForDeviceEntryDuringFetchSchema() { + return queryPlanStatistics == null + ? 0 + : queryPlanStatistics.getDiskIOSizeForDeviceEntryDuringFetchSchema(); + } + + public long getDiskIOTimeCostForDeviceEntryDuringFetchSchema() { + return queryPlanStatistics == null + ? 0 + : queryPlanStatistics.getDiskIOTimeCostForDeviceEntryDuringFetchSchema(); + } + + public long getDeviceEntryCount() { + return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntryCount(); + } + + 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/analyze/ClusterPartitionFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ClusterPartitionFetcher.java index 7bb7478950b9..9b4cc79d9474 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ClusterPartitionFetcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ClusterPartitionFetcher.java @@ -48,6 +48,8 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.queryengine.plan.analyze.cache.partition.PartitionCache; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryReader; import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; import org.apache.iotdb.rpc.TSStatusCode; @@ -57,6 +59,7 @@ import javax.annotation.Nullable; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -207,28 +210,26 @@ public DataPartition getDataPartition( final Map> sgNameToQueryParamsMap) { DataPartition dataPartition = partitionCache.getDataPartition(sgNameToQueryParamsMap); if (null == dataPartition) { - try (ConfigNodeClient client = - configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { - final TDataPartitionTableResp dataPartitionTableResp = - client.getDataPartitionTable(constructDataPartitionReqForQuery(sgNameToQueryParamsMap)); - if (dataPartitionTableResp.getStatus().getCode() - == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - dataPartition = parseDataPartitionResp(dataPartitionTableResp); - partitionCache.updateDataPartitionCache(dataPartitionTableResp.getDataPartitionTable()); - } else { - throw new StatementAnalyzeException( - String.format( - DataNodeQueryMessages - .QUERY_EXCEPTION_AN_ERROR_OCCURRED_WHEN_EXECUTING_GETDATAPARTITION_S_D21A0011, - dataPartitionTableResp.getStatus().getMessage())); - } - } catch (final ClientManagerException | TException e) { - throw new StatementAnalyzeException( - String.format( - DataNodeQueryMessages - .QUERY_EXCEPTION_AN_ERROR_OCCURRED_WHEN_EXECUTING_GETDATAPARTITION_S_D21A0011, - e.getMessage())); - } + dataPartition = + fetchDataPartition(constructDataPartitionReqForQuery(sgNameToQueryParamsMap), true); + } + return dataPartition; + } + + @Override + public DataPartition getDataPartition( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots) { + final Set seriesPartitionSlots = collectSeriesPartitionSlots(dataSet); + DataPartition dataPartition = + partitionCache.getDataPartition(database, seriesPartitionSlots, timePartitionSlots); + if (null == dataPartition) { + dataPartition = + fetchDataPartition( + constructDataPartitionReqForQuery( + database, seriesPartitionSlots, timePartitionSlots, false, false), + true); } return dataPartition; } @@ -239,20 +240,41 @@ public DataPartition getDataPartitionWithUnclosedTimeRange( // In this method, we must fetch from config node because it contains -oo or +oo // and there is no need to update cache because since we will never fetch it from cache, the // update operation will be only time waste + return fetchDataPartition(constructDataPartitionReqForQuery(sgNameToQueryParamsMap), false); + } + + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots, + final boolean needLeftAll, + final boolean needRightAll) { + final Set seriesPartitionSlots = collectSeriesPartitionSlots(dataSet); + return fetchDataPartition( + constructDataPartitionReqForQuery( + database, seriesPartitionSlots, timePartitionSlots, needLeftAll, needRightAll), + false); + } + + private DataPartition fetchDataPartition( + final TDataPartitionReq request, final boolean updateCache) { try (final ConfigNodeClient client = configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { - final TDataPartitionTableResp dataPartitionTableResp = - client.getDataPartitionTable(constructDataPartitionReqForQuery(sgNameToQueryParamsMap)); + final TDataPartitionTableResp dataPartitionTableResp = client.getDataPartitionTable(request); if (dataPartitionTableResp.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - return parseDataPartitionResp(dataPartitionTableResp); - } else { - throw new StatementAnalyzeException( - String.format( - DataNodeQueryMessages - .QUERY_EXCEPTION_AN_ERROR_OCCURRED_WHEN_EXECUTING_GETDATAPARTITION_S_D21A0011, - dataPartitionTableResp.getStatus().getMessage())); + final DataPartition dataPartition = parseDataPartitionResp(dataPartitionTableResp); + if (updateCache) { + partitionCache.updateDataPartitionCache(dataPartitionTableResp.getDataPartitionTable()); + } + return dataPartition; } + throw new StatementAnalyzeException( + String.format( + DataNodeQueryMessages + .QUERY_EXCEPTION_AN_ERROR_OCCURRED_WHEN_EXECUTING_GETDATAPARTITION_S_D21A0011, + dataPartitionTableResp.getStatus().getMessage())); } catch (final ClientManagerException | TException e) { throw new StatementAnalyzeException( String.format( @@ -543,6 +565,34 @@ private TDataPartitionReq constructDataPartitionReqForQuery( return new TDataPartitionReq(partitionSlotsMap); } + private TDataPartitionReq constructDataPartitionReqForQuery( + final String database, + final Set seriesPartitionSlots, + final List timePartitionSlots, + final boolean needLeftAll, + final boolean needRightAll) { + final TTimeSlotList sharedTimeSlotList = + new TTimeSlotList(timePartitionSlots, needLeftAll, needRightAll); + final Map seriesSlotToTimeSlots = new HashMap<>(); + for (final TSeriesPartitionSlot seriesPartitionSlot : seriesPartitionSlots) { + seriesSlotToTimeSlots.put(seriesPartitionSlot, sharedTimeSlotList); + } + return new TDataPartitionReq(Collections.singletonMap(database, seriesSlotToTimeSlots)); + } + + private Set collectSeriesPartitionSlots(final DeviceEntryDataSet dataSet) { + final Set seriesPartitionSlots = new HashSet<>(); + try (final DeviceEntryReader reader = dataSet.openReader()) { + while (reader.hasNext()) { + seriesPartitionSlots.add( + partitionExecutor.getSeriesPartitionSlot(reader.next().getDeviceID())); + } + } catch (final IOException e) { + throw new UncheckedIOException(e); + } + return seriesPartitionSlots; + } + private SchemaPartition parseSchemaPartitionTableResp( final TSchemaPartitionTableResp schemaPartitionTableResp) { final Map> regionReplicaMap = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/IPartitionFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/IPartitionFetcher.java index 0549ec396474..56315b45638e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/IPartitionFetcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/IPartitionFetcher.java @@ -19,11 +19,13 @@ package org.apache.iotdb.db.queryengine.plan.analyze; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.partition.DataPartition; import org.apache.iotdb.commons.partition.DataPartitionQueryParam; import org.apache.iotdb.commons.partition.SchemaNodeManagementPartition; import org.apache.iotdb.commons.partition.SchemaPartition; import org.apache.iotdb.commons.path.PathPatternTree; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; import org.apache.tsfile.file.metadata.IDeviceID; @@ -59,6 +61,9 @@ default SchemaPartition getSchemaPartition( */ DataPartition getDataPartition(Map> sgNameToQueryParamsMap); + DataPartition getDataPartition( + String database, DeviceEntryDataSet dataSet, List timePartitionSlots); + /** * Get data partition, used in query scenarios which contains time filter like: time < XX or time * > XX @@ -68,6 +73,13 @@ default SchemaPartition getSchemaPartition( DataPartition getDataPartitionWithUnclosedTimeRange( Map> sgNameToQueryParamsMap); + DataPartition getDataPartitionWithUnclosedTimeRange( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots, + boolean needLeftAll, + boolean needRightAll); + /** * Get or create data partition, used in standalone write scenarios. if enableAutoCreateSchema is * true and database/series/time slots not exists, then automatically create. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java index 4b8ee72ffd48..e7bdf88ca6a0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java @@ -873,93 +873,80 @@ public void invalidAllSchemaPartitionCache() { */ public DataPartition getDataPartition( Map> databaseToQueryParamsMap) { + final Map>> querySlots = + new HashMap<>(); + for (final Map.Entry> entry : + databaseToQueryParamsMap.entrySet()) { + if (entry.getValue() == null) { + return null; + } + final Map> seriesSlots = new HashMap<>(); + for (final DataPartitionQueryParam param : entry.getValue()) { + if (param.getDeviceID() == null) { + return null; + } + seriesSlots.put( + partitionExecutor.getSeriesPartitionSlot(param.getDeviceID()), + param.getTimePartitionSlotList()); + } + querySlots.put(entry.getKey(), seriesSlots); + } + return getDataPartitionBySlots(querySlots); + } + + public DataPartition getDataPartition( + final String database, + final Set seriesPartitionSlots, + final List timePartitionSlots) { + final Map> querySlots = new HashMap<>(); + for (final TSeriesPartitionSlot seriesPartitionSlot : seriesPartitionSlots) { + querySlots.put(seriesPartitionSlot, timePartitionSlots); + } + return getDataPartitionBySlots(Collections.singletonMap(database, querySlots)); + } + + private DataPartition getDataPartitionBySlots( + final Map>> querySlots) { dataPartitionCacheLock.readLock().lock(); try { failIfMetadataLeaseFenced(); - if (databaseToQueryParamsMap.isEmpty()) { + if (querySlots.isEmpty()) { cacheMetrics.record(false, CacheMetrics.DATA_PARTITION_CACHE_NAME); return null; } - final Set allConsensusGroupIds = new HashSet<>(); final Map> consensusGroupToTimeSlotMap = new HashMap<>(); - - for (Map.Entry> entry : - databaseToQueryParamsMap.entrySet()) { - String databaseName = entry.getKey(); - List params = entry.getValue(); - - if (null == params || params.isEmpty()) { + for (final Map.Entry>> + databaseEntry : querySlots.entrySet()) { + final DataPartitionTable dataPartitionTable = + dataPartitionCache.getIfPresent(databaseEntry.getKey()); + if (dataPartitionTable == null || databaseEntry.getValue().isEmpty()) { cacheMetrics.record(false, CacheMetrics.DATA_PARTITION_CACHE_NAME); return null; } - - DataPartitionTable dataPartitionTable = dataPartitionCache.getIfPresent(databaseName); - if (null == dataPartitionTable) { - if (logger.isDebugEnabled()) { - logger.debug( - DataNodeQueryMessages.ARG_CACHE_MISS_WHEN_SEARCH_DATABASE_ARG, - CacheMetrics.DATA_PARTITION_CACHE_NAME, - databaseName); - } - cacheMetrics.record(false, CacheMetrics.DATA_PARTITION_CACHE_NAME); - return null; - } - - Map cachedDatabasePartitionMap = - dataPartitionTable.getDataPartitionMap(); - - for (DataPartitionQueryParam param : params) { - TSeriesPartitionSlot seriesPartitionSlot; - if (null != param.getDeviceID()) { - seriesPartitionSlot = partitionExecutor.getSeriesPartitionSlot(param.getDeviceID()); - } else { - return null; - } - - SeriesPartitionTable cachedSeriesPartitionTable = - cachedDatabasePartitionMap.get(seriesPartitionSlot); - if (null == cachedSeriesPartitionTable) { - if (logger.isDebugEnabled()) { - logger.debug( - DataNodeQueryMessages.ARG_CACHE_MISS_WHEN_SEARCH_DEVICE_ARG, - CacheMetrics.DATA_PARTITION_CACHE_NAME, - param.getDeviceID()); - } + for (final Map.Entry> seriesEntry : + databaseEntry.getValue().entrySet()) { + final SeriesPartitionTable cachedSeriesPartitionTable = + dataPartitionTable.getDataPartitionMap().get(seriesEntry.getKey()); + if (cachedSeriesPartitionTable == null || seriesEntry.getValue().isEmpty()) { cacheMetrics.record(false, CacheMetrics.DATA_PARTITION_CACHE_NAME); return null; } - - Map> cachedTimePartitionSlot = - cachedSeriesPartitionTable.getSeriesPartitionMap(); - - if (param.getTimePartitionSlotList().isEmpty()) { - return null; - } - - for (TTimePartitionSlot timePartitionSlot : param.getTimePartitionSlotList()) { - List cacheConsensusGroupIds = - cachedTimePartitionSlot.get(timePartitionSlot); - if (null == cacheConsensusGroupIds - || cacheConsensusGroupIds.isEmpty() - || null == timePartitionSlot) { - if (logger.isDebugEnabled()) { - logger.debug( - DataNodeQueryMessages.ARG_CACHE_MISS_WHEN_SEARCH_TIME_PARTITION_ARG, - CacheMetrics.DATA_PARTITION_CACHE_NAME, - timePartitionSlot); - } + for (final TTimePartitionSlot timePartitionSlot : seriesEntry.getValue()) { + final List cachedConsensusGroupIds = + cachedSeriesPartitionTable.getSeriesPartitionMap().get(timePartitionSlot); + if (cachedConsensusGroupIds == null || cachedConsensusGroupIds.isEmpty()) { cacheMetrics.record(false, CacheMetrics.DATA_PARTITION_CACHE_NAME); return null; } - - for (TConsensusGroupId groupId : cacheConsensusGroupIds) { + for (final TConsensusGroupId groupId : cachedConsensusGroupIds) { allConsensusGroupIds.add(groupId); consensusGroupToTimeSlotMap - .computeIfAbsent(groupId, k -> new HashSet<>()) + .computeIfAbsent(groupId, key -> new HashSet<>()) .add( - new TimeSlotRegionInfo(databaseName, seriesPartitionSlot, timePartitionSlot)); + new TimeSlotRegionInfo( + databaseEntry.getKey(), seriesEntry.getKey(), timePartitionSlot)); } } } @@ -967,23 +954,19 @@ public DataPartition getDataPartition( final List consensusGroupIds = new ArrayList<>(allConsensusGroupIds); final List allRegionReplicaSets = getRegionReplicaSet(consensusGroupIds); - - Map>>> + final Map>>> dataPartitionMap = new HashMap<>(); - for (int i = 0; i < allRegionReplicaSets.size(); i++) { - TConsensusGroupId groupId = consensusGroupIds.get(i); - TRegionReplicaSet replicaSet = allRegionReplicaSets.get(i); - - for (TimeSlotRegionInfo info : consensusGroupToTimeSlotMap.get(groupId)) { + final TConsensusGroupId groupId = consensusGroupIds.get(i); + final TRegionReplicaSet replicaSet = allRegionReplicaSets.get(i); + for (final TimeSlotRegionInfo info : consensusGroupToTimeSlotMap.get(groupId)) { dataPartitionMap - .computeIfAbsent(info.databaseName, k -> new HashMap<>()) - .computeIfAbsent(info.seriesPartitionSlot, k -> new HashMap<>()) - .computeIfAbsent(info.timePartitionSlot, k -> new ArrayList<>()) + .computeIfAbsent(info.databaseName, key -> new HashMap<>()) + .computeIfAbsent(info.seriesPartitionSlot, key -> new HashMap<>()) + .computeIfAbsent(info.timePartitionSlot, key -> new ArrayList<>()) .add(replicaSet); } } - if (logger.isDebugEnabled()) { logger.debug(DataNodeQueryMessages.CACHE_HIT, CacheMetrics.DATA_PARTITION_CACHE_NAME); } 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 4701e0e9c62b..1b5e6ab693b1 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.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 void serialize(final DataOutputStream stream) throws IOException { 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 static DeviceEntry deserialize(final ByteBuffer byteBuffer) { 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 8b7f0f32656b..04db571083f5 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 @@ -20,11 +20,13 @@ package org.apache.iotdb.db.queryengine.plan.relational.metadata; import org.apache.iotdb.calc.plan.relational.metadata.ITypeMetadata; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.exception.SemanticException; 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 +39,14 @@ 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.DeviceEntryDataSet; +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 +89,12 @@ default boolean isWindowFunction( * index scanning * @param attributeColumns attribute column names */ - Map> indexScan( + DeviceEntryDataSetResult indexScan( final QualifiedObjectName tableName, final List expressionList, final List 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. @@ -203,6 +207,11 @@ SchemaPartition getOrCreateSchemaPartition( DataPartition getDataPartition( final String database, final List sgNameToQueryParamsMap); + DataPartition getDataPartition( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots); + /** * Get data partition, used in query scenarios which contains time filter like: time < XX or time * > XX @@ -212,4 +221,11 @@ DataPartition getDataPartition( */ DataPartition getDataPartitionWithUnclosedTimeRange( final String database, final List sgNameToQueryParamsMap); + + DataPartition getDataPartitionWithUnclosedTimeRange( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots, + final boolean needLeftAll, + final boolean needRightAll); } 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 9e736a35c581..89a53d18bd43 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 @@ -21,11 +21,13 @@ import org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils; import org.apache.iotdb.calc.utils.constant.SqlConstant; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.exception.SemanticException; 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 +59,8 @@ 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.DeviceEntryDataSet; +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 +79,6 @@ 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 +1442,6 @@ && isNumericType(argumentTypes.get(0)) 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 +1483,6 @@ && isNumericType(argumentTypes.get(0)) 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 +1612,6 @@ && isNumericType(argumentTypes.get(0)) throw new SemanticException(DataNodeQueryMessages.UNKNOWN_FUNCTION + functionName); } - private static void validateRateFunctionArguments( - String functionName, - List 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 +1631,20 @@ public boolean canCoerce(final Type from, final Type to) { } @Override - public Map> indexScan( + public DeviceEntryDataSetResult indexScan( final QualifiedObjectName tableName, final List expressionList, final List 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 @@ -1764,10 +1720,29 @@ public DataPartition getDataPartition( Collections.singletonMap(database, sgNameToQueryParamsMap)); } + @Override + public DataPartition getDataPartition( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots) { + return partitionFetcher.getDataPartition(database, dataSet, timePartitionSlots); + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( String database, List sgNameToQueryParamsMap) { return partitionFetcher.getDataPartitionWithUnclosedTimeRange( Collections.singletonMap(database, sgNameToQueryParamsMap)); } + + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots, + final boolean needLeftAll, + final boolean needRightAll) { + return partitionFetcher.getDataPartitionWithUnclosedTimeRange( + database, dataSet, timePartitionSlots, needLeftAll, needRightAll); + } } 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 024b9ef44ac8..1ac6e1789a4f 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 @@ 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.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.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,7 @@ public Map> fetchDeviceSchemaForDataQuery( mayContainDuplicateDevice, false)) { fetchMissingDeviceSchemaForQuery( - database, tableInstance, attributeColumns, statement, deviceEntryMap, queryContext); + database, tableInstance, attributeColumns, statement, deviceEntryMap, null, queryContext); } // TODO table metadata: implement deduplicate during schemaRegion execution @@ -253,6 +260,129 @@ public Map> fetchDeviceSchemaForDataQuery( : deviceEntryMap; } + public DeviceEntryDataSetResult fetchDeviceSchemaForDataQueryAsDataSet( + final String database, + final String table, + final List expressionList, + final List attributeColumns, + final MPPQueryContext queryContext, + final PlanNodeId planNodeId) { + final TsTable tableInstance = DataNodeTableCache.getInstance().getTable(database, table); + if (TreeViewSchema.isTreeViewTable(tableInstance)) { + final Map> deviceEntryMap = new HashMap<>(); + final AtomicBoolean mayContainDuplicateDevice = new AtomicBoolean(false); + boolean containsNonAlignedDevice = 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 entries : deviceEntryMap.values()) { + for (DeviceEntry entry : entries) { + appendToMaterializer(materializer, entry, queryContext, true); + if (entry instanceof NonAlignedDeviceEntry) { + containsNonAlignedDevice = true; + } + } + entries.clear(); + } + if (needRemoteFetch) { + containsNonAlignedDevice |= + fetchMissingDeviceSchemaForQuery( + database, + tableInstance, + attributeColumns, + statement, + deviceEntryMap, + materializer, + queryContext); + } + 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); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + final Map> 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); + } + 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); + } + 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( @@ -489,14 +619,16 @@ public static IDeviceID convertTagValuesToDeviceID( return IDeviceID.Factory.DEFAULT_FACTORY.create(deviceIdNodes); } - private void fetchMissingDeviceSchemaForQuery( + private boolean fetchMissingDeviceSchemaForQuery( final String database, final TsTable tableInstance, final List attributeColumns, final ShowDevice statement, final Map> deviceEntryMap, + final DeviceEntryMaterializer materializer, final MPPQueryContext mppQueryContext) { Throwable t = null; + boolean containsNonAlignedDevice = false; final long queryId = SessionManager.getInstance().requestQueryId(); // For the correctness of attribute remote update @@ -567,10 +699,17 @@ private void fetchMissingDeviceSchemaForQuery( statement, mppQueryContext, attributeColumns, - deviceEntryMap.get(database)); + deviceEntryMap.get(database), + materializer); } else { - constructTreeResults( - tsBlock.get(), columnHeaderList, tableInstance, mppQueryContext, deviceEntryMap); + containsNonAlignedDevice |= + constructTreeResults( + tsBlock.get(), + columnHeaderList, + tableInstance, + mppQueryContext, + deviceEntryMap, + materializer); } } } else { @@ -583,6 +722,7 @@ private void fetchMissingDeviceSchemaForQuery( TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); } } + return containsNonAlignedDevice; } catch (final Throwable throwable) { t = throwable; throw throwable; @@ -602,7 +742,8 @@ private void constructTableResults( final ShowDevice statement, final MPPQueryContext mppQueryContext, final List attributeColumns, - final List deviceEntryList) { + final List 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 +760,48 @@ private void constructTableResults( 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 void constructTreeResults( + 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 boolean constructTreeResults( final TsBlock tsBlock, final List columnHeaderList, final TsTable tableInstance, final MPPQueryContext mppQueryContext, - final Map> deviceEntryMap) { + final Map> deviceEntryMap, + final DeviceEntryMaterializer materializer) { + boolean containsNonAlignedDevice = false; final Column[] columns = tsBlock.getValueColumns(); for (int i = 0; i < tsBlock.getPositionCount(); i++) { final String[] nodes = new String[tableInstance.getTagNum()]; @@ -646,13 +813,19 @@ private void constructTreeResults( columns[columns.length - 2].getBoolean(i) ? new AlignedDeviceEntry(deviceID, new Binary[0]) : new NonAlignedDeviceEntry(deviceID, new Binary[0]); - mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed()); - deviceEntryMap - .computeIfAbsent( + containsNonAlignedDevice |= deviceEntry instanceof NonAlignedDeviceEntry; + final List 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); + } } + return containsNonAlignedDevice; } private void constructNodesArrayAndAttributeMap( 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 000000000000..76aaa9036751 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java @@ -0,0 +1,167 @@ +/* + * 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.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 bufferedEntries = new ArrayList<>(); + + private long entryCount; + private Path ownerDirectory; + private boolean ownerRegistered; + private boolean finished; + private DeviceEntryIOContext ioContext; + private MPPQueryContext queryContext; + + 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 bufferedEntries() { + return bufferedEntries; + } + + protected final boolean isBufferEmpty() { + return bufferedEntries.isEmpty(); + } + + protected final List copyBufferedEntries() { + return new ArrayList<>(bufferedEntries); + } + + protected final void sortBufferedEntries(Comparator comparator) { + bufferedEntries.sort(comparator); + } + + public abstract void forceSpill() throws IOException; + + protected final void setQueryContext(MPPQueryContext queryContext) { + this.queryContext = queryContext; + } + + protected final DeviceEntryIOContext ioContext() { + return ioContext; + } + + protected final DeviceEntryIOContext createIOContextOnSpill(boolean duringFetchSchema) { + if (ioContext == null && queryContext != null) { + ioContext = queryContext.getOrCreateDeviceEntryIOContext(duringFetchSchema); + } + 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 void checkNotFinished() { + if (finished) { + throw new IllegalStateException(); + } + } + + public MPPQueryContext getQueryContext() { + return queryContext; + } + + 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 000000000000..92a9caf6654a --- /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 { + + long getEntryCount(); + + boolean isSpilled(); + + DeviceEntryReader openReader() throws IOException; + + default DeviceEntryReader openConsumingReader() throws IOException { + throw new UnsupportedOperationException("Open consuming reader is not supported"); + } + + default List getInlineEntries() { + throw new UnsupportedOperationException( + "Only InMemoryDeviceEntryDataSet supports get inline device entries"); + } + + @Override + 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 000000000000..7faed3c5a93f --- /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 000000000000..a7bcdfbcc387 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java @@ -0,0 +1,126 @@ +/* + * 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 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); + recordDiskIO(recordBytes, startNanos); + currentBytes += recordBytes; + } + + public List 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; + } + 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); + 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 000000000000..935c9b51b592 --- /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 segments; + private final boolean deleteSegmentAfterRead; + private final DeviceEntryIOContext ioContext; + private int segmentIndex; + private DataInputStream input; + private Path currentSegment; + private DeviceEntry next; + + public DeviceEntryFileSpillerReader(List segments) { + this(segments, false, null); + } + + public DeviceEntryFileSpillerReader(List segments, boolean deleteSegmentAfterRead) { + this(segments, deleteSegmentAfterRead, null); + } + + public DeviceEntryFileSpillerReader( + List 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 000000000000..5376478a8f88 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java @@ -0,0 +1,45 @@ +/* + * 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.common.MPPQueryContext; + +public final class DeviceEntryIOContext { + + private final MPPQueryContext queryContext; + private final boolean duringFetchSchema; + + public DeviceEntryIOContext(MPPQueryContext queryContext, boolean duringFetchSchema) { + this.queryContext = queryContext; + this.duringFetchSchema = duringFetchSchema; + } + + public void checkTimeout() { + queryContext.checkTimeOut(); + } + + public void recordDiskIO(long bytes, long startNanos) { + long timeCost = System.nanoTime() - startNanos; + if (duringFetchSchema) { + queryContext.recordDeviceEntryDiskIODuringFetchSchema(bytes, timeCost); + } + checkTimeout(); + } +} 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 000000000000..0a6603695019 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java @@ -0,0 +1,133 @@ +/* + * 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 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(), true); + } + if (rawSegment && getQueryContext() != null) { + getQueryContext().recordDeviceEntryCount(entryCount()); + } + markFinished(); + return dataSet; + } + + private void ensureSpiller() throws IOException { + if (spiller != null) { + return; + } + Path ownerDirectory = ensureOwnerDirectory(); + if (rawSegment) { + createIOContextOnSpill(true); + } + 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 000000000000..5fa2e044974a --- /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 000000000000..65ea3fd517fe --- /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> 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 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 listSegments(String queryId, String planNodeId) throws IOException { + Path dataSetDirectory = resolveRegisteredDataSetDirectory(queryId, planNodeId); + try (java.util.stream.Stream 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 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().getSortTmpDir(), "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 000000000000..831cbe351f36 --- /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 entries; + + public InMemoryDeviceEntryDataSet(List entries) { + this.entries = Collections.unmodifiableList(entries); + } + + @Override + public long getEntryCount() { + return entries.size(); + } + + @Override + public boolean isSpilled() { + return false; + } + + @Override + public DeviceEntryReader openReader() { + Iterator 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 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 000000000000..0380f210f343 --- /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 segments; + private final long entryCount; + private final boolean managedBySpillManager; + + public SpilledDeviceEntryDataSet( + String queryId, + Path ownerDirectory, + List 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 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/iterative/rule/PruneTableScanColumns.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java index 819efbbc4aff..526c20abdd17 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java @@ -117,7 +117,7 @@ public static Optional pruneColumns(TableScanNode node, Set re treeDeviceViewScanNode.getTreeDBName(), treeDeviceViewScanNode.getMeasurementColumnNameMap()); prunedNode.setRegionReplicaSet(deviceTableScanNode.getRegionReplicaSet()); - return Optional.of(prunedNode); + return Optional.of(deviceTableScanNode.copyDeviceEntryDataSetTo(prunedNode)); } else if (node instanceof TreeNonAlignedDeviceViewScanNode) { TreeNonAlignedDeviceViewScanNode treeDeviceViewScanNode = (TreeNonAlignedDeviceViewScanNode) deviceTableScanNode; @@ -139,11 +139,11 @@ public static Optional pruneColumns(TableScanNode node, Set re treeDeviceViewScanNode.getTreeDBName(), treeDeviceViewScanNode.getMeasurementColumnNameMap()); prunedNode.setRegionReplicaSet(deviceTableScanNode.getRegionReplicaSet()); - return Optional.of(prunedNode); + return Optional.of(deviceTableScanNode.copyDeviceEntryDataSetTo(prunedNode)); } else if (node instanceof TreeDeviceViewScanNode) { TreeDeviceViewScanNode treeDeviceViewScanNode = (TreeDeviceViewScanNode) deviceTableScanNode; - return Optional.of( + TreeDeviceViewScanNode prunedNode = new TreeDeviceViewScanNode( deviceTableScanNode.getPlanNodeId(), deviceTableScanNode.getQualifiedObjectName(), @@ -159,7 +159,9 @@ public static Optional pruneColumns(TableScanNode node, Set re deviceTableScanNode.isPushLimitToEachDevice(), deviceTableScanNode.containsNonAlignedDevice(), treeDeviceViewScanNode.getTreeDBName(), - treeDeviceViewScanNode.getMeasurementColumnNameMap())); + treeDeviceViewScanNode.getMeasurementColumnNameMap()); + prunedNode.setRegionReplicaSet(deviceTableScanNode.getRegionReplicaSet()); + return Optional.of(deviceTableScanNode.copyDeviceEntryDataSetTo(prunedNode)); } else if (node instanceof ExternalTsFileScanNode externalTsFileScanNode) { ExternalTsFileScanNode prunedNode = new ExternalTsFileScanNode( @@ -179,7 +181,7 @@ public static Optional pruneColumns(TableScanNode node, Set re externalTsFileScanNode.getDeviceTaskPartitionIndex(), externalTsFileScanNode.getSchemaFilter()); prunedNode.setRegionReplicaSet(deviceTableScanNode.getRegionReplicaSet()); - return Optional.of(prunedNode); + return Optional.of(deviceTableScanNode.copyDeviceEntryDataSetTo(prunedNode)); } else { DeviceTableScanNode prunedNode = new DeviceTableScanNode( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java index 70a4198fff93..58d85026b7d5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java @@ -268,26 +268,27 @@ public Optional getGroupIdSymbol() { @Override public AggregationTableScanNode clone() { - return new AggregationTableScanNode( - id, - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice, - projection, - aggregations, - groupingSets, - preGroupedSymbols, - step, - groupIdSymbol); + return copyDeviceEntryDataSetTo( + new AggregationTableScanNode( + id, + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice, + projection, + aggregations, + groupingSets, + preGroupedSymbols, + step, + groupIdSymbol)); } @Override @@ -330,50 +331,52 @@ public static AggregationTableScanNode combineAggregationAndTableScan( } if (tableScanNode instanceof TreeDeviceViewScanNode) { TreeDeviceViewScanNode treeDeviceViewScanNode = (TreeDeviceViewScanNode) tableScanNode; - return new AggregationTreeDeviceViewScanNode( - id, - tableScanNode.getQualifiedObjectName(), - tableScanNode.getOutputSymbols(), - tableScanNode.getAssignments(), - tableScanNode.getDeviceEntries(), - tableScanNode.getTagAndAttributeIndexMap(), - tableScanNode.getScanOrder(), - tableScanNode.getTimePredicate().orElse(null), - tableScanNode.getPushDownPredicate(), - tableScanNode.getPushDownLimit(), - tableScanNode.getPushDownOffset(), - tableScanNode.isPushLimitToEachDevice(), - tableScanNode.containsNonAlignedDevice(), - projectNode == null ? null : projectNode.getAssignments(), - aggregationNode.getAggregations(), - aggregationNode.getGroupingSets(), - aggregationNode.getPreGroupedSymbols(), - aggregationNode.getStep(), - aggregationNode.getGroupIdSymbol(), - treeDeviceViewScanNode.getTreeDBName(), - treeDeviceViewScanNode.getMeasurementColumnNameMap()); + return tableScanNode.copyDeviceEntryDataSetTo( + new AggregationTreeDeviceViewScanNode( + id, + tableScanNode.getQualifiedObjectName(), + tableScanNode.getOutputSymbols(), + tableScanNode.getAssignments(), + tableScanNode.getDeviceEntries(), + tableScanNode.getTagAndAttributeIndexMap(), + tableScanNode.getScanOrder(), + tableScanNode.getTimePredicate().orElse(null), + tableScanNode.getPushDownPredicate(), + tableScanNode.getPushDownLimit(), + tableScanNode.getPushDownOffset(), + tableScanNode.isPushLimitToEachDevice(), + tableScanNode.containsNonAlignedDevice(), + projectNode == null ? null : projectNode.getAssignments(), + aggregationNode.getAggregations(), + aggregationNode.getGroupingSets(), + aggregationNode.getPreGroupedSymbols(), + aggregationNode.getStep(), + aggregationNode.getGroupIdSymbol(), + treeDeviceViewScanNode.getTreeDBName(), + treeDeviceViewScanNode.getMeasurementColumnNameMap())); } - return new AggregationTableScanNode( - id, - tableScanNode.getQualifiedObjectName(), - tableScanNode.getOutputSymbols(), - tableScanNode.getAssignments(), - tableScanNode.getDeviceEntries(), - tableScanNode.getTagAndAttributeIndexMap(), - tableScanNode.getScanOrder(), - tableScanNode.getTimePredicate().orElse(null), - tableScanNode.getPushDownPredicate(), - tableScanNode.getPushDownLimit(), - tableScanNode.getPushDownOffset(), - tableScanNode.isPushLimitToEachDevice(), - tableScanNode.containsNonAlignedDevice(), - projectNode == null ? null : projectNode.getAssignments(), - aggregationNode.getAggregations(), - aggregationNode.getGroupingSets(), - aggregationNode.getPreGroupedSymbols(), - aggregationNode.getStep(), - aggregationNode.getGroupIdSymbol()); + return tableScanNode.copyDeviceEntryDataSetTo( + new AggregationTableScanNode( + id, + tableScanNode.getQualifiedObjectName(), + tableScanNode.getOutputSymbols(), + tableScanNode.getAssignments(), + tableScanNode.getDeviceEntries(), + tableScanNode.getTagAndAttributeIndexMap(), + tableScanNode.getScanOrder(), + tableScanNode.getTimePredicate().orElse(null), + tableScanNode.getPushDownPredicate(), + tableScanNode.getPushDownLimit(), + tableScanNode.getPushDownOffset(), + tableScanNode.isPushLimitToEachDevice(), + tableScanNode.containsNonAlignedDevice(), + projectNode == null ? null : projectNode.getAssignments(), + aggregationNode.getAggregations(), + aggregationNode.getGroupingSets(), + aggregationNode.getPreGroupedSymbols(), + aggregationNode.getStep(), + aggregationNode.getGroupIdSymbol())); } public static AggregationTableScanNode combineAggregationAndTableScan( @@ -410,50 +413,52 @@ public static AggregationTableScanNode combineAggregationAndTableScan( } if (tableScanNode instanceof TreeDeviceViewScanNode) { TreeDeviceViewScanNode treeDeviceViewScanNode = (TreeDeviceViewScanNode) tableScanNode; - return new AggregationTreeDeviceViewScanNode( - id, - tableScanNode.getQualifiedObjectName(), - tableScanNode.getOutputSymbols(), - tableScanNode.getAssignments(), - tableScanNode.getDeviceEntries(), - tableScanNode.getTagAndAttributeIndexMap(), - tableScanNode.getScanOrder(), - tableScanNode.getTimePredicate().orElse(null), - tableScanNode.getPushDownPredicate(), - tableScanNode.getPushDownLimit(), - tableScanNode.getPushDownOffset(), - tableScanNode.isPushLimitToEachDevice(), - tableScanNode.containsNonAlignedDevice(), - projectNode == null ? null : projectNode.getAssignments(), - aggregationNode.getAggregations(), - aggregationNode.getGroupingSets(), - aggregationNode.getPreGroupedSymbols(), - step, - aggregationNode.getGroupIdSymbol(), - treeDeviceViewScanNode.getTreeDBName(), - treeDeviceViewScanNode.getMeasurementColumnNameMap()); + return tableScanNode.copyDeviceEntryDataSetTo( + new AggregationTreeDeviceViewScanNode( + id, + tableScanNode.getQualifiedObjectName(), + tableScanNode.getOutputSymbols(), + tableScanNode.getAssignments(), + tableScanNode.getDeviceEntries(), + tableScanNode.getTagAndAttributeIndexMap(), + tableScanNode.getScanOrder(), + tableScanNode.getTimePredicate().orElse(null), + tableScanNode.getPushDownPredicate(), + tableScanNode.getPushDownLimit(), + tableScanNode.getPushDownOffset(), + tableScanNode.isPushLimitToEachDevice(), + tableScanNode.containsNonAlignedDevice(), + projectNode == null ? null : projectNode.getAssignments(), + aggregationNode.getAggregations(), + aggregationNode.getGroupingSets(), + aggregationNode.getPreGroupedSymbols(), + step, + aggregationNode.getGroupIdSymbol(), + treeDeviceViewScanNode.getTreeDBName(), + treeDeviceViewScanNode.getMeasurementColumnNameMap())); } - return new AggregationTableScanNode( - id, - tableScanNode.getQualifiedObjectName(), - tableScanNode.getOutputSymbols(), - tableScanNode.getAssignments(), - tableScanNode.getDeviceEntries(), - tableScanNode.getTagAndAttributeIndexMap(), - tableScanNode.getScanOrder(), - tableScanNode.getTimePredicate().orElse(null), - tableScanNode.getPushDownPredicate(), - tableScanNode.getPushDownLimit(), - tableScanNode.getPushDownOffset(), - tableScanNode.isPushLimitToEachDevice(), - tableScanNode.containsNonAlignedDevice(), - projectNode == null ? null : projectNode.getAssignments(), - aggregationNode.getAggregations(), - aggregationNode.getGroupingSets(), - aggregationNode.getPreGroupedSymbols(), - step, - aggregationNode.getGroupIdSymbol()); + return tableScanNode.copyDeviceEntryDataSetTo( + new AggregationTableScanNode( + id, + tableScanNode.getQualifiedObjectName(), + tableScanNode.getOutputSymbols(), + tableScanNode.getAssignments(), + tableScanNode.getDeviceEntries(), + tableScanNode.getTagAndAttributeIndexMap(), + tableScanNode.getScanOrder(), + tableScanNode.getTimePredicate().orElse(null), + tableScanNode.getPushDownPredicate(), + tableScanNode.getPushDownLimit(), + tableScanNode.getPushDownOffset(), + tableScanNode.isPushLimitToEachDevice(), + tableScanNode.containsNonAlignedDevice(), + projectNode == null ? null : projectNode.getAssignments(), + aggregationNode.getAggregations(), + aggregationNode.getGroupingSets(), + aggregationNode.getPreGroupedSymbols(), + step, + aggregationNode.getGroupIdSymbol())); } public boolean mayUseLastCache() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTreeDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTreeDeviceViewScanNode.java index 3b35876a2c3e..f6e65beddc53 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTreeDeviceViewScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTreeDeviceViewScanNode.java @@ -137,28 +137,29 @@ public String toString() { @Override public AggregationTreeDeviceViewScanNode clone() { - return new AggregationTreeDeviceViewScanNode( - id, - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice, - projection, - aggregations, - groupingSets, - preGroupedSymbols, - step, - groupIdSymbol, - treeDBName, - measurementColumnNameMap); + return copyDeviceEntryDataSetTo( + new AggregationTreeDeviceViewScanNode( + id, + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice, + projection, + aggregations, + groupingSets, + preGroupedSymbols, + step, + groupIdSymbol, + treeDBName, + measurementColumnNameMap)); } protected PlanNodeType getPlanNodeType() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AlignedAggregationTreeDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AlignedAggregationTreeDeviceViewScanNode.java index 1e577e812800..19d1013420f3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AlignedAggregationTreeDeviceViewScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AlignedAggregationTreeDeviceViewScanNode.java @@ -94,28 +94,29 @@ public R accept(IPlanVisitor visitor, C context) { @Override public AlignedAggregationTreeDeviceViewScanNode clone() { - return new AlignedAggregationTreeDeviceViewScanNode( - getPlanNodeId(), - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice, - projection, - aggregations, - groupingSets, - preGroupedSymbols, - step, - groupIdSymbol, - getTreeDBName(), - getMeasurementColumnNameMap()); + return copyDeviceEntryDataSetTo( + new AlignedAggregationTreeDeviceViewScanNode( + getPlanNodeId(), + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice, + projection, + aggregations, + groupingSets, + preGroupedSymbols, + step, + groupIdSymbol, + getTreeDBName(), + getMeasurementColumnNameMap())); } protected PlanNodeType getPlanNodeType() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java index 3b7b365c3292..d3d8af862bd7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java @@ -30,6 +30,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry; import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.tsfile.read.filter.basic.Filter; @@ -41,6 +42,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -48,7 +50,9 @@ public class DeviceTableScanNode extends TableScanNode { - protected List deviceEntries; + protected List deviceEntries = Collections.emptyList(); + + @Nullable protected transient DeviceEntryDataSet deviceEntryDataSet; // Indicates the respective index order of tag and attribute columns in DeviceEntry. // For example, for DeviceEntry `table1.tag1.tag2.attribute1.attribute2.s1.s2`, the content of @@ -148,7 +152,7 @@ public DeviceTableScanNode clone() { pushLimitToEachDevice, containsNonAlignedDevice); cloned.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId; - return cloned; + return copyDeviceEntryDataSetTo(cloned); } protected static void serializeMemberVariables( @@ -267,6 +271,28 @@ public void setDeviceEntries(List deviceEntries) { this.deviceEntries = deviceEntries; } + public void setDeviceEntryDataSet(final DeviceEntryDataSet deviceEntryDataSet) { + this.deviceEntryDataSet = deviceEntryDataSet; + this.deviceEntries = + deviceEntryDataSet.isSpilled() + ? Collections.emptyList() + : deviceEntryDataSet.getInlineEntries(); + } + + @Nullable + public DeviceEntryDataSet getDeviceEntryDataSet() { + return deviceEntryDataSet; + } + + public T copyDeviceEntryDataSetTo(final T target) { + target.deviceEntryDataSet = deviceEntryDataSet; + return target; + } + + public long getDeviceEntryCount() { + return deviceEntryDataSet == null ? deviceEntries.size() : deviceEntryDataSet.getEntryCount(); + } + public Map getTagAndAttributeIndexMap() { return this.tagAndAttributeIndexMap; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/NonAlignedAggregationTreeDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/NonAlignedAggregationTreeDeviceViewScanNode.java index af851b82c3be..592b4a5c381c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/NonAlignedAggregationTreeDeviceViewScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/NonAlignedAggregationTreeDeviceViewScanNode.java @@ -95,28 +95,29 @@ public R accept(IPlanVisitor visitor, C context) { @Override public NonAlignedAggregationTreeDeviceViewScanNode clone() { - return new NonAlignedAggregationTreeDeviceViewScanNode( - getPlanNodeId(), - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice, - projection, - aggregations, - groupingSets, - preGroupedSymbols, - step, - groupIdSymbol, - getTreeDBName(), - getMeasurementColumnNameMap()); + return copyDeviceEntryDataSetTo( + new NonAlignedAggregationTreeDeviceViewScanNode( + getPlanNodeId(), + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice, + projection, + aggregations, + groupingSets, + preGroupedSymbols, + step, + groupIdSymbol, + getTreeDBName(), + getMeasurementColumnNameMap())); } protected PlanNodeType getPlanNodeType() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeAlignedDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeAlignedDeviceViewScanNode.java index 7246f04dacc6..162de5084211 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeAlignedDeviceViewScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeAlignedDeviceViewScanNode.java @@ -81,22 +81,23 @@ public R accept(IPlanVisitor visitor, C context) { @Override public TreeAlignedDeviceViewScanNode clone() { - return new TreeAlignedDeviceViewScanNode( - getPlanNodeId(), - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice, - treeDBName, - measurementColumnNameMap); + return copyDeviceEntryDataSetTo( + new TreeAlignedDeviceViewScanNode( + getPlanNodeId(), + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice, + treeDBName, + measurementColumnNameMap)); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeDeviceViewScanNode.java index da308b749914..59109d5df369 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeDeviceViewScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeDeviceViewScanNode.java @@ -112,22 +112,23 @@ public R accept(IPlanVisitor visitor, C context) { @Override public TreeDeviceViewScanNode clone() { - return new TreeDeviceViewScanNode( - getPlanNodeId(), - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice, - treeDBName, - measurementColumnNameMap); + return copyDeviceEntryDataSetTo( + new TreeDeviceViewScanNode( + getPlanNodeId(), + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice, + treeDBName, + measurementColumnNameMap)); } protected static void serializeMemberVariables( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java index 60db4f93f499..bd56fdc0e77a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java @@ -81,22 +81,23 @@ public R accept(IPlanVisitor visitor, C context) { @Override public TreeNonAlignedDeviceViewScanNode clone() { - return new TreeNonAlignedDeviceViewScanNode( - getPlanNodeId(), - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice, - treeDBName, - measurementColumnNameMap); + return copyDeviceEntryDataSetTo( + new TreeNonAlignedDeviceViewScanNode( + getPlanNodeId(), + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice, + treeDBName, + measurementColumnNameMap)); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java index 9f3c5c32e25a..984a711cc1e7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java @@ -221,7 +221,7 @@ private boolean isSingleDeviceEntry(DeviceTableScanNode tableScanNode) { // optimizer cannot safely use the single-device shortcut here. return false; } - return tableScanNode.getDeviceEntries().size() < 2; + return tableScanNode.getDeviceEntryCount() < 2; } private List getTagColumnsInTableStore( 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 1ceb4c86f6c1..f24d48db78fd 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 @@ -22,7 +22,6 @@ import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.partition.DataPartition; -import org.apache.iotdb.commons.partition.DataPartitionQueryParam; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.TableScanNode; import org.apache.iotdb.commons.queryengine.plan.relational.analyzer.NodeRef; @@ -67,9 +66,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.analyzer.predicate.PredicateCombineIntoTableScanChecker; import org.apache.iotdb.db.queryengine.plan.relational.analyzer.predicate.PredicatePushIntoMetadataChecker; import org.apache.iotdb.db.queryengine.plan.relational.analyzer.predicate.schema.ConvertSchemaPredicateToFilterVisitor; -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.planner.EqualityInference; import org.apache.iotdb.db.queryengine.plan.relational.planner.IrExpressionInterpreter; import org.apache.iotdb.db.queryengine.plan.relational.planner.IrTypeAnalyzer; @@ -91,6 +90,8 @@ 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 +747,7 @@ private void getDeviceEntriesWithDataPartitions( } long startTime = System.nanoTime(); - final Map> deviceEntriesMap = + final DeviceEntryDataSetResult deviceEntryDataSetResult = metadata.indexScan( tableScanNode.getQualifiedObjectName(), metadataExpressions.stream() @@ -756,84 +757,85 @@ private void getDeviceEntriesWithDataPartitions( expression, tableScanNode.getAssignments())) .collect(Collectors.toList()), attributeColumns, - queryContext); - if (deviceEntriesMap.size() > 1) { - throw new SemanticException( - DataNodeQueryMessages.TREE_DEVICE_VIEW_WITH_MULTIPLE_DATABASES - + deviceEntriesMap.keySet() - + DataNodeQueryMessages.IS_UNSUPPORTED_YET); - } - final String deviceDatabase = - !deviceEntriesMap.isEmpty() ? deviceEntriesMap.keySet().iterator().next() : null; - final List deviceEntries = - Objects.nonNull(deviceDatabase) - ? deviceEntriesMap.get(deviceDatabase) - : Collections.emptyList(); - - tableScanNode.setDeviceEntries(deviceEntries); - if (deviceEntries.stream() - .anyMatch(deviceEntry -> deviceEntry instanceof NonAlignedDeviceEntry)) { - tableScanNode.setContainsNonAlignedDevice(); - } - - if (tableScanNode instanceof TreeDeviceViewScanNode) { - ((TreeDeviceViewScanNode) tableScanNode).setTreeDBName(deviceDatabase); - } - - final long schemaFetchCost = System.nanoTime() - startTime; - QueryPlanCostMetricSet.getInstance().recordTablePlanCost(SCHEMA_FETCHER, schemaFetchCost); - queryContext.setFetchSchemaCost(schemaFetchCost); - - if (deviceEntries.isEmpty()) { - if (analysis.noAggregates() && !analysis.hasJoinNode()) { - // no device entries, queries(except aggregation and join) can be finished - analysis.setEmptyDataSource(true); - analysis.setFinishQueryAfterAnalyze(); + queryContext, + tableScanNode.getPlanNodeId()); + final DeviceEntryDataSet deviceEntryDataSet = deviceEntryDataSetResult.getDataSet(); + boolean dataSetTransferred = false; + try { + final String deviceDatabase = deviceEntryDataSetResult.getDatabase(); + if (deviceEntryDataSetResult.containsNonAlignedDevice()) { + tableScanNode.setContainsNonAlignedDevice(); } - } else { - final Filter timeFilter = - tableScanNode - .getTimePredicate() - .map( - value -> - value.accept( - new ConvertPredicateToTimeFilterVisitor( - queryContext.getZoneId(), TimestampPrecisionUtils.currPrecision), - null)) - .orElse(null); - - tableScanNode.setTimeFilter(timeFilter); - - startTime = System.nanoTime(); - final DataPartition dataPartition = - fetchDataPartitionByDevices( - // for tree view, we need to pass actual tree db name to this method - tableScanNode instanceof TreeDeviceViewScanNode - ? deviceDatabase - : tableScanNode.getQualifiedObjectName().getDatabaseName(), - deviceEntries, - timeFilter); - - if (dataPartition.getDataPartitionMap().size() > 1) { - throw new IllegalStateException( - DataNodeQueryMessages - .QUERY_EXCEPTION_TABLE_MODEL_CAN_ONLY_PROCESS_DATA_ONLY_IN_ONE_DATABASE_YET_AB8C1EF5); + if (tableScanNode instanceof TreeDeviceViewScanNode) { + ((TreeDeviceViewScanNode) tableScanNode).setTreeDBName(deviceDatabase); } - if (dataPartition.getDataPartitionMap().isEmpty()) { + final long schemaFetchCost = System.nanoTime() - startTime; + QueryPlanCostMetricSet.getInstance().recordTablePlanCost(SCHEMA_FETCHER, schemaFetchCost); + queryContext.setFetchSchemaCost(schemaFetchCost); + + if (deviceEntryDataSet.getEntryCount() == 0) { if (analysis.noAggregates() && !analysis.hasJoinNode()) { - // no data partitions, queries(except aggregation and join) can be finished + // no device entries, queries(except aggregation and join) can be finished analysis.setEmptyDataSource(true); analysis.setFinishQueryAfterAnalyze(); } } else { - analysis.upsertDataPartition(dataPartition); + final Filter timeFilter = + tableScanNode + .getTimePredicate() + .map( + value -> + value.accept( + new ConvertPredicateToTimeFilterVisitor( + queryContext.getZoneId(), TimestampPrecisionUtils.currPrecision), + null)) + .orElse(null); + + tableScanNode.setTimeFilter(timeFilter); + + startTime = System.nanoTime(); + final DataPartition dataPartition = + fetchDataPartitionByDeviceDataSet( + // for tree view, we need to pass actual tree db name to this method + tableScanNode instanceof TreeDeviceViewScanNode + ? deviceDatabase + : tableScanNode.getQualifiedObjectName().getDatabaseName(), + deviceEntryDataSet, + timeFilter); + + if (dataPartition.getDataPartitionMap().size() > 1) { + throw new IllegalStateException( + DataNodeQueryMessages + .QUERY_EXCEPTION_TABLE_MODEL_CAN_ONLY_PROCESS_DATA_ONLY_IN_ONE_DATABASE_YET_AB8C1EF5); + } + + if (dataPartition.getDataPartitionMap().isEmpty()) { + if (analysis.noAggregates() && !analysis.hasJoinNode()) { + // no data partitions, queries(except aggregation and join) can be finished + analysis.setEmptyDataSource(true); + analysis.setFinishQueryAfterAnalyze(); + } + } else { + analysis.upsertDataPartition(dataPartition); + } + + final long fetchPartitionCost = System.nanoTime() - startTime; + QueryPlanCostMetricSet.getInstance() + .recordTablePlanCost(PARTITION_FETCHER, fetchPartitionCost); + queryContext.setFetchPartitionCost(fetchPartitionCost); } - final long fetchPartitionCost = System.nanoTime() - startTime; - QueryPlanCostMetricSet.getInstance() - .recordTablePlanCost(PARTITION_FETCHER, fetchPartitionCost); - queryContext.setFetchPartitionCost(fetchPartitionCost); + tableScanNode.setDeviceEntryDataSet(deviceEntryDataSet); + dataSetTransferred = true; + } finally { + if (!dataSetTransferred) { + try { + deviceEntryDataSet.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } } } @@ -1363,10 +1365,10 @@ public PlanNode visitUnion(UnionNode node, RewriteContext context) { return node; } - private DataPartition fetchDataPartitionByDevices( + private DataPartition fetchDataPartitionByDeviceDataSet( final String database, // for tree view, database should be the real tree db name with `root.` prefix - final List deviceEntries, + final DeviceEntryDataSet deviceEntryDataSet, final Filter globalTimeFilter) { final Pair, Pair> res = getTimePartitionSlotList(globalTimeFilter, queryContext); @@ -1379,18 +1381,11 @@ private DataPartition fetchDataPartitionByDevices( CONFIG.getSeriesPartitionSlotNum()); } - final List dataPartitionQueryParams = - deviceEntries.stream() - .map( - deviceEntry -> - new DataPartitionQueryParam( - deviceEntry.getDeviceID(), res.left, res.right.left, res.right.right)) - .collect(Collectors.toList()); - if (res.right.left || res.right.right) { - return metadata.getDataPartitionWithUnclosedTimeRange(database, dataPartitionQueryParams); + return metadata.getDataPartitionWithUnclosedTimeRange( + database, deviceEntryDataSet, res.left, res.right.left, res.right.right); } else { - return metadata.getDataPartition(database, dataPartitionQueryParams); + return metadata.getDataPartition(database, deviceEntryDataSet, res.left); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/QueryCardinalityUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/QueryCardinalityUtil.java index 0d8427c985f9..a59b2e0b4f83 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/QueryCardinalityUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/QueryCardinalityUtil.java @@ -218,10 +218,10 @@ public Range visitAggregationTableScan(AggregationTableScanNode node, Void && !node.getProjection().getMap().isEmpty()) { // also exist date_bin return Range.atLeast(0L); } else { - return Range.atMost((long) node.getDeviceEntries().size()); + return Range.atMost(node.getDeviceEntryCount()); } } else { - return Range.singleton((long) node.getDeviceEntries().size()); + return Range.singleton(node.getDeviceEntryCount()); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java index b5ed662cdae4..f4af36eebea8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java @@ -141,7 +141,7 @@ public PlanNode visitStreamSort(StreamSortNode node, Context context) { @Override public PlanNode visitDeviceTableScan(DeviceTableScanNode node, Context context) { - context.addDeviceEntrySize(node.getDeviceEntries().size()); + context.addDeviceEntrySize(node.getDeviceEntryCount()); context.setTimeColumnName(node.getTimeColumn().map(Symbol::getName).orElse(null)); return node; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java index a938793cd2a5..131ba8af383c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java @@ -200,7 +200,7 @@ public PlanAndMappings visitTreeDeviceViewScan( newAssignments.put(newSymbol, handle); }); - return new PlanAndMappings( + TreeDeviceViewScanNode rewrittenNode = new TreeDeviceViewScanNode( node.getPlanNodeId(), node.getQualifiedObjectName(), @@ -216,8 +216,8 @@ public PlanAndMappings visitTreeDeviceViewScan( node.isPushLimitToEachDevice(), node.containsNonAlignedDevice(), node.getTreeDBName(), - node.getMeasurementColumnNameMap()), - mapping); + node.getMeasurementColumnNameMap()); + return new PlanAndMappings(node.copyDeviceEntryDataSetTo(rewrittenNode), mapping); } @Override @@ -235,7 +235,7 @@ public PlanAndMappings visitDeviceTableScan(DeviceTableScanNode node, UnaliasCon newAssignments.put(newSymbol, handle); }); - return new PlanAndMappings( + DeviceTableScanNode rewrittenNode = new DeviceTableScanNode( node.getPlanNodeId(), node.getQualifiedObjectName(), @@ -249,8 +249,8 @@ public PlanAndMappings visitDeviceTableScan(DeviceTableScanNode node, UnaliasCon node.getPushDownLimit(), node.getPushDownOffset(), node.isPushLimitToEachDevice(), - node.containsNonAlignedDevice()), - mapping); + node.containsNonAlignedDevice()); + return new PlanAndMappings(node.copyDeviceEntryDataSetTo(rewrittenNode), mapping); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java index b18af7913851..9922c8d78969 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java @@ -214,6 +214,7 @@ public static Pair split( node.getGroupIdSymbol(), aggregationTreeDeviceViewScanNode.getTreeDBName(), aggregationTreeDeviceViewScanNode.getMeasurementColumnNameMap()); + node.copyDeviceEntryDataSetTo(rightResult); return new Pair<>( new AggregationNode( 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 02999d57b010..7c4a45cfcacd 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,19 @@ public void renderPlanStatistics(MPPQueryContext context) { 0, String.format( "Fetch Schema Cost: %.3f ms", context.getFetchSchemaCost() * NS_TO_MS_FACTOR)); + addLine( + planHeader, + 1, + String.format( + "Disk IO Size for DeviceEntry During FetchSchema: %d bytes", + context.getDiskIOSizeForDeviceEntryDuringFetchSchema())); + addLine( + planHeader, + 1, + String.format( + "Disk IO Time Cost for DeviceEntry During FetchSchema: %.3f ms", + context.getDiskIOTimeCostForDeviceEntryDuringFetchSchema() * NS_TO_MS_FACTOR)); + addLine(planHeader, 1, String.format("DeviceEntry Count: %d", context.getDeviceEntryCount())); 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 74571ad152f9..ef887034dac1 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,13 @@ public void renderPlanStatistics(MPPQueryContext context) { "fetchPartitionCostMs", formatMs(context.getFetchPartitionCost() * NS_TO_MS_FACTOR)); planStatistics.addProperty( "fetchSchemaCostMs", formatMs(context.getFetchSchemaCost() * NS_TO_MS_FACTOR)); + planStatistics.addProperty( + "diskIOSizeForDeviceEntryDuringFetchSchema", + context.getDiskIOSizeForDeviceEntryDuringFetchSchema()); + planStatistics.addProperty( + "diskIOTimeCostForDeviceEntryDuringFetchSchemaMs", + formatMs(context.getDiskIOTimeCostForDeviceEntryDuringFetchSchema() * NS_TO_MS_FACTOR)); + planStatistics.addProperty("deviceEntryCount", context.getDeviceEntryCount()); 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 edb13217db22..aa46addce37c 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 @@ -27,6 +27,9 @@ public class QueryPlanStatistics { private long logicalOptimizationCost; private long distributionPlanCost; private long dispatchCost = 0; + private long diskIOSizeForDeviceEntryDuringFetchSchema; + private long diskIOTimeCostForDeviceEntryDuringFetchSchema; + private long deviceEntryCount; public void setAnalyzeCost(long analyzeCost) { this.analyzeCost = analyzeCost; @@ -83,4 +86,25 @@ public void recordDispatchCost(long dispatchCost) { public long getDispatchCost() { return dispatchCost; } + + public void recordDeviceEntryDiskIODuringFetchSchema(long bytes, long timeCost) { + diskIOSizeForDeviceEntryDuringFetchSchema += bytes; + diskIOTimeCostForDeviceEntryDuringFetchSchema += timeCost; + } + + public void recordDeviceEntryCount(long count) { + deviceEntryCount += count; + } + + public long getDiskIOSizeForDeviceEntryDuringFetchSchema() { + return diskIOSizeForDeviceEntryDuringFetchSchema; + } + + public long getDiskIOTimeCostForDeviceEntryDuringFetchSchema() { + return diskIOTimeCostForDeviceEntryDuringFetchSchema; + } + + public long getDeviceEntryCount() { + return deviceEntryCount; + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakePartitionFetcherImpl.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakePartitionFetcherImpl.java index 4c0c39107fef..85c2673cc91b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakePartitionFetcherImpl.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/FakePartitionFetcherImpl.java @@ -33,6 +33,7 @@ import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor; import org.apache.iotdb.commons.path.PathPatternTree; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; import org.apache.tsfile.file.metadata.IDeviceID; @@ -216,12 +217,28 @@ public DataPartition getDataPartition( return dataPartition; } + @Override + public DataPartition getDataPartition( + String database, DeviceEntryDataSet dataSet, List timePartitionSlots) { + throw new UnsupportedOperationException(); + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( Map> sgNameToQueryParamsMap) { return getDataPartition(sgNameToQueryParamsMap); } + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots, + boolean needLeftAll, + boolean needRightAll) { + throw new UnsupportedOperationException(); + } + @Override public DataPartition getOrCreateDataPartition( Map> sgNameToQueryParamsMap) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util.java index e4d3a0c4ba9c..03615e558065 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util.java @@ -53,6 +53,7 @@ import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand; import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; import org.apache.iotdb.db.queryengine.plan.planner.LogicalPlanner; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; import org.apache.iotdb.db.queryengine.plan.statement.Statement; import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; @@ -406,12 +407,30 @@ public DataPartition getDataPartition( return ANALYSIS.getDataPartitionInfo(); } + @Override + public DataPartition getDataPartition( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots) { + return ANALYSIS.getDataPartitionInfo(); + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( Map> sgNameToQueryParamsMap) { return ANALYSIS.getDataPartitionInfo(); } + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots, + boolean needLeftAll, + boolean needRightAll) { + return ANALYSIS.getDataPartitionInfo(); + } + @Override public DataPartition getOrCreateDataPartition( Map> sgNameToQueryParamsMap) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util2.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util2.java index bb37c554eab7..303c2268911d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util2.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/Util2.java @@ -50,6 +50,7 @@ import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; import org.apache.iotdb.db.queryengine.plan.planner.LogicalPlanner; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet; import org.apache.iotdb.db.queryengine.plan.statement.Statement; import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; @@ -299,12 +300,30 @@ public DataPartition getDataPartition( return ANALYSIS.getDataPartitionInfo(); } + @Override + public DataPartition getDataPartition( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots) { + return ANALYSIS.getDataPartitionInfo(); + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( Map> sgNameToQueryParamsMap) { return ANALYSIS.getDataPartitionInfo(); } + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots, + boolean needLeftAll, + boolean needRightAll) { + return ANALYSIS.getDataPartitionInfo(); + } + @Override public DataPartition getOrCreateDataPartition( Map> sgNameToQueryParamsMap) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java index 4794b19994e8..4750ae6867b0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java @@ -19,12 +19,14 @@ package org.apache.iotdb.db.queryengine.plan.relational.analyzer; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.partition.DataPartition; import org.apache.iotdb.commons.partition.DataPartitionQueryParam; import org.apache.iotdb.commons.partition.SchemaNodeManagementPartition; import org.apache.iotdb.commons.partition.SchemaPartition; import org.apache.iotdb.commons.path.PathPatternTree; 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.metadata.ColumnMetadata; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; @@ -46,6 +48,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; import org.apache.iotdb.db.queryengine.plan.relational.metadata.OperatorNotFoundException; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator; +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.InMemoryDeviceEntryDataSet; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; import org.apache.iotdb.udf.api.relational.TableFunction; @@ -279,7 +284,20 @@ public boolean canCoerce(Type from, Type to) { } @Override - public Map> indexScan( + public DeviceEntryDataSetResult indexScan( + QualifiedObjectName tableName, + List expressionList, + List attributeColumns, + MPPQueryContext context, + PlanNodeId planNodeId) { + final Map> deviceEntries = + indexScanEntries(tableName, expressionList, attributeColumns, context); + final String database = deviceEntries.keySet().iterator().next(); + return new DeviceEntryDataSetResult( + database, new InMemoryDeviceEntryDataSet(deviceEntries.get(database)), false); + } + + private Map> indexScanEntries( QualifiedObjectName tableName, List expressionList, List attributeColumns, @@ -387,12 +405,28 @@ public DataPartition getDataPartition( return DATA_PARTITION; } + @Override + public DataPartition getDataPartition( + String database, DeviceEntryDataSet dataSet, List timePartitionSlots) { + return DATA_PARTITION; + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( String database, List sgNameToQueryParamsMap) { return DATA_PARTITION; } + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots, + boolean needLeftAll, + boolean needRightAll) { + return DATA_PARTITION; + } + @Override public TableFunction getTableFunction(String functionName) { return null; @@ -430,12 +464,30 @@ public DataPartition getDataPartition( return DATA_PARTITION; } + @Override + public DataPartition getDataPartition( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots) { + return DATA_PARTITION; + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( Map> sgNameToQueryParamsMap) { return DATA_PARTITION; } + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots, + boolean needLeftAll, + boolean needRightAll) { + return DATA_PARTITION; + } + @Override public DataPartition getOrCreateDataPartition( Map> sgNameToQueryParamsMap) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java index ac266f19398c..b371bbaa45a0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.queryengine.plan.relational.analyzer; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.partition.DataPartition; import org.apache.iotdb.commons.partition.DataPartitionQueryParam; @@ -26,6 +27,7 @@ import org.apache.iotdb.commons.partition.SchemaPartition; import org.apache.iotdb.commons.path.PathPatternTree; 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.TableBuiltinTableFunction; import org.apache.iotdb.commons.queryengine.plan.relational.function.arithmetic.SubtractionResolver; @@ -59,6 +61,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.OperatorNotFoundException; import org.apache.iotdb.db.queryengine.plan.relational.metadata.TreeDeviceViewSchema; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator; +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.InMemoryDeviceEntryDataSet; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.iotdb.db.schemaengine.table.InformationSchemaUtils; import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq; @@ -332,7 +337,23 @@ public boolean canCoerce(final Type from, final Type to) { } @Override - public Map> indexScan( + public DeviceEntryDataSetResult indexScan( + final QualifiedObjectName tableName, + final List expressionList, + final List attributeColumns, + final MPPQueryContext context, + final PlanNodeId planNodeId) { + final Map> deviceEntries = + indexScanEntries(tableName, expressionList, attributeColumns, context); + final String database = deviceEntries.keySet().iterator().next(); + final List entries = deviceEntries.get(database); + return new DeviceEntryDataSetResult( + database, + new InMemoryDeviceEntryDataSet(entries), + entries.stream().anyMatch(NonAlignedDeviceEntry.class::isInstance)); + } + + private Map> indexScanEntries( final QualifiedObjectName tableName, final List expressionList, final List attributeColumns, @@ -541,12 +562,30 @@ public DataPartition getDataPartition( return TREE_DB1.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION; } + @Override + public DataPartition getDataPartition( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots) { + return TREE_DB1.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION; + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( final String database, final List sgNameToQueryParamsMap) { return TREE_DB1.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION; } + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + final String database, + final DeviceEntryDataSet dataSet, + final List timePartitionSlots, + final boolean needLeftAll, + final boolean needRightAll) { + return TREE_DB1.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION; + } + @Override public TableFunction getTableFunction(String functionName) { if ("EXCLUDE".equalsIgnoreCase(functionName)) { @@ -606,6 +645,14 @@ public DataPartition getDataPartition( : TABLE_DATA_PARTITION; } + @Override + public DataPartition getDataPartition( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots) { + return TREE_VIEW_DB.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION; + } + @Override public DataPartition getDataPartitionWithUnclosedTimeRange( Map> sgNameToQueryParamsMap) { @@ -614,6 +661,16 @@ public DataPartition getDataPartitionWithUnclosedTimeRange( : TABLE_DATA_PARTITION; } + @Override + public DataPartition getDataPartitionWithUnclosedTimeRange( + String database, + DeviceEntryDataSet dataSet, + List timePartitionSlots, + boolean needLeftAll, + boolean needRightAll) { + return TREE_VIEW_DB.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION; + } + @Override public DataPartition getOrCreateDataPartition( Map> sgNameToQueryParamsMap) { 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 000000000000..d643a91c8311 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializerTest.java @@ -0,0 +1,157 @@ +/* + * 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.ByteBuffer; +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; + private String originalSortTmpDir; + + @Before + public void setUp() throws Exception { + queryDirectory = Files.createTempDirectory("device-entry-spill-test"); + originalSortTmpDir = IoTDBDescriptor.getInstance().getConfig().getSortTmpDir(); + IoTDBDescriptor.getInstance().getConfig().setSortTmpDir(queryDirectory.toString()); + } + + @After + public void tearDown() throws Exception { + DeviceEntrySpillManager.getInstance().clearStaleData(); + Files.deleteIfExists(queryDirectory.resolve("device-entry")); + Files.deleteIfExists(queryDirectory); + IoTDBDescriptor.getInstance().getConfig().setSortTmpDir(originalSortTmpDir); + } + + @Test + public void testKeepSmallDataSetInline() throws Exception { + List 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 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 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 segments; + try (java.util.stream.Stream 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 createEntries(int count) { + List 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 b83501a277c4..4dd52b45d710 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,8 @@ public void testRenderPlanStatistics() { context.setLogicalPlanCost(4000000L); // 4ms context.setLogicalOptimizationCost(5000000L); // 5ms context.setDistributionPlanCost(6000000L); // 6ms + context.recordDeviceEntryDiskIODuringFetchSchema(8192L, 7000000L); // 8 KiB, 7ms + context.recordDeviceEntryCount(3); drawer.renderPlanStatistics(context); @@ -78,6 +80,10 @@ public void testRenderPlanStatistics() { 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("diskIOSizeForDeviceEntryDuringFetchSchema").getAsLong()); + assertEquals( + 7.0, planStats.get("diskIOTimeCostForDeviceEntryDuringFetchSchemaMs").getAsDouble(), 0.01); + assertEquals(3L, planStats.get("deviceEntryCount").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 c4a06b68b149..81573e47f541 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