Skip to content

Commit 10b30f1

Browse files
shirly121shirly121
committed
[GIE Compiler] fix bug of operator
1 parent b37ab90 commit 10b30f1

File tree

5 files changed

+377
-69
lines changed

5 files changed

+377
-69
lines changed

‎interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/FfiPhysicalBuilder.java

+15-61
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818

1919
import com.alibaba.graphscope.common.config.Configs;
2020
import com.alibaba.graphscope.common.config.PegasusConfig;
21-
import com.alibaba.graphscope.common.intermediate.ArgUtils;
2221
import com.alibaba.graphscope.common.ir.rel.GraphLogicalAggregate;
2322
import com.alibaba.graphscope.common.ir.rel.GraphLogicalProject;
2423
import com.alibaba.graphscope.common.ir.rel.GraphLogicalSort;
@@ -29,30 +28,20 @@
2928
import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalSource;
3029
import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalMultiMatch;
3130
import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalSingleMatch;
32-
import com.alibaba.graphscope.common.ir.rel.type.group.GraphGroupKeys;
33-
import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
3431
import com.alibaba.graphscope.common.ir.runtime.RegularPhysicalBuilder;
35-
import com.alibaba.graphscope.common.ir.runtime.proto.RexToProtoConverter;
3632
import com.alibaba.graphscope.common.ir.runtime.type.PhysicalNode;
37-
import com.alibaba.graphscope.common.ir.tools.AliasInference;
3833
import com.alibaba.graphscope.common.ir.tools.LogicalPlan;
39-
import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
4034
import com.alibaba.graphscope.common.jna.IrCoreLibrary;
4135
import com.alibaba.graphscope.common.jna.type.FfiData;
42-
import com.alibaba.graphscope.common.jna.type.FfiPbPointer;
4336
import com.alibaba.graphscope.common.jna.type.FfiResult;
4437
import com.alibaba.graphscope.common.jna.type.ResultCode;
4538
import com.alibaba.graphscope.common.store.IrMeta;
46-
import com.alibaba.graphscope.gaia.proto.OuterExpression;
4739
import com.google.common.base.Preconditions;
4840
import com.sun.jna.Pointer;
4941
import com.sun.jna.ptr.IntByReference;
5042

5143
import org.apache.calcite.rel.RelNode;
5244
import org.apache.calcite.rel.logical.LogicalFilter;
53-
import org.apache.calcite.rel.type.RelDataTypeField;
54-
import org.apache.calcite.rex.RexNode;
55-
import org.apache.calcite.rex.RexVariable;
5645

5746
import java.util.List;
5847

@@ -106,7 +95,7 @@ public void appendNode(PhysicalNode<Pointer> node) {
10695
} else if (original instanceof GraphLogicalAggregate) {
10796
// transform aggregate to project + dedup by key
10897
if (((GraphLogicalAggregate) original).getAggCalls().isEmpty()) {
109-
appendProjectDedup((GraphLogicalAggregate) original, oprIdx);
98+
appendProjectDedup(node, oprIdx);
11099
} else {
111100
checkFfiResult(
112101
LIB.appendGroupbyOperator(
@@ -164,60 +153,25 @@ private void checkFfiResult(FfiResult res) {
164153
}
165154
}
166155

167-
private boolean isColumnId() {
168-
return this.irMeta.getSchema().isColumnId();
169-
}
170-
171156
private void appendMatch(PhysicalNode<Pointer> node, IntByReference oprIdx) {
172-
// append dummy source
173-
Pointer ptrScan = LIB.initScanOperator(Utils.ffiScanOpt(GraphOpt.Source.VERTEX));
174-
checkFfiResult(LIB.appendScanOperator(ptrPlan, ptrScan, oprIdx.getValue(), oprIdx));
157+
List<Pointer> ffiNodes = node.getNodes();
158+
Preconditions.checkArgument(
159+
ffiNodes.size() == 2,
160+
"should have 2 ffi nodes, one is `scan` and the other is `match`");
161+
checkFfiResult(LIB.appendScanOperator(ptrPlan, ffiNodes.get(0), oprIdx.getValue(), oprIdx));
175162
checkFfiResult(
176-
LIB.appendPatternOperator(ptrPlan, node.getNode(), oprIdx.getValue(), oprIdx));
163+
LIB.appendPatternOperator(ptrPlan, ffiNodes.get(1), oprIdx.getValue(), oprIdx));
177164
}
178165

179-
private void appendProjectDedup(GraphLogicalAggregate aggregate, IntByReference oprIdx) {
180-
GraphGroupKeys keys = aggregate.getGroupKey();
166+
private void appendProjectDedup(PhysicalNode<Pointer> node, IntByReference oprIdx) {
167+
List<Pointer> ffiNodes = node.getNodes();
181168
Preconditions.checkArgument(
182-
keys.groupKeyCount() > 0 && aggregate.getAggCalls().isEmpty(),
183-
"group keys should not be empty while group calls should be empty if need dedup");
184-
List<RelDataTypeField> fields = aggregate.getRowType().getFieldList();
185-
Pointer ptrProject = LIB.initProjectOperator(false);
186-
for (int i = 0; i < keys.groupKeyCount(); ++i) {
187-
RexNode var = keys.getVariables().get(i);
188-
Preconditions.checkArgument(
189-
var instanceof RexGraphVariable,
190-
"each group key should be type %s, but is %s",
191-
RexGraphVariable.class,
192-
var.getClass());
193-
OuterExpression.Expression expr =
194-
var.accept(new RexToProtoConverter(true, isColumnId()));
195-
int aliasId;
196-
if (i >= fields.size()
197-
|| (aliasId = fields.get(i).getIndex()) == AliasInference.DEFAULT_ID) {
198-
throw new IllegalArgumentException(
199-
"each group key should have an alias if need dedup");
200-
}
201-
checkFfiResult(
202-
LIB.addProjectExprPbAlias(
203-
ptrProject,
204-
new FfiPbPointer.ByValue(expr.toByteArray()),
205-
ArgUtils.asAlias(aliasId)));
206-
}
207-
Pointer ptrDedup = LIB.initDedupOperator();
208-
for (int i = 0; i < keys.groupKeyCount(); ++i) {
209-
RelDataTypeField field = fields.get(i);
210-
RexVariable rexVar =
211-
RexGraphVariable.of(field.getIndex(), field.getName(), field.getType());
212-
OuterExpression.Variable exprVar =
213-
rexVar.accept(new RexToProtoConverter(true, isColumnId()))
214-
.getOperators(0)
215-
.getVar();
216-
checkFfiResult(
217-
LIB.addDedupKeyPb(ptrDedup, new FfiPbPointer.ByValue(exprVar.toByteArray())));
218-
}
219-
checkFfiResult(LIB.appendProjectOperator(ptrPlan, ptrProject, oprIdx.getValue(), oprIdx));
220-
checkFfiResult(LIB.appendDedupOperator(ptrPlan, ptrDedup, oprIdx.getValue(), oprIdx));
169+
ffiNodes.size() == 2,
170+
"should have 2 ffi nodes, one is `project` and the other is `dedup`");
171+
checkFfiResult(
172+
LIB.appendProjectOperator(ptrPlan, ffiNodes.get(0), oprIdx.getValue(), oprIdx));
173+
checkFfiResult(
174+
LIB.appendDedupOperator(ptrPlan, ffiNodes.get(1), oprIdx.getValue(), oprIdx));
221175
}
222176

223177
private void appendSink(IntByReference oprIdx) {

‎interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java

+49-3
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalMultiMatch;
2626
import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalSingleMatch;
2727
import com.alibaba.graphscope.common.ir.rel.type.group.GraphAggCall;
28+
import com.alibaba.graphscope.common.ir.rel.type.group.GraphGroupKeys;
2829
import com.alibaba.graphscope.common.ir.rel.type.order.GraphFieldCollation;
2930
import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
3031
import com.alibaba.graphscope.common.ir.runtime.proto.RexToProtoConverter;
@@ -50,6 +51,7 @@
5051
import org.apache.calcite.rex.RexCall;
5152
import org.apache.calcite.rex.RexLiteral;
5253
import org.apache.calcite.rex.RexNode;
54+
import org.apache.calcite.rex.RexVariable;
5355
import org.apache.calcite.sql.SqlKind;
5456
import org.apache.commons.lang3.ObjectUtils;
5557
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -168,7 +170,9 @@ public RelNode visit(GraphLogicalSingleMatch match) {
168170
ptrPattern,
169171
new FfiPbPointer.ByValue(k.toByteArray())));
170172
});
171-
return new PhysicalNode(match, ptrPattern);
173+
// add dummy source
174+
Pointer ptrScan = LIB.initScanOperator(Utils.ffiScanOpt(GraphOpt.Source.VERTEX));
175+
return new PhysicalNode(match, ImmutableList.of(ptrScan, ptrPattern));
172176
case OPTIONAL:
173177
case ANTI:
174178
default:
@@ -192,7 +196,9 @@ public RelNode visit(GraphLogicalMultiMatch match) {
192196
LIB.addPatternMeta(
193197
ptrPattern, new FfiPbPointer.ByValue(k.toByteArray())));
194198
});
195-
return new PhysicalNode(match, ptrPattern);
199+
// add dummy source
200+
Pointer ptrScan = LIB.initScanOperator(Utils.ffiScanOpt(GraphOpt.Source.VERTEX));
201+
return new PhysicalNode(match, ImmutableList.of(ptrScan, ptrPattern));
196202
}
197203

198204
@Override
@@ -239,7 +245,47 @@ public PhysicalNode visit(GraphLogicalProject project) {
239245
public PhysicalNode visit(GraphLogicalAggregate aggregate) {
240246
List<GraphAggCall> groupCalls = aggregate.getAggCalls();
241247
if (groupCalls.isEmpty()) { // transform to project + dedup by keys
242-
return new PhysicalNode(aggregate, null);
248+
GraphGroupKeys keys = aggregate.getGroupKey();
249+
Preconditions.checkArgument(
250+
keys.groupKeyCount() > 0,
251+
"group keys should not be empty while group calls is empty");
252+
List<RelDataTypeField> fields = aggregate.getRowType().getFieldList();
253+
Pointer ptrProject = LIB.initProjectOperator(false);
254+
for (int i = 0; i < keys.groupKeyCount(); ++i) {
255+
RexNode var = keys.getVariables().get(i);
256+
Preconditions.checkArgument(
257+
var instanceof RexGraphVariable,
258+
"each group key should be type %s, but is %s",
259+
RexGraphVariable.class,
260+
var.getClass());
261+
OuterExpression.Expression expr =
262+
var.accept(new RexToProtoConverter(true, isColumnId));
263+
int aliasId;
264+
if (i >= fields.size()
265+
|| (aliasId = fields.get(i).getIndex()) == AliasInference.DEFAULT_ID) {
266+
throw new IllegalArgumentException(
267+
"each group key should have an alias if need dedup");
268+
}
269+
checkFfiResult(
270+
LIB.addProjectExprPbAlias(
271+
ptrProject,
272+
new FfiPbPointer.ByValue(expr.toByteArray()),
273+
ArgUtils.asAlias(aliasId)));
274+
}
275+
Pointer ptrDedup = LIB.initDedupOperator();
276+
for (int i = 0; i < keys.groupKeyCount(); ++i) {
277+
RelDataTypeField field = fields.get(i);
278+
RexVariable rexVar =
279+
RexGraphVariable.of(field.getIndex(), field.getName(), field.getType());
280+
OuterExpression.Variable exprVar =
281+
rexVar.accept(new RexToProtoConverter(true, isColumnId))
282+
.getOperators(0)
283+
.getVar();
284+
checkFfiResult(
285+
LIB.addDedupKeyPb(
286+
ptrDedup, new FfiPbPointer.ByValue(exprVar.toByteArray())));
287+
}
288+
return new PhysicalNode(aggregate, ImmutableList.of(ptrProject, ptrDedup));
243289
}
244290
Pointer ptrGroup = LIB.initGroupbyOperator();
245291
List<RelDataTypeField> fields = aggregate.getRowType().getFieldList();

‎interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/type/PhysicalNode.java

+19-5
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,16 @@
1616

1717
package com.alibaba.graphscope.common.ir.runtime.type;
1818

19+
import com.google.common.collect.ImmutableList;
20+
1921
import org.apache.calcite.plan.RelOptCluster;
2022
import org.apache.calcite.plan.RelTraitSet;
2123
import org.apache.calcite.rel.AbstractRelNode;
2224
import org.apache.calcite.rel.RelNode;
25+
import org.apache.commons.lang3.ObjectUtils;
2326

27+
import java.util.Collections;
28+
import java.util.List;
2429
import java.util.Objects;
2530

2631
/**
@@ -29,23 +34,32 @@
2934
*/
3035
public class PhysicalNode<T> extends AbstractRelNode {
3136
private final RelNode original;
32-
private final T node;
37+
private final List<T> nodes;
3338

34-
protected PhysicalNode(RelOptCluster cluster, RelTraitSet traitSet, RelNode original, T node) {
39+
protected PhysicalNode(
40+
RelOptCluster cluster, RelTraitSet traitSet, RelNode original, List<T> nodes) {
3541
super(cluster, traitSet);
3642
this.original = Objects.requireNonNull(original);
37-
this.node = Objects.requireNonNull(node);
43+
this.nodes = ObjectUtils.requireNonEmpty(nodes);
3844
}
3945

4046
public PhysicalNode(RelNode original, T node) {
41-
this(original.getCluster(), RelTraitSet.createEmpty(), original, node);
47+
this(original.getCluster(), RelTraitSet.createEmpty(), original, ImmutableList.of(node));
48+
}
49+
50+
public PhysicalNode(RelNode original, List<T> nodes) {
51+
this(original.getCluster(), RelTraitSet.createEmpty(), original, nodes);
4252
}
4353

4454
public RelNode getOriginal() {
4555
return original;
4656
}
4757

4858
public T getNode() {
49-
return node;
59+
return nodes.get(0);
60+
}
61+
62+
public List<T> getNodes() {
63+
return Collections.unmodifiableList(nodes);
5064
}
5165
}

‎interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java

+26
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,32 @@ public void logical_plan_2_test() throws Exception {
121121
}
122122
}
123123

124+
// Match (n) Return distinct n;
125+
@Test
126+
public void logical_plan_3_test() throws Exception {
127+
GraphBuilder builder = Utils.mockGraphBuilder();
128+
RelNode project =
129+
builder.source(
130+
new SourceConfig(
131+
GraphOpt.Source.VERTEX,
132+
new LabelConfig(false).addLabel("person"),
133+
"n"))
134+
.aggregate(builder.groupKey(builder.variable("n")))
135+
.build();
136+
Assert.assertEquals(
137+
"GraphLogicalAggregate(keys=[{variables=[n], aliases=[n]}], values=[[]])\n"
138+
+ " GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
139+
+ " alias=[n], opt=[VERTEX])",
140+
project.explain().trim());
141+
try (PhysicalBuilder<byte[]> ffiBuilder =
142+
new FfiPhysicalBuilder(
143+
getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(project, false))) {
144+
Assert.assertEquals(
145+
FileUtils.readJsonFromResource("ffi_logical_plan_3.json"),
146+
ffiBuilder.explain());
147+
}
148+
}
149+
124150
private Configs getMockGraphConfig() {
125151
return new Configs(ImmutableMap.of("servers", "1", "workers", "1"));
126152
}

0 commit comments

Comments
 (0)