[flink] branch master updated: [FLINK-30586][table-runtime] Fix calcCodeGen failed if calc with 'like' condition contains double quotation mark

[godfrey]

This is an automated email from the ASF dual-hosted git repository.

godfrey 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 68b37fb8673 [FLINK-30586][table-runtime] Fix calcCodeGen failed if 
calc with 'like' condition contains double quotation mark
68b37fb8673 is described below

commit 68b37fb867374df5a201f0b170e35c21266e5d7b
Author: zhengyunhong.zyh <337361...@qq.com>
AuthorDate: Fri Jan 6 16:25:32 2023 +0800

    [FLINK-30586][table-runtime] Fix calcCodeGen failed if calc with 'like' 
condition contains double quotation mark
    
    This closes #21609
---
 .../table/planner/codegen/GenerateUtils.scala      |  7 +++--
 .../planner/runtime/batch/sql/CalcITCase.scala     | 29 ++++++++++++++++++--
 .../planner/runtime/stream/sql/CalcITCase.scala    | 32 +++++++++++++++++++++-
 3 files changed, 61 insertions(+), 7 deletions(-)

diff --git 
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
 
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
index 9708ca70e66..92a45ab14b6 100644
--- 
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
+++ 
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
@@ -304,9 +304,10 @@ object GenerateUtils {
       // as they're not cheap to construct. For the other types, the return 
term is directly
       // the literal value
       case CHAR | VARCHAR =>
-        val str = literalValue.asInstanceOf[BinaryStringData]
-        val field = 
ctx.addReusableEscapedStringConstant(EncodingUtils.escapeJava(str.toString))
-        generateNonNullLiteral(literalType, field, str)
+        val escapedValue =
+          
EncodingUtils.escapeJava(literalValue.asInstanceOf[BinaryStringData].toString)
+        val field = ctx.addReusableEscapedStringConstant(escapedValue)
+        generateNonNullLiteral(literalType, field, 
StringData.fromString(escapedValue))
 
       case BINARY | VARBINARY =>
         val bytesVal = literalValue.asInstanceOf[Array[Byte]]
diff --git 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
index b95df8c1a89..ef4ebd507f2 100644
--- 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
+++ 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
@@ -50,11 +50,9 @@ import org.junit.rules.ExpectedException
 
 import java.nio.charset.StandardCharsets
 import java.sql.{Date, Time, Timestamp}
-import java.time.{Instant, LocalDate, LocalDateTime, LocalTime, ZoneId}
+import java.time._
 import java.util
 
-import scala.collection.Seq
-
 class CalcITCase extends BatchTestBase {
 
   var _expectedEx: ExpectedException = ExpectedException.none
@@ -2115,4 +2113,29 @@ class CalcITCase extends BatchTestBase {
     tEnv.useDatabase("db1")
     checkResult("SELECT CURRENT_DATABASE()", Seq(row(tEnv.getCurrentDatabase)))
   }
+
+  @Test
+  def testLikeWithConditionContainsDoubleQuotationMark(): Unit = {
+    val rows = Seq(row(42, "abc"), row(2, "cbc\"ddd"))
+    val dataId = TestValuesTableFactory.registerData(rows)
+
+    val ddl =
+      s"""
+         |CREATE TABLE MyTable (
+         |  a int,
+         |  b string
+         |) WITH (
+         |  'connector' = 'values',
+         |  'data-id' = '$dataId',
+         |  'bounded' = 'true'
+         |)
+         |""".stripMargin
+    tEnv.executeSql(ddl)
+
+    checkResult(
+      """
+        | SELECT * FROM MyTable WHERE b LIKE '%"%'
+        |""".stripMargin,
+      Seq(row(2, "cbc\"ddd")))
+  }
 }
diff --git 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
index 1381851eb40..886a862d867 100644
--- 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
+++ 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala
@@ -46,7 +46,6 @@ import java.time.Instant
 import java.util
 
 import scala.collection.JavaConversions._
-import scala.collection.Seq
 
 class CalcITCase extends StreamingTestBase {
 
@@ -714,4 +713,35 @@ class CalcITCase extends StreamingTestBase {
     val result2 = tEnv.sqlQuery("SELECT 
CURRENT_DATABASE()").execute().collect().toList
     assertEquals(Seq(row(tEnv.getCurrentDatabase)), result2)
   }
+
+  @Test
+  def testLikeWithConditionContainsDoubleQuotationMark(): Unit = {
+    val rows = Seq(row(42, "abc"), row(2, "cbc\"ddd"))
+    val dataId = TestValuesTableFactory.registerData(rows)
+
+    val ddl =
+      s"""
+         |CREATE TABLE MyTable (
+         |  a int,
+         |  b string
+         |) WITH (
+         |  'connector' = 'values',
+         |  'data-id' = '$dataId',
+         |  'bounded' = 'true'
+         |)
+       """.stripMargin
+    tEnv.executeSql(ddl)
+
+    val result = tEnv
+      .sqlQuery("""
+                  | SELECT * FROM MyTable WHERE b LIKE '%"%'
+                  |""".stripMargin)
+      .toAppendStream[Row]
+    val sink = new TestingAppendSink
+    result.addSink(sink)
+    env.execute()
+
+    val expected = List("2,cbc\"ddd")
+    assertEquals(expected.sorted, sink.getAppendResults.sorted)
+  }
 }