-
Notifications
You must be signed in to change notification settings - Fork 516
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Rust] add substriat for flink and be compatible for other engines
Signed-off-by: mag1c1an1 <[email protected]> add flink expression to substrait Signed-off-by: mag1c1an1 <[email protected]> add more functions Signed-off-by: mag1c1an1 <[email protected]> add more tests Signed-off-by: mag1c1an1 <[email protected]> add base schema for namedscan, substriat type to arrow type Signed-off-by: mag1c1an1 <[email protected]> compatibility Signed-off-by: mag1c1an1 <[email protected]> switch to java8 Signed-off-by: mag1c1an1 <[email protected]> before apply cargo fix Signed-off-by: mag1c1an1 <[email protected]> cargo clippy && cargo fmt Signed-off-by: mag1c1an1 <[email protected]> fix ci Signed-off-by: mag1c1an1 <[email protected]> rebase Signed-off-by: mag1c1an1 <[email protected]> refactor Signed-off-by: mag1c1an1 <[email protected]>
- Loading branch information
Showing
41 changed files
with
2,007 additions
and
499 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
68 changes: 68 additions & 0 deletions
68
lakesoul-flink/src/main/java/org/apache/flink/lakesoul/substrait/SubstraitFlinkUtil.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| package org.apache.flink.lakesoul.substrait; | ||
|
|
||
| import io.substrait.expression.Expression; | ||
| import io.substrait.expression.ExpressionCreator; | ||
| import io.substrait.extension.SimpleExtension; | ||
| import io.substrait.plan.Plan; | ||
| import io.substrait.type.TypeCreator; | ||
| import org.apache.arrow.vector.types.pojo.Schema; | ||
| import org.apache.flink.api.java.tuple.Tuple2; | ||
| import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; | ||
| import org.apache.flink.table.expressions.CallExpression; | ||
| import org.apache.flink.table.expressions.FieldReferenceExpression; | ||
| import org.apache.flink.table.expressions.ResolvedExpression; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
|
|
||
| import static com.dmetasoul.lakesoul.lakesoul.io.substrait.SubstraitUtil.*; | ||
|
|
||
| public class SubstraitFlinkUtil { | ||
|
|
||
| public static Tuple2<SupportsFilterPushDown.Result, io.substrait.proto.Plan> flinkExprToSubStraitPlan( | ||
| List<ResolvedExpression> exprs, | ||
| List<ResolvedExpression> remaining, | ||
| String tableName, | ||
| String tableSchema | ||
| ) throws IOException { | ||
| List<ResolvedExpression> accepted = new ArrayList<>(); | ||
| Schema arrowSchema = Schema.fromJSON(tableSchema); | ||
| Expression last = null; | ||
| for (ResolvedExpression expr : exprs) { | ||
| Expression e = doTransform(expr,arrowSchema); | ||
| if (e == null) { | ||
| remaining.add(expr); | ||
| } else { | ||
| accepted.add(expr); | ||
| if (last != null) { | ||
| SimpleExtension.FunctionAnchor fa = SimpleExtension.FunctionAnchor.of(BooleanNamespace, "and:bool"); | ||
| last = ExpressionCreator.scalarFunction(Se.getScalarFunction(fa), TypeCreator.NULLABLE.BOOLEAN, last, e); | ||
| } else { | ||
| last = e; | ||
| } | ||
| } | ||
| } | ||
| Plan filter = exprToFilter(last, tableName, arrowSchema); | ||
| return Tuple2.of(SupportsFilterPushDown.Result.of(accepted, remaining), planToProto(filter)); | ||
| } | ||
|
|
||
| public static Expression doTransform(ResolvedExpression flinkExpression, Schema arrow_schema) { | ||
| SubstraitVisitor substraitVisitor = new SubstraitVisitor(arrow_schema); | ||
| return flinkExpression.accept(substraitVisitor); | ||
| } | ||
|
|
||
| public static boolean filterContainsPartitionColumn(ResolvedExpression expression, Set<String> partitionCols) { | ||
| if (expression instanceof FieldReferenceExpression) { | ||
| return partitionCols.contains(((FieldReferenceExpression) expression).getName()); | ||
| } else if (expression instanceof CallExpression) { | ||
| for (ResolvedExpression child : expression.getResolvedChildren()) { | ||
| if (filterContainsPartitionColumn(child, partitionCols)) { | ||
| return true; | ||
| } | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
| } |
Oops, something went wrong.