[
https://issues.apache.org/jira/browse/IMPALA-13533?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18104837#comment-18104837
]
ASF subversion and git services commented on IMPALA-13533:
----------------------------------------------------------
Commit 558c3f685f8164db516a2cc2e9fa184ae8c937d8 in impala's branch
refs/heads/master from Michael Smith
[ https://gitbox.apache.org/repos/asf?p=impala.git;h=558c3f685 ]
IMPALA-13533: Calcite CTE backend
Adds support for CTEs in distributed planning. CTEs are structured like
an exchange, where one CTE fragment can feed multiple CTE exchanges.
Creates a LocalMultiSink as the sink of a CTE fragment, and Sequence
nodes are discarded in the distributed plan.
Adds support for CTE backend execution. Single-node plans only have
dedicated intra-fragment scheduling.
The multi-cast nature of CTEs creates a directed acyclic graph in our
fragment structure that Impala has not previously dealt with. A guard is
added to avoid scheduling the same fragment multiple times.
Unions require special handling, as they're the only instance where
multiple CTEs and/or exchanges can appear in a fragment. Each consumer
is associated with a producer; the fragment must be scheduled on all
nodes which have producers scheduled to them, which is handled similar
to how exchanges are scheduled. Each consumer PlanNode then has one (or
more with MT_DOP>1) fragment instances scheduled on each node in the
fragment.
Note that not all nodes may have a producer, and there may not be as
many producer fragment instances on a node as there are consumer
fragment instances (when insufficient scan ranges exist to support
multiple readers on a ScanNode under the producer). So consumers need to
allow that a producer may not exist (and return 0 rows), and we need to
map each producer to one or more consumers (to better distribute results
among threads).
This implies that the number of consumers for a producer is specific to
each fragment instance, and can only be determined at runtime. We must
also map each PlanNode in a fragment instance to the producer
per_fragment_instance_idx, which is used to distinguish between multiple
instances of the producer when creating LocalExchangers.
Implements backend for CTEs in the Calcite planner. CTE output is added
to a LocalExchanger, then pulled concurrently. LocalExchangers are
registered with QueryState so all fragments can access them;
registration is done during plan fragment construction so all instances
can find the LocalExchanger or identify it's absence.
Mimics UnionNode's MaterializeBatch for translating the CTE tuple to the
expected output tuple, with passthrough for cases where input and output
row layouts match.
Adds metrics for LocalExchanger tracking the number of batches queued,
and adds memory tracking.
Adds cte-distributed showing distributed planning and basic execution.
Tested with TPC-DS queries (DecimalV2 version). Setup
start-impala-cluster.py --use_calcite_planner=true \
--impalad_args=--default_query_options=cte_threshold=1
impala-py.test tests/query_test/test_tpcds_queries.py::TestTpcdsDecimalV2Query
Also tested with mt_dop=5.
Change-Id: I48f16d495d4b37be97e6a913f0eb5b94d70e199a
Reviewed-on: http://gerrit.cloudera.org:8080/22094
Reviewed-by: Joe McDonnell <[email protected]>
Tested-by: Joe McDonnell <[email protected]>
> Execute CTE plans
> -----------------
>
> Key: IMPALA-13533
> URL: https://issues.apache.org/jira/browse/IMPALA-13533
> Project: IMPALA
> Issue Type: Sub-task
> Components: Backend
> Reporter: Michael Smith
> Assignee: Michael Smith
> Priority: Major
>
> Add backend support for CTE plans.
> * Add PlanNode/ExecNode for CTE producer, consumer, and sequence nodes.
> * Implement backend node-local dataflow from CTE producers to consumers. Use
> a shared buffer for CTE producer results until each consumer is ready to
> process them; initially this will be a batch operation, where CTE producer
> puts all results into the buffer, then consumers can start reading them.
> * Should be able to execute all TPC-DS queries.
> For distributed plans: generate CTE producer fragments, removing
> SequenceNodes. CTE buffer must remain active until all consumers have
> finished reading.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
