From 0c0b2cd64226e88692eed426ae20bafe1fdb755a Mon Sep 17 00:00:00 2001 From: wgcn <1026688210@qq.com> Date: Sat, 25 May 2024 21:39:29 +0800 Subject: [PATCH] support parse filter sql --- paimon-flink/paimon-flink-common/pom.xml | 6 + .../SimpleSqlPredicateConvertor.java | 158 ++++++++++++ .../SimpleSqlPredicateConvertorTest.java | 228 ++++++++++++++++++ 3 files changed, 392 insertions(+) create mode 100644 paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertor.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertorTest.java diff --git a/paimon-flink/paimon-flink-common/pom.xml b/paimon-flink/paimon-flink-common/pom.xml index 09ea2bb9c678d..22dd0b0e96c1e 100644 --- a/paimon-flink/paimon-flink-common/pom.xml +++ b/paimon-flink/paimon-flink-common/pom.xml @@ -77,6 +77,12 @@ under the License. test + + org.apache.flink + flink-table-planner_2.12 + ${flink.version} + provided + org.apache.flink flink-connector-test-utils diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertor.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertor.java new file mode 100644 index 0000000000000..1cc5912ac79c0 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertor.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.predicate; + +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.TypeUtils; + +import org.apache.calcite.sql.SqlBasicCall; +import org.apache.calcite.sql.SqlBinaryOperator; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlLiteral; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.SqlOperator; +import org.apache.calcite.sql.SqlPostfixOperator; +import org.apache.calcite.sql.SqlPrefixOperator; +import org.apache.calcite.sql.parser.SqlParseException; +import org.apache.calcite.sql.parser.SqlParser; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.BiFunction; + +import static org.apache.calcite.avatica.util.Casing.UNCHANGED; + +/** convert sql to predicate. */ +public class SimpleSqlPredicateConvertor { + + private final PredicateBuilder builder; + private final RowType rowType; + + public SimpleSqlPredicateConvertor(RowType type) { + this.rowType = type; + this.builder = new PredicateBuilder(type); + } + + public Predicate convertSqlToPredicate(String conditionSql) throws SqlParseException { + SqlParser parser = + SqlParser.create(conditionSql, SqlParser.config().withUnquotedCasing(UNCHANGED)); + SqlNode sqlNode = parser.parseExpression(); + return convert((SqlBasicCall) sqlNode); + } + + public Predicate convert(SqlBasicCall sqlBasicCall) { + SqlOperator operator = sqlBasicCall.getOperator(); + SqlKind kind = operator.getKind(); + if (operator instanceof SqlBinaryOperator) { + List operandList = sqlBasicCall.getOperandList(); + SqlNode left = operandList.get(0); + SqlNode right = operandList.get(1); + switch (kind) { + case OR: + return PredicateBuilder.or( + convert((SqlBasicCall) left), convert((SqlBasicCall) right)); + case AND: + return PredicateBuilder.and( + convert((SqlBasicCall) left), convert((SqlBasicCall) right)); + case EQUALS: + return visitBiFunction(left, right, builder::equal, builder::equal); + case NOT_EQUALS: + return visitBiFunction(left, right, builder::notEqual, builder::notEqual); + case LESS_THAN: + return visitBiFunction(left, right, builder::lessThan, builder::greaterThan); + case LESS_THAN_OR_EQUAL: + return visitBiFunction( + left, right, builder::lessOrEqual, builder::greaterOrEqual); + case GREATER_THAN: + return visitBiFunction(left, right, builder::greaterThan, builder::lessThan); + case GREATER_THAN_OR_EQUAL: + return visitBiFunction( + left, right, builder::greaterOrEqual, builder::lessOrEqual); + case IN: + { + int index = getfieldIndex(left.toString()); + SqlNodeList Elementslist = (SqlNodeList) right; + + List list = new ArrayList<>(); + for (SqlNode sqlNode : Elementslist) { + Object literal = + TypeUtils.castFromString( + ((SqlLiteral) sqlNode).toValue(), + rowType.getFieldTypes().get(index)); + list.add(literal); + } + return builder.in(index, list); + } + } + } else if (operator instanceof SqlPostfixOperator) { + SqlNode child = sqlBasicCall.getOperandList().get(0); + switch (kind) { + case IS_NULL: + { + String field = String.valueOf(child); + return builder.isNull(getfieldIndex(field)); + } + case IS_NOT_NULL: + String field = String.valueOf(child); + return builder.isNotNull(getfieldIndex(field)); + } + } else if (operator instanceof SqlPrefixOperator) { + if (kind == SqlKind.NOT) { + SqlBasicCall child = (SqlBasicCall) sqlBasicCall.getOperandList().get(0); + return convert(child).negate().get(); + } + } + + throw new UnsupportedOperationException(String.format("%s not been supported.", kind)); + } + + public Predicate visitBiFunction( + SqlNode left, + SqlNode right, + BiFunction visitLeft, + BiFunction visitRight) { + if (left instanceof SqlIdentifier && right instanceof SqlLiteral) { + int index = getfieldIndex(left.toString()); + String value = ((SqlLiteral) right).toValue(); + DataType type = rowType.getFieldTypes().get(index); + return visitLeft.apply(index, TypeUtils.castFromString(value, type)); + } else if (right instanceof SqlIdentifier && left instanceof SqlLiteral) { + int index = getfieldIndex(right.toString()); + return visitRight.apply( + index, + TypeUtils.castFromString( + ((SqlLiteral) left).toValue(), rowType.getFieldTypes().get(index))); + } + + throw new UnsupportedOperationException(String.format("%s or %s not been supported.", left, right)); + } + + public int getfieldIndex(String field) { + int index = builder.indexOf(field); + if (index == -1) { + throw new RuntimeException(String.format("Field `%s` not found", field)); + } + return index; + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertorTest.java new file mode 100644 index 0000000000000..515d25c23bc3d --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/predicate/SimpleSqlPredicateConvertorTest.java @@ -0,0 +1,228 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.predicate; + +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.apache.paimon.shade.guava30.com.google.common.collect.Lists; + +import org.apache.calcite.sql.parser.SqlParseException; +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.List; + +/** test for {@link SimpleSqlPredicateConvertor} */ +class SimpleSqlPredicateConvertorTest { + RowType rowType; + PredicateBuilder predicateBuilder; + + SimpleSqlPredicateConvertor simpleSqlPredicateConvertor; + + @BeforeEach + public void init() { + rowType = + RowType.builder() + .field("a", DataTypes.INT()) + .field("b", DataTypes.STRING()) + .field("c", DataTypes.DATE()) + .build(); + predicateBuilder = new PredicateBuilder(rowType); + simpleSqlPredicateConvertor = new SimpleSqlPredicateConvertor(rowType); + } + + @Test + public void testSql() throws SqlParseException { + String conditionSql = + "b is null and c is not null and 100<> a and not (a > 100) or a >1 or a<2 or a=1 or a in ('1','2')"; + + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate(conditionSql); + System.out.println(predicate); + } + + @Test + public void testEqual() throws SqlParseException { + { + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate("a ='1'"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.equal(predicateBuilder.indexOf("a"), 1)); + } + + { + Predicate predicate = + simpleSqlPredicateConvertor.convertSqlToPredicate(" '2024-05-25' = c"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.equal(predicateBuilder.indexOf("c"), 19868)); + } + } + + @Test + public void testNotEqual() throws SqlParseException { + { + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate("a <>'1'"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.notEqual(predicateBuilder.indexOf("a"), 1)); + } + + { + Predicate predicate = + simpleSqlPredicateConvertor.convertSqlToPredicate("'2024-05-25' <> c"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.notEqual(predicateBuilder.indexOf("c"), 19868)); + } + } + + @Test + public void testLessThan() throws SqlParseException { + { + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate("a <'1'"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.lessThan(predicateBuilder.indexOf("a"), 1)); + } + + { + Predicate predicate = + simpleSqlPredicateConvertor.convertSqlToPredicate("'2024-05-25' '1'"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.greaterThan(predicateBuilder.indexOf("a"), 1)); + } + + { + Predicate predicate = + simpleSqlPredicateConvertor.convertSqlToPredicate("'2024-05-25' > c"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.lessThan(predicateBuilder.indexOf("c"), 19868)); + } + } + + @Test + public void testGreatEqual() throws SqlParseException { + { + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate("a >='1'"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.greaterOrEqual(predicateBuilder.indexOf("a"), 1)); + } + + { + Predicate predicate = + simpleSqlPredicateConvertor.convertSqlToPredicate(" '2024-05-25' >= c"); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.lessOrEqual(predicateBuilder.indexOf("c"), 19868)); + } + } + + @Test + public void testIN() throws SqlParseException { + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate("a in ('1','2')"); + List elements = Lists.newArrayList(1, 2); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.in(predicateBuilder.indexOf("a"), elements)); + } + + @Test + public void testIsNull() throws SqlParseException { + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate("a is null "); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.isNull(predicateBuilder.indexOf("a"))); + } + + @Test + public void testIsNotNull() throws SqlParseException { + Predicate predicate = simpleSqlPredicateConvertor.convertSqlToPredicate("a is not null "); + Assertions.assertThat(predicate) + .isEqualTo(predicateBuilder.isNotNull(predicateBuilder.indexOf("a"))); + } + + @Test + public void testAnd() throws SqlParseException { + Predicate actual = + simpleSqlPredicateConvertor.convertSqlToPredicate("a is not null and c is null"); + Predicate expected = + PredicateBuilder.and( + predicateBuilder.isNotNull(predicateBuilder.indexOf("a")), + predicateBuilder.isNull(predicateBuilder.indexOf("c"))); + Assertions.assertThat(actual).isEqualTo(expected); + } + + @Test + public void testOr() throws SqlParseException { + Predicate actual = + simpleSqlPredicateConvertor.convertSqlToPredicate("a is not null or c is null "); + Predicate expected = + PredicateBuilder.or( + predicateBuilder.isNotNull(predicateBuilder.indexOf("a")), + predicateBuilder.isNull(predicateBuilder.indexOf("c"))); + Assertions.assertThat(actual).isEqualTo(expected); + } + + @Test + public void testNOT() throws SqlParseException { + Predicate actual = simpleSqlPredicateConvertor.convertSqlToPredicate("not (a is null) "); + Predicate expected = predicateBuilder.isNull(predicateBuilder.indexOf("a")).negate().get(); + Assertions.assertThat(actual).isEqualTo(expected); + } + + @Test + public void testFieldNoFound() throws SqlParseException { + Assertions.assertThatThrownBy( + () -> simpleSqlPredicateConvertor.convertSqlToPredicate("f =1")) + .hasMessage("Field `f` not found"); + } + + @Test + public void testSqlNoSupport(){ + // function not supported + Assertions.assertThatThrownBy(()-> simpleSqlPredicateConvertor.convertSqlToPredicate("substring(f,0,1) =1")) + .hasMessage("SUBSTRING(`f` FROM 0 FOR 1) or 1 not been supported."); + // like not supported + Assertions.assertThatThrownBy(()-> simpleSqlPredicateConvertor.convertSqlToPredicate("b like 'x'")) + .hasMessage("LIKE not been supported."); + + } +}