diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/LikeFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/LikeFunctionITCase.java index f2c29eae578f8..453dfba76b3ea 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/LikeFunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/LikeFunctionITCase.java @@ -148,8 +148,22 @@ private Stream withEscape() { DataTypes.STRING(), DataTypes.STRING()) // Empty strings in pattern or escape - .testSqlResult("f0 LIKE 'test\"end' ESCAPE ''", false, DataTypes.BOOLEAN()) .testSqlResult("f0 LIKE '' ESCAPE ''", false, DataTypes.BOOLEAN()) + .testSqlResult("f0 LIKE '' ESCAPE '!'", false, DataTypes.BOOLEAN()) + .testSqlResult("f0 LIKE 'test\"end' ESCAPE ''", false, DataTypes.BOOLEAN()) + // Escaped _ in quick path: startsWith (BEGIN_PATTERN) + .testSqlResult("f2 LIKE 'te!_%' ESCAPE '!'", true, DataTypes.BOOLEAN()) + .testSqlResult("f0 LIKE 'te!_%' ESCAPE '!'", false, DataTypes.BOOLEAN()) + + // Escaped _ in quick path: endsWith (END_PATTERN) + .testSqlResult("f2 LIKE '%!_st' ESCAPE '!'", true, DataTypes.BOOLEAN()) + + // Escaped _ in quick path: contains (MIDDLE_PATTERN) + .testSqlResult("f2 LIKE '%!_s%' ESCAPE '!'", true, DataTypes.BOOLEAN()) + + // Escaped _ in quick path: multi-segment (ChainChecker) + .testSqlResult("f2 LIKE 'te!_%st' ESCAPE '!'", true, DataTypes.BOOLEAN()) + .testSqlResult("f0 LIKE 'te!_%st' ESCAPE '!'", false, DataTypes.BOOLEAN()) // Escaping with emoji .testSqlResult("f0 LIKE 'test' ESCAPE '✅'", true, DataTypes.BOOLEAN()) .testSqlResult("f1 LIKE 'test✅%' ESCAPE '✅'", true, DataTypes.BOOLEAN()) diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlLikeChainChecker.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlLikeChainChecker.java index 06068031bfc1a..2be97b48a0d54 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlLikeChainChecker.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlLikeChainChecker.java @@ -45,13 +45,13 @@ public class SqlLikeChainChecker { private final int[] midLens; private final int beginLen; private final int endLen; - private final boolean leftAnchor; - private final boolean rightAnchor; + private final boolean isEmptyPattern; public SqlLikeChainChecker(String pattern) { final StringTokenizer tokens = new StringTokenizer(pattern, "%"); - leftAnchor = !pattern.startsWith("%"); - rightAnchor = !pattern.endsWith("%"); + boolean leftAnchor = !pattern.startsWith("%"); + boolean rightAnchor = !pattern.endsWith("%"); + isEmptyPattern = pattern.isEmpty(); int len = 0; // at least 2 checkers always BinaryStringData leftPattern = null; @@ -63,7 +63,7 @@ public SqlLikeChainChecker(String pattern) { for (int i = 0; tokens.hasMoreTokens(); i++) { String chunk = tokens.nextToken(); - if (chunk.length() == 0) { + if (chunk.isEmpty()) { // %% is folded in the .*?.*? regex usually into .*? continue; } @@ -97,13 +97,8 @@ public boolean check(BinaryStringData str) { int mark = str.getSizeInBytes(); // Returns false early if either: // the input is too short to match the pattern, or - // the pattern is empty (or anchored with no literals) but the input is not empty. - if (mark < minLen - || beginPattern == null - && endPattern == null - && middlePatterns.length == 0 - && mark > 0 - && (leftAnchor || rightAnchor)) { + // the pattern is empty but the input is not. + if (mark < minLen || mark > 0 && isEmptyPattern) { return false; } // prefix, extend start