This is an automated email from the ASF dual-hosted git repository.
yao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 1a6ef61351 Add a Shim layer for PartitionedFileUtil of Spark 3.5
(#9748)
1a6ef61351 is described below
commit 1a6ef61351f2f5472158f00177f596f29b04959d
Author: Kent Yao <[email protected]>
AuthorDate: Tue May 27 10:13:40 2025 +0800
Add a Shim layer for PartitionedFileUtil of Spark 3.5 (#9748)
* Add a Shim layer for PartitionedFileUtil in Spark 3.5
* Add a Shim layer for PartitionedFileUtil in Spark 3.5
---
.../gluten/execution/PartitionedFileUtilShim.scala | 149 +++++++++++++++++++++
.../gluten/sql/shims/spark35/Spark35Shims.scala | 5 +-
.../sql/execution/AbstractFileSourceScanExec.scala | 6 +-
3 files changed, 156 insertions(+), 4 deletions(-)
diff --git
a/shims/spark35/src/main/scala/org/apache/gluten/execution/PartitionedFileUtilShim.scala
b/shims/spark35/src/main/scala/org/apache/gluten/execution/PartitionedFileUtilShim.scala
new file mode 100644
index 0000000000..838932add7
--- /dev/null
+++
b/shims/spark35/src/main/scala/org/apache/gluten/execution/PartitionedFileUtilShim.scala
@@ -0,0 +1,149 @@
+/*
+ * 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.gluten.execution
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.PartitionedFileUtil
+import org.apache.spark.sql.execution.datasources.{FileStatusWithMetadata,
PartitionedFile}
+
+import org.apache.hadoop.fs.Path
+
+import java.lang.reflect.Method
+
+object PartitionedFileUtilShim {
+
+ private val clz: Class[_] = PartitionedFileUtil.getClass
+ private val module = clz.getField("MODULE$").get(null)
+
+ private lazy val getPartitionedFileMethod: Method = {
+ try {
+ val m = clz.getDeclaredMethod(
+ "getPartitionedFile",
+ classOf[FileStatusWithMetadata],
+ classOf[InternalRow])
+ m.setAccessible(true)
+ m
+ } catch {
+ case _: NoSuchMethodException => null
+ }
+ }
+
+ private lazy val getPartitionedFileByPathMethod: Method = {
+ try {
+ val m = clz.getDeclaredMethod(
+ "getPartitionedFile",
+ classOf[FileStatusWithMetadata],
+ classOf[Path],
+ classOf[InternalRow])
+ m.setAccessible(true)
+ m
+ } catch {
+ case _: NoSuchMethodException => null
+ }
+ }
+
+ def getPartitionedFile(
+ file: FileStatusWithMetadata,
+ partitionValues: InternalRow): PartitionedFile = {
+ if (getPartitionedFileMethod != null) {
+ getPartitionedFileMethod
+ .invoke(module, file, partitionValues)
+ .asInstanceOf[PartitionedFile]
+ } else if (getPartitionedFileByPathMethod != null) {
+ getPartitionedFileByPathMethod
+ .invoke(module, file, file.getPath, partitionValues)
+ .asInstanceOf[PartitionedFile]
+ } else {
+ val params = clz.getDeclaredMethods
+ .find(_.getName == "getPartitionedFile")
+ .map(_.getGenericParameterTypes.mkString(", "))
+ throw new RuntimeException(
+ s"getPartitionedFile with $params is not correctly shimmed " +
+ "in PartitionedFileUtilShim")
+ }
+ }
+
+ private lazy val splitFilesMethod: Method = {
+ try {
+ val m = clz.getDeclaredMethod(
+ "splitFiles",
+ classOf[SparkSession],
+ classOf[FileStatusWithMetadata],
+ classOf[Boolean],
+ classOf[Long],
+ classOf[InternalRow])
+ m.setAccessible(true)
+ m
+ } catch {
+ case _: NoSuchMethodException => null
+ }
+ }
+
+ private lazy val splitFilesByPathMethod: Method = {
+ try {
+ val m = clz.getDeclaredMethod(
+ "splitFiles",
+ classOf[FileStatusWithMetadata],
+ classOf[Path],
+ classOf[Boolean],
+ classOf[Long],
+ classOf[InternalRow])
+ m.setAccessible(true)
+ m
+ } catch {
+ case _: NoSuchMethodException => null
+ }
+ }
+
+ def splitFiles(
+ sparkSession: SparkSession,
+ file: FileStatusWithMetadata,
+ isSplitable: Boolean,
+ maxSplitBytes: Long,
+ partitionValues: InternalRow): Seq[PartitionedFile] = {
+ if (splitFilesMethod != null) {
+ splitFilesMethod
+ .invoke(
+ module,
+ sparkSession,
+ file,
+ java.lang.Boolean.valueOf(isSplitable),
+ java.lang.Long.valueOf(maxSplitBytes),
+ partitionValues)
+ .asInstanceOf[Seq[PartitionedFile]]
+ } else if (splitFilesByPathMethod != null) {
+ splitFilesByPathMethod
+ .invoke(
+ module,
+ file,
+ file.getPath,
+ java.lang.Boolean.valueOf(isSplitable),
+ java.lang.Long.valueOf(maxSplitBytes),
+ partitionValues)
+ .asInstanceOf[Seq[PartitionedFile]]
+ } else {
+ val params = clz.getDeclaredMethods
+ .find(_.getName == "splitFiles")
+ .map(_.getGenericParameterTypes.mkString(", "))
+ throw new RuntimeException(
+ s"splitFiles with $params is not correctly shimmed " +
+ "in PartitionedFileUtilShim")
+ }
+ }
+
+}
diff --git
a/shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala
b/shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala
index badf1ca66a..db9601235c 100644
---
a/shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala
+++
b/shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala
@@ -16,6 +16,7 @@
*/
package org.apache.gluten.sql.shims.spark35
+import org.apache.gluten.execution.PartitionedFileUtilShim
import org.apache.gluten.expression.{ExpressionNames, Sig}
import org.apache.gluten.sql.shims.SparkShims
import org.apache.gluten.utils.ExceptionUtils
@@ -164,7 +165,7 @@ class Spark35Shims extends SparkShims {
override def filesGroupedToBuckets(
selectedPartitions: Array[PartitionDirectory]): Map[Int,
Array[PartitionedFile]] = {
selectedPartitions
- .flatMap(p => p.files.map(f => PartitionedFileUtil.getPartitionedFile(f,
p.values)))
+ .flatMap(p => p.files.map(f =>
PartitionedFileUtilShim.getPartitionedFile(f, p.values)))
.groupBy {
f =>
BucketingUtils
@@ -430,7 +431,7 @@ class Spark35Shims extends SparkShims {
maxSplitBytes: Long,
partitionValues: InternalRow,
metadata: Map[String, Any] = Map.empty): Seq[PartitionedFile] = {
- PartitionedFileUtil.splitFiles(
+ PartitionedFileUtilShim.splitFiles(
sparkSession,
FileStatusWithMetadata(file, metadata),
isSplitable,
diff --git
a/shims/spark35/src/main/scala/org/apache/spark/sql/execution/AbstractFileSourceScanExec.scala
b/shims/spark35/src/main/scala/org/apache/spark/sql/execution/AbstractFileSourceScanExec.scala
index a83c763c45..a9bb032500 100644
---
a/shims/spark35/src/main/scala/org/apache/spark/sql/execution/AbstractFileSourceScanExec.scala
+++
b/shims/spark35/src/main/scala/org/apache/spark/sql/execution/AbstractFileSourceScanExec.scala
@@ -16,6 +16,8 @@
*/
package org.apache.spark.sql.execution
+import org.apache.gluten.execution.PartitionedFileUtilShim
+
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow,
TableIdentifier}
import org.apache.spark.sql.catalyst.catalog.BucketSpec
@@ -181,7 +183,7 @@ abstract class AbstractFileSourceScanExec(
logInfo(s"Planning with ${bucketSpec.numBuckets} buckets")
val filesGroupedToBuckets =
selectedPartitions
- .flatMap(p => p.files.map(f =>
PartitionedFileUtil.getPartitionedFile(f, p.values)))
+ .flatMap(p => p.files.map(f =>
PartitionedFileUtilShim.getPartitionedFile(f, p.values)))
.groupBy {
f =>
BucketingUtils
@@ -269,7 +271,7 @@ abstract class AbstractFileSourceScanExec(
relation.sparkSession,
relation.options,
file.getPath)
- PartitionedFileUtil.splitFiles(
+ PartitionedFileUtilShim.splitFiles(
sparkSession = relation.sparkSession,
file = file,
isSplitable = isSplitable,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
