[
https://issues.apache.org/jira/browse/PIG-1038?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=12775214#action_12775214
]
Pradeep Kamath commented on PIG-1038:
-------------------------------------
Review comments:
In JobControlCompiler:
======================
The OutputValueGroupingComparator is a RawComparator -
how are we ensuring that compare(WritableComparable a, WritableComparable b) is
called?
{code}
622 if ((wa.getIndex() & PigNullableWritable.mqFlag) != 0) { //
this is a multi-query index
{code}
Why do we only compare on index if this is true? The if-else in this block does
not consider the
case where both indices are same - is that by design?
{code}
653 } else if (wa.isNull() && wb.isNull()) {
{code}
In this block the case where both indices are same is not considered - is that
by design?
The change in src/org/apache/pig/impl/io/PigNullableWritable.java seems
unrelated to the patch
In SecondaryKeyOptimizer.java:
==============================
{code}
154 else if (mapLeaf instanceof POUnion || mapLeaf instanceof
POSplit) {
155 List<PhysicalOperator> preds =
mr.mapPlan.getPredecessors(mapLeaf);
156 for (PhysicalOperator pred:preds)
157 {
158 if (pred instanceof POLocalRearrange)
159 {
160 SortKeyInfo sortKeyInfo =
getSortKeyInfo((POLocalRearrange)pred);
161 sortKeyInfos.add(sortKeyInfo);
162 }
163 }
{code}
If mapLeaf is a POSplit, the POSplit may have POLocalRearrange as the leaf (in
multi query optimized queries) - should we be handling those?
Also, getSortKeyInfo() can return a null - so all places where getSortKeyInfo()
is called, return value should be checked for null
{code}
98 List<Integer> columns = new ArrayList<Integer>();
99
columns.add(rearrange.getIndex()&PigNullableWritable.idxSpace);
100 columnChainInfo.insert(false, columns, DataType.TUPLE)
{code}
Why does the column chain start with the index of LocalRearrange and why is the
type tuple?
{code}
102 PhysicalOperator node = plan.getRoots().get(0);
103 while (node!=null)
104 {
105 if (node instanceof POProject) {
106 POProject project = (POProject)node;
107
108 columnChainInfo.insert(project.isStar(),
project.getColumns(), project.getResultType());
109
110 if (plan.getPredecessors(node)==null)
111 node = null;
{code}
If node is initially the root, wouldn't plan.getPredecessors(node) always ==
null?
{code}
175 List<PhysicalOperator> reduceRoots = mr.reducePlan.getRoots();
176 if (reduceRoots.size() != 1) {
177 log.debug("Expected reduce to have single leaf");
178 return;
179 }
{code}
Did you mean to say "Expected reduce to have single root" ?
{code}
209 // Removed POSort, if the predecessor require a databag, we
need to add a PORelationToExprProject
{code}
Should the above comment read "if the successor requires ..."
{code}
247 throw new VisitorException("Sort on columns from
different inputs.");
{code}
Should this exception follow Error Handling guidelines to include errorcode,
and error source?
{code}
253 } else if (mapLeaf instanceof POUnion || mapLeaf instanceof
POSplit){
254 List<PhysicalOperator> preds =
mr.mapPlan.getPredecessors(mapLeaf);
255 for (PhysicalOperator pred:preds) {
256 POLocalRearrange rearrange = (POLocalRearrange)pred;
257 rearrange.setUseSecondaryKey(true);
258 if (rearrange.getIndex()==indexOfRearrangeToChange)
259 setSecondaryPlan(mr.mapPlan, rearrange,
secondarySortKeyInfo);
260 }
261 }
{code}
If mapLeaf is a POSplit, the POSplit may have POLocalRearrange as the leaf (in
multi query optimized queries) - should we be handling those?
Also in the if statement on line 258, what if the condition evaluates to false
- shouwl we throw an Exception like earlier in the same
method?
{code}
274 for (int i=1;i<columnChainInfo.size();i++) {
{code}
A comment explaining why this loop starts from 1 would be useful
{code}
286 if (secondaryPlan.isEmpty())
287 {
{code}
A comment explaining when this would happen would be useful
There is ForEachVisitor which visits *all* ForEach ops in the reduce plan - is
a visitor needed? Shouldn't only the
foreach following the POPackage be processed?
372 SecondaryKeyDiscoverVisitor(PhysicalPlan plan, List<SortKeyInfo>
sortKeyInfos, SortKeyInfo secondarySortKeyInfo) {
It seems like the secondarySortKeyInfo passed in the constructor call is always
null - is that argument needed in the constructor?
431 throw new VisitorException("POForEach has more than 1 input
plans");
Should this exception follow Error Handling guidelines to include errorcode,
and error source?
A test case should be added for the case of a non Project group by key like
group by $0 + $1 - I did not follow the code path for this case - we should
ensure this works with
a nested sort in the foreach.
In ColumnChainInfo.java:
=========================
A comment on ColumnChainInfo and SortKeyInfo explaining how it tracks to
POProjects in the plan would be useful
{code}
42 ColumnInfo newColumnChainInfo = new ColumnInfo(star, columns, type);
{code}
I think it is better to name this variable as newColumnInfo since it the name
'newColumnChainInfo' seems to
suggest the object is of type ColumnChainInfo.
52 newColumns.add(columns.get(0)-1);
The above code suggests that we rely on columns only having 1 element - at
position 0. If that is not
true, should the code handle the case where columns has > 1 element?
53 ColumnInfo newColumnChainInfo = new ColumnInfo(star,
newColumns, type);
I think it is better to name this variable as newColumnInfo since it the name
'newColumnChainInfo' seems to
suggest the object is of type ColumnChainInfo.
{code}
85 public void normalize() {
86 int i=size()-1;
87 while (i>=0 && getColumnInfo(i).star) {
88 columnInfos.remove(i);
89 i--;
90 }
91 }
{code}
What is the above method used for?
In SortKeyInfo.java:
=====================
{code}
public void insertColumnInfo(int index, ColumnChainInfo columnChainInfo)
{
if (columnChains.size()<=index)
{
columnChains.add(columnChainInfo);
return;
}
ColumnChainInfo chain = columnChains.get(index);
chain.insertColumnChainInfo(columnChainInfo);
}
{code}
Should this method be named "insertColumnChainInfo" instead? When would the if
condition evaluate to false?
{code}
public void normalize() {
if (columnChains.size()==1)
columnChains.get(0).normalize();
}
{code}
What does the above do?
In POSortedDistinct.java:
=========================
{code}
75 if (in.returnStatus == POStatus.STATUS_EOP) {
76 if (!isAccumulative() || !isAccumStarted()) {
77 lastTuple = null;
78 }
79 return in;
80 }
{code}
what do the checks for isAccumulative() and is AccumStarted do?
In POLocalRearrange.java:
=========================
Is there a way to refactor the code so that setPlans() and setSecondayPlans()
can share common code?
In POMultiQueryPackage.java:
============================
In setUseSecondaryKey(), the code, sets the flag to true in all underlying
packages irrespective of whether the argument to
the method is true or false. Also I am wondering if in multiquery optimization,
some subqueries need secondary key and some don't,
is that use case handled?
> Optimize nested distinct/sort to use secondary key
> --------------------------------------------------
>
> Key: PIG-1038
> URL: https://issues.apache.org/jira/browse/PIG-1038
> Project: Pig
> Issue Type: Improvement
> Components: impl
> Affects Versions: 0.4.0
> Reporter: Olga Natkovich
> Assignee: Daniel Dai
> Fix For: 0.6.0
>
> Attachments: PIG-1038-1.patch, PIG-1038-2.patch
>
>
> If nested foreach plan contains sort/distinct, it is possible to use hadoop
> secondary sort instead of SortedDataBag and DistinctDataBag to optimize the
> query.
> Eg1:
> A = load 'mydata';
> B = group A by $0;
> C = foreach B {
> D = order A by $1;
> generate group, D;
> }
> store C into 'myresult';
> We can specify a secondary sort on A.$1, and drop "order A by $1".
> Eg2:
> A = load 'mydata';
> B = group A by $0;
> C = foreach B {
> D = A.$1;
> E = distinct D;
> generate group, E;
> }
> store C into 'myresult';
> We can specify a secondary sort key on A.$1, and simplify "D=A.$1; E=distinct
> D" to a special version of distinct, which does not do the sorting.
--
This message is automatically generated by JIRA.
-
You can reply to this email to add a comment to the issue online.
