thomasrebele commented on code in PR #6916:
URL: https://github.com/apache/hive/pull/6916#discussion_r4413020399
##########
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java:
##########
@@ -1883,6 +1884,42 @@ private static boolean compareOperator(ParseContext
pctx, Operator<?> op1, Opera
return op1.logicalEquals(op2);
}
+ /**
+ * The discardable inputs of a merge are deleted afterwards, usually as a
chain ending in a
+ * ReduceSink (e.g. SEL -> RS). If such a chain holds the last sink
(ReduceSink/FileSink) of the
+ * work its surviving parent belongs to, the rest of that work no longer
ends in a sink - e.g.
+ * only a DPP event branch is left. GenTezWork only cuts a new vertex when
it reaches a sink, so
+ * such a pipeline never gets a vertex of its own and stays attached below
the ReduceSink that
+ * feeds it, inside the upstream vertex. Its reduce-side operators (e.g. a
mergepartial GroupBy
+ * reading KEY._colN) are then initialised against map-side rows and fail
with
+ * "cannot find field key from [0:_col0, ...]".
+ */
+ @VisibleForTesting
+ static boolean removalLeavesWorkWithoutSink(SharedWorkOptimizerCache
optimizerCache,
+ Collection<Operator<?>> discardableInputOps, Collection<Operator<?>>
discardableOps) {
+ for (Operator<?> op : discardableInputOps) {
+ // Check where a removed chain is cut off from what stays
+ for (Operator<?> parent : op.getParentOperators()) {
+ if (discardableInputOps.contains(parent) ||
discardableOps.contains(parent)) {
+ continue;
+ }
+ boolean hasSink = false;
+ for (Operator<?> workOp : findWorkOperators(optimizerCache, parent)) {
+ if (!discardableInputOps.contains(workOp) &&
!discardableOps.contains(workOp)
+ && (workOp instanceof ReduceSinkOperator || workOp instanceof
FileSinkOperator)) {
Review Comment:
Shouldn't org.apache.hadoop.hive.ql.exec.TerminalOperator be used instead?
##########
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java:
##########
@@ -1883,6 +1884,42 @@ private static boolean compareOperator(ParseContext
pctx, Operator<?> op1, Opera
return op1.logicalEquals(op2);
}
+ /**
+ * The discardable inputs of a merge are deleted afterwards, usually as a
chain ending in a
+ * ReduceSink (e.g. SEL -> RS). If such a chain holds the last sink
(ReduceSink/FileSink) of the
+ * work its surviving parent belongs to, the rest of that work no longer
ends in a sink - e.g.
+ * only a DPP event branch is left. GenTezWork only cuts a new vertex when
it reaches a sink, so
+ * such a pipeline never gets a vertex of its own and stays attached below
the ReduceSink that
+ * feeds it, inside the upstream vertex. Its reduce-side operators (e.g. a
mergepartial GroupBy
+ * reading KEY._colN) are then initialised against map-side rows and fail
with
+ * "cannot find field key from [0:_col0, ...]".
+ */
+ @VisibleForTesting
+ static boolean removalLeavesWorkWithoutSink(SharedWorkOptimizerCache
optimizerCache,
+ Collection<Operator<?>> discardableInputOps, Collection<Operator<?>>
discardableOps) {
+ for (Operator<?> op : discardableInputOps) {
+ // Check where a removed chain is cut off from what stays
+ for (Operator<?> parent : op.getParentOperators()) {
+ if (discardableInputOps.contains(parent) ||
discardableOps.contains(parent)) {
+ continue;
+ }
+ boolean hasSink = false;
+ for (Operator<?> workOp : findWorkOperators(optimizerCache, parent)) {
Review Comment:
`findWorkOperators` searches in both directions (parents and children). As
far as I understand the algorithm, we only want to search in one direction here?
##########
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SharedWorkOptimizer.java:
##########
@@ -1883,6 +1884,42 @@ private static boolean compareOperator(ParseContext
pctx, Operator<?> op1, Opera
return op1.logicalEquals(op2);
}
+ /**
+ * The discardable inputs of a merge are deleted afterwards, usually as a
chain ending in a
+ * ReduceSink (e.g. SEL -> RS). If such a chain holds the last sink
(ReduceSink/FileSink) of the
+ * work its surviving parent belongs to, the rest of that work no longer
ends in a sink - e.g.
+ * only a DPP event branch is left. GenTezWork only cuts a new vertex when
it reaches a sink, so
+ * such a pipeline never gets a vertex of its own and stays attached below
the ReduceSink that
+ * feeds it, inside the upstream vertex. Its reduce-side operators (e.g. a
mergepartial GroupBy
+ * reading KEY._colN) are then initialised against map-side rows and fail
with
+ * "cannot find field key from [0:_col0, ...]".
+ */
Review Comment:
I find the description a bit difficult to understand. Would it be possible
to rephrase it illustrating the problem with an example? The example could
include the GroupBy, if adding it does not make the example too complicated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
