This is an automated email from the ASF dual-hosted git repository.
hongze 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 acbdce981 [VL] Fix case-class inheritance for
VeloxColumnarWriteFilesExec (#5480)
acbdce981 is described below
commit acbdce9810186d0aa26ac72f3974a1310e67e6cd
Author: Hongze Zhang <[email protected]>
AuthorDate: Tue Apr 23 18:44:36 2024 +0800
[VL] Fix case-class inheritance for VeloxColumnarWriteFilesExec (#5480)
---
.../clickhouse/CHSparkPlanExecApi.scala | 4 +-
.../backendsapi/velox/SparkPlanExecApiImpl.scala | 6 +-
.../execution/VeloxColumnarWriteFilesExec.scala | 64 ++++++++++++++++++----
.../gluten/backendsapi/SparkPlanExecApi.scala | 4 +-
.../org/apache/spark/util/SparkDirectoryUtil.scala | 6 +-
.../datasources/GlutenV1WriteCommandSuite.scala | 10 ++--
.../spark/sql/sources/GlutenInsertSuite.scala | 2 +-
.../datasources/GlutenV1WriteCommandSuite.scala | 10 ++--
.../spark/sql/sources/GlutenInsertSuite.scala | 2 +-
9 files changed, 74 insertions(+), 34 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
index 55983171b..7ea12ffe7 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
@@ -49,7 +49,7 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.delta.files.TahoeFileIndex
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.AQEShuffleReadExec
-import org.apache.spark.sql.execution.datasources.{FileFormat,
HadoopFsRelation, WriteFilesExec}
+import org.apache.spark.sql.execution.datasources.{FileFormat,
HadoopFsRelation}
import
org.apache.spark.sql.execution.datasources.GlutenWriterColumnarRules.NativeWritePostRule
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import
org.apache.spark.sql.execution.datasources.v2.clickhouse.source.DeltaMergeTreeFileFormat
@@ -677,7 +677,7 @@ class CHSparkPlanExecApi extends SparkPlanExecApi {
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
- staticPartitions: TablePartitionSpec): WriteFilesExec = {
+ staticPartitions: TablePartitionSpec): SparkPlan = {
throw new GlutenNotSupportException("ColumnarWriteFilesExec is not support
in ch backend.")
}
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/SparkPlanExecApiImpl.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/SparkPlanExecApiImpl.scala
index 9952147b9..64a2dc2ce 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/SparkPlanExecApiImpl.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/SparkPlanExecApiImpl.scala
@@ -45,7 +45,7 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.plans.physical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution._
-import org.apache.spark.sql.execution.datasources.{FileFormat, WriteFilesExec}
+import org.apache.spark.sql.execution.datasources.FileFormat
import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
ShuffleExchangeExec}
import org.apache.spark.sql.execution.joins.BuildSideRelation
import org.apache.spark.sql.execution.metric.SQLMetric
@@ -506,8 +506,8 @@ class SparkPlanExecApiImpl extends SparkPlanExecApi {
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
- staticPartitions: TablePartitionSpec): WriteFilesExec = {
- new VeloxColumnarWriteFilesExec(
+ staticPartitions: TablePartitionSpec): SparkPlan = {
+ VeloxColumnarWriteFilesExec(
child,
fileFormat,
partitionColumns,
diff --git
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/VeloxColumnarWriteFilesExec.scala
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/VeloxColumnarWriteFilesExec.scala
index 57fbc8104..26d249f90 100644
---
a/backends-velox/src/main/scala/org/apache/spark/sql/execution/VeloxColumnarWriteFilesExec.scala
+++
b/backends-velox/src/main/scala/org/apache/spark/sql/execution/VeloxColumnarWriteFilesExec.scala
@@ -18,6 +18,7 @@ package org.apache.spark.sql.execution
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.columnarbatch.ColumnarBatches
+import org.apache.gluten.exception.GlutenException
import org.apache.gluten.extension.GlutenPlan
import org.apache.gluten.memory.arrowalloc.ArrowBufferAllocators
import org.apache.gluten.sql.shims.SparkShimLoader
@@ -32,7 +33,7 @@ import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet,
GenericInternalRow}
import org.apache.spark.sql.connector.write.WriterCommitMessage
-import org.apache.spark.sql.execution.datasources.{BasicWriteTaskStats,
DynamicPartitionDataSingleWriter, EmptyDirectoryDataWriter,
ExecutedWriteSummary, FileFormat, PartitioningUtils, SingleDirectoryDataWriter,
WriteFilesExec, WriteFilesSpec, WriteTaskResult}
+import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.vectorized.ColumnarBatch
import org.apache.spark.util.Utils
@@ -254,20 +255,32 @@ class VeloxColumnarWriteFilesRDD(
}
}
-class VeloxColumnarWriteFilesExec(
+// The class inherits from "BinaryExecNode" instead of "UnaryExecNode" because
+// we need to expose a dummy child (as right child) with type "WriteFilesExec"
to let Spark
+// choose the new write code path (version >= 3.4). The actual plan to write
is the left child
+// of this operator.
+case class VeloxColumnarWriteFilesExec(
child: SparkPlan,
fileFormat: FileFormat,
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
staticPartitions: TablePartitionSpec)
- extends WriteFilesExec(child, fileFormat, partitionColumns, bucketSpec,
options, staticPartitions)
- with GlutenPlan {
+ extends BinaryExecNode
+ with GlutenPlan
+ with VeloxColumnarWriteFilesExec.ExecuteWriteCompatible {
+ import VeloxColumnarWriteFilesExec._
override lazy val references: AttributeSet = AttributeSet.empty
override def supportsColumnar(): Boolean = true
+ override def output: Seq[Attribute] = Seq.empty
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ throw new GlutenException(s"$nodeName does not support doExecute")
+ }
+
/** Fallback to use vanilla Spark write files to generate an empty file for
metadata only. */
private def writeFilesForEmptyRDD(
writeFilesSpec: WriteFilesSpec,
@@ -308,12 +321,39 @@ class VeloxColumnarWriteFilesExec(
}
}
- override protected def withNewChildInternal(newChild: SparkPlan):
WriteFilesExec =
- new VeloxColumnarWriteFilesExec(
- newChild,
- fileFormat,
- partitionColumns,
- bucketSpec,
- options,
- staticPartitions)
+ override def left: SparkPlan = child
+
+ // This is a workaround for FileFormatWriter#write. Vanilla Spark (version
>= 3.4) requires for
+ // a plan that has at least one node exactly of type `WriteFilesExec` that
is a Scala case-class,
+ // to decide to choose new `#executeWrite` code path over the legacy
`#execute` for write
+ // operation.
+ //
+ // So we add a no-op `WriteFilesExec` child to let Spark pick the new code
path.
+ //
+ // See: FileFormatWriter#write
+ // See: V1Writes#getWriteFilesOpt
+ override val right: SparkPlan =
+ WriteFilesExec(NoopLeaf(), fileFormat, partitionColumns, bucketSpec,
options, staticPartitions)
+
+ override protected def withNewChildrenInternal(
+ newLeft: SparkPlan,
+ newRight: SparkPlan): SparkPlan =
+ copy(newLeft, fileFormat, partitionColumns, bucketSpec, options,
staticPartitions)
+}
+
+object VeloxColumnarWriteFilesExec {
+ private case class NoopLeaf() extends LeafExecNode {
+ override protected def doExecute(): RDD[InternalRow] =
+ throw new GlutenException(s"$nodeName does not support doExecute")
+ override def output: Seq[Attribute] = Seq.empty
+ }
+
+ sealed trait ExecuteWriteCompatible {
+ // To be compatible with Spark (version < 3.4)
+ protected def doExecuteWrite(writeFilesSpec: WriteFilesSpec):
RDD[WriterCommitMessage] = {
+ throw new GlutenException(
+ s"Internal Error ${this.getClass} has write support" +
+ s" mismatch:\n${this}")
+ }
+ }
}
diff --git
a/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
b/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
index 74e03e329..8dfb1e641 100644
---
a/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
+++
b/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.plans.physical.{BroadcastMode,
Partitioning}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{FileSourceScanExec, GenerateExec,
LeafExecNode, SparkPlan}
-import org.apache.spark.sql.execution.datasources.{FileFormat, WriteFilesExec}
+import org.apache.spark.sql.execution.datasources.FileFormat
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileScan}
import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
import org.apache.spark.sql.execution.joins.BuildSideRelation
@@ -357,7 +357,7 @@ trait SparkPlanExecApi {
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
- staticPartitions: TablePartitionSpec): WriteFilesExec
+ staticPartitions: TablePartitionSpec): SparkPlan
/**
* Generate extended DataSourceV2 Strategies. Currently only for ClickHouse
backend.
diff --git
a/gluten-core/src/main/scala/org/apache/spark/util/SparkDirectoryUtil.scala
b/gluten-core/src/main/scala/org/apache/spark/util/SparkDirectoryUtil.scala
index 435ee8df7..fbc59edfd 100644
--- a/gluten-core/src/main/scala/org/apache/spark/util/SparkDirectoryUtil.scala
+++ b/gluten-core/src/main/scala/org/apache/spark/util/SparkDirectoryUtil.scala
@@ -30,8 +30,8 @@ import java.nio.file.Paths
* Manages Gluten's local directories, for storing jars, libs, spill files, or
other temporary
* stuffs.
*/
-class SparkDirectoryUtil private (roots: Array[String]) extends Logging {
- private val ROOTS = roots.flatMap {
+class SparkDirectoryUtil private (val roots: Array[String]) extends Logging {
+ private val ROOTS: Array[File] = roots.flatMap {
rootDir =>
try {
val localDir = Utils.createDirectory(rootDir, "gluten")
@@ -78,7 +78,7 @@ object SparkDirectoryUtil extends Logging {
INSTANCE = new SparkDirectoryUtil(roots)
return
}
- if (INSTANCE.ROOTS.toSet != roots.toSet) {
+ if (INSTANCE.roots.toSet != roots.toSet) {
logWarning(
s"Reinitialize SparkDirectoryUtil with different root dirs: old:
${INSTANCE.ROOTS
.mkString("Array(", ", ", ")")}, new: ${roots.mkString("Array(",
", ", ")")}"
diff --git
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
index 7799252ba..3d277b94c 100644
---
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
+++
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
@@ -21,7 +21,7 @@ import org.apache.gluten.execution.SortExecTransformer
import org.apache.spark.sql.GlutenSQLTestsBaseTrait
import org.apache.spark.sql.catalyst.expressions.{Ascending,
AttributeReference, NullsFirst, SortOrder}
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Sort}
-import org.apache.spark.sql.execution.{QueryExecution, SortExec}
+import org.apache.spark.sql.execution.{QueryExecution, SortExec,
VeloxColumnarWriteFilesExec}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StringType}
@@ -122,8 +122,8 @@ class GlutenV1WriteCommandSuite
val executedPlan = FileFormatWriter.executedPlan.get
val plan = if (enabled) {
- assert(executedPlan.isInstanceOf[WriteFilesExec])
- executedPlan.asInstanceOf[WriteFilesExec].child
+ assert(executedPlan.isInstanceOf[VeloxColumnarWriteFilesExec])
+ executedPlan.asInstanceOf[VeloxColumnarWriteFilesExec].child
} else {
executedPlan.transformDown { case a: AdaptiveSparkPlanExec =>
a.executedPlan }
}
@@ -204,8 +204,8 @@ class GlutenV1WriteCommandSuite
val executedPlan = FileFormatWriter.executedPlan.get
val plan = if (enabled) {
- assert(executedPlan.isInstanceOf[WriteFilesExec])
- executedPlan.asInstanceOf[WriteFilesExec].child
+ assert(executedPlan.isInstanceOf[VeloxColumnarWriteFilesExec])
+ executedPlan.asInstanceOf[VeloxColumnarWriteFilesExec].child
} else {
executedPlan.transformDown { case a: AdaptiveSparkPlanExec =>
a.executedPlan }
}
diff --git
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
index cf6f35288..b951da84a 100644
---
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
+++
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
@@ -141,7 +141,7 @@ class GlutenInsertSuite
.find(_.isInstanceOf[SortExecTransformer])
.isEmpty)
// all operators should be transformed
- assert(writeFiles.find(!_.isInstanceOf[GlutenPlan]).isEmpty)
+ assert(writeFiles.child.find(!_.isInstanceOf[GlutenPlan]).isEmpty)
val parts =
spark.sessionState.catalog.listPartitionNames(TableIdentifier("pt")).toSet
assert(parts == expectedPartitionNames)
diff --git
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
index 7799252ba..3d277b94c 100644
---
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
+++
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
@@ -21,7 +21,7 @@ import org.apache.gluten.execution.SortExecTransformer
import org.apache.spark.sql.GlutenSQLTestsBaseTrait
import org.apache.spark.sql.catalyst.expressions.{Ascending,
AttributeReference, NullsFirst, SortOrder}
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Sort}
-import org.apache.spark.sql.execution.{QueryExecution, SortExec}
+import org.apache.spark.sql.execution.{QueryExecution, SortExec,
VeloxColumnarWriteFilesExec}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StringType}
@@ -122,8 +122,8 @@ class GlutenV1WriteCommandSuite
val executedPlan = FileFormatWriter.executedPlan.get
val plan = if (enabled) {
- assert(executedPlan.isInstanceOf[WriteFilesExec])
- executedPlan.asInstanceOf[WriteFilesExec].child
+ assert(executedPlan.isInstanceOf[VeloxColumnarWriteFilesExec])
+ executedPlan.asInstanceOf[VeloxColumnarWriteFilesExec].child
} else {
executedPlan.transformDown { case a: AdaptiveSparkPlanExec =>
a.executedPlan }
}
@@ -204,8 +204,8 @@ class GlutenV1WriteCommandSuite
val executedPlan = FileFormatWriter.executedPlan.get
val plan = if (enabled) {
- assert(executedPlan.isInstanceOf[WriteFilesExec])
- executedPlan.asInstanceOf[WriteFilesExec].child
+ assert(executedPlan.isInstanceOf[VeloxColumnarWriteFilesExec])
+ executedPlan.asInstanceOf[VeloxColumnarWriteFilesExec].child
} else {
executedPlan.transformDown { case a: AdaptiveSparkPlanExec =>
a.executedPlan }
}
diff --git
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
index 51b034ca8..69e075b78 100644
---
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
+++
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala
@@ -141,7 +141,7 @@ class GlutenInsertSuite
.find(_.isInstanceOf[SortExecTransformer])
.isEmpty)
// all operators should be transformed
- assert(writeFiles.find(!_.isInstanceOf[GlutenPlan]).isEmpty)
+ assert(writeFiles.child.find(!_.isInstanceOf[GlutenPlan]).isEmpty)
val parts =
spark.sessionState.catalog.listPartitionNames(TableIdentifier("pt")).toSet
assert(parts == expectedPartitionNames)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
