Hi all,
I would like to propose an optimization for the memory usage of table-model
full-table queries. I plan to work on this for current version and would
appreciate feedback from the community before finalizing the implementation.
Background
==========
For table-model queries without a highly selective TAG or ATTRIBUTE predicate,
the Coordinator may need to enumerate metadata for all matching devices.
Typical queries include:
SELECT * FROM t1 WHERE time >= ? AND time < ?;
SELECT count(*) FROM t1 WHERE time >= ? AND time < ?;
SELECT tag1, count(*) FROM t1 GROUP BY tag1;
The Coordinator already fetches DeviceEntry objects from SchemaRegions in a
streaming manner. However, the results are eventually collected into an
in-memory List<DeviceEntry>. The list is then used to obtain DataPartitions,
split scan nodes by DataRegion, identify cross-region devices, derive
aggregation and ordering properties, and send device metadata to
FragmentInstances.
For tables containing millions or tens of millions of devices, the DeviceEntry
list will be very large and cause MemoryNotEnough exception.The goal of this
work is to bound DeviceEntry memory usage on both the Coordinator and
execution DataNodes while preserving the current in-memory path for small
queries.
Configuration and memory budget
===============================
The proposed configuration item is:
table_query_device_entry_batch_size_in_bytes
The configuration is intended to be a SYSTEM-level setting and supports hot
reload. No restart should be required for a new value to take effect.
The default value is 0 and it means that IoTDB
calculates the effective batch target from the currently available query
execution memory and query_thread_count. In simplified form:
table_query_device_entry_batch_size_in_bytes = available query execution
memory / query_thread_count
The implementation will apply the configured target as a query-wide memory
budget during distributed planning, rather than independently granting the
full target to every DataRegion. All active Region materializers and sort
buffers share that budget. This prevents a query with many Regions from
allocating one full batch budget per Region. When the shared budget is under
pressure, the active Region materializer with the largest retained data is
spilled first, and subsequent data is written directly to disk.
The exact available-memory source and the SYSTEM configuration reload path will
follow the existing IoTDB query-memory configuration mechanisms. The effective
value and the reason for automatic or explicit selection should be observable
in query diagnostics.
Four-module architecture
========================
Module 1: Streaming DeviceEntry materialization and spill on the Coordinator
-------------------------------------------------------------------------------
Each DeviceEntry returned by the existing SchemaRegion stream is appended to a
query-scoped materializer.
1. Before the memory target is exceeded, entries remain in the existing
in-memory list.
2. Once the shared query budget requires spilling, the buffered entries are
written to immutable Raw Segments and the list memory is released.
3. Subsequent entries are accumulated in a bounded buffer and written to
additional Raw Segments.
4. If the query stays below the target, no temporary file is created and the
current planning and execution path remains unchanged.
The memory target is configured by:
table_query_device_entry_batch_size_in_bytes
Segment boundaries are aligned with DeviceEntry records, so an individual
entry is never split across files. A single entry larger than the target may
occupy one oversized Segment.
Module 2: Streaming grouping by DataRegion and optional external sorting
------------------------------------------------------------------------
During distributed planning, the Coordinator reads the Raw Dataset once and
routes each DeviceEntry to one or more TRegionReplicaSet values according to
the DataPartition.
This single pass performs the following work together:
1. Generate an independent Region Dataset for every target DataRegion.
2. Handle devices spanning multiple DataRegions.
3. Record per-region entry counts and cross-region information.
4. Derive the information required by scan-node distribution and Partial/Final
aggregation planning.
Each Region gets an independent DeviceTableScanNode. The distributed PlanNode
carries only a Dataset Handle instead of all DeviceEntry objects.
Region materializers participate in one query-wide memory budget. They do not
each receive an independent full threshold. Without expected ordering, entries
are written directly to bounded Region Segments. If expected ordering is
required, a query-scoped external sorter creates sorted runs and performs
bounded fan-in merge passes before publishing the final ordered Region
Segments.
Module 3: DeviceEntry Segment service
--------------------------------------
FragmentInstances obtain Region Segments through an DeviceEntry file
service.
1. A request identifies a Segment by controlled IDs such as query ID, dataset
ID, and segment ID; arbitrary file paths are not accepted.
2. A local FragmentInstance reads a local Segment directly.
3. A remote FragmentInstance fetches one complete immutable Segment through
internal RPC.
4. Fetch retries operate on the complete Segment.
Temporary files are query-scoped. Normal completion, failure, cancellation,
and timeout trigger cleanup. A DataNode also removes stale Fragment temporary
directories during startup to cover a crash before normal cleanup.
Module 4: Batched FI scanning and QueryDataSource lifecycle
------------------------------------------------------------
On the execution side, table scan operators consume DeviceEntry objects
through a common batch source:
1. The current in-memory path uses a list-backed source and returns only one
batch.
2. The spill path uses a Segment-backed source and returns one Segment as one
batch.
For every batch, the FI performs this lifecycle:
fetch/read one DeviceEntry Segment
-> deserialize the current DeviceEntry batch
-> construct paths for the batch
-> initialize a batch-scoped QueryDataSource
-> retain the referenced TsFileResources
-> scan all devices in the batch
-> release the QueryDataSource and file references
-> release DeviceEntry objects and continue with the next Segment
The QueryDataSource is therefore no longer assumed to be initialized only once
for the entire FI in spill mode. Each batch owns an independent lease, and the
FragmentInstanceContext tracks active leases so that normal completion and
failure cleanup are symmetric. If DataRegion initialization cannot acquire the
required lock, the fetched DeviceEntry batch is retained and only
QueryDataSource initialization is retried; the batch must not be consumed or
treated as finished before initialization succeeds.
End-to-end flow
===============
SchemaRegion DeviceEntry stream
-> query-scoped in-memory materializer
-> shared budget exceeded: Raw Segments
-> one-pass DataPartition routing by DataRegion
-> Region materialization or external sorting
-> Region Dataset Handles in distributed PlanNodes
-> local read or internal RPC Segment transfer
-> FI batch-scoped QueryDataSource initialization
-> raw/aggregation scan
-> batch release and next Segment
-> query/fragment cleanup
Compatibility and scope
=======================
1. Small queries remain on the current in-memory path and do not create files,
invoke the file service, or repeatedly initialize QueryDataSource.
2. The proposal does not reduce the SchemaRegion enumeration cost; it bounds
the memory required to materialize and distribute the result.
3. No persistent data format is changed. All new files are temporary and may
be removed during rollback.
4. Mixed-version clusters require a capability check. The Handle and
batched-scan path should only be enabled when the target execution node
supports it.
5. The first version covers full raw queries and full aggregation queries. The
Coordinator and distributed-planning portions also cover Last Query and
TreeDeviceView, while their existing BE execution paths remain unchanged as
described above. External TsFile keeps its DeviceTask Run path.
Thanks,
Weihao Li
