Repository navigation
Conversation
Validate unordered results and EXPLAIN semantics independently on every DataNode while retaining ordered assertions. Add duplicate-preserving checks and deterministic regressions, and resolve receiver column names from metadata.
Read the address column rather than the socket state from lsof output and cover a real listening socket. Allow ClientManagerTest to use an explicitly configured port when local services occupy its default.
Retain independent plan-node quotas for dependent operators and internal pipelines, and fall back when a split side cannot hold one TsBlock. Make memory admission atomic without speculative charges and add regression coverage for MergeSort, queue forwarding, and contention.
Determine charged directions from remote sources and fragment sinks after planning. Split shared budgets only for bidirectional fragments and validate the maximum TsBlock size only for active directions. Add full-budget and local-input ownership regressions.
Cancel pending reservations before FI deregistration and keep legacy accounting independent per plan node. Guard late Collect mode changes, cover planner entry points, and order pipe verification queries by their complete primary key.
ab11600 to
5d24b4c
Compare
JackieTien97
left a comment
There was a problem hiding this comment.
Thanks for the quick follow-up in 788a995. I re-ran my probes on it. The shared account is now registered under the same fragmentId.instanceId key that the handles and deregistration use. With four real SourceHandles and F = 1024 B, the FI now reserves at most 1024 B (4 × 1024 B before), and the query entry is gone after deregistration.
Requesting changes for two problems that are still present at 788a995:
- Ready consumption under order-sensitive operators (inline on
SharedExchangeMemoryEligibility):FILL METHOD PREVIOUS/LINEARwithout FILL_GROUP anddiff()sit directly on a ready Collect.- The child FI that feeds COPY TO is eligible as well.
- Filled and
diff()values become wrong and can change between runs. No IT covers this.
- Cancelling pending reservations at FI deregistration (inline on
MemoryPool):- It fixes a base bug where another query's
free()hits the dead FI's request, throws an NPE and leaks pool memory. - But now
SourceHandle.abort()fails for a handle that only has a pending request. - A request published while the FI is being deregistered still makes every later
free()on the DataNode throw.
- It fixes a base bug where another query's
Non-blocking:
- FIs that never enable ready mode still lose
GroupNodeand streaming aggregation: they get a full sort, or a hash aggregation that cannot spill (inline onvisitGroup). - Many multi-region queries without ORDER BY now return rows in an order that varies between runs; several ITs had to switch to unordered or per-node checks. This is worth a line in the release notes.
- Nits:
MemoryPool'smaxBytesPerFragmentInstanceconstructor parameter is now only validated, never used.EXCEPTION_BYTESTORESERVE_SHOULD_BE_IN_LEFT_PAREN_0_COMMA_MAXBYTESPERFI_RIGHT_BRACKET_DOT_M_0753BB69(en and zh) is no longer referenced.FragmentInstanceContext.inputMaxBytes/outputMaxBytesare only read back insidesetExchangeMemoryLimits; local variables would do.
How I checked:
| // TopK's input ordering assumptions remain unchanged, including its single-child form. | ||
| if (node instanceof TopKNode) { | ||
| return false; | ||
| } |
There was a problem hiding this comment.
Order-sensitive single-input operators need an ordered Collect too. TopK is excluded here because it depends on input order. Fill without FILL_GROUP and diff() depend on input order as well, but they are not excluded.
PlanTester on 788a995 (the plans are the same as on base; only the runtime mode changed):
| query | root FI | eligible |
|---|---|---|
SELECT time, tag1, tag2, tag3, s1 FROM table1 FILL METHOD PREVIOUS |
PreviousFill → Collect(3 × Exchange) |
true |
SELECT time, tag1, tag2, tag3, diff(s1) AS d FROM table1 |
Project[diff(s1)] → Collect(3 × Exchange) |
true |
SELECT time, s1 FROM table1 WHERE tag1='shenzhen' AND tag2='B2' AND tag3='ZZ' FILL METHOD PREVIOUS |
PreviousFill → Collect(3 × Exchange) |
true |
same, FILL METHOD LINEAR |
LinearFill → Collect |
true |
same, FILL METHOD PREVIOUS ORDER BY time |
Sort → PreviousFill → Collect |
true |
SELECT time, diff(s1) AS d FROM table1 WHERE tag1='shenzhen' AND tag2='B2' AND tag3='ZZ' |
Project[diff(s1)] → Collect |
true |
Why the values go wrong:
- In ready mode the Collect hands out one block per ready child in turn;
CollectOperatorTest#testAllReadyChildrenHaveLinearSelectionCostasserts exactly that. So a device's rows from one region are interrupted by blocks from other regions. - Fill without FILL_GROUP and
diff()carry state over from the previous row. After a block boundary, a NULL is filled from a row of whatever block arrived in between, anddiff()is computed against that row. The values can change from run to run. - This hits multi-device queries without FILL_GROUP, and single devices whose time partitions live in different regions.
ORDER BY timedoes not help, because the Sort sits above the fill.- On base the children were drained one after another. Each region's rows stayed contiguous, and the same query returned the same values on every run.
COPY TO: the check cannot see order requirements outside its own FI. COPY table1 TO 'x.tsfile' plans CopyTo → Exchange in the root FI, and IdentitySink → Collect(3 × Exchange) in a child FI that is eligible. The TsFile writer rejects non-increasing timestamps per device (AlignedChunkGroupWriterImpl#checkIsHistoryData). So COPY TO of a device whose partitions live in two regions can now fail depending on block arrival order. Base already depends on the fixed child order there, but it at least behaves the same on every run.
Coverage: IoTDBCollectFanInIT only has a CONSTANT fill. It has no PREVIOUS/LINEAR fill, diff() or COPY TO case, so CI does not catch this.
Suggestion: decide ordering in the distributed planner instead of inferring it from the FI shape on the DataNode.
- Mark a Collect/Union as unordered only when nothing above it relies on input order, in this FI or in the parent FI.
- Serialize that flag, with ordered as the default, and enable ready consumption only for unordered Collects.
- With an ordered default, plans from a coordinator without this PR keep today's semantics during a rolling upgrade.
- It also lets
visitGroupandvalidateAggregationGroupingkeep their streaming forms where no ready Collect is involved (see the comment there).
A narrower fix would be to reject PreviousFillNode/LinearFillNode without grouping keys, and expressions that contain diff(), right here. COPY TO would still need information from the planner.
| } else if (groupingInvalidatedByCollect( | ||
| child, node.getOrderingScheme().getOrderBy().subList(0, node.getPartitionKeyCount()))) { | ||
| SortNode sort = | ||
| new SortNode(queryId.genPlanNodeId(), child, node.getOrderingScheme(), false, false); | ||
| nodeOrderingMap.put(sort.getPlanNodeId(), sort.getOrderingScheme()); | ||
| result.add(sort); |
There was a problem hiding this comment.
Non-blocking. groupingInvalidatedByCollect treats every multi-child Collect/Union as unordered. But the DataNode only enables ready consumption in eligible FIs. In an FI that is not eligible (TopK, joins, more than one pipeline), the Collect still drains its children one after another, as on base. There, this conversion mostly adds cost, and so does the preGrouped reset in validateAggregationGrouping.
PlanTester, base d2b97bc vs 788a995. In both queries the root FI has eligible=false because of the TopK:
-- base: GroupNode, this PR: SortNode (blocking full sort of the Collect output)
SELECT tag1, tag2, tag3, s1, SUM(s1) OVER (PARTITION BY tag1, tag2, tag3 ORDER BY s1) rs
FROM (SELECT tag1, tag2, tag3, s1, ROW_NUMBER() OVER () n FROM table1) t
WHERE n > 0 ORDER BY rs DESC LIMIT 10
-- base: preGrouped=[tag1, tag2, tag3], this PR: preGrouped=[]
SELECT tag1, tag2, tag3, SUM(s1) AS s
FROM (SELECT tag1, tag2, tag3, s1, COUNT(*) OVER () AS n FROM table1 WHERE tag1 != 'shenzhen') t
WHERE n > 0 GROUP BY tag1, tag2, tag3 ORDER BY s DESC LIMIT 10The second case matters more. HashAggregationOperator has no spill path (its spillEnabled branch throws UnsupportedOperationException) and keeps every group in memory. So a high-cardinality GROUP BY that streamed on base can now run out of operator memory.
If some of these resets are also meant to fix sequential cases, for example devices that span regions, please limit them to those cases. Otherwise this goes away with the planner-side ordering decision suggested on SharedExchangeMemoryEligibility.
| // Pending requests must not outlive their FI registration. Cancel before removing the | ||
| // account so another query's free() cannot retry a request against a missing FI. Do not | ||
| // hold the registration lock while notifying cancellation listeners. | ||
| if (fragmentRelatedMemory != null) { | ||
| for (MemoryReservationFuture<Void> future : | ||
| fragmentRelatedMemory.pendingReservations.values()) { | ||
| tryCancel(future); | ||
| } | ||
| } |
There was a problem hiding this comment.
Cancelling here breaks the code that owns these requests. It does fix a real base bug. On d2b97bc a pending request outlives its FI, and the next free() from any other query hits it: tryReserve throws a NullPointerException, and the request's bytes stay allocated in the pool for good.
The owners (SourceHandle, SinkChannel, SharedTsBlockQueue) rely on two rules: a pending future completes only after a successful reservation, and it is cancelled only by the owner itself through tryCancel(), which returns the pending bytes. This cancellation breaks both rules.
Probe setup: a 64 B pool fully held by query other. A SourceHandle for (q, 0.0) has one 64 B block pending. Then deRegisterFragmentInstanceFromQueryMemoryMap("q", "0.0", true), then free() by other, then handle.abort():
| base d2b97bc | 788a995 | |
|---|---|---|
free() by other |
NPE; 64 B stay reserved with no live reservation | ok |
handle.abort() |
ok, onAborted called |
IllegalArgumentException: RelatedMemoryReserved can't be null when freeing memory; aborted stays false, onAborted not called |
What happens on 788a995:
- The cancellation runs the handle's
blockedOnMemorylistener while the handle is still live. That submits aGetDataBlocksTaskfor a block that has no reservation. - Later,
tryCancel()returns 0 for the already-cancelled future. Soabort()tries tofree()the pending bytes against the removed account, and throws. - In
MPPDataExchangeManager#forceDeregisterFragmentInstance, that exception escapes the loop;DriverScheduleronly logs it. The FI's remaining handles are not aborted, and itssourceHandlesentry is not removed.
Same pattern elsewhere (from reading the code):
SinkChannel.abort()/close()calltryCancel(blocked)and thenfree.- In
SharedTsBlockQueue, the memory listener enqueues the TsBlock withreservedBytesthat were never reserved.
SourceHandleTest#reproduceAbortAfterFragmentInstanceMemoryIsForceDeregistered already covers the case where the handle holds reserved bytes. This change extends the failure to handles that only have a pending request.
Suggestion: detach instead of cancel. Remove the FI's requests from both indexes without completing them. Then no listener fires, and the owner's later tryCancel() still returns the pending bytes. If you want to keep cancelling, the three owners need to treat a future cancelled by the pool as unreserved: no fetch or enqueue in the listener, and no free of its bytes. Either way, please add a regression test: a real SourceHandle blocked on memory, then force deregistration, then abort().
| Iterator<MemoryReservationFuture<Void>> iterator = memoryReservationFutures.values().iterator(); | ||
| while (iterator.hasNext()) { | ||
| MemoryReservationFuture<Void> future = iterator.next(); | ||
| synchronized (future) { | ||
| if (future.isCancelled() || future.isDone()) { | ||
| iterator.remove(); | ||
| continue; | ||
| } | ||
| long bytesToReserve = future.getBytesToReserve(); | ||
| String curQueryId = future.getQueryId(); | ||
| String curFragmentInstanceId = future.getFragmentInstanceId(); | ||
| String curPlanNodeId = future.getPlanNodeId(); | ||
| long maxBytesCanReserve = future.getMaxBytesCanReserve(); | ||
| if (tryReserve( | ||
| curQueryId, curFragmentInstanceId, curPlanNodeId, bytesToReserve, maxBytesCanReserve)) { | ||
| future.set(null); | ||
| iterator.remove(); | ||
| } else { | ||
| rollbackReserve(curQueryId, curFragmentInstanceId, curPlanNodeId, bytesToReserve); | ||
| } | ||
| if (tryCompleteReservation(future)) { | ||
| removePendingReservation(future); | ||
| } |
There was a problem hiding this comment.
A waiter whose FI account is gone makes every free() on the DataNode throw. The cancellation in deRegisterFragmentInstanceFromQueryMemoryMap does not shut out reserveWithPriority. The race:
- Deregistration reads
pendingReservations(311–316). - A handle then publishes a request (401–402), and its retry at 406 fails without throwing.
- Deregistration removes the account at 318. The request is still in
memoryReservationFutures.
From then on, every free() releases its bytes and then throws from tryCompleteReservation → tryReserve → getFragmentReservation. Waiters queued behind that request are never reached. The try/catch at 409 only covers a retry that throws.
I forced this interleaving in a probe on 788a995, by hooking the pendingReservations.values() read. Another query's free() then threw IllegalArgumentException: Memory reservation is not registered for query x, fragment 0.0, plan node src. The pool was back to 0 used bytes, but a later 32 B waiter from query y was never woken. The window is narrow: a handle has to still be reserving while its FI is force-deregistered. But the failure lands on unrelated queries.
Suggestion: in this loop, treat a missing FI account as a dead waiter: remove it from the index and continue, instead of throwing. Alternatively, publish and deregister under the same FragmentMemoryReservation monitor, and have reserveWithPriority refuse to publish into a removed account.
Description
Per-handle exchange quotas and sequential Collect consumption can leave ready inputs waiting while other inputs hold unused capacity. This change enables ready-input Collect/MappingCollect consumption together with shared exchange quotas only within eligible table-model fragment instances (FIs). Tree-model queries and excluded table FIs retain sequential Collect consumption and independent quotas.
Exchange accounting
inputMaxBytesandoutputMaxBytes. Give the fullFto the only active direction; splitINPUT = F / 2,OUTPUT = F - INPUTwhen both directions charge the FI. Unused directions get zero. Remote SourceHandles charge INPUT; fragment sinks and local queues charge their producing FI's OUTPUT. Fall back if an active direction cannot hold the configured maximum TsBlock.Collect and plan properties
OperatorTreeGeneratorandvisitTopK()identical to the PR base. TopK planning and runtime filter ID handling, along with MergeSort and Join consumption algorithms, remain unchanged. TopK-specific fixes are deferred; this PR excludes those FIs from the new mode.SQL assertions
TestUtilsResultSet assertion overloads. Add explicitCollectQueryAssertionsfor the affected tests, using the existing SQL parser to inspect final-result ordering before accepting unordered assertions.Validation
test-compile -DskipTestspassed.git diff --checkpassed. Maven validation used offline mode.This PR has: