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 2e3b41cf26 [spark] Clamp Format Table ANALYZE parallelism (#9482)
2e3b41cf26 is described below
commit 2e3b41cf26075414ae8779b4293a63c6a70e8f5d
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Mon Aug 31 12:05:35 2026 +0800
[spark] Clamp Format Table ANALYZE parallelism (#9482)
---
...PaimonAnalyzeFormatTablePartitionsCommand.scala | 5 ++--
.../sql/CatalogManagedPartitionAnalyzeTest.scala | 34 ++++++++++++++++++++++
2 files changed, 37 insertions(+), 2 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
index 06aee2ffa3..0f68c63773 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
@@ -105,8 +105,9 @@ case class PaimonAnalyzeFormatTablePartitionsCommand(
sparkSession: SparkSession,
partitions: List[JMap[String, String]],
parallelism: Int): JList[PartitionStatistics] = {
- val tasks = math.min(parallelism, partitions.size)
- val perTask = math.max(1, parallelism / tasks)
+ val effectiveParallelism = math.max(1, parallelism)
+ val tasks = math.min(effectiveParallelism, partitions.size)
+ val perTask = math.max(1, effectiveParallelism / tasks)
// The table, not this command: a Spark table cannot be shipped to an
executor.
val table = v2Table.table
val measured = sparkSession.sparkContext
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
index e2e85ae64e..3b2a2d1b9a 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionAnalyzeTest.scala
@@ -338,6 +338,40 @@ class CatalogManagedPartitionAnalyzeTest extends
PaimonSparkTestWithRestCatalogB
}
}
+ test("a full ANALYZE clamps non-positive statistics parallelism") {
+ Seq("zero" -> 0, "negative" -> -1).foreach {
+ case (label, parallelism) =>
+ val tableName = s"analyze_${label}_parallelism"
+ withTable(tableName) {
+ sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING,
hour STRING)
+ |USING PARQUET
+ |PARTITIONED BY (dt, hour)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+ sql(s"""INSERT INTO ${qualified(tableName)}
+ |VALUES (1, 'a', '20260101', '00'), (2, 'b', '20260101', '00')
+ |""".stripMargin)
+ copyPartitionFiles(tableName, "20260101", "20260102")
+ repair(tableName)
+ assert(
+ !PartitionStatistics.isKnown(statisticsOf(tableName, "20260102",
"00").recordCount()))
+
+ withSparkSQLConf(
+ "spark.paimon.format-table.statistics.parallelism" ->
parallelism.toString) {
+ sql(
+ s"ANALYZE TABLE ${qualified(tableName)} PARTITION (dt =
'20260102') " +
+ s"COMPUTE STATISTICS").collect()
+ }
+
+ val scanned = statisticsOf(tableName, "20260102", "00")
+ // The exact row count confirms that ANALYZE completed the Parquet
footer scan.
+ assert(scanned.recordCount() == 2L, scanned.toString)
+ }
+ }
+ }
+
test("catalog partition row counts feed scan statistics after partition
pruning") {
val tableName = "analyze_scan_statistics"
withTable(tableName) {
