This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f26a23fdba [spark] Fix partition predicate conversion in procedures
(#8390)
f26a23fdba is described below
commit f26a23fdbacf76fdab47edffdbadde94fb49f286
Author: huangxiaoping <[email protected]>
AuthorDate: Tue Jun 30 17:20:15 2026 +0800
[spark] Fix partition predicate conversion in procedures (#8390)
---
.../paimon/codegen/CodeGeneratorContext.scala | 11 +-
.../procedure/CompactChainTableProcedure.java | 19 +--
.../paimon/spark/procedure/CompactProcedure.java | 20 ++-
.../procedure/CreateGlobalIndexProcedure.java | 12 +-
.../spark/procedure/DropGlobalIndexProcedure.java | 11 +-
.../paimon/spark/procedure/RescaleProcedure.java | 21 +--
.../paimon/spark/utils/SparkProcedureUtils.java | 143 ++++++++++++++++-----
.../spark/procedure/CompactProcedureTestBase.scala | 41 +++++-
.../spark/procedure/RescaleProcedureTest.scala | 31 +++++
9 files changed, 230 insertions(+), 79 deletions(-)
diff --git
a/paimon-codegen/src/main/scala/org/apache/paimon/codegen/CodeGeneratorContext.scala
b/paimon-codegen/src/main/scala/org/apache/paimon/codegen/CodeGeneratorContext.scala
index 0c33c8a6e6..5f137ea25e 100644
---
a/paimon-codegen/src/main/scala/org/apache/paimon/codegen/CodeGeneratorContext.scala
+++
b/paimon-codegen/src/main/scala/org/apache/paimon/codegen/CodeGeneratorContext.scala
@@ -57,8 +57,9 @@ class CodeGeneratorContext {
private var currentMethodNameForLocalVariables = "DEFAULT"
// method_name -> local_variable_statements
- private val reusableLocalVariableStatements = mutable.Map[String,
mutable.LinkedHashSet[String]](
- (currentMethodNameForLocalVariables, mutable.LinkedHashSet[String]()))
+ private val reusableLocalVariableStatements = mutable.Map[String,
mutable.LinkedHashSet[String]]()
+ reusableLocalVariableStatements(currentMethodNameForLocalVariables) =
+ new mutable.LinkedHashSet[String]()
/**
* Starts a new local variable statements for a generated class with the
given method name.
@@ -68,7 +69,7 @@ class CodeGeneratorContext {
*/
def startNewLocalVariableStatement(methodName: String): Unit = {
currentMethodNameForLocalVariables = methodName
- reusableLocalVariableStatements(methodName) =
mutable.LinkedHashSet[String]()
+ reusableLocalVariableStatements(methodName) = new
mutable.LinkedHashSet[String]()
}
/**
@@ -86,7 +87,7 @@ class CodeGeneratorContext {
def addReusableLocalVariable(fieldTypeTerm: String, fieldName: String):
String = {
val fieldTerm = newName(fieldName)
reusableLocalVariableStatements
- .getOrElse(currentMethodNameForLocalVariables,
mutable.LinkedHashSet[String]())
+ .getOrElse(currentMethodNameForLocalVariables, new
mutable.LinkedHashSet[String]())
.add(s"$fieldTypeTerm $fieldTerm;")
fieldTerm
}
@@ -105,7 +106,7 @@ class CodeGeneratorContext {
fieldTypeAndNames.map(_._1).zip(fieldTerms).foreach {
case (fieldTypeTerm, fieldTerm) =>
reusableLocalVariableStatements
- .getOrElse(currentMethodNameForLocalVariables,
mutable.LinkedHashSet[String]())
+ .getOrElse(currentMethodNameForLocalVariables, new
mutable.LinkedHashSet[String]())
.add(s"$fieldTypeTerm $fieldTerm;")
}
fieldTerms
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
index ad992cf505..6594dabd38 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
@@ -132,21 +132,20 @@ public class CompactChainTableProcedure extends
BaseProcedure {
DataSourceV2Relation relation,
String partitionStr,
boolean overwrite) {
- String partition = SparkProcedureUtils.toWhere(partitionStr);
FileStoreTable snapshotTable = table.wrapped();
ChainGroupReadTable.ChainTableBatchScan scan =
(ChainGroupReadTable.ChainTableBatchScan) table.newScan();
PartitionPredicate partitionPredicate =
- SparkProcedureUtils.convertToPartitionPredicate(
- partition, table.schema().logicalPartitionType(),
spark(), relation);
+ SparkProcedureUtils.convertPartitionsToPartitionPredicate(
+ partitionStr, snapshotTable, spark());
// Check if target partition already exists in snapshot branch
- boolean partitionExists = checkPartitionExists(snapshotTable,
partition, relation);
+ boolean partitionExists = checkPartitionExists(snapshotTable,
partitionStr);
if (partitionExists) {
if (overwrite) {
scan.skipPreloadTargetSnapshot().withPartitionFilter(partitionPredicate);
- LOG.info("Found existing partition {}, will overwrite it.",
partition);
+ LOG.info("Found existing partition {}, will overwrite it.",
partitionStr);
} else {
LOG.info(
"Partition {} already exists in snapshot branch,
skipping compaction.",
@@ -187,14 +186,10 @@ public class CompactChainTableProcedure extends
BaseProcedure {
return true;
}
- private boolean checkPartitionExists(
- FileStoreTable snapshotTable, String partition,
DataSourceV2Relation relation) {
+ private boolean checkPartitionExists(FileStoreTable snapshotTable, String
partition) {
PartitionPredicate snapshotPartitionPredicate =
- SparkProcedureUtils.convertToPartitionPredicate(
- partition,
- snapshotTable.schema().logicalPartitionType(),
- spark(),
- relation);
+ SparkProcedureUtils.convertPartitionsToPartitionPredicate(
+ partition, snapshotTable, spark());
return !snapshotTable
.newScan()
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index b07fbd26be..6854633b74 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -183,7 +183,6 @@ public class CompactProcedure extends BaseProcedure {
checkArgument(
partitions == null || where == null,
"partitions and where cannot be used together.");
- String finalWhere = partitions != null ?
SparkProcedureUtils.toWhere(partitions) : where;
return modifySparkTable(
tableIdent,
sparkTable -> {
@@ -197,12 +196,19 @@ public class CompactProcedure extends BaseProcedure {
sortColumns,
table.partitionKeys());
DataSourceV2Relation relation = createRelation(tableIdent,
sparkTable);
- PartitionPredicate partitionPredicate =
- SparkProcedureUtils.convertToPartitionPredicate(
- finalWhere,
- table.schema().logicalPartitionType(),
- spark(),
- relation);
+ PartitionPredicate partitionPredicate;
+ if (partitions != null) {
+ partitionPredicate =
+
SparkProcedureUtils.convertPartitionsToPartitionPredicate(
+ partitions, table, spark());
+ } else {
+ partitionPredicate =
+
SparkProcedureUtils.convertToPartitionPredicate(
+ where,
+ table.schema().logicalPartitionType(),
+ spark(),
+ relation);
+ }
HashMap<String, String> dynamicOptions = new HashMap<>();
ProcedureUtils.putIfNotEmpty(
dynamicOptions, CoreOptions.WRITE_ONLY.key(),
"false");
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
index adceb783af..cff2fece60 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
@@ -116,9 +116,8 @@ public class CreateGlobalIndexProcedure extends
BaseProcedure {
: args.getString(3);
String optionString = args.isNullAt(4) ? null : args.getString(4);
- String finalWhere = partitions != null ?
SparkProcedureUtils.toWhere(partitions) : null;
-
- LOG.info("Starting to build index for table " + tableIdent + " WHERE:
" + finalWhere);
+ LOG.info(
+ "Starting to build index for table {} with partitions: {}",
tableIdent, partitions);
return modifySparkTable(
tableIdent,
@@ -160,11 +159,8 @@ public class CreateGlobalIndexProcedure extends
BaseProcedure {
}
DataSourceV2Relation relation =
createRelation(tableIdent, sparkTable);
PartitionPredicate partitionPredicate =
-
SparkProcedureUtils.convertToPartitionPredicate(
- finalWhere,
- table.schema().logicalPartitionType(),
- spark(),
- relation);
+
SparkProcedureUtils.convertPartitionsToPartitionPredicate(
+ partitions, table, spark());
List<DataField> indexFields =
indexColumns.stream()
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
index ec8b1a2c85..3aae599186 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
@@ -108,9 +108,7 @@ public class DropGlobalIndexProcedure extends BaseProcedure
{
: args.getString(3);
boolean dryRun = !args.isNullAt(4) && args.getBoolean(4);
- String finalWhere = partitions != null ?
SparkProcedureUtils.toWhere(partitions) : null;
-
- LOG.info("Starting to drop index for table " + tableIdent + " WHERE: "
+ finalWhere);
+ LOG.info("Starting to drop index for table {} with partitions: {}",
tableIdent, partitions);
List<String> indexColumns =
Arrays.stream(column.split(","))
@@ -142,11 +140,8 @@ public class DropGlobalIndexProcedure extends
BaseProcedure {
.collect(Collectors.toList());
DataSourceV2Relation relation =
createRelation(tableIdent);
PartitionPredicate partitionPredicate =
-
SparkProcedureUtils.convertToPartitionPredicate(
- finalWhere,
- table.schema().logicalPartitionType(),
- spark(),
- relation);
+
SparkProcedureUtils.convertPartitionsToPartitionPredicate(
+ partitions, table, spark());
Snapshot snapshot =
t.latestSnapshot()
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
index 4a4c15ceff..404a78db78 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java
@@ -104,8 +104,6 @@ public class RescaleProcedure extends BaseProcedure {
checkArgument(
partitions == null || where == null,
"partitions and where cannot be used together.");
- String finalWhere = partitions != null ?
SparkProcedureUtils.toWhere(partitions) : where;
-
return modifySparkTable(
tableIdent,
sparkTable -> {
@@ -132,12 +130,19 @@ public class RescaleProcedure extends BaseProcedure {
fileStoreTable = fileStoreTable.copy(dynamicOptions);
DataSourceV2Relation relation = createRelation(tableIdent,
sparkTable);
- PartitionPredicate partitionPredicate =
- SparkProcedureUtils.convertToPartitionPredicate(
- finalWhere,
-
fileStoreTable.schema().logicalPartitionType(),
- spark(),
- relation);
+ PartitionPredicate partitionPredicate;
+ if (partitions != null) {
+ partitionPredicate =
+
SparkProcedureUtils.convertPartitionsToPartitionPredicate(
+ partitions, fileStoreTable, spark());
+ } else {
+ partitionPredicate =
+
SparkProcedureUtils.convertToPartitionPredicate(
+ where,
+
fileStoreTable.schema().logicalPartitionType(),
+ spark(),
+ relation);
+ }
if (bucketNum == null) {
checkArgument(
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
index 59f93f92fb..923c8319df 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
@@ -20,7 +20,9 @@ package org.apache.paimon.spark.utils;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionUtils;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ParameterUtils;
import org.apache.paimon.utils.StringUtils;
@@ -35,10 +37,12 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
-import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -77,6 +81,114 @@ public class SparkProcedureUtils {
return PartitionPredicate.fromPredicate(partitionType, predicate);
}
+ @Nullable
+ public static PartitionPredicate convertPartitionsToPartitionPredicate(
+ @Nullable String partitions, FileStoreTable table, SparkSession
spark) {
+ // `partitions` is a structured partition spec path such as
+ // `dt=2024-01-01,hh=0;dt=2024-01-02,hh=1`, not a SQL expression.
+ // Values that are valid Spark SQL literals are normalized with
Spark's parser, so quoted
+ // partition values follow Spark partition spec behavior. Values that
are not literals are
+ // kept as path-style typed strings for backward compatibility.
+ //
+ // Invalid partition keys are rejected explicitly here so procedures
can fail with a clear
+ // error message before converting values to Paimon internal literals.
+ if (StringUtils.isNullOrWhitespaceOnly(partitions)) {
+ return null;
+ }
+
+ RowType partitionType = table.schema().logicalPartitionType();
+ List<String> partitionKeys = partitionType.getFieldNames();
+ checkArgument(
+ !partitionKeys.isEmpty(),
+ "Table should be a partitioned table when using partition
predicate.");
+
+ List<Map<String, String>> partitionSpecs =
+ ParameterUtils.getPartitions(partitions.split(";"));
+ validatePartitionKeys(partitionSpecs, partitionKeys);
+ partitionSpecs = normalizePartitionValuesWithSparkParser(spark,
partitionSpecs);
+
+ Predicate predicate =
+ PredicateBuilder.partitions(
+ partitionSpecs, partitionType,
table.coreOptions().partitionDefaultName());
+ return PartitionPredicate.fromPredicate(partitionType, predicate);
+ }
+
+ private static void validatePartitionKeys(
+ List<Map<String, String>> partitionSpecs, List<String>
partitionKeys) {
+ Set<String> invalidKeys = new HashSet<>();
+ for (Map<String, String> partitionSpec : partitionSpecs) {
+ for (String partitionKey : partitionSpec.keySet()) {
+ if (!partitionKeys.contains(partitionKey)) {
+ invalidKeys.add(partitionKey);
+ }
+ }
+ }
+
+ checkArgument(
+ invalidKeys.isEmpty(),
+ "Partition keys %s are invalid. Available partition keys are
%s",
+ invalidKeys,
+ partitionKeys);
+ }
+
+ private static List<Map<String, String>>
normalizePartitionValuesWithSparkParser(
+ SparkSession spark, List<Map<String, String>> partitionSpecs) {
+ return partitionSpecs.stream()
+ .map(partitionSpec -> parsePartitionSpec(spark, partitionSpec,
false))
+ .collect(java.util.stream.Collectors.toList());
+ }
+
+ /**
+ * Parse static partition spec values by evaluating them as Spark SQL
literal expressions. This
+ * keeps overwrite partition handling aligned with Spark SQL partition
specs, while rejecting
+ * non-literal or null values.
+ *
+ * @param spark the Spark session
+ * @param partitionSpec the partition spec with raw values (e.g., {"date":
"\"20260225\""})
+ * @return the static partition map with parsed literal values (e.g.,
{"date": "20260225"})
+ */
+ public static Map<String, String> parseStaticPartition(
+ SparkSession spark, Map<String, String> partitionSpec) {
+ return parsePartitionSpec(spark, partitionSpec, true);
+ }
+
+ private static Map<String, String> parsePartitionSpec(
+ SparkSession spark, Map<String, String> partitionSpec, boolean
requireLiteral) {
+ Map<String, String> staticPartition = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : partitionSpec.entrySet()) {
+ staticPartition.put(
+ entry.getKey(), parsePartitionValue(spark,
entry.getValue(), requireLiteral));
+ }
+ return staticPartition;
+ }
+
+ private static String parsePartitionValue(
+ SparkSession spark, String value, boolean requireLiteral) {
+ Expression expr;
+ try {
+ expr = spark.sessionState().sqlParser().parseExpression(value);
+ } catch (Exception e) {
+ if (requireLiteral) {
+ throw new RuntimeException(e);
+ }
+ return value;
+ }
+ if (!(expr instanceof Literal)) {
+ checkArgument(
+ !requireLiteral,
+ "Partition value must be a literal expression, but got:
%s",
+ value);
+ return value;
+ }
+
+ Object literalValue = ((Literal) expr).value();
+ if (literalValue == null) {
+ checkArgument(!requireLiteral, "Partition value cannot be null");
+ return value;
+ }
+ return literalValue.toString();
+ }
+
public static int readParallelism(List<?> groupedTasks, SparkSession
spark) {
int sparkParallelism =
Math.max(
@@ -109,33 +221,4 @@ public class SparkProcedureUtils {
.reduce((a, b) -> a + " OR " + b)
.orElse(null);
}
-
- /**
- * Parse partition spec values by evaluating them as Spark SQL literal
expressions. This strips
- * quotes from string literals and validates that values are non-null
literals.
- *
- * @param spark the Spark session
- * @param partitionSpec the partition spec with raw values (e.g., {"date":
"\"20260225\""})
- * @return the static partition map with unquoted literal values (e.g.,
{"date": "20260225"})
- */
- public static Map<String, String> parseStaticPartition(
- SparkSession spark, Map<String, String> partitionSpec) {
- Map<String, String> staticPartition = new HashMap<>();
- for (Map.Entry<String, String> entry : partitionSpec.entrySet()) {
- Expression expr;
- try {
- expr =
spark.sessionState().sqlParser().parseExpression(entry.getValue());
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- checkArgument(
- expr instanceof Literal,
- "Partition value must be a literal expression, but got:
%s",
- entry.getValue());
- Object value = ((Literal) expr).value();
- checkArgument(value != null, "Partition value cannot be null");
- staticPartition.put(entry.getKey(), value.toString());
- }
- return staticPartition;
- }
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 7cbe2b6659..76218f19ef 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -30,6 +30,7 @@ import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.paimon.shims.memstream.MemoryStream
import org.apache.spark.sql.streaming.StreamTest
import org.assertj.core.api.Assertions
+import org.assertj.core.api.Assertions.assertThatThrownBy
import org.scalatest.time.Span
import java.util
@@ -537,7 +538,7 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
assert(intercept[IllegalArgumentException] {
spark.sql("CALL sys.compact(table => 'T', partitions => 'id = 1')")
- }.getMessage.contains("Only partition predicate is supported"))
+ }.getMessage.contains("Partition keys [id] are invalid"))
assert(intercept[IllegalArgumentException] {
spark.sql("CALL sys.compact(table => 'T', where => 'id > 1 AND pt =
\"p1\"')")
@@ -1533,4 +1534,42 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
def lastSnapshotId(table: FileStoreTable): Long = {
table.snapshotManager().latestSnapshotId()
}
+
+ test("Paimon Procedure: compact partitions accept unquoted string values") {
+ withTable("T") {
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, value STRING, dt STRING, hh INT)
+ |TBLPROPERTIES ('bucket'='-1', 'write-only'='true')
+ |PARTITIONED BY (dt, hh)
+ |""".stripMargin)
+
+ val table = loadTable("T")
+
+ spark.sql(s"INSERT INTO T VALUES (1, 'a', '2024-01-01', 0), (2, 'b',
'2024-01-01', 0)")
+ spark.sql(s"INSERT INTO T VALUES (3, 'c', '2024-01-02', 0), (4, 'd',
'2024-01-02', 0)")
+
+ val before = lastSnapshotId(table)
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact(table => 'T', partitions => 'dt=2024-01-01,hh=0',
options => 'compaction.min.file-num=2')"),
+ Row(true) :: Nil)
+
+
Assertions.assertThat(lastSnapshotId(loadTable("T"))).isGreaterThan(before)
+ }
+ }
+
+ test("Paimon Procedure: compact with invalid partition key") {
+ withTable("T") {
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, value STRING, dt STRING, hh INT)
+ |TBLPROPERTIES ('bucket'='-1', 'write-only'='true')
+ |PARTITIONED BY (dt, hh)
+ |""".stripMargin)
+
+ val e = intercept[IllegalArgumentException] {
+ spark.sql("CALL sys.compact(table => 'T', partitions =>
'pt=2024-01-01')")
+ }
+ Assertions.assertThat(e.getMessage.contains("Partition keys [pt] are
invalid"))
+ }
+ }
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
index 986b1f9b6b..b4a6525116 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RescaleProcedureTest.scala
@@ -146,6 +146,37 @@ class RescaleProcedureTest extends PaimonSparkTestBase {
}
}
+ test("Paimon Procedure: rescale partitions accept unquoted string values") {
+ withTable("T") {
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, value STRING, dt STRING, hh INT)
+ |TBLPROPERTIES ('primary-key'='id, dt, hh', 'bucket'='2')
+ |PARTITIONED BY (dt, hh)
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO T VALUES (1, 'a', '2024-01-01', 0), (2, 'b',
'2024-01-01', 0)")
+ spark.sql(s"INSERT INTO T VALUES (3, 'c', '2024-01-02', 0), (4, 'd',
'2024-01-02', 0)")
+
+ spark.sql("ALTER TABLE T SET TBLPROPERTIES ('bucket' = '4')")
+ checkAnswer(
+ spark.sql(
+ "CALL sys.rescale(table => 'T', bucket_num => 4, partitions =>
'dt=2024-01-01,hh=0')"),
+ Row(true) :: Nil)
+
+ val reloadedTable = loadTable("T")
+ val predicate = PartitionPredicate.fromMap(
+ reloadedTable.schema().logicalPartitionType(),
+ Map("dt" -> "2024-01-01", "hh" -> "0").asJava,
+ reloadedTable.coreOptions().partitionDefaultName())
+ reloadedTable.newSnapshotReader
+ .withPartitionFilter(predicate)
+ .read
+ .dataSplits
+ .asScala
+ .foreach(split => Assertions.assertThat(split.bucket()).isLessThan(4))
+ }
+ }
+
test("Paimon Procedure: rescale with where clause") {
withTable("T") {
spark.sql(s"""
