Skip to content

Commit 4ae6d73

Browse files
juripetersenjuripetersen
authored
Merge pull request #546 from mspruc/main
Restructure visitors & add filters w. where clauses for sql-api
2 parents 91573ae + 166022b commit 4ae6d73

17 files changed

Lines changed: 432 additions & 337 deletions

‎wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangAggregateVisitor.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818

1919
package org.apache.wayang.api.sql.calcite.converter;
2020

21-
import java.util.ArrayList;
2221
import java.util.List;
2322
import java.util.Set;
2423

‎wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangCrossJoinVisitor.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020

2121
import java.io.Serializable;
2222

23-
import org.apache.wayang.api.sql.calcite.converter.functions.FlattenJoinResult;
23+
import org.apache.wayang.api.sql.calcite.converter.functions.JoinFlattenResult;
2424
import org.apache.wayang.api.sql.calcite.rel.WayangJoin;
2525
import org.apache.wayang.basic.data.Record;
2626
import org.apache.wayang.basic.data.Tuple2;
@@ -48,7 +48,7 @@ Operator visit(final WayangJoin wayangRelNode) {
4848
childOpLeft.connectTo(0, join, 0);
4949
childOpRight.connectTo(0, join, 1);
5050

51-
final SerializableFunction<Tuple2<Record, Record>, Record> mp = new FlattenJoinResult();
51+
final SerializableFunction<Tuple2<Record, Record>, Record> mp = new JoinFlattenResult();
5252

5353
final MapOperator<Tuple2<Record, Record>, Record> mapOperator = new MapOperator<Tuple2<Record, Record>, Record>(
5454
mp,

‎wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangFilterVisitor.java‎

Lines changed: 2 additions & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -19,18 +19,13 @@
1919
package org.apache.wayang.api.sql.calcite.converter;
2020

2121
import org.apache.calcite.rel.core.Filter;
22-
import org.apache.calcite.rex.RexCall;
23-
import org.apache.calcite.rex.RexInputRef;
24-
import org.apache.calcite.rex.RexLiteral;
2522
import org.apache.calcite.rex.RexNode;
26-
import org.apache.calcite.rex.RexVisitorImpl;
27-
import org.apache.calcite.runtime.SqlFunctions;
2823
import org.apache.calcite.sql.SqlKind;
2924

25+
import org.apache.wayang.api.sql.calcite.converter.functions.FilterPredicateImpl;
3026
import org.apache.wayang.api.sql.calcite.rel.WayangFilter;
3127
import org.apache.wayang.basic.data.Record;
3228
import org.apache.wayang.basic.operators.FilterOperator;
33-
import org.apache.wayang.core.function.FunctionDescriptor;
3429
import org.apache.wayang.core.plan.wayangplan.Operator;
3530

3631
import java.util.EnumSet;
@@ -42,7 +37,6 @@ public class WayangFilterVisitor extends WayangRelNodeVisitor<WayangFilter> {
4237

4338
@Override
4439
Operator visit(final WayangFilter wayangRelNode) {
45-
4640
final Operator childOp = wayangRelConverter.convert(wayangRelNode.getInput(0));
4741

4842
final RexNode condition = ((Filter) wayangRelNode).getCondition();
@@ -56,106 +50,8 @@ Operator visit(final WayangFilter wayangRelNode) {
5650
return filter;
5751
}
5852

59-
private class FilterPredicateImpl implements FunctionDescriptor.SerializablePredicate<Record> {
60-
61-
private final RexNode condition;
62-
63-
private FilterPredicateImpl(final RexNode condition) {
64-
this.condition = condition;
65-
}
66-
67-
@Override
68-
public boolean test(final Record record) {
69-
return condition.accept(new EvaluateFilterCondition(true, record));
70-
}
71-
}
72-
73-
private class EvaluateFilterCondition extends RexVisitorImpl<Boolean> {
74-
75-
final Record record;
76-
77-
protected EvaluateFilterCondition(final boolean deep, final Record record) {
78-
super(deep);
79-
this.record = record;
80-
}
81-
82-
@Override
83-
public Boolean visitCall(final RexCall call) {
84-
final SqlKind kind = call.getKind();
85-
86-
if (!kind.belongsTo(WayangFilterVisitor.SUPPORTED_OPS))
87-
throw new IllegalStateException(
88-
"Cannot handle this filter predicate yet: " + kind + " during RexCall: " + call);
89-
90-
switch (kind) {
91-
// Since NOT captures only one operand we just get
92-
// the first
93-
case NOT:
94-
assert (call.getOperands().size() == 1) : "SqlKind.NOT should only have 1 operand in call got: " + call.getOperands().size() + ", call: " + call;
95-
return !(call.getOperands().get(0).accept(this));
96-
case AND:
97-
return call.getOperands().stream().allMatch(operator -> operator.accept(this));
98-
case OR:
99-
return call.getOperands().stream().anyMatch(operator -> operator.accept(this));
100-
default:
101-
assert (call.getOperands().size() == 2);
102-
return eval(record, kind, call.getOperands().get(0), call.getOperands().get(1));
103-
}
104-
}
105-
106-
public boolean eval(final Record record, final SqlKind kind, final RexNode leftOperand,
107-
final RexNode rightOperand) {
108-
109-
if (leftOperand instanceof RexInputRef && rightOperand instanceof RexLiteral) {
110-
final RexInputRef rexInputRef = (RexInputRef) leftOperand;
111-
final int index = rexInputRef.getIndex();
112-
final Object field = record.getField(index);
113-
final RexLiteral rexLiteral = (RexLiteral) rightOperand;
114-
switch (kind) {
115-
case LIKE:
116-
return SqlFunctions.like(field.toString(), rexLiteral.toString().replace("'", ""));
117-
case GREATER_THAN:
118-
return isGreaterThan(field, rexLiteral);
119-
case LESS_THAN:
120-
return isLessThan(field, rexLiteral);
121-
case EQUALS:
122-
return isEqualTo(field, rexLiteral);
123-
case GREATER_THAN_OR_EQUAL:
124-
return isGreaterThan(field, rexLiteral) || isEqualTo(field, rexLiteral);
125-
case LESS_THAN_OR_EQUAL:
126-
return isLessThan(field, rexLiteral) || isEqualTo(field, rexLiteral);
127-
default:
128-
throw new IllegalStateException("Predicate not supported yet");
129-
130-
}
131-
132-
} else {
133-
throw new IllegalStateException("Predicate not supported yet");
134-
}
135-
136-
}
137-
138-
private boolean isGreaterThan(final Object o, final RexLiteral rexLiteral) {
139-
// return rexLiteral.getValue().compareTo(o)< 0;
140-
return ((Comparable) o).compareTo(rexLiteral.getValueAs(o.getClass())) > 0;
141-
142-
}
143-
144-
private boolean isLessThan(final Object o, final RexLiteral rexLiteral) {
145-
return ((Comparable) o).compareTo(rexLiteral.getValueAs(o.getClass())) < 0;
146-
}
147-
148-
private boolean isEqualTo(final Object o, final RexLiteral rexLiteral) {
149-
try {
150-
return ((Comparable) o).compareTo(rexLiteral.getValueAs(o.getClass())) == 0;
151-
} catch (final Exception e) {
152-
throw new IllegalStateException("Predicate not supported yet");
153-
}
154-
}
155-
}
156-
15753
/** for quick sanity check **/
158-
private static final EnumSet<SqlKind> SUPPORTED_OPS = EnumSet.of(SqlKind.AND, SqlKind.OR, SqlKind.NOT,
54+
public static final EnumSet<SqlKind> SUPPORTED_OPS = EnumSet.of(SqlKind.AND, SqlKind.OR, SqlKind.NOT,
15955
SqlKind.EQUALS, SqlKind.NOT_EQUALS,
16056
SqlKind.LESS_THAN, SqlKind.GREATER_THAN,
16157
SqlKind.GREATER_THAN_OR_EQUAL, SqlKind.LESS_THAN_OR_EQUAL, SqlKind.LIKE);

‎wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangJoinVisitor.java‎

Lines changed: 34 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -18,131 +18,74 @@
1818

1919
package org.apache.wayang.api.sql.calcite.converter;
2020

21+
import java.util.List;
22+
import java.util.stream.Collectors;
23+
2124
import org.apache.calcite.rel.core.Join;
2225
import org.apache.calcite.rex.RexCall;
2326
import org.apache.calcite.rex.RexInputRef;
2427
import org.apache.calcite.rex.RexNode;
25-
import org.apache.calcite.rex.RexVisitorImpl;
2628
import org.apache.calcite.sql.SqlKind;
29+
30+
import org.apache.wayang.api.sql.calcite.converter.functions.JoinFlattenResult;
31+
import org.apache.wayang.api.sql.calcite.converter.functions.JoinKeyExtractor;
2732
import org.apache.wayang.api.sql.calcite.rel.WayangJoin;
33+
2834
import org.apache.wayang.basic.data.Record;
2935
import org.apache.wayang.basic.data.Tuple2;
3036
import org.apache.wayang.basic.operators.JoinOperator;
3137
import org.apache.wayang.basic.operators.MapOperator;
32-
import org.apache.wayang.core.function.FunctionDescriptor;
3338
import org.apache.wayang.core.function.TransformationDescriptor;
3439
import org.apache.wayang.core.plan.wayangplan.Operator;
40+
import org.apache.wayang.core.util.ReflectionUtils;
3541

3642
public class WayangJoinVisitor extends WayangRelNodeVisitor<WayangJoin> {
3743

38-
WayangJoinVisitor(WayangRelConverter wayangRelConverter) {
44+
WayangJoinVisitor(final WayangRelConverter wayangRelConverter) {
3945
super(wayangRelConverter);
4046
}
4147

4248
@Override
43-
Operator visit(WayangJoin wayangRelNode) {
44-
Operator childOpLeft = wayangRelConverter.convert(wayangRelNode.getInput(0));
45-
Operator childOpRight = wayangRelConverter.convert(wayangRelNode.getInput(1));
49+
Operator visit(final WayangJoin wayangRelNode) {
50+
final Operator childOpLeft = wayangRelConverter.convert(wayangRelNode.getInput(0));
51+
final Operator childOpRight = wayangRelConverter.convert(wayangRelNode.getInput(1));
52+
53+
final RexNode condition = ((Join) wayangRelNode).getCondition();
54+
final RexCall call = (RexCall) condition;
4655

47-
RexNode condition = ((Join) wayangRelNode).getCondition();
56+
final List<Integer> keys = call.getOperands().stream()
57+
.map(RexInputRef.class::cast)
58+
.map(RexInputRef::getIndex)
59+
.collect(Collectors.toList());
4860

61+
assert (keys.size() == 2) : "Amount of keys found in join was not 2, got: " + keys.size();
62+
4963
if (!condition.isA(SqlKind.EQUALS)) {
5064
throw new UnsupportedOperationException("Only equality joins supported");
5165
}
5266

53-
//offset of the index in the right child
54-
int offset = wayangRelNode.getInput(0).getRowType().getFieldCount();
67+
// offset of the index in the right child
68+
final int offset = wayangRelNode.getInput(0).getRowType().getFieldCount();
5569

56-
int leftKeyIndex = condition.accept(new KeyIndex(false, Child.LEFT));
57-
int rightKeyIndex = condition.accept(new KeyIndex(false, Child.RIGHT)) - offset;
70+
final int leftKeyIndex = keys.get(0);
71+
final int rightKeyIndex = keys.get(1) - offset;
5872

59-
JoinOperator<Record, Record, Object> join = new JoinOperator<>(
60-
new TransformationDescriptor<>(new KeyExtractor(leftKeyIndex), Record.class, Object.class),
61-
new TransformationDescriptor<>(new KeyExtractor(rightKeyIndex), Record.class, Object.class)
62-
);
73+
final JoinOperator<Record, Record, Object> join = new JoinOperator<>(
74+
new TransformationDescriptor<>(new JoinKeyExtractor(leftKeyIndex), Record.class, Object.class),
75+
new TransformationDescriptor<>(new JoinKeyExtractor(rightKeyIndex), Record.class, Object.class));
6376

64-
//call connectTo on both operators (left and right)
77+
// call connectTo on both operators (left and right)
6578
childOpLeft.connectTo(0, join, 0);
6679
childOpRight.connectTo(0, join, 1);
6780

6881
// Join returns Tuple2 - map to a Record
69-
MapOperator<Tuple2, Record> mapOperator = new MapOperator(
70-
new MapFunctionImpl(),
71-
Tuple2.class,
72-
Record.class
73-
);
82+
final MapOperator<Tuple2<Record, Record>, Record> mapOperator = new MapOperator<Tuple2<Record, Record>, Record>(
83+
new JoinFlattenResult(),
84+
ReflectionUtils.specify(Tuple2.class),
85+
Record.class);
86+
7487
join.connectTo(0, mapOperator, 0);
7588

7689
return mapOperator;
7790
}
78-
79-
/**
80-
* Extracts key index from the call
81-
*/
82-
private class KeyIndex extends RexVisitorImpl<Integer> {
83-
final Child child;
84-
85-
protected KeyIndex(boolean deep, Child child) {
86-
super(deep);
87-
this.child = child;
88-
}
89-
90-
@Override
91-
public Integer visitCall(RexCall call) {
92-
RexNode operand = call.getOperands().get(child.ordinal());
93-
if (!(operand instanceof RexInputRef)) {
94-
throw new UnsupportedOperationException("Unsupported operation");
95-
}
96-
RexInputRef rexInputRef = (RexInputRef) operand;
97-
return rexInputRef.getIndex();
98-
}
99-
}
100-
101-
/**
102-
* Extracts the key
103-
*/
104-
private class KeyExtractor implements FunctionDescriptor.SerializableFunction<Record, Object> {
105-
private final int index;
106-
107-
public KeyExtractor(int index) {
108-
this.index = index;
109-
}
110-
111-
public Object apply(final Record record) {
112-
return record.getField(index);
113-
}
114-
}
115-
116-
/**
117-
* Flattens Tuple2<Record, Record> to Record
118-
*/
119-
private class MapFunctionImpl implements FunctionDescriptor.SerializableFunction<Tuple2<Record, Record>, Record> {
120-
public MapFunctionImpl() {
121-
super();
122-
}
123-
124-
@Override
125-
public Record apply(final Tuple2<Record, Record> tuple2) {
126-
int length1 = tuple2.getField0().size();
127-
int length2 = tuple2.getField1().size();
128-
129-
int totalLength = length1 + length2;
130-
131-
Object[] fields = new Object[totalLength];
132-
133-
for (int i = 0; i < length1; i++) {
134-
fields[i] = tuple2.getField0().getField(i);
135-
}
136-
for (int j = length1; j < totalLength; j++) {
137-
fields[j] = tuple2.getField1().getField(j - length1);
138-
}
139-
return new Record(fields);
140-
141-
}
142-
}
143-
144-
// Helpers
145-
private enum Child {
146-
LEFT, RIGHT
147-
}
14891
}

‎wayang-api/wayang-api-sql/src/main/java/org/apache/wayang/api/sql/calcite/converter/WayangMultiConditionJoinVisitor.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,11 +21,13 @@
2121
import java.io.Serializable;
2222
import java.util.List;
2323
import java.util.stream.Collectors;
24+
2425
import org.apache.calcite.rel.core.Join;
2526
import org.apache.calcite.rex.RexCall;
2627
import org.apache.calcite.rex.RexInputRef;
2728
import org.apache.calcite.rex.RexNode;
28-
import org.apache.wayang.api.sql.calcite.converter.functions.MultiConditionJoinFuncImpl;
29+
30+
import org.apache.wayang.api.sql.calcite.converter.functions.JoinFlattenResult;
2931
import org.apache.wayang.api.sql.calcite.converter.functions.MultiConditionJoinKeyExtractor;
3032
import org.apache.wayang.api.sql.calcite.rel.WayangJoin;
3133
import org.apache.wayang.basic.data.Record;
@@ -119,7 +121,7 @@ Operator visit(WayangJoin wayangRelNode) {
119121
childOpRight.connectTo(0, join, 1);
120122

121123
// Join returns Tuple2 - map to a Record
122-
final SerializableFunction<Tuple2<Record, Record>, Record> mp = new MultiConditionJoinFuncImpl();
124+
final SerializableFunction<Tuple2<Record, Record>, Record> mp = new JoinFlattenResult();
123125

124126
final MapOperator<Tuple2<Record, Record>, Record> mapOperator = new MapOperator<Tuple2<Record, Record>, Record>(
125127
mp,

0 commit comments

Comments
 (0)