[Rust] add substrait for flink and be compatible for other engines #454

mag1c1an1 · 2024-03-19T10:42:37Z

add flink expression to substrait

add more functions

add more tests

add base schema for namedscan, substriat type to arrow type

compatibility

switch to java8

Ceng23333 · 2024-03-21T07:34:45Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/source/LakeSoulLookupTableSource.java

@@ -235,6 +235,7 @@ public DynamicTableSource copy() {
        lsts.projectedFields = this.projectedFields;
        lsts.remainingPartitions = this.remainingPartitions;
        lsts.filter = this.filter;
+        lsts.filter = this.filter;


duplicate code

Ceng23333 · 2024-03-21T07:35:47Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/source/LakeSoulSource.java

@@ -52,15 +55,18 @@ public LakeSoulSource(TableId tableId,
                          List<String> pkColumns,
                          Map<String, String> optionParams,
                          @Nullable List<Map<String, String>> remainingPartitions,
-                          @Nullable FilterPredicate filter) {
+                          @Nullable FilterPredicate filterStr,


improper name 'filterStr'

Ceng23333 · 2024-03-21T07:39:08Z

...soul-flink/src/main/java/org/apache/flink/lakesoul/source/LakeSoulOneSplitRecordsReader.java

@@ -129,7 +133,11 @@ private void initializeReader() throws IOException {
        }

        if (filter != null) {


Will two kinds of filter cause conflict?

No, filterSter is null forever in above code. It is used to debug the difference of result datafusion::expr of two kinds filters.

Ceng23333 · 2024-03-21T07:42:27Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/substrait/SubstraitUtil.java

+        return Tuple2.of(SupportsFilterPushDown.Result.of(accepted, remaining), planToProto(filter));
+    }
+
+    static Schema toArrowSchema(String tableSchema) {


No diff from Schema.fromJSON

Ceng23333 · 2024-03-21T07:46:53Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/substrait/SubstraitUtil.java

+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class SubstraitUtil {


Move to lakesoul-io-java maybe better.

Ceng23333 · 2024-03-21T07:53:42Z

native-io/lakesoul-io-java/src/main/java/com/dmetasoul/lakesoul/lakesoul/io/NativeIOReader.java

+     *
+     * @param plan Filter{}
+     */
+    public void addFilterProto(Plan plan) {


refer com.dmetasoul.lakesoul.meta.jnr.NativeMetadataJavaClient#executeInsert

Ceng23333 · 2024-03-21T07:58:08Z

rust/lakesoul-io/src/datasource/physical_plan/merge.rs

@@ -255,20 +255,36 @@ pub async fn prune_filter_and_execute(
    df: DataFrame,
    request_schema: SchemaRef,
    filter_str: Vec<String>,


use Vec as input directly

mag1c1an1 · 2024-03-21T16:00:01Z

rebase from main, please review the dependency

Ceng23333

Please provide some flink test cases.

Ceng23333 · 2024-03-25T07:56:19Z

...soul-flink/src/main/java/org/apache/flink/lakesoul/source/LakeSoulOneSplitRecordsReader.java

@@ -73,7 +74,8 @@ public class LakeSoulOneSplitRecordsReader implements RecordsWithSplitIds<RowDat
    // arrow batch -> row, with requested schema
    private ArrowReader curArrowReaderRequestedSchema;

-    private final FilterPredicate filter;
+    private final FilterPredicate _filterPredicate;


Why use underline as var name

Ceng23333 · 2024-03-25T08:02:26Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/substrait/SubstraitFlinkUtil.java

+        return Tuple2.of(SupportsFilterPushDown.Result.of(accepted, remaining), planToProto(filter));
+    }
+
+    public static Expression doTransform(ResolvedExpression flinkExpression, Schema arrow_schema) {


arrow_schema should use CamelCase name in Java

Ceng23333 · 2024-03-25T08:08:50Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/substrait/SubstraitVisitor.java

+                }
+                return ExpressionCreator.binary(nullable, b);
+            }
+            case TINYINT:


Use integer type with exactly bit-width

Ceng23333 · 2024-03-25T08:12:43Z

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/substrait/SubstraitVisitor.java

+                }
+                return ExpressionCreator.fp64(nullable, d);
+            }
+            case DATE: {


Any unit test for date/timestamp case?

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]>

Signed-off-by: mag1c1an1 <[email protected]>

Ceng23333 · 2024-04-02T10:41:10Z

...kesoul-io-java/src/main/java/com/dmetasoul/lakesoul/lakesoul/io/substrait/SubstraitUtil.java

+import java.util.stream.Stream;
+
+public class SubstraitUtil {
+    public static final SimpleExtension.ExtensionCollection Se;


Replace 'Se' with recognizable name.

Ceng23333 · 2024-04-02T10:42:58Z

...kesoul-io-java/src/main/java/com/dmetasoul/lakesoul/lakesoul/io/substrait/SubstraitUtil.java

+
+public class SubstraitUtil {
+    public static final SimpleExtension.ExtensionCollection Se;
+    public static final SubstraitBuilder Builder;


Rename static constants with UPPERCASE_UNDERSCORE format

Ceng23333 · 2024-04-02T10:45:16Z

lakesoul-flink/pom.xml

@@ -466,6 +466,8 @@ SPDX-License-Identifier: Apache-2.0
                            <include>com.google.code.gson:gson</include>
                            <include>dev.failsafe:failsafe</include>
                            <include>com.google.protobuf:protobuf-java</include>
+                            <!--substrait-->


Try removing org.apache.parquet:parquet-column from pom

Ceng23333 · 2024-04-02T10:58:43Z

lakesoul-flink/src/test/java/org/apache/flink/lakesoul/test/SubstraitTest.java

+        createLakeSoulSourceTableWithDateType(createTableEnv);
+        // not supported
+//        String testSql = "select * from type_info where modifyTime=TO_TIMESTAMP_LTZ(1612176000,0)";
+        String testSql = "select * from type_info " +


Mark not supported datatype with comment.

Ceng23333 · 2024-04-02T11:00:44Z

lakesoul-flink/src/test/java/org/apache/flink/lakesoul/test/SubstraitTest.java

+
+public class SubstraitTest extends AbstractTestBase {
+
+    private final String BATCH_TYPE = "batch";


Add tests which filter on hash column/range column

Signed-off-by: mag1c1an1 <[email protected]>

mag1c1an1 force-pushed the sub branch 3 times, most recently from fdfad6b to f06e75f Compare March 21, 2024 06:02

Ceng23333 reviewed Mar 21, 2024

View reviewed changes

mag1c1an1 force-pushed the sub branch 3 times, most recently from 21bd868 to b7104da Compare March 21, 2024 15:58

mag1c1an1 force-pushed the sub branch from b7104da to 9aed0a0 Compare March 22, 2024 11:17

Ceng23333 reviewed Mar 26, 2024

View reviewed changes

mag1c1an1 force-pushed the sub branch 5 times, most recently from 2873000 to 1e49766 Compare March 29, 2024 09:55

mag1c1an1 added 6 commits April 2, 2024 10:30

before rollback fmt

f8abcf4

Signed-off-by: mag1c1an1 <[email protected]>

add some test

8561c42

Signed-off-by: mag1c1an1 <[email protected]>

fix ci clippy

b9e4a6e

Signed-off-by: mag1c1an1 <[email protected]>

clean code

ebc5a86

Signed-off-by: mag1c1an1 <[email protected]>

add rules for lakesoul io reader

3a92001

Signed-off-by: mag1c1an1 <[email protected]>

mag1c1an1 force-pushed the sub branch from 0c04036 to 3a92001 Compare April 2, 2024 02:30

Ceng23333 reviewed Apr 2, 2024

View reviewed changes

change field name and delete parquetfilter

17e184e

Signed-off-by: mag1c1an1 <[email protected]>

Ceng23333 approved these changes Apr 3, 2024

View reviewed changes

xuchen-plus merged commit 499de72 into lakesoul-io:main Apr 7, 2024
17 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Rust] add substrait for flink and be compatible for other engines #454

[Rust] add substrait for flink and be compatible for other engines #454

mag1c1an1 commented Mar 19, 2024

Ceng23333 Mar 21, 2024

Ceng23333 Mar 21, 2024

Ceng23333 Mar 21, 2024

mag1c1an1 Mar 21, 2024

Ceng23333 Mar 21, 2024

Ceng23333 Mar 21, 2024

Ceng23333 Mar 21, 2024

Ceng23333 Mar 21, 2024

mag1c1an1 commented Mar 21, 2024

Ceng23333 left a comment

Ceng23333 Mar 25, 2024

Ceng23333 Mar 25, 2024

Ceng23333 Mar 25, 2024

Ceng23333 Mar 25, 2024

Ceng23333 Apr 2, 2024

Ceng23333 Apr 2, 2024

Ceng23333 Apr 2, 2024

Ceng23333 Apr 2, 2024

Ceng23333 Apr 2, 2024

		@@ -129,7 +133,11 @@ private void initializeReader() throws IOException {
		}

		if (filter != null) {


		public class SubstraitTest extends AbstractTestBase {

		private final String BATCH_TYPE = "batch";

[Rust] add substrait for flink and be compatible for other engines #454

[Rust] add substrait for flink and be compatible for other engines #454

Conversation

mag1c1an1 commented Mar 19, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mag1c1an1 commented Mar 21, 2024

Ceng23333 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment