Skip to content

Commit e6e043b

Browse files
comnetworkcomnetwork
committed
PHOENIX-4820 Optimize OrderBy for ClientAggregatePlan
1 parent 3671097 commit e6e043b

18 files changed

+705
-85
lines changed

‎phoenix-core/src/it/java/org/apache/phoenix/end2end/AggregateIT.java

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -227,5 +227,109 @@ protected void testCountNullInNonEmptyKeyValueCF(int columnEncodedBytes) throws
227227
assertEquals(4, rs.getLong(1));
228228
}
229229
}
230+
231+
@Test
232+
public void testOrderByOptimizeForClientAggregatePlanBug4820() throws Exception {
233+
doTestOrderByOptimizeForClientAggregatePlanBug4820(false,false);
234+
doTestOrderByOptimizeForClientAggregatePlanBug4820(false,true);
235+
doTestOrderByOptimizeForClientAggregatePlanBug4820(true,false);
236+
doTestOrderByOptimizeForClientAggregatePlanBug4820(true,true);
237+
}
238+
239+
private void doTestOrderByOptimizeForClientAggregatePlanBug4820(boolean desc ,boolean salted) throws Exception {
240+
Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
241+
Connection conn = null;
242+
try {
243+
conn = DriverManager.getConnection(getUrl(), props);
244+
String tableName = generateUniqueName();
245+
String sql = "create table " + tableName + "( "+
246+
" pk1 varchar not null , " +
247+
" pk2 varchar not null, " +
248+
" pk3 varchar not null," +
249+
" v1 varchar, " +
250+
" v2 varchar, " +
251+
" CONSTRAINT TEST_PK PRIMARY KEY ( "+
252+
"pk1 "+(desc ? "desc" : "")+", "+
253+
"pk2 "+(desc ? "desc" : "")+", "+
254+
"pk3 "+(desc ? "desc" : "")+
255+
" )) "+(salted ? "SALT_BUCKETS =4" : "split on('b')");
256+
conn.createStatement().execute(sql);
257+
258+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES ('a11','a12','a13','a14','a15')");
259+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES ('a21','a22','a23','a24','a25')");
260+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES ('a31','a32','a33','a34','a35')");
261+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES ('b11','b12','b13','b14','b15')");
262+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES ('b21','b22','b23','b24','b25')");
263+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES ('b31','b32','b33','b34','b35')");
264+
conn.commit();
265+
266+
sql = "select a.ak3 "+
267+
"from (select pk1 ak1,pk2 ak2,pk3 ak3, substr(v1,1,1) av1,substr(v2,1,1) av2 from "+tableName+" order by pk2,pk3 limit 10) a "+
268+
"group by a.ak3,a.av1 order by a.ak3 desc,a.av1";
269+
ResultSet rs = conn.prepareStatement(sql).executeQuery();
270+
assertResultSet(rs, new Object[][]{{"b33"},{"b23"},{"b13"},{"a33"},{"a23"},{"a13"}});
271+
272+
sql = "select a.ak3 "+
273+
"from (select pk1 ak1,pk2 ak2,pk3 ak3, substr(v1,1,1) av1,substr(v2,1,1) av2 from "+tableName+" order by pk2,pk3 limit 10) a "+
274+
"group by a.ak3,a.av1 order by a.ak3,a.av1";
275+
rs = conn.prepareStatement(sql).executeQuery();
276+
assertResultSet(rs, new Object[][]{{"a13"},{"a23"},{"a33"},{"b13"},{"b23"},{"b33"}});
277+
278+
sql = "select a.ak3 "+
279+
"from (select pk1 ak1,pk2 ak2,pk3 ak3,substr(v1,1,1) av1,substr(v2,1,1) av2 from "+tableName+" order by pk2,pk3 limit 10) a "+
280+
"where a.av1 = 'a' group by a.av1,a.ak3 order by a.ak3 desc";
281+
rs = conn.prepareStatement(sql).executeQuery();
282+
assertResultSet(rs, new Object[][]{{"a33"},{"a23"},{"a13"}});
283+
284+
sql = "select a.ak3 "+
285+
"from (select pk1 ak1,pk2 ak2,pk3 ak3,substr(v1,1,1) av1,substr(v2,1,1) av2 from "+tableName+" order by pk2,pk3 limit 10) a "+
286+
"where a.av1 = 'a' group by a.av1,a.ak3 order by a.ak3";
287+
rs = conn.prepareStatement(sql).executeQuery();
288+
assertResultSet(rs, new Object[][]{{"a13"},{"a23"},{"a33"}});
289+
290+
sql = "select a.ak3 "+
291+
"from (select pk1 ak1,pk2 ak2,pk3 ak3,substr(v1,1,1) av1,substr(v2,1,1) av2 from "+tableName+" order by pk2,pk3 limit 10) a "+
292+
"where a.av1 = 'b' and a.av2= 'b' group by CASE WHEN a.av1 > a.av2 THEN a.av1 ELSE a.av2 END,a.ak3,a.ak2 order by a.ak3 desc,a.ak2 desc";
293+
rs = conn.prepareStatement(sql).executeQuery();
294+
assertResultSet(rs, new Object[][]{{"b33"},{"b23"},{"b13"}});
295+
296+
sql = "select a.ak3 "+
297+
"from (select pk1 ak1,pk2 ak2,pk3 ak3,substr(v1,1,1) av1,substr(v2,1,1) av2 from "+tableName+" order by pk2,pk3 limit 10) a "+
298+
"where a.av1 = 'b' and a.av2= 'b' group by CASE WHEN a.av1 > a.av2 THEN a.av1 ELSE a.av2 END,a.ak3,a.ak2 order by a.ak3,a.ak2 desc";
299+
rs = conn.prepareStatement(sql).executeQuery();
300+
assertResultSet(rs, new Object[][]{{"b13"},{"b23"},{"b33"}});
301+
302+
tableName = generateUniqueName();
303+
sql = "create table " + tableName + "( "+
304+
" pk1 double not null , " +
305+
" pk2 double not null, " +
306+
" pk3 double not null," +
307+
" v1 varchar, " +
308+
" CONSTRAINT TEST_PK PRIMARY KEY ( "+
309+
"pk1 "+(desc ? "desc" : "")+", "+
310+
"pk2 "+(desc ? "desc" : "")+", "+
311+
"pk3 "+(desc ? "desc" : "")+
312+
" )) "+(salted ? "SALT_BUCKETS =4" : "split on(2.3)");
313+
conn.createStatement().execute(sql);
314+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES (2.1,2.11,2.12,'e')");
315+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES (2.2,2.21,2.23,'d')");
316+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES (2.3,2.31,2.32,'c')");
317+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES (2.4,2.41,2.42,'b')");
318+
conn.createStatement().execute("UPSERT INTO "+tableName+" VALUES (2.5,2.51,2.52,'a')");
319+
conn.commit();
320+
321+
sql = "select a.av1 "+
322+
"from (select pk1 ak1,pk2 ak2,pk3 ak3, substr(v1,1,1) av1 from "+tableName+" order by pk1,pk2 limit 10) a "+
323+
"where cast(a.ak1 as integer)=2 group by a.ak1,a.av1 order by a.av1";
324+
rs = conn.prepareStatement(sql).executeQuery();
325+
assertResultSet(rs, new Object[][]{{"a"},{"b"},{"c"},{"d"},{"e"}});
326+
327+
} finally {
328+
if(conn != null) {
329+
conn.close();
330+
}
331+
}
332+
}
333+
230334
}
231335

‎phoenix-core/src/main/java/org/apache/phoenix/compile/GroupByCompiler.java

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,13 @@ public GroupBy compile(StatementContext context, TupleProjector tupleProjector)
143143
boolean isOrderPreserving = this.isOrderPreserving;
144144
int orderPreservingColumnCount = 0;
145145
if (isOrderPreserving) {
146-
OrderPreservingTracker tracker = new OrderPreservingTracker(context, GroupBy.EMPTY_GROUP_BY, Ordering.UNORDERED, expressions.size(), tupleProjector);
146+
OrderPreservingTracker tracker = new OrderPreservingTracker(
147+
context,
148+
GroupBy.EMPTY_GROUP_BY,
149+
Ordering.UNORDERED,
150+
expressions.size(),
151+
tupleProjector,
152+
null);
147153
for (int i = 0; i < expressions.size(); i++) {
148154
Expression expression = expressions.get(i);
149155
tracker.track(expression);

‎phoenix-core/src/main/java/org/apache/phoenix/compile/OrderByCompiler.java

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,8 @@ public static OrderBy compile(StatementContext context,
8888
Integer offset,
8989
RowProjector rowProjector,
9090
TupleProjector tupleProjector,
91-
boolean isInRowKeyOrder) throws SQLException {
91+
boolean isInRowKeyOrder,
92+
Expression whereExpression) throws SQLException {
9293
List<OrderByNode> orderByNodes = statement.getOrderBy();
9394
if (orderByNodes.isEmpty()) {
9495
return OrderBy.EMPTY_ORDER_BY;
@@ -105,9 +106,22 @@ protected void addColumn(PColumn column) {}
105106
} else {
106107
compiler = new ExpressionCompiler(context, groupBy);
107108
}
109+
110+
if(groupBy != GroupBy.EMPTY_GROUP_BY) {
111+
//if there is groupBy,the groupBy.expressions are viewed as new rowKey columns,so
112+
//tupleProjector and isInRowKeyOrder is cleared
113+
tupleProjector = null;
114+
isInRowKeyOrder = true;
115+
}
108116
// accumulate columns in ORDER BY
109117
OrderPreservingTracker tracker =
110-
new OrderPreservingTracker(context, groupBy, Ordering.ORDERED, orderByNodes.size(), tupleProjector);
118+
new OrderPreservingTracker(
119+
context,
120+
groupBy,
121+
Ordering.ORDERED,
122+
orderByNodes.size(),
123+
tupleProjector,
124+
whereExpression);
111125
LinkedHashSet<OrderByExpression> orderByExpressions = Sets.newLinkedHashSetWithExpectedSize(orderByNodes.size());
112126
for (OrderByNode node : orderByNodes) {
113127
ParseNode parseNode = node.getNode();

‎phoenix-core/src/main/java/org/apache/phoenix/compile/OrderPreservingTracker.java

Lines changed: 25 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,8 @@
1818
import org.apache.phoenix.compile.GroupByCompiler.GroupBy;
1919
import org.apache.phoenix.execute.TupleProjector;
2020
import org.apache.phoenix.expression.CoerceExpression;
21-
import org.apache.phoenix.expression.Determinism;
2221
import org.apache.phoenix.expression.Expression;
22+
import org.apache.phoenix.expression.KeyValueColumnExpression;
2323
import org.apache.phoenix.expression.LiteralExpression;
2424
import org.apache.phoenix.expression.ProjectedColumnExpression;
2525
import org.apache.phoenix.expression.RowKeyColumnExpression;
@@ -30,6 +30,7 @@
3030
import org.apache.phoenix.expression.visitor.StatelessTraverseNoExpressionVisitor;
3131
import org.apache.phoenix.schema.PTable;
3232
import org.apache.phoenix.schema.SortOrder;
33+
import org.apache.phoenix.util.ExpressionUtil;
3334

3435
import com.google.common.collect.Iterators;
3536
import com.google.common.collect.Lists;
@@ -76,16 +77,22 @@ public Info(Info info, int slotSpan, OrderPreserving orderPreserving) {
7677
private final Ordering ordering;
7778
private final int pkPositionOffset;
7879
private final List<Info> orderPreservingInfos;
79-
private final TupleProjector projector;
8080
private boolean isOrderPreserving = true;
8181
private Boolean isReverse = null;
8282
private int orderPreservingColumnCount = 0;
83+
private Expression whereExpression;
8384

8485
public OrderPreservingTracker(StatementContext context, GroupBy groupBy, Ordering ordering, int nNodes) {
85-
this(context, groupBy, ordering, nNodes, null);
86+
this(context, groupBy, ordering, nNodes, null, null);
8687
}
8788

88-
public OrderPreservingTracker(StatementContext context, GroupBy groupBy, Ordering ordering, int nNodes, TupleProjector projector) {
89+
public OrderPreservingTracker(
90+
StatementContext context,
91+
GroupBy groupBy,
92+
Ordering ordering,
93+
int nNodes,
94+
TupleProjector projector,
95+
Expression whereExpression) {
8996
this.context = context;
9097
if (groupBy.isEmpty()) {
9198
PTable table = context.getResolver().getTables().get(0).getTable();
@@ -103,7 +110,7 @@ public OrderPreservingTracker(StatementContext context, GroupBy groupBy, Orderin
103110
this.visitor = new TrackOrderPreservingExpressionVisitor(projector);
104111
this.orderPreservingInfos = Lists.newArrayListWithExpectedSize(nNodes);
105112
this.ordering = ordering;
106-
this.projector = projector;
113+
this.whereExpression = whereExpression;
107114
}
108115

109116
public void track(Expression node) {
@@ -208,20 +215,14 @@ private boolean hasEqualityConstraints(int startPos, int endPos) {
208215
// not by the original row key order of the table (see PHOENIX-3451).
209216
// We check each GROUP BY expression to see if it only references columns that are
210217
// matched by equality constraints, in which case the expression itself would be constant.
211-
// FIXME: this only recognizes row key columns that are held constant, not all columns.
212-
// FIXME: we should optimize out any GROUP BY or ORDER BY expression which is deemed to
213-
// be held constant based on the WHERE clause.
214218
if (!groupBy.isEmpty()) {
215219
for (int pos = startPos; pos < endPos; pos++) {
216-
IsConstantVisitor visitor = new IsConstantVisitor(this.projector, ranges);
220+
IsConstantVisitor visitor = new IsConstantVisitor(ranges, whereExpression);
217221
List<Expression> groupByExpressions = groupBy.getExpressions();
218222
if (pos >= groupByExpressions.size()) { // sanity check - shouldn't be necessary
219223
return false;
220224
}
221225
Expression groupByExpression = groupByExpressions.get(pos);
222-
if ( groupByExpression.getDeterminism().ordinal() > Determinism.PER_STATEMENT.ordinal() ) {
223-
return false;
224-
}
225226
Boolean isConstant = groupByExpression.accept(visitor);
226227
if (!Boolean.TRUE.equals(isConstant)) {
227228
return false;
@@ -248,17 +249,17 @@ public boolean isReverse() {
248249
*
249250
*/
250251
private static class IsConstantVisitor extends StatelessTraverseAllExpressionVisitor<Boolean> {
251-
private final TupleProjector projector;
252252
private final ScanRanges scanRanges;
253+
private final Expression whereExpression;
253254

254-
public IsConstantVisitor(TupleProjector projector, ScanRanges scanRanges) {
255-
this.projector = projector;
256-
this.scanRanges = scanRanges;
257-
}
255+
public IsConstantVisitor(ScanRanges scanRanges, Expression whereExpression) {
256+
this.scanRanges = scanRanges;
257+
this.whereExpression = whereExpression;
258+
}
258259

259260
@Override
260261
public Boolean defaultReturn(Expression node, List<Boolean> returnValues) {
261-
if (node.getDeterminism().ordinal() > Determinism.PER_STATEMENT.ordinal() ||
262+
if (!ExpressionUtil.isContantForStatement(node) ||
262263
returnValues.size() < node.getChildren().size()) {
263264
return Boolean.FALSE;
264265
}
@@ -281,16 +282,12 @@ public Boolean visit(LiteralExpression node) {
281282
}
282283

283284
@Override
284-
public Boolean visit(ProjectedColumnExpression node) {
285-
if (projector == null) {
286-
return super.visit(node);
287-
}
288-
Expression expression = projector.getExpressions()[node.getPosition()];
289-
// Only look one level down the projection.
290-
if (expression instanceof ProjectedColumnExpression) {
291-
return super.visit(node);
292-
}
293-
return expression.accept(this);
285+
public Boolean visit(KeyValueColumnExpression keyValueColumnExpression) {
286+
return ExpressionUtil.isColumnExpressionConstant(keyValueColumnExpression, whereExpression);
287+
}
288+
@Override
289+
public Boolean visit(ProjectedColumnExpression projectedColumnExpression) {
290+
return ExpressionUtil.isColumnExpressionConstant(projectedColumnExpression, whereExpression);
294291
}
295292
}
296293
/**

‎phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -562,8 +562,16 @@ protected QueryPlan compileSingleFlatQuery(StatementContext context, SelectState
562562
groupBy = groupBy.compile(context, innerPlanTupleProjector);
563563
context.setResolver(resolver); // recover resolver
564564
RowProjector projector = ProjectionCompiler.compile(context, select, groupBy, asSubquery ? Collections.<PDatum>emptyList() : targetColumns, where);
565-
OrderBy orderBy = OrderByCompiler.compile(context, select, groupBy, limit, offset, projector,
566-
groupBy == GroupBy.EMPTY_GROUP_BY ? innerPlanTupleProjector : null, isInRowKeyOrder);
565+
OrderBy orderBy = OrderByCompiler.compile(
566+
context,
567+
select,
568+
groupBy,
569+
limit,
570+
offset,
571+
projector,
572+
innerPlanTupleProjector,
573+
isInRowKeyOrder,
574+
where);
567575
context.getAggregationManager().compile(context, groupBy);
568576
// Final step is to build the query plan
569577
if (!asSubquery) {

‎phoenix-core/src/main/java/org/apache/phoenix/compile/RowProjector.java

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,8 @@
2222
import java.util.Collections;
2323
import java.util.List;
2424

25-
import org.apache.phoenix.expression.Determinism;
2625
import org.apache.phoenix.expression.Expression;
27-
import org.apache.phoenix.expression.visitor.CloneNonDeterministicExpressionVisitor;
26+
import org.apache.phoenix.expression.visitor.CloneExpressionVisitor;
2827
import org.apache.phoenix.schema.ColumnNotFoundException;
2928
import org.apache.phoenix.util.SchemaUtil;
3029

@@ -97,17 +96,17 @@ public RowProjector(List<? extends ColumnProjector> columnProjectors, int estima
9796
this.isProjectEmptyKeyValue = isProjectEmptyKeyValue;
9897
this.isProjectAll = isProjectAll;
9998
this.hasUDFs = hasUDFs;
100-
boolean hasPerInvocationExpression = false;
99+
boolean cloneRequired = false;
101100
if (!hasUDFs) {
102101
for (int i = 0; i < this.columnProjectors.size(); i++) {
103102
Expression expression = this.columnProjectors.get(i).getExpression();
104-
if (expression.getDeterminism() == Determinism.PER_INVOCATION) {
105-
hasPerInvocationExpression = true;
103+
if (expression.isCloneExpression()) {
104+
cloneRequired = true;
106105
break;
107106
}
108107
}
109108
}
110-
this.cloneRequired = hasPerInvocationExpression || hasUDFs;
109+
this.cloneRequired = cloneRequired || hasUDFs;
111110
}
112111

113112
public RowProjector cloneIfNecessary() {
@@ -118,8 +117,8 @@ public RowProjector cloneIfNecessary() {
118117
for (int i = 0; i < this.columnProjectors.size(); i++) {
119118
ColumnProjector colProjector = columnProjectors.get(i);
120119
Expression expression = colProjector.getExpression();
121-
if (expression.getDeterminism() == Determinism.PER_INVOCATION) {
122-
CloneNonDeterministicExpressionVisitor visitor = new CloneNonDeterministicExpressionVisitor();
120+
if (expression.isCloneExpression()) {
121+
CloneExpressionVisitor visitor = new CloneExpressionVisitor();
123122
Expression clonedExpression = expression.accept(visitor);
124123
clonedColProjectors.add(new ExpressionProjector(colProjector.getName(),
125124
colProjector.getTableName(),

‎phoenix-core/src/main/java/org/apache/phoenix/expression/BaseCompoundExpression.java

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ public abstract class BaseCompoundExpression extends BaseExpression {
3535
private boolean isStateless;
3636
private Determinism determinism;
3737
private boolean requiresFinalEvaluation;
38+
private boolean cloneExpression;
3839

3940
public BaseCompoundExpression() {
4041
init(Collections.<Expression>emptyList());
@@ -49,17 +50,20 @@ private void init(List<Expression> children) {
4950
boolean isStateless = true;
5051
boolean isNullable = false;
5152
boolean requiresFinalEvaluation = false;
53+
boolean cloneExpression = false;
5254
this.determinism = Determinism.ALWAYS;
5355
for (int i = 0; i < children.size(); i++) {
5456
Expression child = children.get(i);
5557
isNullable |= child.isNullable();
5658
isStateless &= child.isStateless();
5759
this.determinism = this.determinism.combine(child.getDeterminism());
5860
requiresFinalEvaluation |= child.requiresFinalEvaluation();
61+
cloneExpression |= child.isCloneExpression();
5962
}
6063
this.isStateless = isStateless;
6164
this.isNullable = isNullable;
6265
this.requiresFinalEvaluation = requiresFinalEvaluation;
66+
this.cloneExpression = cloneExpression;
6367
}
6468

6569
@Override
@@ -72,7 +76,12 @@ public List<Expression> getChildren() {
7276
public Determinism getDeterminism() {
7377
return determinism;
7478
}
75-
79+
80+
@Override
81+
public boolean isCloneExpression() {
82+
return this.cloneExpression;
83+
}
84+
7685
@Override
7786
public boolean isStateless() {
7887
return isStateless;

0 commit comments

Comments
 (0)