Joe McDonnell has posted comments on this change. ( http://gerrit.cloudera.org:8080/22094 )
Change subject: IMPALA-13533: Calcite CTE backend ...................................................................... Patch Set 52: (10 comments) I'm still wrapping my head around this, but I'm starting to understand it. I haven't looked at scheduling yet. I think we're a little bit light on comments for some class fields and some of the interesting methods. I don't care as much about the trivial / boilerplate methods. To use this, someone would need to use the Calcite planner and then set cte_threshold. We can add additional controls if we need them. http://gerrit.cloudera.org:8080/#/c/22094/52//COMMIT_MSG Commit Message: http://gerrit.cloudera.org:8080/#/c/22094/52//COMMIT_MSG@36 PS52, Line 36: map map Nit: extra map http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/cte-consumer-node.cc File be/src/exec/cte-consumer-node.cc: http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/cte-consumer-node.cc@167 PS52, Line 167: BufferedTupleStream Nit: Comment out of date http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/cte-consumer-node.cc@188 PS52, Line 188: // Resources will be released once the buffer is closed. : output_batch->MarkNeedsDeepCopy(); Nit: If I'm understanding this right, this is critical for having the right memory lifetime. LocalExchanger is relying on the memory being cleaned up before the next GetNext() call, because if this was the last reference to the Cell, it would be freed on the next GetNext() call. Let's add a bit to the comment here to make it clear that this is critical. In reading the code, I think there is a downside for marking it needing deep copy. If this feeds into a selective operator like a selective hash join, usually that would be assembling a single row batch out of multiple row batches (because many rows are eliminated). However, this forces it to flush, so a particularly selective join could end up producing many small or empty row batches. Hopefully CTEs are small, so that's not the end of the world. A reference counting system could make that unnecessary. In theory, we might be able to skip setting this if the row batch doesn't have any attached memory. Either way, I don't see this as a blocker for the first patch. http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/cte-producer-node.h File be/src/exec/cte-producer-node.h: http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/cte-producer-node.h@40 PS52, Line 40: /// Node that buffers results produced by a Common Table Expression into a : /// LocalExchanger. Let's add some more detail here. Right now, everything accumulates in Open() and GetNext() waits for the consumers to finish. http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/cte-producer-node.cc File be/src/exec/cte-producer-node.cc: http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/cte-producer-node.cc@97 PS52, Line 97: RETURN_IF_ERROR(exchanger_->Push(std::move(child_batch))); I think either here or in the local exchanger or both, we should DCHECK that the incoming row batch does not have needs_deep_copy() == true. We'll be holding on to memory, so we need to know it isn't set. http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/sequence-node.h File be/src/exec/sequence-node.h: http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/exec/sequence-node.h@32 PS52, Line 32: /// Node that consumes terminal children in-order, before passing through rows from the : /// first child. So, a SequenceNode runs Open() on the non-passthrough children (but never GetNext()), then runs its regular passthrough child. http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/runtime/local-exchanger.h File be/src/runtime/local-exchanger.h: http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/runtime/local-exchanger.h@36 PS52, Line 36: /// In-memory exchange. One producer pushes RowBatches which are consumed by multiple : /// consumers. Pull returns the original RowBatch; creating a copy is the responsibility : /// of the caller. Thread-safe. Inspired by StarRocks' multi_cast_local_exchange. Let me check my understanding. Some upsides: - Avoid copying the tuple when it is passthrough - Stream things through to consumers on a very granular level - With quick consumers, doesn't need to retain memory. Some downsides: - Since we don't make a copy, we need to hold the plan tree open until everything has been consumed. - We hold on to memory in a way that children may not expect. Even if we receive a row batch where AtCapacity() is true due to memory, we'll still just hold on to it. - Can't spill yet - Memory is not particularly bounded. Some refinements (not necessarily in this change) would look like: - Limit the number of batches to be at most N batches ahead of the fastest consumer rather than accumulating more. This doesn't really bound memory, but it avoids some cases where the extra memory doesn't help much. (But this can't be used with the SequenceNode.) - There may need to be an option to switch to a BufferedTupleStream at runtime (e.g. if we see memory accumulate past a certain point). http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/runtime/local-exchanger.h@100 PS52, Line 100: const int32_t consumer_count_; : RuntimeProfile* runtime_profile_ = nullptr; : std::unique_ptr<MemTracker> mem_tracker_; : RuntimeProfile::HighWaterMarkCounter* num_cells_counter_ = nullptr; : mutable std::mutex mutex_; : mutable std::condition_variable all_consumers_done_cv_; : std::condition_variable batch_available_cv_; : int32_t consumers_done_ = 0; : int32_t next_consumer_index_ = 0; : Cell* head_; : Cell* tail_; : std::vector<Cell*> progress_; : bool eos_ = false; A bunch of these would benefit from comments. What's being protected by the mutex, etc. http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/runtime/local-exchanger.cc File be/src/runtime/local-exchanger.cc: http://gerrit.cloudera.org:8080/#/c/22094/52/be/src/runtime/local-exchanger.cc@39 PS52, Line 39: Status LocalExchanger::Push(std::unique_ptr<RowBatch> batch) { I think we'll want a way to see how much memory is accumulated in this local exchange (and high water mark of memory accumulated). The main contribution would be buffers attached to the row batches. http://gerrit.cloudera.org:8080/#/c/22094/52/testdata/workloads/functional-query/queries/QueryTest/cte-distributed.test File testdata/workloads/functional-query/queries/QueryTest/cte-distributed.test: http://gerrit.cloudera.org:8080/#/c/22094/52/testdata/workloads/functional-query/queries/QueryTest/cte-distributed.test@52 PS52, Line 52: WITH v1 AS (SELECT n_name, n_nationkey FROM tpch.nation WHERE n_regionkey = 0) : SELECT v1.n_name, v1_next.n_name FROM v1, v1 v1_next WHERE v1.n_nationkey + 1 = v1_next.n_nationkey; The test passes, but if I run this manually in impala-shell, it hits this DCHECK: F20260716 09:47:05.238353 1244811 cte-producer-node.cc:39] e94ecd396dfa959e:e807ae9500000000] Check failed: instance_ctx->num_cte_consumers > 0 (0 vs. 0) It looks like EXEC_SINGLE_NODE_ROWS_THRESHOLD=0 is important for this to work, and the error case happens when that isn't set and it uses the single-node code. -- To view, visit http://gerrit.cloudera.org:8080/22094 To unsubscribe, visit http://gerrit.cloudera.org:8080/settings Gerrit-Project: Impala-ASF Gerrit-Branch: master Gerrit-MessageType: comment Gerrit-Change-Id: I48f16d495d4b37be97e6a913f0eb5b94d70e199a Gerrit-Change-Number: 22094 Gerrit-PatchSet: 52 Gerrit-Owner: Michael Smith <[email protected]> Gerrit-Reviewer: Anonymous Coward (816) Gerrit-Reviewer: Balazs Hevele <[email protected]> Gerrit-Reviewer: Csaba Ringhofer <[email protected]> Gerrit-Reviewer: Impala Public Jenkins <[email protected]> Gerrit-Reviewer: Joe McDonnell <[email protected]> Gerrit-Reviewer: Michael Smith <[email protected]> Gerrit-Reviewer: Steve Carlin <[email protected]> Gerrit-Comment-Date: Fri, 17 Jul 2026 01:45:16 +0000 Gerrit-HasComments: Yes
