This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new 8145cc94f [AURON #2366] fix: Handle Paimon metadata columns in V2
native scan (#2367)
8145cc94f is described below
commit 8145cc94fc7261f4bfaf1bbf9f95baabc2013f2d
Author: linfeng <[email protected]>
AuthorDate: Tue Jun 30 19:15:57 2026 +0800
[AURON #2366] fix: Handle Paimon metadata columns in V2 native scan (#2367)
# Which issue does this PR close?
Closes #2366
# Rationale for this change
Paimon metadata columns are produced by the Paimon scan layer rather
than stored as physical columns in data files. The Paimon V2 native scan
was passing these columns to the native Parquet/ORC reader as file
columns, which can return incorrect values.
For example:
```sql
create table paimon.db.t_metadata (id int, v string) using paimon;
insert into paimon.db.t_metadata values (1, 'a');
select id, __paimon_file_path from paimon.db.t_metadata;
```
The native path returned null for `__paimon_file_path`, while
Spark/Paimon's scan path returns the actual file path.
# What changes are included in this PR?
- Recognize Paimon metadata columns using PaimonMetadataColumn.
- Materialize supported file-level metadata columns (__paimon_file_path,
__paimon_bucket) as per-file constants.
- Keep unsupported Paimon metadata columns on Spark/Paimon's scan path
instead of reading them from Parquet/ORC files.
- Cover metadata columns both with and without table partition columns.
# Are there any user-facing changes?
No API changes. This is a correctness fix for Paimon V2 native scan.
# How was this patch tested?
Adds Paimon V2 integration tests
---
.../spark/sql/auron/paimon/PaimonScanSupport.scala | 131 ++++++++++---
.../paimon/AuronPaimonV2IntegrationSuite.scala | 206 ++++++++++++++++++++-
2 files changed, 306 insertions(+), 31 deletions(-)
diff --git
a/thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala
b/thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala
index f5c23d3a2..fddc3df5d 100644
---
a/thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala
+++
b/thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala
@@ -16,11 +16,15 @@
*/
package org.apache.spark.sql.auron.paimon
+import java.util.Locale
+
import scala.collection.JavaConverters._
import scala.util.control.NonFatal
import org.apache.commons.lang3.reflect.MethodUtils
+import org.apache.paimon.fs.Path
import org.apache.paimon.spark.DataConverter
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.source.{DataSplit, Split}
import org.apache.paimon.types.RowType
@@ -34,6 +38,7 @@ import
org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.hive.auron.paimon.PaimonUtil
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
+import org.apache.spark.unsafe.types.UTF8String
final case class PaimonFile(filePath: String, fileSize: Long, partitionValues:
InternalRow)
@@ -49,6 +54,10 @@ object PaimonScanSupport extends Logging {
private val PaimonBaseScanClassName =
"org.apache.paimon.spark.PaimonBaseScan"
private val PaimonInputPartitionClassName =
"org.apache.paimon.spark.PaimonInputPartition"
+ private val PaimonMetadataColumnPrefix = "__paimon_"
+ private val PaimonFilePathColumn = PaimonMetadataColumn.FILE_PATH_COLUMN
+ private val PaimonBucketColumn = PaimonMetadataColumn.BUCKET_COLUMN
+ private val PaimonMetadataColumns =
PaimonMetadataColumn.SUPPORTED_METADATA_COLUMNS.toSet
// Planning a Paimon scan performs split-planning I/O (reading metadata
files). The conversion
// pipeline calls plan() twice on the same exec (once in isSupported, once
in convert), so we
@@ -107,10 +116,26 @@ object PaimonScanSupport extends Logging {
logDebug("Skip native Paimon scan: unsupported column data type in read
schema.")
return None
}
+ val physicalColumnSet = table.schema().fieldNames().asScala.toSet
+ def isPhysicalColumn(name: String): Boolean =
containsName(physicalColumnSet, name)
+ val unsupportedMetadataColumns = readSchema.fields.filter { f =>
+ !isPhysicalColumn(f.name) && isPaimonMetadataColumn(f.name) &&
!isSupportedMetadataColumn(
+ f.name)
+ }
+ if (unsupportedMetadataColumns.nonEmpty) {
+ logDebug(
+ s"Skip native Paimon scan: unsupported metadata columns " +
+ s"${unsupportedMetadataColumns.map(_.name).mkString(", ")}.")
+ return None
+ }
- val partitionKeys = table.schema().partitionKeys().asScala.toSet
- val partitionFields = readSchema.fields.filter(f =>
containsName(partitionKeys, f.name))
- val fileFields = readSchema.fields.filterNot(f =>
containsName(partitionKeys, f.name))
+ val partitionKeys = table.schema().partitionKeys().asScala.toSeq
+ val partitionKeySet = partitionKeys.toSet
+ def isPartitionValueField(name: String): Boolean =
+ containsName(partitionKeySet, name) ||
+ (!isPhysicalColumn(name) && isSupportedMetadataColumn(name))
+ val partitionFields = readSchema.fields.filter(f =>
isPartitionValueField(f.name))
+ val fileFields = readSchema.fields.filterNot(f =>
isPartitionValueField(f.name))
val partitionSchema = StructType(partitionFields)
val fileSchema = StructType(fileFields)
@@ -149,20 +174,40 @@ object PaimonScanSupport extends Logging {
val partitionRowType = table.schema().logicalPartitionType()
val partitionConverter = new
RowDataToObjectArrayConverter(partitionRowType)
+ val partitionKeyIndexByName = partitionKeys.zipWithIndex.toMap
+ val resolver = SQLConf.get.resolver
+ val partitionFieldIndexes = partitionSchema.fields.map { field =>
+ partitionKeyIndexByName
+ .find { case (k, _) => resolver(k, field.name) }
+ .map(_._2)
+ .getOrElse(-1)
+ }
+ val filePathMetadataIndex = partitionSchema.fields.indexWhere { field =>
+ !isPhysicalColumn(field.name) && isFilePathMetadataColumn(field.name)
+ }
val files = splits.flatMap { split =>
- val partitionValues = if (partitionSchema.isEmpty) {
- InternalRow.empty
+ val partitionValueTemplate = if (partitionSchema.isEmpty) {
+ Array.empty[Any]
} else {
- toPartitionRow(
+ toPartitionValueTemplate(
partitionConverter.convert(split.partition()),
partitionRowType,
partitionSchema,
- table.schema().partitionKeys().asScala.toSeq)
+ partitionFieldIndexes,
+ split.bucket())
}
split.dataFiles().asScala.map { dataFile =>
- val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
- PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ val rawFilePath = dataFile
+ .externalPath()
+ .orElse(s"${split.bucketPath()}/${dataFile.fileName()}")
+ val metadataFilePath = new Path(rawFilePath).toUri.toString
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(partitionValueTemplate, filePathMetadataIndex,
metadataFilePath)
+ }
+ PaimonFile(rawFilePath, dataFile.fileSize(), partitionValues)
}
}
@@ -174,30 +219,62 @@ object PaimonScanSupport extends Logging {
names.exists(n => resolver(n, target))
}
- // Build a Spark InternalRow for partition values matching partitionSchema's
data types.
- // Partition values from Paimon are returned in the table's partition-key
order; we reorder
- // them to match partitionSchema and convert each value into the Spark
catalyst representation
- // expected for the field's type via Paimon's own DataConverter. This
handles dates (epoch
- // days), timestamps (micros), decimals and binary correctly, rather than
the lossy
- // String round-trip a Cast(Literal(value.toString), dataType) would perform.
- private def toPartitionRow(
+ private def isFilePathMetadataColumn(name: String): Boolean = {
+ SQLConf.get.resolver(name, PaimonFilePathColumn)
+ }
+
+ private def isBucketMetadataColumn(name: String): Boolean = {
+ SQLConf.get.resolver(name, PaimonBucketColumn)
+ }
+
+ private def isSupportedMetadataColumn(name: String): Boolean = {
+ isFilePathMetadataColumn(name) || isBucketMetadataColumn(name)
+ }
+
+ private def isPaimonMetadataColumn(name: String): Boolean = {
+ containsName(PaimonMetadataColumns, name) ||
+ name.toLowerCase(Locale.ROOT).startsWith(PaimonMetadataColumnPrefix)
+ }
+
+ // Build split-invariant constants for partition columns and supported
metadata columns in
+ // partitionSchema order. Paimon's DataConverter preserves Catalyst
representations for typed
+ // partition values such as dates, timestamps, decimals and binary.
+ private def toPartitionValueTemplate(
paimonValues: Array[AnyRef],
partitionRowType: RowType,
partitionSchema: StructType,
- partitionKeys: Seq[String]): InternalRow = {
- val resolver = SQLConf.get.resolver
- val indexByName = partitionKeys.zipWithIndex.toMap
- InternalRow.fromSeq(partitionSchema.fields.map { field =>
- val idx = indexByName
- .find { case (k, _) => resolver(k, field.name) }
- .map(_._2)
- .getOrElse(-1)
- if (idx >= 0 && idx < paimonValues.length && paimonValues(idx) != null) {
- DataConverter.fromPaimon(paimonValues(idx),
partitionRowType.getTypeAt(idx))
+ partitionFieldIndexes: Array[Int],
+ bucket: Int): Array[Any] = {
+ partitionSchema.fields.zip(partitionFieldIndexes).map { case (field,
partitionColumnIndex) =>
+ if (partitionColumnIndex >= 0 && partitionColumnIndex <
paimonValues.length) {
+ if (paimonValues(partitionColumnIndex) == null) {
+ null
+ } else {
+ DataConverter.fromPaimon(
+ paimonValues(partitionColumnIndex),
+ partitionRowType.getTypeAt(partitionColumnIndex))
+ }
+ } else if (isFilePathMetadataColumn(field.name)) {
+ null
+ } else if (isBucketMetadataColumn(field.name)) {
+ bucket
} else {
null
}
- })
+ }
+ }
+
+ private def toPartitionRow(
+ partitionValueTemplate: Array[Any],
+ filePathMetadataIndex: Int,
+ filePath: String): InternalRow = {
+ if (filePathMetadataIndex < 0) {
+ InternalRow.fromSeq(partitionValueTemplate)
+ } else {
+ val partitionValues = partitionValueTemplate.clone()
+ partitionValues(filePathMetadataIndex) = UTF8String.fromString(filePath)
+ InternalRow.fromSeq(partitionValues)
+ }
}
private def collectSplits(partitions: Seq[InputPartition]):
Option[Seq[DataSplit]] = {
diff --git
a/thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala
b/thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala
index 6aa2cd319..a36b06441 100644
---
a/thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala
+++
b/thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala
@@ -22,9 +22,13 @@ import java.util.concurrent.TimeUnit
import scala.collection.JavaConverters._
+import org.apache.paimon.spark.PaimonInputPartition
+import org.apache.paimon.table.source.DataSplit
import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.auron.paimon.PaimonScanSupport
import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
class AuronPaimonV2IntegrationSuite
@@ -228,6 +232,178 @@ class AuronPaimonV2IntegrationSuite
}
}
+ test("paimon v2 native scan supports file-level metadata columns") {
+ withTable("paimon.db.t_metadata") {
+ sql("create table paimon.db.t_metadata (id int, v string) using paimon")
+ sql("insert into paimon.db.t_metadata values (1, 'a')")
+
+ checkSparkAnswerAndNativePaimonScan(
+ "select id, __paimon_file_path, __paimon_bucket from
paimon.db.t_metadata")
+ }
+ }
+
+ test("paimon v2 native scan supports metadata-only projection") {
+ withTable("paimon.db.t_metadata_only") {
+ sql("create table paimon.db.t_metadata_only (id int, v string) using
paimon")
+ sql("insert into paimon.db.t_metadata_only values (1, 'a'), (2, 'b')")
+ sql("insert into paimon.db.t_metadata_only values (3, 'c'), (4, 'd')")
+
+ withSQLConf(
+ "spark.sql.files.maxPartitionBytes" -> "256",
+ "spark.sql.files.openCostInBytes" -> "1") {
+ checkSparkAnswerAndNativePaimonScan(
+ "select __paimon_file_path from paimon.db.t_metadata_only")
+ }
+ }
+ }
+
+ test(
+ "paimon v2 native scan planning supports metadata across multiple data
files in one split") {
+ withTable("paimon.db.t_metadata_multi_file_split") {
+ sql("""
+ |create table paimon.db.t_metadata_multi_file_split (id int, v
string)
+ |using paimon
+ |tblproperties (
+ | 'source.split.target-size' = '1 gb',
+ | 'source.split.open-file-cost' = '1 b'
+ |)
+ |""".stripMargin)
+ withSQLConf("spark.sql.shuffle.partitions" -> "2") {
+ spark
+ .range(0, 20)
+ .repartition(2)
+ .selectExpr("cast(id as int) as id", "cast(id as string) as v")
+ .writeTo("paimon.db.t_metadata_multi_file_split")
+ .append()
+ }
+
+ val sqlText =
+ "select id, v, __paimon_file_path, __paimon_bucket " +
+ "from paimon.db.t_metadata_multi_file_split"
+ withSQLConf("spark.sql.files.minPartitionNum" -> "1") {
+ var expected: Seq[Row] = Nil
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ expected = sql(sqlText).collect().toSeq
+ }
+ assert(
+ expected.map(_.getString(2)).distinct.size > 1,
+ s"expected rows from multiple data files, got $expected")
+
+ val df = sql(sqlText)
+ checkAnswer(df, expected)
+ val plan = df.queryExecution.sparkPlan
+ val batchScan = plan.collectFirst { case scan: BatchScanExec => scan }
+ assert(batchScan.nonEmpty, s"expected BatchScanExec in spark
plan:\n$plan")
+ val splits = paimonDataSplits(batchScan.get)
+ assert(
+ splits.exists(_.dataFiles().size() > 1),
+ s"expected at least one Paimon split with multiple data files, got "
+
+ s"${splits.map(_.dataFiles().size()).mkString("[", ", ", "]")}")
+
+ val scanPlan = PaimonScanSupport.plan(batchScan.get)
+ assert(scanPlan.nonEmpty, s"expected native Paimon scan plan
for:\n$batchScan")
+ val filePathIndex =
scanPlan.get.partitionSchema.fieldIndex("__paimon_file_path")
+ val plannedFilePaths = scanPlan.get.files.map { file =>
+ file.partitionValues.getUTF8String(filePathIndex).toString
+ }
+ assert(
+ plannedFilePaths.distinct.size > 1,
+ s"expected per-file metadata paths in native scan plan, got
$plannedFilePaths")
+ }
+ }
+ }
+
+ test("paimon v2 native scan reads physical columns that share metadata
names") {
+ withTable("paimon.db.t_metadata_name_collision") {
+ sql("""
+ |create table paimon.db.t_metadata_name_collision
+ |(`__paimon_bucket` int, id int)
+ |using paimon
+ |""".stripMargin)
+ sql("insert into paimon.db.t_metadata_name_collision values (10, 1),
(20, 2)")
+
+ checkSparkAnswerAndNativePaimonScan(
+ "select `__paimon_bucket`, id from
paimon.db.t_metadata_name_collision")
+ }
+ }
+
+ test("paimon v2 native scan reads partition columns that share metadata
names") {
+ withTable("paimon.db.t_metadata_partition_name_collision") {
+ sql("""
+ |create table paimon.db.t_metadata_partition_name_collision
+ |(`__paimon_bucket` int, `__paimon_file_path` string, id int)
+ |using paimon
+ |partitioned by (`__paimon_bucket`, `__paimon_file_path`)
+ |""".stripMargin)
+ sql("""
+ |insert into paimon.db.t_metadata_partition_name_collision values
+ |(10, 'path-a', 1),
+ |(20, 'path-b', 2)
+ |""".stripMargin)
+
+ checkSparkAnswerAndNativePaimonScan(
+ "select `__paimon_bucket`, `__paimon_file_path`, id " +
+ "from paimon.db.t_metadata_partition_name_collision")
+ }
+ }
+
+ test("paimon v2 native scan supports metadata columns with table
partitions") {
+ withTable("paimon.db.t_metadata_part") {
+ sql("""
+ |create table paimon.db.t_metadata_part (id int, v string, p
string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_metadata_part values (1, 'a', 'p1'), (2,
'b', '50%')")
+
+ checkSparkAnswerAndNativePaimonScan(
+ "select p, __paimon_file_path, id, __paimon_bucket from
paimon.db.t_metadata_part")
+ }
+ }
+
+ test("paimon v2 native scan supports non-zero bucket metadata columns") {
+ withTable("paimon.db.t_metadata_bucketed") {
+ sql("""
+ |create table paimon.db.t_metadata_bucketed (id int, v string)
+ |using paimon
+ |tblproperties (
+ | 'primary-key' = 'id',
+ | 'bucket' = '2',
+ | 'full-compaction.delta-commits' = '1'
+ |)
+ |""".stripMargin)
+ sql(
+ "insert into paimon.db.t_metadata_bucketed " +
+ "select cast(id as int), cast(id as string) from range(0, 100)")
+
+ var expected: Seq[Row] = Nil
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ expected =
+ sql("select id, __paimon_bucket from
paimon.db.t_metadata_bucketed").collect().toSeq
+ }
+ assert(
+ expected.exists(_.getInt(1) != 0),
+ s"expected at least one non-zero Paimon bucket, got $expected")
+
+ val df = sql("select id, __paimon_bucket from
paimon.db.t_metadata_bucketed")
+ checkAnswer(df, expected)
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan falls back for unsupported metadata columns") {
+ withTable("paimon.db.t_metadata_unsupported") {
+ sql("create table paimon.db.t_metadata_unsupported (id int, v string)
using paimon")
+ sql("insert into paimon.db.t_metadata_unsupported values (1, 'a')")
+
+ val df = sql("select id, __paimon_row_index from
paimon.db.t_metadata_unsupported")
+ val plan = df.queryExecution.executedPlan.toString()
+
+ assert(!plan.contains("NativePaimonV2TableScan"))
+ assert(df.collect().length === 1)
+ }
+ }
+
private def assertNativePaimonScanApplied(df: DataFrame): Unit = {
val plan = df.queryExecution.executedPlan.toString()
assert(
@@ -235,11 +411,33 @@ class AuronPaimonV2IntegrationSuite
s"plan should use native paimon scan:\n$plan")
}
- private def executedNativeScan(df: DataFrame): NativePaimonV2TableScanExec =
{
- val nativeScan = df.queryExecution.executedPlan.collectFirst {
- case scan: NativePaimonV2TableScanExec => scan
+ private def checkSparkAnswerAndNativePaimonScan(sqlText: String): Unit = {
+ var expected: Seq[Row] = Nil
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ expected = sql(sqlText).collect().toSeq
}
- assert(nativeScan.nonEmpty, "expected NativePaimonV2TableScanExec in
executed plan")
+
+ val df = sql(sqlText)
+ checkAnswer(df, expected)
+ assertNativePaimonScanApplied(df)
+ }
+
+ private def executedNativeScan(df: DataFrame): NativePaimonV2TableScanExec =
{
+ val plan = df.queryExecution.executedPlan
+ val nativeScan = plan.collectFirst { case scan:
NativePaimonV2TableScanExec => scan }
+ assert(nativeScan.nonEmpty, s"expected NativePaimonV2TableScanExec in
executed plan:\n$plan")
nativeScan.get
}
+
+ private def paimonDataSplits(batchScan: BatchScanExec): Seq[DataSplit] = {
+ batchScan.scan.toBatch.planInputPartitions().toSeq.flatMap { partition =>
+ assert(
+ partition.isInstanceOf[PaimonInputPartition],
+ s"expected Paimon input partition, got $partition")
+ partition
+ .asInstanceOf[PaimonInputPartition]
+ .splits
+ .collect { case split: DataSplit => split }
+ }
+ }
}
