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 f87bf749df [spark] Make max_pt work on format tables (#9024)
f87bf749df is described below
commit f87bf749df776642b4ce113bb3de65c58a9301d2
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Tue Aug 4 18:52:45 2026 +0800
[spark] Make max_pt work on format tables (#9024)
---
.../catalyst/analysis/ReplacePaimonFunctions.scala | 65 ++++++++++++---
.../paimon/spark/table/PaimonFormatTableTest.scala | 97 ++++++++++++++++++++++
2 files changed, 150 insertions(+), 12 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala
index 5a11384246..d81560ffe6 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala
@@ -18,12 +18,15 @@
package org.apache.paimon.spark.catalyst.analysis
-import org.apache.paimon.spark.{DataConverter, SparkTable, SparkTypeUtils,
SparkUtils}
+import org.apache.paimon.partition.PartitionPredicate
+import org.apache.paimon.predicate.PredicateBuilder
+import org.apache.paimon.spark.{BaseTable, DataConverter, SparkTable,
SparkTypeUtils, SparkUtils}
import org.apache.paimon.spark.catalog.SparkBaseCatalog
import org.apache.paimon.spark.catalog.functions.PaimonFunctions
import org.apache.paimon.spark.function.{BlobViewFieldIdSparkFunction,
BlobViewSparkFunction, DescriptorToPresignedUrlFunction,
ResolvedDescriptorToPresignedUrlFunction}
import org.apache.paimon.spark.utils.CatalogUtils
import org.apache.paimon.table.DataTable
+import org.apache.paimon.table.FormatTable
import org.apache.paimon.types.DataTypeRoot
import org.apache.paimon.utils.{InternalRowUtils, TypeUtils}
@@ -151,21 +154,59 @@ case class ReplacePaimonFunctions(spark: SparkSession)
extends Rule[LogicalPlan]
val table =
catalogAndIdentifier.catalog.asTableCatalog.loadTable(catalogAndIdentifier.identifier())
- assert(table.isInstanceOf[SparkTable])
- val sparkTable = table.asInstanceOf[SparkTable]
- if (sparkTable.table.partitionKeys().size() == 0) {
+ // Every Paimon table wrapper exposes the underlying table through
BaseTable. Asserting the
+ // narrower SparkTable used to fail a format table with an AssertionError
rather than telling
+ // the user anything.
+ if (!table.isInstanceOf[BaseTable]) {
+ throw new UnsupportedOperationException(s"$table is not a Paimon table")
+ }
+ val paimonTable = table.asInstanceOf[BaseTable].table
+ if (paimonTable.partitionKeys().size() == 0) {
throw new UnsupportedOperationException(s"$table is not a partitioned
table")
}
val toplevelPartitionType =
- TypeUtils.project(sparkTable.table.rowType,
sparkTable.table.partitionKeys()).getTypeAt(0)
- val partitionValues = sparkTable.table.newReadBuilder.newScan
- .listPartitionEntries()
- .asScala
- .filter(_.fileCount() > 0)
- .map {
- partitionEntry => InternalRowUtils.get(partitionEntry.partition(), 0,
toplevelPartitionType)
- }
+ TypeUtils.project(paimonTable.rowType,
paimonTable.partitionKeys()).getTypeAt(0)
+ val partitions = paimonTable match {
+ case formatTable: FormatTable =>
+ val partitionType = TypeUtils.project(paimonTable.rowType,
paimonTable.partitionKeys())
+ val candidates = formatTable.newReadBuilder.newScan
+ .listPartitionEntries()
+ .asScala
+ .map(entry => InternalRowUtils.get(entry.partition(), 0,
toplevelPartitionType))
+ .filter(_ != null)
+ .distinct
+ .sortWith(InternalRowUtils.compare(_, _,
toplevelPartitionType.getTypeRoot) > 0)
+
+ // Catalog metadata uses zero both for an empty format table partition
and for a
+ // partition whose file count has not been reported. Check candidates
from largest to
+ // smallest and stop at the first value whose filtered scan produces
data. This also avoids
+ // treating an empty filesystem directory as data without planning
splits for the whole
+ // table.
+ candidates.find {
+ candidate =>
+ val predicate = new PredicateBuilder(partitionType).equal(0,
candidate)
+ val partitionFilter =
+ PartitionPredicate.fromPredicate(partitionType, predicate)
+ !formatTable.newReadBuilder
+ .withPartitionFilter(partitionFilter)
+ .newScan
+ .plan()
+ .splits()
+ .isEmpty
+ }.toSeq
+ case _ =>
+ // FileStoreTable manifests carry an exact file count, so keep the
cheaper metadata path.
+ paimonTable.newReadBuilder.newScan
+ .listPartitionEntries()
+ .asScala
+ .filter(_.fileCount() > 0)
+ .map(entry => InternalRowUtils.get(entry.partition(), 0,
toplevelPartitionType))
+ }
+ val partitionValues = partitions
+ // The default partition comes back as a real null, which
InternalRowUtils.compare would
+ // dereference. It is also not a value anyone means by "the max
partition".
+ .filter(_ != null)
.sortWith(InternalRowUtils.compare(_, _,
toplevelPartitionType.getTypeRoot) < 0)
.map(DataConverter.fromPaimon(_, toplevelPartitionType))
if (partitionValues.isEmpty) {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
index adac70df2e..fedf8363d0 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
@@ -245,6 +245,103 @@ class PaimonFormatTableTest extends
PaimonSparkTestWithRestCatalogBase {
}
}
+ test("PaimonFormatTable: max_pt picks the largest partition") {
+ val tableName = "max_pt_t"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (f0 INT) USING CSV PARTITIONED BY (ds
STRING) " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon', " +
+ "'metastore.partitioned-table'='true')")
+ val table =
+ paimonCatalog.getTable(Identifier.create("test_db",
tableName)).asInstanceOf[FormatTable]
+ assert(table.partitionManager() != null)
+ sql(s"INSERT INTO $tableName VALUES (1, '20240101')")
+ sql(s"INSERT INTO $tableName VALUES (2, '20240103')")
+ sql(s"INSERT INTO $tableName VALUES (3, '20240102')")
+
+ // Two things used to stop max_pt on a format table: it is not a
SparkTable, and
+ // catalog partitions report zero before the partition-statistics
contract exists.
+ checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"),
Seq(Row("20240103")))
+ checkAnswer(
+ sql(s"SELECT * FROM $tableName WHERE ds =
sys.max_pt('test_db.$tableName')"),
+ Seq(Row(2, "20240103")))
+ }
+ }
+
+ test("PaimonFormatTable: max_pt skips the default partition") {
+ val tableName = "max_pt_null"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (f0 INT) USING CSV PARTITIONED BY (ds
STRING) " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon', " +
+ "'metastore.partitioned-table'='true')")
+ sql(s"INSERT INTO $tableName VALUES (1, '20240101')")
+ // A null partition value comes back as a real null, which the typed
comparator would
+ // dereference; it is also not what anyone means by the max partition.
+ sql(s"INSERT INTO $tableName VALUES (2, null)")
+
+ checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"),
Seq(Row("20240101")))
+ }
+ }
+
+ test("PaimonFormatTable: max_pt skips an empty registered partition") {
+ val tableName = "max_pt_empty"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (f0 INT) USING CSV PARTITIONED BY (ds
STRING) " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon', " +
+ "'metastore.partitioned-table'='true')")
+ sql(s"INSERT INTO $tableName VALUES (1, '20240101')")
+ sql(s"ALTER TABLE $tableName ADD PARTITION (ds='20240103')")
+
+ checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"),
Seq(Row("20240101")))
+ }
+ }
+
+ test("PaimonFormatTable: max_pt skips an empty filesystem partition
directory") {
+ val tableName = "max_pt_empty_dir"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (f0 INT) USING CSV PARTITIONED BY (ds
STRING) " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon')")
+ sql(s"INSERT INTO $tableName VALUES (1, '20240101')")
+ val table =
+ paimonCatalog.getTable(Identifier.create("test_db",
tableName)).asInstanceOf[FormatTable]
+ assert(table.partitionManager() == null)
+ table.fileIO().mkdirs(new Path(table.location(), "ds=20240103"))
+
+ checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"),
Seq(Row("20240101")))
+ }
+ }
+
+ test("max_pt on a FileStoreTable skips a null partition instead of failing")
{
+ val tableName = "max_pt_fst_null"
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (id INT, ds STRING) USING paimon
PARTITIONED BY (ds)")
+ sql(s"INSERT INTO $tableName VALUES (1, '20240101')")
+ sql(s"INSERT INTO $tableName VALUES (2, null)")
+
+ // This is a behaviour change for FileStoreTable, not only for format
tables: a null
+ // top-level partition value used to reach InternalRowUtils.compare and
throw, and is now
+ // skipped. Pinned here so the change is visible rather than incidental.
+ checkAnswer(sql(s"SELECT sys.max_pt('test_db.$tableName')"),
Seq(Row("20240101")))
+ }
+ }
+
+ test("PaimonFormatTable: max_pt rejects an unpartitioned table") {
+ val tableName = "max_pt_flat"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (f0 INT) USING CSV " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon')")
+ sql(s"INSERT INTO $tableName VALUES (1)")
+ val e = intercept[Exception] {
+ sql(s"SELECT sys.max_pt('test_db.$tableName')").collect()
+ }
+ assert(e.getMessage.contains("not a partitioned table"))
+ }
+ }
+
test("PaimonFormatTable: non-partitioned table") {
for {
(format, compression) <- Seq(
