Repository: pingcap/tispark (language: python)

Issue:
[BUG] Read from range column partition table error
**Describe the bug**
<!-- A clear and concise description of what the bug is. -->

Read from range column partition table error.

**What did you do**
1. DDL
```
create table range_column_test (id varchar(10)) partition by RANGE COLUMNS(`id`) (PARTITION `p1` VALUES LESS THAN ('CN001'),PARTITION `p2` VALUES LESS THAN ('CN002'))
```

2. use tispark to read

```
select * from range_column_test where id = 'CN001'
```

3. error occurs
<img width="1002" alt="image" src="https://user-images.githubusercontent.com/52435083/223045869-2a9a054b-3332-49cc-971a-d4dd4a93d8ac.png">


**Spark and TiSpark version info**
TiSpark : <= v3.2.0


Discussion / hints:

Fix patch:
diff --git a/core/src/main/scala/org/apache/spark/sql/execution/CoprocessorRDD.scala b/core/src/main/scala/org/apache/spark/sql/execution/CoprocessorRDD.scala
index c815fcd75c..57fcc84e40 100644
--- a/core/src/main/scala/org/apache/spark/sql/execution/CoprocessorRDD.scala
+++ b/core/src/main/scala/org/apache/spark/sql/execution/CoprocessorRDD.scala
@@ -59,9 +59,12 @@ trait LeafColumnarExecRDD extends LeafExecNode {
       }
       b.append("]")
       b.toString
-    } else {
+    } else if (tiRDDs.lengthCompare(1) == 0) {
       s"${dagRequest.getStoreType.name()} $nodeName{$dagRequest}" +
         s"${TiUtil.getReqEstCountStr(dagRequest)}"
+    } else {
+      // return empty TiRDD when there is no tiRDDs
+      "Empty TiRDD"
     }
 
   def dagRequest: TiDAGRequest = tiRDDs.head.dagRequest
diff --git a/docs/features/partition_table.md b/docs/features/partition_table.md
index 68b3532b2e..6bd7406671 100644
--- a/docs/features/partition_table.md
+++ b/docs/features/partition_table.md
@@ -16,7 +16,10 @@ TiSpark only supports partition pruning with the following partition expression
 + `YEAR(col)` and its type is datetime/string/date literal that can be parsed as datetime.
 + `TO_DAYS(col)` and its type is datetime/string/date literal that can be parsed as datetime.
 
-TiSpark does not support partition pruning in hash and list partition.
+### Limitations
+
+- TiSpark does not support partition pruning in hash and list partition.
+- TiSpark can not apply partition pruning for some special characters in partition definition. For example, Partition definition with `""` can not be pruned. such as `partition p0 values less than ('"string"')`.
 
 ## Write into partition table
 
diff --git a/tikv-client/src/main/java/com/pingcap/tikv/expression/PartitionPruner.java b/tikv-client/src/main/java/com/pingcap/tikv/expression/PartitionPruner.java
index 088d9dbde0..8e3c747a1a 100644
--- a/tikv-client/src/main/java/com/pingcap/tikv/expression/PartitionPruner.java
+++ b/tikv-client/src/main/java/com/pingcap/tikv/expression/PartitionPruner.java
@@ -70,6 +70,13 @@ public static List<TiPartitionDef> prune(TiTableInfo tableInfo, List<Expression>
             return tableInfo.getPartitionInfo().getDefs();
           }
 
+          // prune can not handle \" now.
+          for (int i = 0; i < tableInfo.getPartitionInfo().getDefs().size(); i++) {
+            TiPartitionDef pDef = tableInfo.getPartitionInfo().getDefs().get(i);
+            if (pDef.getLessThan().get(0).contains("\"")) {
+              return tableInfo.getPartitionInfo().getDefs();
+            }
+          }
           RangeColumnPartitionPruner pruner = new RangeColumnPartitionPruner(tableInfo);
           return pruner.prune(filters);
         }
@@ -104,7 +111,7 @@ public static void generateRangeExprs(
     // partExprColRefs.addAll(PredicateUtils.extractColumnRefFromExpression(partExpr));
     for (int i = 0; i < partInfo.getDefs().size(); i++) {
       TiPartitionDef pDef = partInfo.getDefs().get(i);
-      String current = pDef.getLessThan().get(lessThanIdx);
+      String current = wrapValue(pDef.getLessThan().get(lessThanIdx));
       String leftHand;
       if (current.equals("MAXVALUE")) {
         leftHand = "true";
@@ -114,7 +121,7 @@ public static void generateRangeExprs(
       if (i == 0) {
         partExprs.add(parser.parseExpression(leftHand));
       } else {
-        String previous = partInfo.getDefs().get(i - 1).getLessThan().get(lessThanIdx);
+        String previous = wrapValue(partInfo.getDefs().get(i - 1).getLessThan().get(lessThanIdx));
         String and =
             String.format("%s >= %s and %s", wrapColumnName(partExprStr), previous, leftHand);
         partExprs.add(parser.parseExpression(and));
@@ -132,4 +139,41 @@ private static String wrapColumnName(String columnName) {
       return String.format("`%s`", columnName);
     }
   }
+
+  /**
+   * Spark SQL will parse string literal without escape, So we need to parse partition definition
+   * without escape too.
+   *
+   * <p>wrapValue will replace the first '' to "", so that antlr will not regard the first '' as a
+   * part of string literal.
+   *
+   * <p>wrapValue will also delete the escape character in string literal.
+   *
+   * <p>e.g. 'string' -> "string" '''string''' -> "'string'" 'string''' -> "string'"
+   *
+   * <p>Can't handle '""'. e.g. '"string"' -> ""string"". parseExpression will parse ""string"" to
+   * empty string, parse '"string"' to 'string'
+   *
+   * @param value
+   * @return
+   */
+  private static String wrapValue(String value) {
+    if (value.startsWith("'") && value.endsWith("'")) {
+      String newValue = String.format("\"%s\"", value.substring(1, value.length() - 1));
+      StringBuilder valueWithoutEscape = new StringBuilder();
+      for (int i = 0; i < newValue.length(); i++) {
+        if (newValue.charAt(i) != '\'') {
+          valueWithoutEscape.append(newValue.charAt(i));
+        } else {
+          if (i + 1 < newValue.length()) {
+            valueWithoutEscape.append(newValue.charAt(i + 1));
+          }
+          i++;
+        }
+      }
+      return valueWithoutEscape.toString();
+    } else {
+      return value;
+    }
+  }
 }


Test patch:
diff --git a/core/src/test/scala/org/apache/spark/sql/PartitionTableSuite.scala b/core/src/test/scala/org/apache/spark/sql/PartitionTableSuite.scala
index eb9192b845..5723e2e231 100644
--- a/core/src/test/scala/org/apache/spark/sql/PartitionTableSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/PartitionTableSuite.scala
@@ -33,6 +33,18 @@ class PartitionTableSuite extends BasePlanTest {
       super.afterAll()
     }
 
+  test("reading from range column partition") {
+    tidbStmt.execute("drop table if exists range_column_test")
+    tidbStmt.execute(
+      "create table range_column_test (id varchar(10)) partition by RANGE COLUMNS(`id`) (PARTITION `p1` VALUES LESS THAN ('''CN001'''),PARTITION `p2` VALUES LESS THAN ('CN002'))")
+    tidbStmt.execute("insert into `range_column_test` values('CN001')")
+    tidbStmt.execute("insert into `range_column_test` values('''CN001''')")
+
+    judge("select * from range_column_test where id = 'CN001'")
+    judge("select * from range_column_test where id = '\\'CN001\\''")
+    judge("select * from range_column_test where id = 'CN002'")
+  }
+
   test("reading from hash partition") {
     enablePartitionForTiDB()
     tidbStmt.execute("drop table if exists t")