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 bc6531dfba [spark] Allow partition and bucket metadata queries during
merge reads (#10481)
bc6531dfba is described below
commit bc6531dfba4e568d65298c5b46e7485376f595eb
Author: Ran Tao <[email protected]>
AuthorDate: Sat Oct 10 14:35:46 2026 +0800
[spark] Allow partition and bucket metadata queries during merge reads
(#10481)
---
.../paimon/spark/PaimonRecordReaderIterator.scala | 8 +--
.../apache/paimon/spark/sql/PaimonQueryTest.scala | 60 +++++++++++++++++-----
2 files changed, 52 insertions(+), 16 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
index 93256df2c0..d2c758e93e 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
@@ -133,7 +133,7 @@ case class PaimonRecordReaderIterator(
currentIterator.asInstanceOf[ScoreRecordIterator[PaimonInternalRow]])
currentResult = joinedRow.replace(dataRow, metadataRow)
} else if (needMetadata) {
-
updateMetadataRow(currentIterator.asInstanceOf[FileRecordIterator[PaimonInternalRow]])
+ updateMetadataRow(currentIterator)
currentResult = joinedRow.replace(dataRow, metadataRow)
} else {
currentResult = dataRow
@@ -160,12 +160,14 @@ case class PaimonRecordReaderIterator(
}
}
- private def updateMetadataRow(fileRecordIterator:
FileRecordIterator[PaimonInternalRow]): Unit = {
+ private def updateMetadataRow(iterator:
RecordReader.RecordIterator[PaimonInternalRow]): Unit = {
metadataColumns.zipWithIndex.foreach {
case (metadataColumn, index) =>
metadataColumn.name match {
case PaimonMetadataColumn.ROW_INDEX_COLUMN =>
- metadataRow.setField(index, fileRecordIterator.returnedPosition())
+ metadataRow.setField(
+ index,
+
iterator.asInstanceOf[FileRecordIterator[PaimonInternalRow]].returnedPosition())
case PaimonMetadataColumn.FILE_PATH_COLUMN =>
metadataRow.setField(index,
BinaryString.fromString(lastFilePath.toString))
case PaimonMetadataColumn.PARTITION_COLUMN =>
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
index f94313f5e7..24287d5417 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
@@ -384,22 +384,56 @@ class PaimonQueryTest extends PaimonSparkTestBase {
)
}
- test("Paimon Query: not support querying metadata columns for pk table") {
- spark.sql("""
- |CREATE TABLE T (id INT, name STRING)
- |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1')
- |""".stripMargin)
+ test("Paimon Query: query partition and bucket metadata columns for pk
table") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT, name STRING, pt STRING) PARTITIONED BY
(pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'pt,id',
+ | 'bucket' = '1',
+ | 'write-only' = 'true',
+ | 'deletion-vectors.enabled' = 'false')
+ |""".stripMargin)
- spark.sql("INSERT INTO T VALUES(1,'a')")
- assertThat(spark.sql("SELECT *,__paimon_file_path FROM
T").collect()).hasSize(1)
+ sql("INSERT INTO T VALUES (1, 'old', 'p1')")
+ checkAnswer(
+ sql("SELECT id, name, __paimon_bucket, __paimon_partition FROM T"),
+ Row(1, "old", 0, Row("p1")))
+
+ // Updating the same key in a separate file requires merging; write-only
disables compaction.
+ sql("INSERT INTO T VALUES (1, 'new', 'p1')")
+ checkAnswer(sql("SELECT id, name FROM T"), Row(1, "new"))
+ checkAnswer(sql("SELECT __paimon_bucket FROM T"), Row(0))
+ checkAnswer(sql("SELECT __paimon_partition FROM T"), Row(Row("p1")))
+ checkAnswer(
+ sql("SELECT id, name, __paimon_bucket, __paimon_partition FROM T"),
+ Row(1, "new", 0, Row("p1")))
+ checkAnswer(
+ sql("SELECT __paimon_bucket, __paimon_partition FROM T WHERE id =
999"),
+ Seq.empty)
+ }
+ }
+
+ test("Paimon Query: reject file path and row index metadata when merging pk
files") {
+ sql("""
+ |CREATE TABLE T (id INT, name STRING)
+ |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1')
+ |""".stripMargin)
+
+ sql("INSERT INTO T VALUES(1,'a')")
+ assertThat(sql("SELECT *, __paimon_file_path, __paimon_row_index FROM
T").collect()).hasSize(1)
// query failed if more than one file in a bucket
- spark.sql("INSERT INTO T VALUES(2,'b')")
- assert(
- intercept[SparkException] {
- spark.sql("SELECT *,__paimon_file_path FROM T").collect()
- }.getMessage
- .contains("Only append table or deletion vector table support querying
metadata columns."))
+ sql("INSERT INTO T VALUES(2,'b')")
+ Seq("__paimon_file_path", "__paimon_row_index").foreach {
+ column =>
+ assert(
+ intercept[SparkException] {
+ sql(s"SELECT *, $column FROM T").collect()
+ }.getMessage
+ .contains(
+ "Only append table or deletion vector table support querying
metadata columns."))
+ }
}
test("Paimon Query: disallow full scan") {
