You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by "zhuxiangyi (via GitHub)" <gi...@apache.org> on 2026/06/08 03:44:33 UTC
[PR] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
zhuxiangyi opened a new pull request, #2313:
URL: https://github.com/apache/auron/pull/2313
# Which issue does this PR close?
N/A — extends the Paimon native scan support to DSv2 catalogs. Happy to file a tracking issue if preferred.
# Rationale for this change
`PaimonConvertProvider` only matches `HiveTableScanExec` today, so Paimon tables registered through a DSv2 catalog (e.g. `spark.sql.catalog.paimon = org.apache.paimon.spark.SparkCatalog`, accessed as `paimon.db.t`) fall back to Spark execution even when the underlying layout is a raw-convertible COW Parquet/ORC dataset that Auron can read natively.
This PR adds a parallel DSv2 path that reuses the existing native file scan infrastructure (the same protobufs used by `NativeIcebergTableScanExec` / native FileSourceScan) so DSv2 Paimon scans get the same native acceleration as the Hive-registered path.
# What changes are included in this PR?
- `PaimonScanSupport` — entry point that:
- detects Paimon DSv2 scans via class-name reflection (`org.apache.paimon.spark.PaimonBaseScan` / `PaimonInputPartition`), so the module does not pull a hard Maven dependency on a particular `paimon-spark-*` artifact at compile time;
- resolves the underlying `FileStoreTable`, verifies COW + supported file format (Parquet/ORC) + supported column types;
- collects `DataSplit`s from input partitions and rejects splits where `!rawConvertible()` or deletion files are present (MOR/MOW with positional / equality deletes);
- reconstructs per-split partition `InternalRow` via `RowDataToObjectArrayConverter` so partition columns land in Auron's `(fileSchema ++ partitionSchema)` layout.
- `NativePaimonV2TableScanExec` — leaf native exec built on the existing `NativeRDD` / `NativeSupports` machinery. Mirrors `NativeIcebergTableScanExec`:
- splits files by `filesMaxPartitionBytes` (both Parquet and ORC are splittable here);
- dispatches to `pb.ParquetScanExecNode` / `pb.OrcScanExecNode`;
- exposes `numPartitions` / `numFiles` driver metrics and forwards `bytes_scanned` / `output_rows` to `TaskMetrics.inputMetrics`;
- returns `EmptyNativeRDD` when there are no partitions to read.
- `PaimonConvertProvider` — adds a `BatchScanExec` branch dispatching to the new path. Reuses the existing `spark.auron.enable.paimon.scan` config so there is no new feature flag.
The Hive-registered Paimon path (`NativePaimonTableScanExec`) is untouched.
# Are there any user-facing changes?
Yes — purely additive: when `spark.auron.enable.paimon.scan=true` (default), Paimon DSv2 catalog reads on COW Parquet/ORC tables that pass the safety checks above will now be executed natively. Tables that are MOR/MOW or use deletion files continue to fall back to Spark execution (logged at DEBUG with the reason).
No new configuration knobs are introduced.
# How was this patch tested?
New integration suite `AuronPaimonV2IntegrationSuite` (extends `SharedSparkSession`, wires `AuronSparkSessionExtension` + `PaimonSparkSessionExtensions`, uses a temp-dir warehouse via the `paimon` DSv2 catalog) covers:
- simple COW Parquet select — correctness + `NativePaimonV2TableScan` in executed plan;
- column projection;
- partitioned COW table with predicate (asserts native scan still applied);
- ORC COW table (`file.format=orc`);
- empty table (asserts `EmptyNativeRDD` path returns no rows);
- driver metrics (`numPartitions` / `numFiles`) actually posted via `SparkListenerDriverAccumUpdates`;
- `spark.auron.enable.paimon.scan=false` falls back to Spark;
- MOR primary-key table falls back to Spark (rawConvertible check).
The existing Paimon CI workflow (`.github/workflows/paimon.yml`, matrix `paimon-1.2 / spark-3.5 / scala-2.12`) picks up the new suite automatically; `paimon-spark-${shortSparkVersion}:${paimonVersion}` is added as a test-scope dependency (uses the shaded `paimon-spark-3.5:1.2.0` bundle published on Maven Central).
`mvn test-compile -pl thirdparty/auron-paimon -am -Pscala-2.12 -Ppaimon-1.2 -Pspark-3.5 -Prelease` builds cleanly with scalastyle passing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas merged PR #2313:
URL: https://github.com/apache/auron/pull/2313
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on code in PR #2313:
URL: https://github.com/apache/auron/pull/2313#discussion_r3434220315
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
Review Comment:
Good catch — renamed to `paimon v2 native scan runs simple append-only select` in a5bebf55. The table has no primary key, so it is indeed an append-only table rather than a primary-key COW table.
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
+ withTable("paimon.db.t1") {
+ sql("create table paimon.db.t1 (id int, v string) using paimon")
+ sql("insert into paimon.db.t1 values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t1")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports projection") {
+ withTable("paimon.db.t_proj") {
+ sql("create table paimon.db.t_proj (id int, v string) using paimon")
+ sql("insert into paimon.db.t_proj values (1, 'a'), (2, 'b')")
+ val df = sql("select id from paimon.db.t_proj")
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports partitioned table with predicate") {
+ withTable("paimon.db.t_part") {
+ sql("""
+ |create table paimon.db.t_part (id int, v string, p string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_part values (1, 'a', 'p1'), (2, 'b', 'p2')")
+ val df = sql("select * from paimon.db.t_part where p = 'p1'")
+ checkAnswer(df, Seq(Row(1, "a", "p1")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports ORC COW table") {
+ withTable("paimon.db.t_orc") {
+ sql("""
+ |create table paimon.db.t_orc (id int, v string)
+ |using paimon
+ |tblproperties ('file.format' = 'orc')
+ |""".stripMargin)
+ sql("insert into paimon.db.t_orc values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_orc")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan handles empty table") {
+ withTable("paimon.db.t_empty") {
+ sql("create table paimon.db.t_empty (id int, v string) using paimon")
+ val df = sql("select * from paimon.db.t_empty")
+ checkAnswer(df, Seq.empty)
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 scan exposes file scan driver metrics") {
+ withTable("paimon.db.t_metrics") {
+ sql("create table paimon.db.t_metrics (id int, v string) using paimon")
+ sql("insert into paimon.db.t_metrics values (1, 'a')")
+ withSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val df = sql("select * from paimon.db.t_metrics")
+ val nativeScan = executedNativeScan(df)
+ val metricIds = Map(
+ "numPartitions" -> nativeScan.metrics("numPartitions").id,
+ "numFiles" -> nativeScan.metrics("numFiles").id)
+ val driverMetricUpdates = new ConcurrentLinkedQueue[(Long, Long)]()
+ val driverMetricUpdatesPosted = new CountDownLatch(1)
+ val listener = new SparkListener {
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case SparkListenerDriverAccumUpdates(_, updates) =>
+ updates.foreach { case (metricId, value) =>
+ driverMetricUpdates.add(metricId -> value)
+ }
+ val updatedMetricIds = driverMetricUpdates.iterator().asScala.map(_._1).toSet
+ if (metricIds.values.forall(updatedMetricIds.contains)) {
+ driverMetricUpdatesPosted.countDown()
+ }
+ case _ =>
+ }
+ }
+
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ checkAnswer(df, Seq(Row(1, "a")))
+ assert(driverMetricUpdatesPosted.await(30, TimeUnit.SECONDS))
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+
+ val driverMetricValues = driverMetricUpdates
+ .iterator()
+ .asScala
+ .toSeq
+ .groupBy(_._1)
+ .mapValues(_.map(_._2).sum)
+ .toMap
+ assert(driverMetricValues.getOrElse(metricIds("numPartitions"), 0L) > 0)
+ assert(driverMetricValues.getOrElse(metricIds("numFiles"), 0L) > 0)
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back when spark.auron.enable.paimon.scan=false") {
+ withTable("paimon.db.t_disable") {
+ sql("create table paimon.db.t_disable (id int, v string) using paimon")
+ sql("insert into paimon.db.t_disable values (1, 'a')")
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ val df = sql("select * from paimon.db.t_disable")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back for MOR (merge-on-read) table") {
Review Comment:
Added `paimon v2 native scan supports COW primary-key table` in a5bebf55, which asserts both that `NativePaimonV2TableScan` is in the plan and `checkAnswer`. It is the positive counterpart to this MOR fallback test: a primary-key table with `full-compaction.delta-commits = 1` is COW and goes native, whereas the MOR table here (primary key without that option) falls back. This fills the previously-missing coverage for the primary-key COW -> native path.
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
+ withTable("paimon.db.t1") {
+ sql("create table paimon.db.t1 (id int, v string) using paimon")
+ sql("insert into paimon.db.t1 values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t1")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports projection") {
+ withTable("paimon.db.t_proj") {
+ sql("create table paimon.db.t_proj (id int, v string) using paimon")
+ sql("insert into paimon.db.t_proj values (1, 'a'), (2, 'b')")
+ val df = sql("select id from paimon.db.t_proj")
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports partitioned table with predicate") {
+ withTable("paimon.db.t_part") {
+ sql("""
+ |create table paimon.db.t_part (id int, v string, p string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_part values (1, 'a', 'p1'), (2, 'b', 'p2')")
+ val df = sql("select * from paimon.db.t_part where p = 'p1'")
+ checkAnswer(df, Seq(Row(1, "a", "p1")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports ORC COW table") {
+ withTable("paimon.db.t_orc") {
+ sql("""
+ |create table paimon.db.t_orc (id int, v string)
+ |using paimon
+ |tblproperties ('file.format' = 'orc')
+ |""".stripMargin)
+ sql("insert into paimon.db.t_orc values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_orc")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan handles empty table") {
+ withTable("paimon.db.t_empty") {
+ sql("create table paimon.db.t_empty (id int, v string) using paimon")
+ val df = sql("select * from paimon.db.t_empty")
+ checkAnswer(df, Seq.empty)
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 scan exposes file scan driver metrics") {
+ withTable("paimon.db.t_metrics") {
+ sql("create table paimon.db.t_metrics (id int, v string) using paimon")
+ sql("insert into paimon.db.t_metrics values (1, 'a')")
+ withSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val df = sql("select * from paimon.db.t_metrics")
+ val nativeScan = executedNativeScan(df)
+ val metricIds = Map(
+ "numPartitions" -> nativeScan.metrics("numPartitions").id,
+ "numFiles" -> nativeScan.metrics("numFiles").id)
+ val driverMetricUpdates = new ConcurrentLinkedQueue[(Long, Long)]()
+ val driverMetricUpdatesPosted = new CountDownLatch(1)
+ val listener = new SparkListener {
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case SparkListenerDriverAccumUpdates(_, updates) =>
+ updates.foreach { case (metricId, value) =>
+ driverMetricUpdates.add(metricId -> value)
+ }
+ val updatedMetricIds = driverMetricUpdates.iterator().asScala.map(_._1).toSet
+ if (metricIds.values.forall(updatedMetricIds.contains)) {
+ driverMetricUpdatesPosted.countDown()
+ }
+ case _ =>
+ }
+ }
+
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ checkAnswer(df, Seq(Row(1, "a")))
+ assert(driverMetricUpdatesPosted.await(30, TimeUnit.SECONDS))
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+
+ val driverMetricValues = driverMetricUpdates
+ .iterator()
+ .asScala
+ .toSeq
+ .groupBy(_._1)
+ .mapValues(_.map(_._2).sum)
+ .toMap
+ assert(driverMetricValues.getOrElse(metricIds("numPartitions"), 0L) > 0)
+ assert(driverMetricValues.getOrElse(metricIds("numFiles"), 0L) > 0)
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back when spark.auron.enable.paimon.scan=false") {
+ withTable("paimon.db.t_disable") {
+ sql("create table paimon.db.t_disable (id int, v string) using paimon")
+ sql("insert into paimon.db.t_disable values (1, 'a')")
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ val df = sql("select * from paimon.db.t_disable")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back for MOR (merge-on-read) table") {
+ withTable("paimon.db.t_mor") {
+ sql("""
+ |create table paimon.db.t_mor (id int, v string)
+ |using paimon
+ |tblproperties (
+ | 'primary-key' = 'id',
+ | 'bucket' = '2'
+ |)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_mor values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_mor")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
Review Comment:
Done in a5bebf55 — replaced `df.collect()` with `checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))` so the Spark fallback path is verified to return correct results, not just that it does not use the native scan.
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
+ withTable("paimon.db.t1") {
+ sql("create table paimon.db.t1 (id int, v string) using paimon")
+ sql("insert into paimon.db.t1 values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t1")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports projection") {
+ withTable("paimon.db.t_proj") {
+ sql("create table paimon.db.t_proj (id int, v string) using paimon")
+ sql("insert into paimon.db.t_proj values (1, 'a'), (2, 'b')")
+ val df = sql("select id from paimon.db.t_proj")
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports partitioned table with predicate") {
+ withTable("paimon.db.t_part") {
+ sql("""
+ |create table paimon.db.t_part (id int, v string, p string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_part values (1, 'a', 'p1'), (2, 'b', 'p2')")
+ val df = sql("select * from paimon.db.t_part where p = 'p1'")
+ checkAnswer(df, Seq(Row(1, "a", "p1")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports ORC COW table") {
+ withTable("paimon.db.t_orc") {
+ sql("""
+ |create table paimon.db.t_orc (id int, v string)
+ |using paimon
+ |tblproperties ('file.format' = 'orc')
+ |""".stripMargin)
+ sql("insert into paimon.db.t_orc values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_orc")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan handles empty table") {
+ withTable("paimon.db.t_empty") {
+ sql("create table paimon.db.t_empty (id int, v string) using paimon")
+ val df = sql("select * from paimon.db.t_empty")
+ checkAnswer(df, Seq.empty)
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 scan exposes file scan driver metrics") {
+ withTable("paimon.db.t_metrics") {
+ sql("create table paimon.db.t_metrics (id int, v string) using paimon")
+ sql("insert into paimon.db.t_metrics values (1, 'a')")
+ withSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val df = sql("select * from paimon.db.t_metrics")
+ val nativeScan = executedNativeScan(df)
+ val metricIds = Map(
+ "numPartitions" -> nativeScan.metrics("numPartitions").id,
+ "numFiles" -> nativeScan.metrics("numFiles").id)
+ val driverMetricUpdates = new ConcurrentLinkedQueue[(Long, Long)]()
+ val driverMetricUpdatesPosted = new CountDownLatch(1)
+ val listener = new SparkListener {
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case SparkListenerDriverAccumUpdates(_, updates) =>
+ updates.foreach { case (metricId, value) =>
+ driverMetricUpdates.add(metricId -> value)
+ }
+ val updatedMetricIds = driverMetricUpdates.iterator().asScala.map(_._1).toSet
+ if (metricIds.values.forall(updatedMetricIds.contains)) {
+ driverMetricUpdatesPosted.countDown()
+ }
+ case _ =>
+ }
+ }
+
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ checkAnswer(df, Seq(Row(1, "a")))
+ assert(driverMetricUpdatesPosted.await(30, TimeUnit.SECONDS))
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+
+ val driverMetricValues = driverMetricUpdates
+ .iterator()
+ .asScala
+ .toSeq
+ .groupBy(_._1)
+ .mapValues(_.map(_._2).sum)
+ .toMap
+ assert(driverMetricValues.getOrElse(metricIds("numPartitions"), 0L) > 0)
+ assert(driverMetricValues.getOrElse(metricIds("numFiles"), 0L) > 0)
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back when spark.auron.enable.paimon.scan=false") {
+ withTable("paimon.db.t_disable") {
+ sql("create table paimon.db.t_disable (id int, v string) using paimon")
+ sql("insert into paimon.db.t_disable values (1, 'a')")
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ val df = sql("select * from paimon.db.t_disable")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
Review Comment:
Done in a5bebf55 — same change applied here: the disabled-scan fallback test now also asserts `checkAnswer(df, Seq(Row(1, "a")))` to confirm the Spark fallback returns correct results.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.commons.lang3.reflect.MethodUtils
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ // 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
+ // memoize the result on the node to avoid doubling planning latency and to guarantee both
+ // calls observe the same splits even if Paimon's split planning is non-deterministic.
+ private val scanPlanTag = TreeNodeTag[Option[PaimonScanPlan]]("auron.paimon.scanPlan")
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ exec.getTagValue(scanPlanTag) match {
+ case Some(cached) => cached
+ case None =>
+ val result = computePlan(exec)
+ exec.setTagValue(scanPlanTag, result)
+ result
+ }
+ }
+
+ private def computePlan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
Review Comment:
Fixed in 1f8da4c3. Rather than hand-rolling per-type `Literal` construction, I switched the partition-value reconstruction from `Cast(Literal(value.toString), dataType)` to Paimon's own `DataConverter.fromPaimon(value, paimonType)`, which produces the correct Spark catalyst value per type (dates as epoch days, timestamps as micros, decimals/binary as-is). The previous string round-trip silently produced `null` for non-string partition columns — e.g. a DATE is stored as an epoch-day int whose `toString` ("19358") cannot be parsed back by `Cast(_, DateType)`. Added an integration test `paimon v2 native scan supports non-string partition columns` covering date + int partition columns. This required promoting `paimon-spark` from `test` to `provided` scope so `DataConverter` is available at compile time (it is always present at runtime wherever a Paimon DSv2 scan exists).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2313:
URL: https://github.com/apache/auron/pull/2313#discussion_r3371568352
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativePaimonV2TableScanExec.scala:
##########
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.auron.plan
+
+import java.net.URI
+import java.security.PrivilegedExceptionAction
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.fs.FileSystem
+import org.apache.spark.Partition
+import org.apache.spark.TaskContext
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.auron.{EmptyNativeRDD, NativeConverters, NativeHelper, NativeRDD, NativeSupports, Shims}
+import org.apache.spark.sql.auron.paimon.{PaimonFile, PaimonScanPlan}
+import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.hive.auron.paimon.PaimonUtil
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.util.SerializableConfiguration
+
+import org.apache.auron.{protobuf => pb}
+import org.apache.auron.jni.JniBridge
+import org.apache.auron.metric.SparkMetricNode
+
+case class NativePaimonV2TableScanExec(basedScan: BatchScanExec, plan: PaimonScanPlan)
+ extends LeafExecNode
+ with NativeSupports
+ with Logging {
+
+ override lazy val metrics: Map[String, SQLMetric] =
+ NativeHelper.getNativeFileScanMetrics(sparkContext) ++ Seq(
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "Native.partitions_read"),
+ "numFiles" -> SQLMetrics.createMetric(sparkContext, "Native.files_read"))
+
+ override val output = basedScan.output
+ override val outputPartitioning = basedScan.outputPartitioning
+
+ private lazy val fileSchema: StructType = plan.fileSchema
+ private lazy val partitionSchema: StructType = plan.partitionSchema
+ private lazy val files: Seq[PaimonFile] = plan.files
+
+ private lazy val partitions: Array[FilePartition] = {
+ val filePartitions = buildFilePartitions()
+ postDriverMetrics(filePartitions)
+ filePartitions
+ }
+ private lazy val fileSizes: Map[String, Long] =
+ files.map(f => f.filePath -> f.fileSize).toMap
+
+ private lazy val nativeFileSchema: pb.Schema = NativeConverters.convertSchema(fileSchema)
+ private lazy val nativePartitionSchema: pb.Schema =
+ NativeConverters.convertSchema(partitionSchema)
+
+ // Project the read schema onto the (fileSchema ++ partitionSchema) layout used by the native scan.
+ private lazy val projection: Seq[Integer] = {
+ val combined = StructType(fileSchema.fields ++ partitionSchema.fields)
+ plan.readSchema.fields.map(f => Integer.valueOf(combined.fieldIndex(f.name)))
+ }
Review Comment:
`projection` is derived via `combined.fieldIndex(...)`, which is case-sensitive and can throw when Spark is running with case-insensitive analysis and the attribute name casing differs from the `StructType` field names. Consider resolving field indexes using Spark's resolver (like NativeIcebergTableScanExec does) and building the projection from `basedScan.output` to match the actual output attributes.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,277 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ if (!PaimonUtil.isPaimonCowTable(table)) {
+ logDebug("Skip native Paimon scan: only Paimon COW tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec)
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
Review Comment:
`inputPartitions(exec)` returns `Seq.empty` both when the table is genuinely empty and when partition planning fails (e.g., `toBatch.planInputPartitions()` throws, reflection can't find a method, or the reflective return type isn't a `Seq`). In `plan`, an empty `partitions` value is treated as "empty table" and returns `Some(PaimonScanPlan(..., files=Seq.empty))`, which would make the scan return *no rows* even for a non-empty table if partition planning failed. This should distinguish "no partitions" from "failed to obtain partitions" and fall back to Spark execution on failure (return `None`).
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
+ withTable("paimon.db.t1") {
+ sql("create table paimon.db.t1 (id int, v string) using paimon")
+ sql("insert into paimon.db.t1 values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t1")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports projection") {
+ withTable("paimon.db.t_proj") {
+ sql("create table paimon.db.t_proj (id int, v string) using paimon")
+ sql("insert into paimon.db.t_proj values (1, 'a'), (2, 'b')")
+ val df = sql("select id from paimon.db.t_proj")
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports partitioned table with predicate") {
+ withTable("paimon.db.t_part") {
+ sql("""
+ |create table paimon.db.t_part (id int, v string, p string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_part values (1, 'a', 'p1'), (2, 'b', 'p2')")
+ val df = sql("select * from paimon.db.t_part where p = 'p1'")
+ checkAnswer(df, Seq(Row(1, "a", "p1")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports ORC COW table") {
+ withTable("paimon.db.t_orc") {
+ sql("""
+ |create table paimon.db.t_orc (id int, v string)
+ |using paimon
+ |tblproperties ('file.format' = 'orc')
+ |""".stripMargin)
+ sql("insert into paimon.db.t_orc values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_orc")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan handles empty table") {
+ withTable("paimon.db.t_empty") {
+ sql("create table paimon.db.t_empty (id int, v string) using paimon")
+ val df = sql("select * from paimon.db.t_empty")
+ checkAnswer(df, Seq.empty)
+ }
Review Comment:
The "empty table" test currently only checks that the result is empty, but it doesn't assert that the native v2 scan was actually applied. Without asserting the executed plan contains `NativePaimonV2TableScan`, this test won't catch regressions where empty-table scans silently fall back to Spark.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on code in PR #2313:
URL: https://github.com/apache/auron/pull/2313#discussion_r3425945351
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/hive/auron/paimon/PaimonConvertProvider.scala:
##########
@@ -43,13 +48,16 @@ class PaimonConvertProvider extends AuronConvertProvider with Logging {
if e.relation.tableMeta.storage.serde.isDefined
&& e.relation.tableMeta.storage.serde.get.contains("Paimon") =>
true
Review Comment:
Fixed in `cb50eb17`: `plan()` now memoizes its result on the `BatchScanExec` via a `TreeNodeTag`, so `isSupported()` and `convert()` share a single split-planning pass and always observe the same splits. The second call is just a tag lookup with no extra I/O.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
+ case _ => return None
+ }
+ splits.foreach {
+ case ds: DataSplit => buf += ds
+ case _: Split => return None
+ case _ => return None
+ }
+ }
+ Some(buf.toSeq)
+ }
+
+ private def paimonTable(scan: AnyRef): Option[FileStoreTable] = {
+ invokeMethod(scan, "table") match {
+ case Some(t: FileStoreTable) => Some(t)
+ case Some(other) =>
+ logDebug(s"Unexpected Paimon table type: ${other.getClass.getName}")
+ None
+ case None => None
+ }
+ }
+
+ // DSv2 BatchScanExec exposes input partitions via Scan.toBatch (preferred) or a method on
+ // the exec itself; the latter varies across Spark versions, so we attempt both.
+ // Returns Some(partitions) on success (possibly empty if the table is empty), or None when
+ // partition planning fails - the caller falls back to Spark execution on None.
+ private def inputPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = {
+ try {
+ val batch = exec.scan.toBatch
+ if (batch != null) {
+ val parts = batch.planInputPartitions()
+ if (parts != null) return Some(parts.toSeq)
+ logWarning("Paimon Scan.toBatch.planInputPartitions() returned null.")
+ return None
+ }
+ logWarning("Paimon Scan.toBatch returned null.")
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to plan Paimon input partitions via Scan.toBatch.", t)
+ return None
+ }
+
+ val methods = exec.getClass.getMethods
+ val m =
+ methods.find(_.getName == "inputPartitions").orElse(methods.find(_.getName == "partitions"))
+ if (m.isEmpty) {
+ logWarning(
+ "BatchScanExec exposes no inputPartitions/partitions method; cannot plan Paimon scan.")
+ return None
+ }
+ try {
+ m.map(_.invoke(exec)) match {
+ case Some(s: scala.collection.Seq[_])
+ if s.nonEmpty && s.head.isInstanceOf[scala.collection.Seq[_]] =>
+ Some(
+ s.asInstanceOf[scala.collection.Seq[scala.collection.Seq[InputPartition]]]
+ .flatten
+ .toSeq)
+ case Some(s: scala.collection.Seq[_]) =>
+ Some(s.asInstanceOf[scala.collection.Seq[InputPartition]].toSeq)
+ case other =>
+ logWarning(
+ s"Unexpected return type from BatchScanExec partitions method: ${other.getClass}.")
+ None
+ }
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to read Paimon input partitions via reflection.", t)
+ None
+ }
+ }
+
+ private def isInstanceOfClass(obj: AnyRef, className: String): Boolean = {
+ if (obj == null) return false
Review Comment:
Fixed in `cb50eb17`: replaced the manual chain walk with `obj.getClass.getClassLoader.loadClass(className).isInstance(obj)`, which walks the full superclass/super-interface hierarchy (so an intermediate interface is handled) and resolves the class from the object's own classloader. Used `loadClass` rather than `Class.forName` because the repo scalastyle rule forbids `Class.forName`.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
Review Comment:
Fixed in `cb50eb17`: `collectSplits` now also matches `java.util.Collection` and converts via `.asScala`, so a future Java-returning `splits()` won't silently disable the native scan.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
+ case _ => return None
+ }
+ splits.foreach {
+ case ds: DataSplit => buf += ds
+ case _: Split => return None
+ case _ => return None
+ }
+ }
+ Some(buf.toSeq)
+ }
+
+ private def paimonTable(scan: AnyRef): Option[FileStoreTable] = {
+ invokeMethod(scan, "table") match {
+ case Some(t: FileStoreTable) => Some(t)
+ case Some(other) =>
+ logDebug(s"Unexpected Paimon table type: ${other.getClass.getName}")
+ None
+ case None => None
+ }
+ }
+
+ // DSv2 BatchScanExec exposes input partitions via Scan.toBatch (preferred) or a method on
+ // the exec itself; the latter varies across Spark versions, so we attempt both.
+ // Returns Some(partitions) on success (possibly empty if the table is empty), or None when
+ // partition planning fails - the caller falls back to Spark execution on None.
+ private def inputPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = {
+ try {
+ val batch = exec.scan.toBatch
+ if (batch != null) {
+ val parts = batch.planInputPartitions()
+ if (parts != null) return Some(parts.toSeq)
+ logWarning("Paimon Scan.toBatch.planInputPartitions() returned null.")
+ return None
+ }
+ logWarning("Paimon Scan.toBatch returned null.")
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to plan Paimon input partitions via Scan.toBatch.", t)
+ return None
+ }
+
+ val methods = exec.getClass.getMethods
+ val m =
+ methods.find(_.getName == "inputPartitions").orElse(methods.find(_.getName == "partitions"))
+ if (m.isEmpty) {
+ logWarning(
+ "BatchScanExec exposes no inputPartitions/partitions method; cannot plan Paimon scan.")
+ return None
+ }
+ try {
+ m.map(_.invoke(exec)) match {
+ case Some(s: scala.collection.Seq[_])
+ if s.nonEmpty && s.head.isInstanceOf[scala.collection.Seq[_]] =>
+ Some(
+ s.asInstanceOf[scala.collection.Seq[scala.collection.Seq[InputPartition]]]
+ .flatten
+ .toSeq)
+ case Some(s: scala.collection.Seq[_]) =>
+ Some(s.asInstanceOf[scala.collection.Seq[InputPartition]].toSeq)
+ case other =>
+ logWarning(
+ s"Unexpected return type from BatchScanExec partitions method: ${other.getClass}.")
+ None
+ }
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to read Paimon input partitions via reflection.", t)
+ None
+ }
+ }
+
+ private def isInstanceOfClass(obj: AnyRef, className: String): Boolean = {
+ if (obj == null) return false
+ var c: Class[_] = obj.getClass
+ while (c != null) {
+ if (c.getName == className) return true
+ c.getInterfaces.foreach { i =>
+ if (i.getName == className) return true
+ }
+ c = c.getSuperclass
+ }
+ false
+ }
+
+ private def invokeMethod(target: AnyRef, methodName: String): Option[Any] = {
+ try {
Review Comment:
Fixed in `2358c551`: `invokeMethod` now delegates to commons-lang3 `MethodUtils.invokeMethod(target, true, name)` (already on the classpath and used in `AuronConvertStrategy`), which resolves the full hierarchy including interface default methods and removes the hand-rolled reflection.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on PR #2313:
URL: https://github.com/apache/auron/pull/2313#issuecomment-4726186293
@SteNicholas thanks for the detailed review! Here is how each comment was handled.
**Addressed in this PR:**
- **Double `plan()` call / split-planning I/O** (`PaimonConvertProvider`): `PaimonScanSupport.plan()` now memoizes its result on the `BatchScanExec` via a `TreeNodeTag`, so the `isSupported()` and `convert()` calls share a single split-planning pass and always observe the same splits. (commit `cb50eb17`)
- **`isInstanceOfClass` missing super-interfaces**: now loads the class from the object's own classloader and uses `isInstance`, which walks the full superclass/super-interface hierarchy. (commit `cb50eb17`)
- **`splits()` returning `java.util.List`**: `collectSplits` now also matches `java.util.Collection` and converts via `.asScala`, so a future Java-returning API won't silently disable the native scan. (commit `cb50eb17`)
- **`invokeMethod` hand-rolled reflection**: replaced with `commons-lang3` `MethodUtils.invokeMethod` (already on the classpath and used elsewhere in auron), which resolves the full hierarchy including interface default methods. (commit `2358c551`)
**Proposed as a follow-up refactor PR** (cross-module dedup with Iceberg/core, orthogonal to this DSv2 feature and would touch already-stable scan paths):
- Extracting a shared `BatchScanInputPartitions.resolve(exec)` utility (Paimon's variant intentionally returns `Option` to distinguish empty-table from planning-failure, while Iceberg's returns `Seq`, so a clean merge changes Iceberg behavior and needs its own validation).
- Extracting `putJniBridgeResource` + `broadcastedHadoopConf` into a shared trait (currently duplicated across `NativeFileSourceScanBase`, `NativeHiveTableScanBase`, `NativeIcebergTableScanExec` and this exec).
**Left as-is:** the case-insensitive `fieldIndexByName` collision — as you noted it mirrors the existing `NativeIcebergTableScanExec` design, so I'd prefer to keep it consistent and address both together if we change it.
PTAL, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #2313:
URL: https://github.com/apache/auron/pull/2313#discussion_r3413352594
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/hive/auron/paimon/PaimonConvertProvider.scala:
##########
@@ -43,13 +48,16 @@ class PaimonConvertProvider extends AuronConvertProvider with Logging {
if e.relation.tableMeta.storage.serde.isDefined
&& e.relation.tableMeta.storage.serde.get.contains("Paimon") =>
true
Review Comment:
`PaimonScanSupport.plan()` is called twice per query — once here in `isSupported()` and again in `convertPaimonV2BatchScanExec()`. Both calls go through `exec.scan.toBatch.planInputPartitions()` which performs Paimon split-planning I/O (scanning metadata files). For tables with many splits this doubles planning latency.
Additionally, if Paimon's split planning is non-deterministic (e.g., concurrent compaction changes available splits between calls), the plan used in `convert()` could differ from what `isSupported()` validated.
Consider caching the plan result (e.g., via a `TreeNodeTag` on the `BatchScanExec`, or a `WeakHashMap[BatchScanExec, Option[PaimonScanPlan]]`) so the second call is free.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
Review Comment:
`invokeMethod(p, "splits")` returns the result of reflective invocation and the match expects `scala.collection.Seq[_]`. If a future Paimon version changes `PaimonInputPartition.splits()` to return `java.util.List[Split]` (common in Scala-to-Java API evolution), the pattern match silently fails and `collectSplits` returns `None`, disabling native scan with only a DEBUG log.
Consider also matching `java.util.Collection` and converting via `.asScala`, or using `MethodUtils.invokeMethod` from commons-lang3 (already on the classpath, used in `AuronConvertStrategy`).
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativePaimonV2TableScanExec.scala:
##########
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.auron.plan
+
+import java.net.URI
+import java.security.PrivilegedExceptionAction
+import java.util.Locale
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.fs.FileSystem
+import org.apache.spark.Partition
+import org.apache.spark.TaskContext
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.auron.{EmptyNativeRDD, NativeConverters, NativeHelper, NativeRDD, NativeSupports, Shims}
+import org.apache.spark.sql.auron.paimon.{PaimonFile, PaimonScanPlan}
+import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+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.util.SerializableConfiguration
+
+import org.apache.auron.{protobuf => pb}
+import org.apache.auron.jni.JniBridge
+import org.apache.auron.metric.SparkMetricNode
+
+case class NativePaimonV2TableScanExec(basedScan: BatchScanExec, plan: PaimonScanPlan)
+ extends LeafExecNode
+ with NativeSupports
+ with Logging {
+
+ override lazy val metrics: Map[String, SQLMetric] =
+ NativeHelper.getNativeFileScanMetrics(sparkContext) ++ Seq(
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "Native.partitions_read"),
+ "numFiles" -> SQLMetrics.createMetric(sparkContext, "Native.files_read"))
+
+ override val output = basedScan.output
+ override val outputPartitioning = basedScan.outputPartitioning
+
+ private lazy val fileSchema: StructType = plan.fileSchema
+ private lazy val partitionSchema: StructType = plan.partitionSchema
+ private lazy val files: Seq[PaimonFile] = plan.files
+
+ private lazy val partitions: Array[FilePartition] = {
+ val filePartitions = buildFilePartitions()
+ postDriverMetrics(filePartitions)
+ filePartitions
+ }
+ private lazy val fileSizes: Map[String, Long] =
+ files.map(f => f.filePath -> f.fileSize).toMap
+
+ private lazy val nativeFileSchema: pb.Schema = NativeConverters.convertSchema(fileSchema)
+ private lazy val nativePartitionSchema: pb.Schema =
+ NativeConverters.convertSchema(partitionSchema)
+
+ // Project the output attributes onto the (fileSchema ++ partitionSchema) layout used by the
+ // native scan. Index lookup follows SQLConf.caseSensitiveAnalysis so that the projection
+ // remains correct under case-insensitive analysis (mirrors NativeIcebergTableScanExec).
+ private lazy val combinedSchema: StructType =
+ StructType(fileSchema.fields ++ partitionSchema.fields)
+
+ private lazy val caseSensitive: Boolean = SQLConf.get.caseSensitiveAnalysis
+
+ private lazy val fieldIndexByName: Map[String, Int] = {
+ if (caseSensitive) {
+ combinedSchema.fieldNames.zipWithIndex.toMap
+ } else {
+ combinedSchema.fieldNames.map(_.toLowerCase(Locale.ROOT)).zipWithIndex.toMap
+ }
Review Comment:
`fieldIndexByName` in case-insensitive mode maps `toLowerCase` field names to indices. If `fileSchema` and `partitionSchema` contain fields that differ only in case (e.g., data column `Status` and partition column `status`), both lowercase to the same key and the `Map` silently drops one entry, causing `fieldIndexFor` to return the wrong index and the projection to read the wrong column.
This is the same pattern as `NativeIcebergTableScanExec`, so it's a pre-existing design choice — but worth noting as it can produce wrong results with mixed-case schemas.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativePaimonV2TableScanExec.scala:
##########
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.auron.plan
+
+import java.net.URI
+import java.security.PrivilegedExceptionAction
+import java.util.Locale
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.fs.FileSystem
+import org.apache.spark.Partition
+import org.apache.spark.TaskContext
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.auron.{EmptyNativeRDD, NativeConverters, NativeHelper, NativeRDD, NativeSupports, Shims}
+import org.apache.spark.sql.auron.paimon.{PaimonFile, PaimonScanPlan}
+import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+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.util.SerializableConfiguration
+
+import org.apache.auron.{protobuf => pb}
+import org.apache.auron.jni.JniBridge
+import org.apache.auron.metric.SparkMetricNode
+
+case class NativePaimonV2TableScanExec(basedScan: BatchScanExec, plan: PaimonScanPlan)
+ extends LeafExecNode
+ with NativeSupports
+ with Logging {
+
+ override lazy val metrics: Map[String, SQLMetric] =
+ NativeHelper.getNativeFileScanMetrics(sparkContext) ++ Seq(
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "Native.partitions_read"),
+ "numFiles" -> SQLMetrics.createMetric(sparkContext, "Native.files_read"))
+
+ override val output = basedScan.output
+ override val outputPartitioning = basedScan.outputPartitioning
+
+ private lazy val fileSchema: StructType = plan.fileSchema
+ private lazy val partitionSchema: StructType = plan.partitionSchema
+ private lazy val files: Seq[PaimonFile] = plan.files
+
+ private lazy val partitions: Array[FilePartition] = {
+ val filePartitions = buildFilePartitions()
+ postDriverMetrics(filePartitions)
+ filePartitions
+ }
+ private lazy val fileSizes: Map[String, Long] =
+ files.map(f => f.filePath -> f.fileSize).toMap
+
+ private lazy val nativeFileSchema: pb.Schema = NativeConverters.convertSchema(fileSchema)
+ private lazy val nativePartitionSchema: pb.Schema =
+ NativeConverters.convertSchema(partitionSchema)
+
+ // Project the output attributes onto the (fileSchema ++ partitionSchema) layout used by the
+ // native scan. Index lookup follows SQLConf.caseSensitiveAnalysis so that the projection
+ // remains correct under case-insensitive analysis (mirrors NativeIcebergTableScanExec).
+ private lazy val combinedSchema: StructType =
+ StructType(fileSchema.fields ++ partitionSchema.fields)
+
+ private lazy val caseSensitive: Boolean = SQLConf.get.caseSensitiveAnalysis
+
+ private lazy val fieldIndexByName: Map[String, Int] = {
+ if (caseSensitive) {
+ combinedSchema.fieldNames.zipWithIndex.toMap
+ } else {
+ combinedSchema.fieldNames.map(_.toLowerCase(Locale.ROOT)).zipWithIndex.toMap
+ }
+ }
+
+ private def fieldIndexFor(name: String): Int = {
+ if (caseSensitive) {
+ fieldIndexByName.getOrElse(name, combinedSchema.fieldIndex(name))
+ } else {
+ fieldIndexByName.getOrElse(name.toLowerCase(Locale.ROOT), combinedSchema.fieldIndex(name))
+ }
+ }
+
+ private lazy val projection: Seq[Integer] =
+ output.map(attr => Integer.valueOf(fieldIndexFor(attr.name)))
+
+ override def doExecuteNative(): NativeRDD = {
+ if (partitions.isEmpty) {
+ return new EmptyNativeRDD(sparkContext)
+ }
+
+ val nativeMetrics = SparkMetricNode(
+ metrics,
+ Nil,
+ Some({
+ case ("bytes_scanned", v) =>
+ val inputMetric = TaskContext.get.taskMetrics().inputMetrics
+ inputMetric.incBytesRead(v)
+ case ("output_rows", v) =>
+ val inputMetric = TaskContext.get.taskMetrics().inputMetrics
+ inputMetric.incRecordsRead(v)
+ case _ =>
+ }))
+
+ val fileFormat = plan.fileFormat
+ val broadcastedHadoopConf = this.broadcastedHadoopConf
+ val numPartitions = partitions.length
+ val nativeFileGroups = this.nativeFileGroups
+ val nativeFileSchema = this.nativeFileSchema
+ val nativePartitionSchema = this.nativePartitionSchema
+ val projection = this.projection
+
+ new NativeRDD(
+ sparkContext,
+ nativeMetrics,
+ partitions.asInstanceOf[Array[Partition]],
+ None,
+ Nil,
+ rddShuffleReadFull = true,
+ (partition, _) => {
+ val resourceId = s"NativePaimonV2TableScan:${UUID.randomUUID().toString}"
+ putJniBridgeResource(resourceId, broadcastedHadoopConf)
+
+ val nativeFileGroup = nativeFileGroups(partition.asInstanceOf[FilePartition])
+ val nativeFileScanConf = pb.FileScanExecConf
+ .newBuilder()
+ .setNumPartitions(numPartitions)
+ .setPartitionIndex(partition.index)
+ .setStatistics(pb.Statistics.getDefaultInstance)
+ .setSchema(nativeFileSchema)
+ .setFileGroup(nativeFileGroup)
+ .addAllProjection(projection.asJava)
+ .setPartitionSchema(nativePartitionSchema)
+ .build()
+
+ if (fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ val nativeOrcScanExecBuilder = pb.OrcScanExecNode
+ .newBuilder()
+ .setBaseConf(nativeFileScanConf)
+ .setFsResourceId(resourceId)
+ .addAllPruningPredicates(new java.util.ArrayList())
+
+ pb.PhysicalPlanNode
+ .newBuilder()
+ .setOrcScan(nativeOrcScanExecBuilder.build())
+ .build()
+ } else {
+ val nativeParquetScanExecBuilder = pb.ParquetScanExecNode
+ .newBuilder()
+ .setBaseConf(nativeFileScanConf)
+ .setFsResourceId(resourceId)
+ .addAllPruningPredicates(new java.util.ArrayList())
+
+ pb.PhysicalPlanNode
+ .newBuilder()
+ .setParquetScan(nativeParquetScanExecBuilder.build())
+ .build()
+ }
+ },
+ friendlyName = "NativeRDD.PaimonV2Scan")
+ }
+
+ override val nodeName: String = "NativePaimonV2TableScan"
+
+ override protected def doCanonicalize(): SparkPlan = basedScan.canonicalized
+
+ private lazy val nativeFileGroups: FilePartition => pb.FileGroup = (partition: FilePartition) =>
+ {
+ val nativePartitionedFile = (file: PartitionedFile) => {
+ val filePath = file.filePath.toString
+ val size = fileSizes.getOrElse(filePath, file.length)
+ val nativePartitionValues = partitionSchema.zipWithIndex.map { case (field, index) =>
+ NativeConverters
+ .convertExpr(
+ org.apache.spark.sql.catalyst.expressions
+ .Literal(file.partitionValues.get(index, field.dataType), field.dataType))
+ .getLiteral
+ }
+ pb.PartitionedFile
+ .newBuilder()
+ .setPath(filePath)
+ .setSize(size)
+ .setLastModifiedNs(0)
+ .addAllPartitionValues(nativePartitionValues.asJava)
+ .setRange(
+ pb.FileRange
+ .newBuilder()
+ .setStart(file.start)
+ .setEnd(file.start + file.length)
+ .build())
+ .build()
+ }
+ pb.FileGroup
+ .newBuilder()
+ .addAllFiles(partition.files.map(nativePartitionedFile).toList.asJava)
+ .build()
+ }
+
+ private def postDriverMetrics(filePartitions: Array[FilePartition]): Unit = {
+ val numPartitions = filePartitions.length
+ metrics("numPartitions").add(numPartitions)
+ val numFiles = filePartitions.foldLeft(0L)(_ + _.files.length)
+ metrics("numFiles").add(numFiles)
+ val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ SQLMetrics.postDriverMetricUpdates(
+ sparkContext,
+ executionId,
+ Seq(metrics("numPartitions"), metrics("numFiles")))
+ }
+
+ private def buildFilePartitions(): Array[FilePartition] = {
+ if (files.isEmpty) {
+ return Array.empty
+ }
+
+ val sparkSession = Shims.get.getSqlContext(basedScan).sparkSession
+ val isSplitable =
+ plan.fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) ||
+ plan.fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)
+ val maxSplitBytes = getMaxSplitBytes(sparkSession, files)
+ val partitionedFiles = files
+ .flatMap { f =>
+ if (isSplitable) {
+ (0L until f.fileSize by maxSplitBytes).map { offset =>
Review Comment:
`putJniBridgeResource` is now the 4th verbatim copy — identical code exists in `NativeIcebergTableScanExec`, `NativeFileSourceScanBase`, and `NativeHiveTableScanBase`. A Hadoop FS caching fix, Kerberos delegation-token change, or metrics-key rename would need to be replicated across all four.
Similarly, `broadcastedHadoopConf` (below) is duplicated across the same four files. Consider extracting both into a shared trait.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
+ case _ => return None
+ }
+ splits.foreach {
+ case ds: DataSplit => buf += ds
+ case _: Split => return None
+ case _ => return None
+ }
+ }
+ Some(buf.toSeq)
+ }
+
+ private def paimonTable(scan: AnyRef): Option[FileStoreTable] = {
+ invokeMethod(scan, "table") match {
+ case Some(t: FileStoreTable) => Some(t)
+ case Some(other) =>
+ logDebug(s"Unexpected Paimon table type: ${other.getClass.getName}")
+ None
+ case None => None
+ }
+ }
+
+ // DSv2 BatchScanExec exposes input partitions via Scan.toBatch (preferred) or a method on
+ // the exec itself; the latter varies across Spark versions, so we attempt both.
+ // Returns Some(partitions) on success (possibly empty if the table is empty), or None when
+ // partition planning fails - the caller falls back to Spark execution on None.
+ private def inputPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = {
+ try {
+ val batch = exec.scan.toBatch
+ if (batch != null) {
+ val parts = batch.planInputPartitions()
+ if (parts != null) return Some(parts.toSeq)
+ logWarning("Paimon Scan.toBatch.planInputPartitions() returned null.")
+ return None
+ }
+ logWarning("Paimon Scan.toBatch returned null.")
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to plan Paimon input partitions via Scan.toBatch.", t)
+ return None
+ }
+
+ val methods = exec.getClass.getMethods
+ val m =
+ methods.find(_.getName == "inputPartitions").orElse(methods.find(_.getName == "partitions"))
+ if (m.isEmpty) {
+ logWarning(
+ "BatchScanExec exposes no inputPartitions/partitions method; cannot plan Paimon scan.")
+ return None
+ }
+ try {
+ m.map(_.invoke(exec)) match {
+ case Some(s: scala.collection.Seq[_])
+ if s.nonEmpty && s.head.isInstanceOf[scala.collection.Seq[_]] =>
+ Some(
+ s.asInstanceOf[scala.collection.Seq[scala.collection.Seq[InputPartition]]]
+ .flatten
+ .toSeq)
+ case Some(s: scala.collection.Seq[_]) =>
+ Some(s.asInstanceOf[scala.collection.Seq[InputPartition]].toSeq)
+ case other =>
+ logWarning(
+ s"Unexpected return type from BatchScanExec partitions method: ${other.getClass}.")
+ None
+ }
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to read Paimon input partitions via reflection.", t)
+ None
+ }
+ }
+
+ private def isInstanceOfClass(obj: AnyRef, className: String): Boolean = {
+ if (obj == null) return false
Review Comment:
`isInstanceOfClass` walks the superclass chain and checks each class's direct interfaces, but does **not** recurse into interfaces' super-interfaces. If a future Paimon version introduces an intermediate interface (e.g., `trait PaimonPartitionAware extends PaimonInputPartition`) and the concrete class implements the intermediate rather than `PaimonInputPartition` directly, this check silently fails — `collectSplits()` returns `None` and the native scan falls back to Spark with no warning.
A simpler and fully correct replacement:
```scala
private def isInstanceOfClass(obj: AnyRef, className: String): Boolean = {
if (obj == null) return false
try { Class.forName(className).isInstance(obj) }
catch { case _: ClassNotFoundException => false }
}
```
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
+ case _ => return None
+ }
+ splits.foreach {
+ case ds: DataSplit => buf += ds
+ case _: Split => return None
+ case _ => return None
+ }
+ }
+ Some(buf.toSeq)
+ }
+
+ private def paimonTable(scan: AnyRef): Option[FileStoreTable] = {
+ invokeMethod(scan, "table") match {
+ case Some(t: FileStoreTable) => Some(t)
+ case Some(other) =>
+ logDebug(s"Unexpected Paimon table type: ${other.getClass.getName}")
+ None
+ case None => None
+ }
+ }
+
+ // DSv2 BatchScanExec exposes input partitions via Scan.toBatch (preferred) or a method on
+ // the exec itself; the latter varies across Spark versions, so we attempt both.
+ // Returns Some(partitions) on success (possibly empty if the table is empty), or None when
+ // partition planning fails - the caller falls back to Spark execution on None.
+ private def inputPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = {
+ try {
Review Comment:
`inputPartitions()` is a near-exact duplicate of `IcebergScanSupport.inputPartitions()` — both try `scan.toBatch.planInputPartitions()`, fall back to reflection on `BatchScanExec`, and handle nested `Seq[Seq[InputPartition]]` flattening.
Consider extracting a shared `BatchScanInputPartitions.resolve(exec)` utility. A Spark version change to `BatchScanExec`'s partition method signature would currently require identical fixes in both files.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
+ case _ => return None
+ }
+ splits.foreach {
+ case ds: DataSplit => buf += ds
+ case _: Split => return None
+ case _ => return None
+ }
+ }
+ Some(buf.toSeq)
+ }
+
+ private def paimonTable(scan: AnyRef): Option[FileStoreTable] = {
+ invokeMethod(scan, "table") match {
+ case Some(t: FileStoreTable) => Some(t)
+ case Some(other) =>
+ logDebug(s"Unexpected Paimon table type: ${other.getClass.getName}")
+ None
+ case None => None
+ }
+ }
+
+ // DSv2 BatchScanExec exposes input partitions via Scan.toBatch (preferred) or a method on
+ // the exec itself; the latter varies across Spark versions, so we attempt both.
+ // Returns Some(partitions) on success (possibly empty if the table is empty), or None when
+ // partition planning fails - the caller falls back to Spark execution on None.
+ private def inputPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = {
+ try {
+ val batch = exec.scan.toBatch
+ if (batch != null) {
+ val parts = batch.planInputPartitions()
+ if (parts != null) return Some(parts.toSeq)
+ logWarning("Paimon Scan.toBatch.planInputPartitions() returned null.")
+ return None
+ }
+ logWarning("Paimon Scan.toBatch returned null.")
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to plan Paimon input partitions via Scan.toBatch.", t)
+ return None
+ }
+
+ val methods = exec.getClass.getMethods
+ val m =
+ methods.find(_.getName == "inputPartitions").orElse(methods.find(_.getName == "partitions"))
+ if (m.isEmpty) {
+ logWarning(
+ "BatchScanExec exposes no inputPartitions/partitions method; cannot plan Paimon scan.")
+ return None
+ }
+ try {
+ m.map(_.invoke(exec)) match {
+ case Some(s: scala.collection.Seq[_])
+ if s.nonEmpty && s.head.isInstanceOf[scala.collection.Seq[_]] =>
+ Some(
+ s.asInstanceOf[scala.collection.Seq[scala.collection.Seq[InputPartition]]]
+ .flatten
+ .toSeq)
+ case Some(s: scala.collection.Seq[_]) =>
+ Some(s.asInstanceOf[scala.collection.Seq[InputPartition]].toSeq)
+ case other =>
+ logWarning(
+ s"Unexpected return type from BatchScanExec partitions method: ${other.getClass}.")
+ None
+ }
+ } catch {
+ case NonFatal(t) =>
+ logWarning("Failed to read Paimon input partitions via reflection.", t)
+ None
+ }
+ }
+
+ private def isInstanceOfClass(obj: AnyRef, className: String): Boolean = {
+ if (obj == null) return false
+ var c: Class[_] = obj.getClass
+ while (c != null) {
+ if (c.getName == className) return true
+ c.getInterfaces.foreach { i =>
+ if (i.getName == className) return true
+ }
+ c = c.getSuperclass
+ }
+ false
+ }
+
+ private def invokeMethod(target: AnyRef, methodName: String): Option[Any] = {
+ try {
Review Comment:
`invokeMethod` is a duplicate of `IcebergScanSupport.invokeDeclaredMethod` (identical pattern: walk superclass chain, find declared method by name, setAccessible, invoke). `MethodUtils.invokeMethod` from commons-lang3 is already on the classpath and used in `AuronConvertStrategy` — using it here would eliminate the duplication and also handle interface default methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "merrily01 (via GitHub)" <gi...@apache.org>.
merrily01 commented on code in PR #2313:
URL: https://github.com/apache/auron/pull/2313#discussion_r3432739379
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
Review Comment:
This table has no primary key, so it is append-only rather than a primary-key COW table. Would it be clearer to rename the test to something like `"append-only select"`?
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
+ withTable("paimon.db.t1") {
+ sql("create table paimon.db.t1 (id int, v string) using paimon")
+ sql("insert into paimon.db.t1 values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t1")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports projection") {
+ withTable("paimon.db.t_proj") {
+ sql("create table paimon.db.t_proj (id int, v string) using paimon")
+ sql("insert into paimon.db.t_proj values (1, 'a'), (2, 'b')")
+ val df = sql("select id from paimon.db.t_proj")
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports partitioned table with predicate") {
+ withTable("paimon.db.t_part") {
+ sql("""
+ |create table paimon.db.t_part (id int, v string, p string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_part values (1, 'a', 'p1'), (2, 'b', 'p2')")
+ val df = sql("select * from paimon.db.t_part where p = 'p1'")
+ checkAnswer(df, Seq(Row(1, "a", "p1")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports ORC COW table") {
+ withTable("paimon.db.t_orc") {
+ sql("""
+ |create table paimon.db.t_orc (id int, v string)
+ |using paimon
+ |tblproperties ('file.format' = 'orc')
+ |""".stripMargin)
+ sql("insert into paimon.db.t_orc values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_orc")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan handles empty table") {
+ withTable("paimon.db.t_empty") {
+ sql("create table paimon.db.t_empty (id int, v string) using paimon")
+ val df = sql("select * from paimon.db.t_empty")
+ checkAnswer(df, Seq.empty)
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 scan exposes file scan driver metrics") {
+ withTable("paimon.db.t_metrics") {
+ sql("create table paimon.db.t_metrics (id int, v string) using paimon")
+ sql("insert into paimon.db.t_metrics values (1, 'a')")
+ withSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val df = sql("select * from paimon.db.t_metrics")
+ val nativeScan = executedNativeScan(df)
+ val metricIds = Map(
+ "numPartitions" -> nativeScan.metrics("numPartitions").id,
+ "numFiles" -> nativeScan.metrics("numFiles").id)
+ val driverMetricUpdates = new ConcurrentLinkedQueue[(Long, Long)]()
+ val driverMetricUpdatesPosted = new CountDownLatch(1)
+ val listener = new SparkListener {
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case SparkListenerDriverAccumUpdates(_, updates) =>
+ updates.foreach { case (metricId, value) =>
+ driverMetricUpdates.add(metricId -> value)
+ }
+ val updatedMetricIds = driverMetricUpdates.iterator().asScala.map(_._1).toSet
+ if (metricIds.values.forall(updatedMetricIds.contains)) {
+ driverMetricUpdatesPosted.countDown()
+ }
+ case _ =>
+ }
+ }
+
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ checkAnswer(df, Seq(Row(1, "a")))
+ assert(driverMetricUpdatesPosted.await(30, TimeUnit.SECONDS))
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+
+ val driverMetricValues = driverMetricUpdates
+ .iterator()
+ .asScala
+ .toSeq
+ .groupBy(_._1)
+ .mapValues(_.map(_._2).sum)
+ .toMap
+ assert(driverMetricValues.getOrElse(metricIds("numPartitions"), 0L) > 0)
+ assert(driverMetricValues.getOrElse(metricIds("numFiles"), 0L) > 0)
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back when spark.auron.enable.paimon.scan=false") {
+ withTable("paimon.db.t_disable") {
+ sql("create table paimon.db.t_disable (id int, v string) using paimon")
+ sql("insert into paimon.db.t_disable values (1, 'a')")
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ val df = sql("select * from paimon.db.t_disable")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back for MOR (merge-on-read) table") {
Review Comment:
Would it make sense to add a happy-path test that checks both `NativePaimonV2TableScan` in the plan and `checkAnswer`?
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
+ withTable("paimon.db.t1") {
+ sql("create table paimon.db.t1 (id int, v string) using paimon")
+ sql("insert into paimon.db.t1 values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t1")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports projection") {
+ withTable("paimon.db.t_proj") {
+ sql("create table paimon.db.t_proj (id int, v string) using paimon")
+ sql("insert into paimon.db.t_proj values (1, 'a'), (2, 'b')")
+ val df = sql("select id from paimon.db.t_proj")
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports partitioned table with predicate") {
+ withTable("paimon.db.t_part") {
+ sql("""
+ |create table paimon.db.t_part (id int, v string, p string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_part values (1, 'a', 'p1'), (2, 'b', 'p2')")
+ val df = sql("select * from paimon.db.t_part where p = 'p1'")
+ checkAnswer(df, Seq(Row(1, "a", "p1")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports ORC COW table") {
+ withTable("paimon.db.t_orc") {
+ sql("""
+ |create table paimon.db.t_orc (id int, v string)
+ |using paimon
+ |tblproperties ('file.format' = 'orc')
+ |""".stripMargin)
+ sql("insert into paimon.db.t_orc values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_orc")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan handles empty table") {
+ withTable("paimon.db.t_empty") {
+ sql("create table paimon.db.t_empty (id int, v string) using paimon")
+ val df = sql("select * from paimon.db.t_empty")
+ checkAnswer(df, Seq.empty)
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 scan exposes file scan driver metrics") {
+ withTable("paimon.db.t_metrics") {
+ sql("create table paimon.db.t_metrics (id int, v string) using paimon")
+ sql("insert into paimon.db.t_metrics values (1, 'a')")
+ withSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val df = sql("select * from paimon.db.t_metrics")
+ val nativeScan = executedNativeScan(df)
+ val metricIds = Map(
+ "numPartitions" -> nativeScan.metrics("numPartitions").id,
+ "numFiles" -> nativeScan.metrics("numFiles").id)
+ val driverMetricUpdates = new ConcurrentLinkedQueue[(Long, Long)]()
+ val driverMetricUpdatesPosted = new CountDownLatch(1)
+ val listener = new SparkListener {
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case SparkListenerDriverAccumUpdates(_, updates) =>
+ updates.foreach { case (metricId, value) =>
+ driverMetricUpdates.add(metricId -> value)
+ }
+ val updatedMetricIds = driverMetricUpdates.iterator().asScala.map(_._1).toSet
+ if (metricIds.values.forall(updatedMetricIds.contains)) {
+ driverMetricUpdatesPosted.countDown()
+ }
+ case _ =>
+ }
+ }
+
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ checkAnswer(df, Seq(Row(1, "a")))
+ assert(driverMetricUpdatesPosted.await(30, TimeUnit.SECONDS))
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+
+ val driverMetricValues = driverMetricUpdates
+ .iterator()
+ .asScala
+ .toSeq
+ .groupBy(_._1)
+ .mapValues(_.map(_._2).sum)
+ .toMap
+ assert(driverMetricValues.getOrElse(metricIds("numPartitions"), 0L) > 0)
+ assert(driverMetricValues.getOrElse(metricIds("numFiles"), 0L) > 0)
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back when spark.auron.enable.paimon.scan=false") {
+ withTable("paimon.db.t_disable") {
+ sql("create table paimon.db.t_disable (id int, v string) using paimon")
+ sql("insert into paimon.db.t_disable values (1, 'a')")
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ val df = sql("select * from paimon.db.t_disable")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back for MOR (merge-on-read) table") {
+ withTable("paimon.db.t_mor") {
+ sql("""
+ |create table paimon.db.t_mor (id int, v string)
+ |using paimon
+ |tblproperties (
+ | 'primary-key' = 'id',
+ | 'bucket' = '2'
+ |)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_mor values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_mor")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
Review Comment:
Should we also add `checkAnswer` here？ Just to make sure the Spark fallback still returns correct results
##########
thirdparty/auron-paimon/src/test/scala/org/apache/auron/paimon/AuronPaimonV2IntegrationSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.paimon
+
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.auron.plan.NativePaimonV2TableScanExec
+import org.apache.spark.sql.execution.ui.SparkListenerDriverAccumUpdates
+
+class AuronPaimonV2IntegrationSuite
+ extends org.apache.spark.sql.QueryTest
+ with BaseAuronPaimonSuite {
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ sql("create database if not exists paimon.db")
+ }
+
+ test("paimon v2 native scan runs simple COW select") {
+ withTable("paimon.db.t1") {
+ sql("create table paimon.db.t1 (id int, v string) using paimon")
+ sql("insert into paimon.db.t1 values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t1")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports projection") {
+ withTable("paimon.db.t_proj") {
+ sql("create table paimon.db.t_proj (id int, v string) using paimon")
+ sql("insert into paimon.db.t_proj values (1, 'a'), (2, 'b')")
+ val df = sql("select id from paimon.db.t_proj")
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports partitioned table with predicate") {
+ withTable("paimon.db.t_part") {
+ sql("""
+ |create table paimon.db.t_part (id int, v string, p string)
+ |using paimon
+ |partitioned by (p)
+ |""".stripMargin)
+ sql("insert into paimon.db.t_part values (1, 'a', 'p1'), (2, 'b', 'p2')")
+ val df = sql("select * from paimon.db.t_part where p = 'p1'")
+ checkAnswer(df, Seq(Row(1, "a", "p1")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan supports ORC COW table") {
+ withTable("paimon.db.t_orc") {
+ sql("""
+ |create table paimon.db.t_orc (id int, v string)
+ |using paimon
+ |tblproperties ('file.format' = 'orc')
+ |""".stripMargin)
+ sql("insert into paimon.db.t_orc values (1, 'a'), (2, 'b')")
+ val df = sql("select * from paimon.db.t_orc")
+ checkAnswer(df, Seq(Row(1, "a"), Row(2, "b")))
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 native scan handles empty table") {
+ withTable("paimon.db.t_empty") {
+ sql("create table paimon.db.t_empty (id int, v string) using paimon")
+ val df = sql("select * from paimon.db.t_empty")
+ checkAnswer(df, Seq.empty)
+ assertNativePaimonScanApplied(df)
+ }
+ }
+
+ test("paimon v2 scan exposes file scan driver metrics") {
+ withTable("paimon.db.t_metrics") {
+ sql("create table paimon.db.t_metrics (id int, v string) using paimon")
+ sql("insert into paimon.db.t_metrics values (1, 'a')")
+ withSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ val df = sql("select * from paimon.db.t_metrics")
+ val nativeScan = executedNativeScan(df)
+ val metricIds = Map(
+ "numPartitions" -> nativeScan.metrics("numPartitions").id,
+ "numFiles" -> nativeScan.metrics("numFiles").id)
+ val driverMetricUpdates = new ConcurrentLinkedQueue[(Long, Long)]()
+ val driverMetricUpdatesPosted = new CountDownLatch(1)
+ val listener = new SparkListener {
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case SparkListenerDriverAccumUpdates(_, updates) =>
+ updates.foreach { case (metricId, value) =>
+ driverMetricUpdates.add(metricId -> value)
+ }
+ val updatedMetricIds = driverMetricUpdates.iterator().asScala.map(_._1).toSet
+ if (metricIds.values.forall(updatedMetricIds.contains)) {
+ driverMetricUpdatesPosted.countDown()
+ }
+ case _ =>
+ }
+ }
+
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ checkAnswer(df, Seq(Row(1, "a")))
+ assert(driverMetricUpdatesPosted.await(30, TimeUnit.SECONDS))
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+
+ val driverMetricValues = driverMetricUpdates
+ .iterator()
+ .asScala
+ .toSeq
+ .groupBy(_._1)
+ .mapValues(_.map(_._2).sum)
+ .toMap
+ assert(driverMetricValues.getOrElse(metricIds("numPartitions"), 0L) > 0)
+ assert(driverMetricValues.getOrElse(metricIds("numFiles"), 0L) > 0)
+ }
+ }
+ }
+
+ test("paimon v2 native scan falls back when spark.auron.enable.paimon.scan=false") {
+ withTable("paimon.db.t_disable") {
+ sql("create table paimon.db.t_disable (id int, v string) using paimon")
+ sql("insert into paimon.db.t_disable values (1, 'a')")
+ withSQLConf("spark.auron.enable.paimon.scan" -> "false") {
+ val df = sql("select * from paimon.db.t_disable")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativePaimonV2TableScan"))
Review Comment:
Same thought as the MOR test
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.commons.lang3.reflect.MethodUtils
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ // 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
+ // memoize the result on the node to avoid doubling planning latency and to guarantee both
+ // calls observe the same splits even if Paimon's split planning is non-deterministic.
+ private val scanPlanTag = TreeNodeTag[Option[PaimonScanPlan]]("auron.paimon.scanPlan")
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ exec.getTagValue(scanPlanTag) match {
+ case Some(cached) => cached
+ case None =>
+ val result = computePlan(exec)
+ exec.setTagValue(scanPlanTag, result)
+ result
+ }
+ }
+
+ private def computePlan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
Review Comment:
Should we consider constructing `Literal` based on the runtime type / target `DataType`, and add an integration test with a non-string partition column？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "merrily01 (via GitHub)" <gi...@apache.org>.
merrily01 commented on PR #2313:
URL: https://github.com/apache/auron/pull/2313#issuecomment-4656862572
> @merrily01 Filed tracking issue #2316 and renamed the PR title to `[AURON #2316] …` to satisfy the title check. The new `pr-title-check` run is in `action_required` state since this PR is from a fork — could you help approve it? Thanks!
Thank you, the CI has been re-triggered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on PR #2313:
URL: https://github.com/apache/auron/pull/2313#issuecomment-4649974310
@merrily01 Filed tracking issue #2316 and renamed the PR title to `[AURON #2316] …` to satisfy the title check. The new `pr-title-check` run is in `action_required` state since this PR is from a fork — could you help approve it? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on PR #2313:
URL: https://github.com/apache/auron/pull/2313#issuecomment-4660030708
@merrily01 I've fixed the CI error; could you please help me trigger CI again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on PR #2313:
URL: https://github.com/apache/auron/pull/2313#issuecomment-4650199306
Pushed a follow-up commit addressing the three Copilot review comments:
1. **`NativePaimonV2TableScanExec`** — projection field-index lookup now respects `SQLConf.caseSensitiveAnalysis` (mirrors `NativeIcebergTableScanExec`) and is derived from `output` attributes, so case-insensitive analysis no longer breaks the projection.
2. **`PaimonScanSupport.inputPartitions`** — return type changed to `Option[Seq[InputPartition]]` so `plan` can distinguish a genuinely empty table (`Some(Seq.empty)` -> empty native scan) from a planning failure (`None` -> fall back to Spark). No more silent zero-row scans on failure.
3. **`AuronPaimonV2IntegrationSuite`** — empty-table test now asserts `NativePaimonV2TableScan` is in the executed plan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2316] Support native scan for Paimon DSv2 (BatchScanExec) COW tables [auron]
Posted by "zhuxiangyi (via GitHub)" <gi...@apache.org>.
zhuxiangyi commented on code in PR #2313:
URL: https://github.com/apache/auron/pull/2313#discussion_r3425946292
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/auron/paimon/PaimonScanSupport.scala:
##########
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.auron.paimon
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.utils.RowDataToObjectArrayConverter
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.auron.NativeConverters
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
+import org.apache.spark.sql.connector.read.InputPartition
+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
+
+final case class PaimonFile(filePath: String, fileSize: Long, partitionValues: InternalRow)
+
+final case class PaimonScanPlan(
+ table: FileStoreTable,
+ files: Seq[PaimonFile],
+ fileFormat: String,
+ readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType)
+
+object PaimonScanSupport extends Logging {
+
+ private val PaimonBaseScanClassName = "org.apache.paimon.spark.PaimonBaseScan"
+ private val PaimonInputPartitionClassName = "org.apache.paimon.spark.PaimonInputPartition"
+
+ def isPaimonScan(exec: BatchScanExec): Boolean = isPaimonScan(exec.scan)
+
+ private def isPaimonScan(scan: AnyRef): Boolean = {
+ isInstanceOfClass(scan, PaimonBaseScanClassName)
+ }
+
+ def plan(exec: BatchScanExec): Option[PaimonScanPlan] = {
+ val scan = exec.scan
+ if (!isPaimonScan(scan)) {
+ return None
+ }
+
+ val table = paimonTable(scan) match {
+ case Some(t) => t
+ case None =>
+ logDebug("Skip native Paimon scan: cannot resolve FileStoreTable from PaimonScan.")
+ return None
+ }
+
+ // Append-only tables (no primary key) have no merge process and can always be read
+ // from raw files; primary-key tables are only supported in COW mode. Either way the
+ // per-split rawConvertible/deletion-files checks below remain the final safety net.
+ val isAppendOnlyTable = table.primaryKeys().isEmpty
+ if (!isAppendOnlyTable && !PaimonUtil.isPaimonCowTable(table)) {
+ logDebug(
+ "Skip native Paimon scan: only append-only or COW primary-key tables are supported.")
+ return None
+ }
+
+ val fileFormat = PaimonUtil.paimonFileFormat(table)
+ if (!fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) &&
+ !fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ logDebug(s"Skip native Paimon scan: unsupported file format $fileFormat.")
+ return None
+ }
+
+ val readSchema = scan.readSchema()
+ if (!readSchema.fields.forall(f => NativeConverters.isTypeSupported(f.dataType))) {
+ logDebug("Skip native Paimon scan: unsupported column data type in read schema.")
+ 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 partitionSchema = StructType(partitionFields)
+ val fileSchema = StructType(fileFields)
+
+ val partitions = inputPartitions(exec) match {
+ case Some(p) => p
+ case None =>
+ logDebug("Skip native Paimon scan: failed to obtain input partitions.")
+ return None
+ }
+ if (partitions.isEmpty) {
+ logDebug("Paimon scan planned with empty input partitions.")
+ return Some(
+ PaimonScanPlan(table, Seq.empty, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ val splitsOpt = collectSplits(partitions)
+ val splits = splitsOpt match {
+ case Some(s) => s
+ case None =>
+ logDebug("Skip native Paimon scan: cannot extract splits from input partitions.")
+ return None
+ }
+
+ // Only allow COW-style raw-readable splits; reject MOR/MOW or splits with deletion vectors.
+ val unsupported = splits.find { s =>
+ !s.rawConvertible() ||
+ (s.deletionFiles().isPresent && {
+ val list = s.deletionFiles().get()
+ list != null && list.asScala.exists(_ != null)
+ })
+ }
+ if (unsupported.isDefined) {
+ logDebug("Skip native Paimon scan: split is not raw-convertible or has deletion files.")
+ return None
+ }
+
+ val partitionConverter = new RowDataToObjectArrayConverter(
+ table.schema().logicalPartitionType())
+ val sessionLocalTimeZone = SQLConf.get.sessionLocalTimeZone
+ val tzOption: String = {
+ val props = CaseInsensitiveMap(table.options().asScala.toMap)
+ props.getOrElse(DateTimeUtils.TIMEZONE_OPTION, sessionLocalTimeZone)
+ }
+
+ val files = splits.flatMap { split =>
+ val partitionValues = if (partitionSchema.isEmpty) {
+ InternalRow.empty
+ } else {
+ toPartitionRow(
+ partitionConverter.convert(split.partition()),
+ partitionSchema,
+ table.schema().partitionKeys().asScala.toSeq,
+ tzOption)
+ }
+ split.dataFiles().asScala.map { dataFile =>
+ val filePath = s"${split.bucketPath()}/${dataFile.fileName()}"
+ PaimonFile(filePath, dataFile.fileSize(), partitionValues)
+ }
+ }
+
+ Some(PaimonScanPlan(table, files, fileFormat, readSchema, fileSchema, partitionSchema))
+ }
+
+ private def containsName(names: Set[String], target: String): Boolean = {
+ val resolver = SQLConf.get.resolver
+ 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 cast strings/temporals into the requested types.
+ private def toPartitionRow(
+ paimonValues: Array[AnyRef],
+ partitionSchema: StructType,
+ partitionKeys: Seq[String],
+ timeZoneId: 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)
+ val raw = if (idx >= 0 && idx < paimonValues.length) paimonValues(idx) else null
+ val literal: Literal = raw match {
+ case null => Literal(null, field.dataType)
+ case v => Literal(v.toString)
+ }
+ Cast(literal, field.dataType, Option(timeZoneId)).eval()
+ })
+ }
+
+ private def collectSplits(partitions: Seq[InputPartition]): Option[Seq[DataSplit]] = {
+ val buf = scala.collection.mutable.ArrayBuffer.empty[DataSplit]
+ partitions.foreach { p =>
+ if (!isInstanceOfClass(p, PaimonInputPartitionClassName)) {
+ return None
+ }
+ val splits = invokeMethod(p, "splits") match {
+ case Some(s: scala.collection.Seq[_]) => s.toSeq
+ case _ => return None
+ }
+ splits.foreach {
+ case ds: DataSplit => buf += ds
+ case _: Split => return None
+ case _ => return None
+ }
+ }
+ Some(buf.toSeq)
+ }
+
+ private def paimonTable(scan: AnyRef): Option[FileStoreTable] = {
+ invokeMethod(scan, "table") match {
+ case Some(t: FileStoreTable) => Some(t)
+ case Some(other) =>
+ logDebug(s"Unexpected Paimon table type: ${other.getClass.getName}")
+ None
+ case None => None
+ }
+ }
+
+ // DSv2 BatchScanExec exposes input partitions via Scan.toBatch (preferred) or a method on
+ // the exec itself; the latter varies across Spark versions, so we attempt both.
+ // Returns Some(partitions) on success (possibly empty if the table is empty), or None when
+ // partition planning fails - the caller falls back to Spark execution on None.
+ private def inputPartitions(exec: BatchScanExec): Option[Seq[InputPartition]] = {
+ try {
Review Comment:
Agreed it's duplicated. I'd prefer to handle this in a follow-up refactor PR rather than here: Paimon's variant intentionally returns `Option[Seq[InputPartition]]` to distinguish an empty table from a planning failure (added earlier in this PR), while `IcebergScanSupport`'s returns `Seq` and collapses failure to empty. A clean shared `BatchScanInputPartitions.resolve(exec)` therefore changes Iceberg's behavior and needs its own validation, which is out of scope for this DSv2 feature PR.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativePaimonV2TableScanExec.scala:
##########
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.auron.plan
+
+import java.net.URI
+import java.security.PrivilegedExceptionAction
+import java.util.Locale
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.fs.FileSystem
+import org.apache.spark.Partition
+import org.apache.spark.TaskContext
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.auron.{EmptyNativeRDD, NativeConverters, NativeHelper, NativeRDD, NativeSupports, Shims}
+import org.apache.spark.sql.auron.paimon.{PaimonFile, PaimonScanPlan}
+import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+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.util.SerializableConfiguration
+
+import org.apache.auron.{protobuf => pb}
+import org.apache.auron.jni.JniBridge
+import org.apache.auron.metric.SparkMetricNode
+
+case class NativePaimonV2TableScanExec(basedScan: BatchScanExec, plan: PaimonScanPlan)
+ extends LeafExecNode
+ with NativeSupports
+ with Logging {
+
+ override lazy val metrics: Map[String, SQLMetric] =
+ NativeHelper.getNativeFileScanMetrics(sparkContext) ++ Seq(
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "Native.partitions_read"),
+ "numFiles" -> SQLMetrics.createMetric(sparkContext, "Native.files_read"))
+
+ override val output = basedScan.output
+ override val outputPartitioning = basedScan.outputPartitioning
+
+ private lazy val fileSchema: StructType = plan.fileSchema
+ private lazy val partitionSchema: StructType = plan.partitionSchema
+ private lazy val files: Seq[PaimonFile] = plan.files
+
+ private lazy val partitions: Array[FilePartition] = {
+ val filePartitions = buildFilePartitions()
+ postDriverMetrics(filePartitions)
+ filePartitions
+ }
+ private lazy val fileSizes: Map[String, Long] =
+ files.map(f => f.filePath -> f.fileSize).toMap
+
+ private lazy val nativeFileSchema: pb.Schema = NativeConverters.convertSchema(fileSchema)
+ private lazy val nativePartitionSchema: pb.Schema =
+ NativeConverters.convertSchema(partitionSchema)
+
+ // Project the output attributes onto the (fileSchema ++ partitionSchema) layout used by the
+ // native scan. Index lookup follows SQLConf.caseSensitiveAnalysis so that the projection
+ // remains correct under case-insensitive analysis (mirrors NativeIcebergTableScanExec).
+ private lazy val combinedSchema: StructType =
+ StructType(fileSchema.fields ++ partitionSchema.fields)
+
+ private lazy val caseSensitive: Boolean = SQLConf.get.caseSensitiveAnalysis
+
+ private lazy val fieldIndexByName: Map[String, Int] = {
+ if (caseSensitive) {
+ combinedSchema.fieldNames.zipWithIndex.toMap
+ } else {
+ combinedSchema.fieldNames.map(_.toLowerCase(Locale.ROOT)).zipWithIndex.toMap
+ }
+ }
+
+ private def fieldIndexFor(name: String): Int = {
+ if (caseSensitive) {
+ fieldIndexByName.getOrElse(name, combinedSchema.fieldIndex(name))
+ } else {
+ fieldIndexByName.getOrElse(name.toLowerCase(Locale.ROOT), combinedSchema.fieldIndex(name))
+ }
+ }
+
+ private lazy val projection: Seq[Integer] =
+ output.map(attr => Integer.valueOf(fieldIndexFor(attr.name)))
+
+ override def doExecuteNative(): NativeRDD = {
+ if (partitions.isEmpty) {
+ return new EmptyNativeRDD(sparkContext)
+ }
+
+ val nativeMetrics = SparkMetricNode(
+ metrics,
+ Nil,
+ Some({
+ case ("bytes_scanned", v) =>
+ val inputMetric = TaskContext.get.taskMetrics().inputMetrics
+ inputMetric.incBytesRead(v)
+ case ("output_rows", v) =>
+ val inputMetric = TaskContext.get.taskMetrics().inputMetrics
+ inputMetric.incRecordsRead(v)
+ case _ =>
+ }))
+
+ val fileFormat = plan.fileFormat
+ val broadcastedHadoopConf = this.broadcastedHadoopConf
+ val numPartitions = partitions.length
+ val nativeFileGroups = this.nativeFileGroups
+ val nativeFileSchema = this.nativeFileSchema
+ val nativePartitionSchema = this.nativePartitionSchema
+ val projection = this.projection
+
+ new NativeRDD(
+ sparkContext,
+ nativeMetrics,
+ partitions.asInstanceOf[Array[Partition]],
+ None,
+ Nil,
+ rddShuffleReadFull = true,
+ (partition, _) => {
+ val resourceId = s"NativePaimonV2TableScan:${UUID.randomUUID().toString}"
+ putJniBridgeResource(resourceId, broadcastedHadoopConf)
+
+ val nativeFileGroup = nativeFileGroups(partition.asInstanceOf[FilePartition])
+ val nativeFileScanConf = pb.FileScanExecConf
+ .newBuilder()
+ .setNumPartitions(numPartitions)
+ .setPartitionIndex(partition.index)
+ .setStatistics(pb.Statistics.getDefaultInstance)
+ .setSchema(nativeFileSchema)
+ .setFileGroup(nativeFileGroup)
+ .addAllProjection(projection.asJava)
+ .setPartitionSchema(nativePartitionSchema)
+ .build()
+
+ if (fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)) {
+ val nativeOrcScanExecBuilder = pb.OrcScanExecNode
+ .newBuilder()
+ .setBaseConf(nativeFileScanConf)
+ .setFsResourceId(resourceId)
+ .addAllPruningPredicates(new java.util.ArrayList())
+
+ pb.PhysicalPlanNode
+ .newBuilder()
+ .setOrcScan(nativeOrcScanExecBuilder.build())
+ .build()
+ } else {
+ val nativeParquetScanExecBuilder = pb.ParquetScanExecNode
+ .newBuilder()
+ .setBaseConf(nativeFileScanConf)
+ .setFsResourceId(resourceId)
+ .addAllPruningPredicates(new java.util.ArrayList())
+
+ pb.PhysicalPlanNode
+ .newBuilder()
+ .setParquetScan(nativeParquetScanExecBuilder.build())
+ .build()
+ }
+ },
+ friendlyName = "NativeRDD.PaimonV2Scan")
+ }
+
+ override val nodeName: String = "NativePaimonV2TableScan"
+
+ override protected def doCanonicalize(): SparkPlan = basedScan.canonicalized
+
+ private lazy val nativeFileGroups: FilePartition => pb.FileGroup = (partition: FilePartition) =>
+ {
+ val nativePartitionedFile = (file: PartitionedFile) => {
+ val filePath = file.filePath.toString
+ val size = fileSizes.getOrElse(filePath, file.length)
+ val nativePartitionValues = partitionSchema.zipWithIndex.map { case (field, index) =>
+ NativeConverters
+ .convertExpr(
+ org.apache.spark.sql.catalyst.expressions
+ .Literal(file.partitionValues.get(index, field.dataType), field.dataType))
+ .getLiteral
+ }
+ pb.PartitionedFile
+ .newBuilder()
+ .setPath(filePath)
+ .setSize(size)
+ .setLastModifiedNs(0)
+ .addAllPartitionValues(nativePartitionValues.asJava)
+ .setRange(
+ pb.FileRange
+ .newBuilder()
+ .setStart(file.start)
+ .setEnd(file.start + file.length)
+ .build())
+ .build()
+ }
+ pb.FileGroup
+ .newBuilder()
+ .addAllFiles(partition.files.map(nativePartitionedFile).toList.asJava)
+ .build()
+ }
+
+ private def postDriverMetrics(filePartitions: Array[FilePartition]): Unit = {
+ val numPartitions = filePartitions.length
+ metrics("numPartitions").add(numPartitions)
+ val numFiles = filePartitions.foldLeft(0L)(_ + _.files.length)
+ metrics("numFiles").add(numFiles)
+ val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ SQLMetrics.postDriverMetricUpdates(
+ sparkContext,
+ executionId,
+ Seq(metrics("numPartitions"), metrics("numFiles")))
+ }
+
+ private def buildFilePartitions(): Array[FilePartition] = {
+ if (files.isEmpty) {
+ return Array.empty
+ }
+
+ val sparkSession = Shims.get.getSqlContext(basedScan).sparkSession
+ val isSplitable =
+ plan.fileFormat.equalsIgnoreCase(PaimonUtil.parquetFormat) ||
+ plan.fileFormat.equalsIgnoreCase(PaimonUtil.orcFormat)
+ val maxSplitBytes = getMaxSplitBytes(sparkSession, files)
+ val partitionedFiles = files
+ .flatMap { f =>
+ if (isSplitable) {
+ (0L until f.fileSize by maxSplitBytes).map { offset =>
Review Comment:
Agreed. Proposing a follow-up refactor PR: `putJniBridgeResource` and `broadcastedHadoopConf` are verbatim across `NativeFileSourceScanBase`, `NativeHiveTableScanBase`, `NativeIcebergTableScanExec` and this exec, so extracting a shared trait touches the core FileSource/Hive scan paths and is best done and tested on its own rather than inside this Paimon feature PR.
##########
thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativePaimonV2TableScanExec.scala:
##########
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.auron.plan
+
+import java.net.URI
+import java.security.PrivilegedExceptionAction
+import java.util.Locale
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.fs.FileSystem
+import org.apache.spark.Partition
+import org.apache.spark.TaskContext
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.auron.{EmptyNativeRDD, NativeConverters, NativeHelper, NativeRDD, NativeSupports, Shims}
+import org.apache.spark.sql.auron.paimon.{PaimonFile, PaimonScanPlan}
+import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+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.util.SerializableConfiguration
+
+import org.apache.auron.{protobuf => pb}
+import org.apache.auron.jni.JniBridge
+import org.apache.auron.metric.SparkMetricNode
+
+case class NativePaimonV2TableScanExec(basedScan: BatchScanExec, plan: PaimonScanPlan)
+ extends LeafExecNode
+ with NativeSupports
+ with Logging {
+
+ override lazy val metrics: Map[String, SQLMetric] =
+ NativeHelper.getNativeFileScanMetrics(sparkContext) ++ Seq(
+ "numPartitions" -> SQLMetrics.createMetric(sparkContext, "Native.partitions_read"),
+ "numFiles" -> SQLMetrics.createMetric(sparkContext, "Native.files_read"))
+
+ override val output = basedScan.output
+ override val outputPartitioning = basedScan.outputPartitioning
+
+ private lazy val fileSchema: StructType = plan.fileSchema
+ private lazy val partitionSchema: StructType = plan.partitionSchema
+ private lazy val files: Seq[PaimonFile] = plan.files
+
+ private lazy val partitions: Array[FilePartition] = {
+ val filePartitions = buildFilePartitions()
+ postDriverMetrics(filePartitions)
+ filePartitions
+ }
+ private lazy val fileSizes: Map[String, Long] =
+ files.map(f => f.filePath -> f.fileSize).toMap
+
+ private lazy val nativeFileSchema: pb.Schema = NativeConverters.convertSchema(fileSchema)
+ private lazy val nativePartitionSchema: pb.Schema =
+ NativeConverters.convertSchema(partitionSchema)
+
+ // Project the output attributes onto the (fileSchema ++ partitionSchema) layout used by the
+ // native scan. Index lookup follows SQLConf.caseSensitiveAnalysis so that the projection
+ // remains correct under case-insensitive analysis (mirrors NativeIcebergTableScanExec).
+ private lazy val combinedSchema: StructType =
+ StructType(fileSchema.fields ++ partitionSchema.fields)
+
+ private lazy val caseSensitive: Boolean = SQLConf.get.caseSensitiveAnalysis
+
+ private lazy val fieldIndexByName: Map[String, Int] = {
+ if (caseSensitive) {
+ combinedSchema.fieldNames.zipWithIndex.toMap
+ } else {
+ combinedSchema.fieldNames.map(_.toLowerCase(Locale.ROOT)).zipWithIndex.toMap
+ }
Review Comment:
Leaving as-is for now: as you noted, this mirrors the existing `NativeIcebergTableScanExec` design. A mixed-case schema that collides under `toLowerCase` is an edge case, and I'd rather keep the two execs consistent and fix both together if we decide to change it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org