This is an automated email from the ASF dual-hosted git repository. libenchao pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 4d37b8c34ff [FLINK-33313][table] Fix RexNodeExtractor#extractConjunctiveConditions throws an Exception when handle binary literal 4d37b8c34ff is described below commit 4d37b8c34ff062b7505ab8c0ca8f2181768aab60 Author: zoudan <zou...@bytedance.com> AuthorDate: Thu Oct 19 17:48:05 2023 +0800 [FLINK-33313][table] Fix RexNodeExtractor#extractConjunctiveConditions throws an Exception when handle binary literal Close apache/flink#23551 --- .../table/planner/plan/utils/RexNodeExtractor.scala | 4 ++++ .../plan/utils/NestedProjectionUtilTest.scala | 10 ++++++---- .../planner/plan/utils/RexNodeExtractorTest.scala | 21 +++++++++++++++++++++ .../planner/plan/utils/RexNodeRewriterTest.scala | 7 ++++--- .../table/planner/plan/utils/RexNodeTestBase.scala | 6 ++++-- 5 files changed, 39 insertions(+), 9 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala index 482ce56dc63..481cbda8b82 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala @@ -37,6 +37,7 @@ import org.apache.flink.table.types.logical.YearMonthIntervalType import org.apache.flink.table.types.utils.TypeConversions import org.apache.flink.util.Preconditions +import org.apache.calcite.avatica.util.ByteString import org.apache.calcite.plan.RelOptUtil import org.apache.calcite.rel.`type`.RelDataType import org.apache.calcite.rex._ @@ -502,6 +503,9 @@ class RexNodeToExpressionConverter( // convert to BigDecimal literal.getValueAs(classOf[java.math.BigDecimal]) + case BINARY | VARBINARY => + literal.getValueAs(classOf[Array[Byte]]) + case _ => literal.getValue } diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/NestedProjectionUtilTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/NestedProjectionUtilTest.scala index ec8214f5b91..9cd44c9fea7 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/NestedProjectionUtilTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/NestedProjectionUtilTest.scala @@ -87,12 +87,13 @@ class NestedProjectionUtilTest extends RexNodeTestBase { "$2", "$3", "$4", + "$5", "*($t2, $t3)", "100", - "<($t5, $t6)", + "<($t6, $t7)", "6", - ">($t1, $t8)", - "AND($t7, $t9)"))) + ">($t1, $t9)", + "AND($t8, $t10)"))) val nestedField = NestedProjectionUtil.build(exprs, rexProgram.getInputRowType) val paths = NestedProjectionUtil.convertToIndexArray(nestedField) @@ -101,7 +102,8 @@ class NestedProjectionUtilTest extends RexNodeTestBase { Array(1), Array(2), Array(3), - Array(4) + Array(4), + Array(5) ) assertArray(paths, orderedPaths) val builder = new FlinkRexBuilder(typeFactory) diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala index 2eb87e35cc8..bd5f15d3bed 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala @@ -33,6 +33,7 @@ import org.apache.flink.table.planner.utils.{DateTimeTestUtil, IntSumAggFunction import org.apache.flink.table.resource.ResourceManager import org.apache.flink.table.utils.CatalogManagerMocks +import org.apache.calcite.avatica.util.ByteString import org.apache.calcite.rel.`type`.RelDataType import org.apache.calcite.rex.{RexBuilder, RexNode} import org.apache.calcite.sql.`type`.SqlTypeName @@ -145,6 +146,26 @@ class RexNodeExtractorTest extends RexNodeTestBase { assertEquals(0, unconvertedRexNodes.length) } + @Test + def testExtractConditionWithBinaryLiteral(): Unit = { + // blob + val t0 = rexBuilder.makeInputRef(allFieldTypes.get(5), 5) + + // X'616263' + val t1 = rexBuilder.makeBinaryLiteral(ByteString.of("616263", 16)) + + // blob = X'616263' + val a = rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, t0, t1) + + val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory) + val (convertedExpressions, unconvertedRexNodes) = + extractConjunctiveConditions(a, -1, allFieldNames, relBuilder, functionCatalog) + + val expected: Array[Expression] = Array($"blob" === Array[Byte](97, 98, 99)) + assertExpressionArrayEquals(expected, convertedExpressions) + assertEquals(0, unconvertedRexNodes.length) + } + // ((a AND b) OR c) AND (NOT d) => (a OR c) AND (b OR c) AND (NOT d) @Test def testExtractCnfCondition(): Unit = { diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeRewriterTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeRewriterTest.scala index 0cea5c8be69..57a0edda39e 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeRewriterTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeRewriterTest.scala @@ -39,12 +39,13 @@ class RexNodeRewriterTest extends RexNodeTestBase { "$2", "$3", "$4", + "$5", "*($t2, $t3)", "100", - "<($t5, $t6)", + "<($t6, $t7)", "6", - ">($t1, $t8)", - "AND($t7, $t9)"))) + ">($t1, $t9)", + "AND($t8, $t10)"))) // use amount, id, price fields to create a new RexProgram val usedFields = Array(2, 3, 1) diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeTestBase.scala index ec326c2a540..1ba968b3253 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeTestBase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeTestBase.scala @@ -39,14 +39,16 @@ abstract class RexNodeTestBase { val typeFactory: FlinkTypeFactory = new FlinkTypeFactory( Thread.currentThread().getContextClassLoader) - val allFieldNames: java.util.List[String] = List("name", "id", "amount", "price", "flag").asJava + val allFieldNames: java.util.List[String] = + List("name", "id", "amount", "price", "flag", "blob").asJava val allFieldTypes: java.util.List[RelDataType] = List( DataTypes.VARCHAR(100), DataTypes.BIGINT(), DataTypes.INT(), DataTypes.DOUBLE(), - DataTypes.BOOLEAN()) + DataTypes.BOOLEAN(), + DataTypes.BYTES()) .map(LogicalTypeDataTypeConverter.fromDataTypeToLogicalType) .map(typeFactory.createFieldTypeFromLogicalType) .asJava