This is an automated email from the ASF dual-hosted git repository.
loneylee 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 62d9696339 [CORE-8569][CH] Support DeltaOptimizedWriterTransformer
(#8570)
62d9696339 is described below
commit 62d96963399268179517d53c64e320dfc856003b
Author: Shuai li <[email protected]>
AuthorDate: Tue Feb 11 14:59:25 2025 +0800
[CORE-8569][CH] Support DeltaOptimizedWriterTransformer (#8570)
* [CORE-8569][CH] Support DeltaOptimizedWriterTransformer
---
backends-clickhouse/pom.xml | 39 ++-
.../org.apache.gluten.sql.shims.DeltaShimProvider | 1 +
.../sql/shims/delta20/Delta20ShimProvider.scala} | 17 +-
.../gluten/sql/shims/delta20/Delta20Shims.scala} | 17 +-
.../sql/delta/catalog/ClickHouseTableV2.scala | 4 +-
.../org.apache.gluten.sql.shims.DeltaShimProvider | 1 +
.../sql/shims/delta23/Delta23ShimProvider.scala} | 17 +-
.../gluten/sql/shims/delta23/Delta23Shims.scala} | 17 +-
.../sql/delta/catalog/ClickHouseTableV2.scala | 4 +-
.../org.apache.gluten.sql.shims.DeltaShimProvider | 1 +
.../sql/shims/delta32/Delta32ShimProvider.scala} | 17 +-
.../gluten/sql/shims/delta32/Delta32Shims.scala} | 22 +-
.../delta/ClickhouseOptimisticTransaction.scala | 1 -
.../sql/delta/catalog/ClickHouseTableV2.scala | 4 +-
.../sql/execution/FileDeltaColumnarWrite.scala | 4 +-
.../execution/datasources/DeltaV1WritesSuite.scala | 53 ++-
.../org.apache.gluten.component.CHDeltaComponent | 0
.../gluten/component/CHDeltaComponent.scala} | 18 +-
.../gluten/execution/OffloadDeltaNode.scala} | 20 +-
.../apache/gluten/sql/shims/DeltaShimLoader.scala | 56 +++
.../gluten/sql/shims/DeltaShimProvider.scala} | 18 +-
.../org/apache/gluten/sql/shims/DeltaShims.scala} | 22 +-
.../gluten/component}/CHKafkaComponent.scala | 0
.../Storages/MergeTree/SparkMergeTreeSink.h | 2 +
gluten-delta/pom.xml | 6 +
.../sql/perf/DeltaOptimizedWriterTransformer.scala | 386 +++++++++++++++++++++
.../shuffle/sort/ColumnarShuffleManager.scala | 2 +-
27 files changed, 622 insertions(+), 127 deletions(-)
diff --git a/backends-clickhouse/pom.xml b/backends-clickhouse/pom.xml
index 9b954c2ccf..0aa22c426d 100644
--- a/backends-clickhouse/pom.xml
+++ b/backends-clickhouse/pom.xml
@@ -92,6 +92,28 @@
</dependency>
</dependencies>
</profile>
+ <profile>
+ <id>delta</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-delta</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-delta</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+
<artifactId>${delta.package.name}_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+ </profile>
</profiles>
<dependencies>
@@ -464,15 +486,18 @@
<includes>
<include>src/main/scala/**/*.scala</include>
<include>src/test/scala/**/*.scala</include>
-
<include>src-delta-${delta.binary.version}/main/delta/**/*.scala</include>
-
<include>src-delta-${delta.binary.version}/test/delta/**/*.scala</include>
+ <include>src-delta/main/scala/**/*.scala</include>
+ <include>src-delta/test/scala/**/*.scala</include>
+
<include>src-delta-${delta.binary.version}/main/scala/**/*.scala</include>
+
<include>src-delta-${delta.binary.version}/test/scala/**/*.scala</include>
</includes>
<excludes>
-
<exclude>src-delta-${delta.binary.version}/main/delta/org/apache/spark/sql/delta/commands/*.scala</exclude>
-
<exclude>src-delta-${delta.binary.version}/main/delta/org/apache/spark/sql/delta/commands/merge/*.scala</exclude>
-
<exclude>src-delta-${delta.binary.version}/main/delta/org/apache/spark/sql/delta/stats/*.scala</exclude>
-
<exclude>src-delta-${delta.binary.version}/main/delta/org/apache/spark/sql/delta/DeltaLog.scala</exclude>
-
<exclude>src-delta-${delta.binary.version}/main/delta/org/apache/spark/sql/delta/Snapshot.scala</exclude>
+
<exclude>src-delta-${delta.binary.version}/main/scala/org/apache/spark/sql/delta/commands/*.scala</exclude>
+
<exclude>src-delta-${delta.binary.version}/main/scala/org/apache/spark/sql/delta/commands/merge/*.scala</exclude>
+
<exclude>src-delta-${delta.binary.version}/main/scala/org/apache/spark/sql/delta/stats/*.scala</exclude>
+
<exclude>src-delta-${delta.binary.version}/main/scala/org/apache/spark/sql/delta/DeltaLog.scala</exclude>
+
<exclude>src-delta-${delta.binary.version}/main/scala/org/apache/spark/sql/delta/Snapshot.scala</exclude>
+
<exclude>src-delta-${delta.binary.version}/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala</exclude>
</excludes>
</scala>
</configuration>
diff --git
a/backends-clickhouse/src-delta-20/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
b/backends-clickhouse/src-delta-20/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
new file mode 100644
index 0000000000..2e9fc7f42f
--- /dev/null
+++
b/backends-clickhouse/src-delta-20/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
@@ -0,0 +1 @@
+org.apache.gluten.sql.shims.delta20.Delta20ShimProvider
\ No newline at end of file
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta-20/main/scala/org/apache/gluten/sql/shims/delta20/Delta20ShimProvider.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta-20/main/scala/org/apache/gluten/sql/shims/delta20/Delta20ShimProvider.scala
index 1e9f0f9d77..65efe745fb 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta-20/main/scala/org/apache/gluten/sql/shims/delta20/Delta20ShimProvider.scala
@@ -14,19 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims.delta20
-package org.apache.gluten.component
+import org.apache.gluten.sql.shims.{DeltaShimProvider, DeltaShims}
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
+class Delta20ShimProvider extends DeltaShimProvider {
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
+ override def createShim: DeltaShims = {
+ new Delta20Shims()
}
+
}
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta-20/main/scala/org/apache/gluten/sql/shims/delta20/Delta20Shims.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta-20/main/scala/org/apache/gluten/sql/shims/delta20/Delta20Shims.scala
index 1e9f0f9d77..2348586e40 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta-20/main/scala/org/apache/gluten/sql/shims/delta20/Delta20Shims.scala
@@ -14,19 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims.delta20
-package org.apache.gluten.component
+import org.apache.gluten.sql.shims.DeltaShims
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
-
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
- }
-}
+class Delta20Shims extends DeltaShims {}
diff --git
a/backends-clickhouse/src-delta-20/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
b/backends-clickhouse/src-delta-20/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
index 9e76a282a5..d2966eafee 100644
---
a/backends-clickhouse/src-delta-20/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
+++
b/backends-clickhouse/src-delta-20/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
@@ -29,12 +29,14 @@ import
org.apache.spark.sql.execution.datasources.{HadoopFsRelation, PartitionDi
import
org.apache.spark.sql.execution.datasources.clickhouse.utils.MergeTreePartsPartitionsUtil
import org.apache.spark.sql.execution.datasources.mergetree.StorageMeta
import
org.apache.spark.sql.execution.datasources.v2.clickhouse.source.DeltaMergeTreeFileFormat
+import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.collection.BitSet
+
import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.types.StructType
import java.{util => ju}
+
import scala.collection.JavaConverters._
@SuppressWarnings(Array("io.github.zhztheplayer.scalawarts.InheritFromCaseClass"))
diff --git
a/backends-clickhouse/src-delta-23/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
b/backends-clickhouse/src-delta-23/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
new file mode 100644
index 0000000000..c512ce8a71
--- /dev/null
+++
b/backends-clickhouse/src-delta-23/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
@@ -0,0 +1 @@
+org.apache.gluten.sql.shims.delta23.Delta23ShimProvider
\ No newline at end of file
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta-23/main/scala/org/apache/gluten/sql/shims/delta23/Delta23ShimProvider.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta-23/main/scala/org/apache/gluten/sql/shims/delta23/Delta23ShimProvider.scala
index 1e9f0f9d77..854a996999 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta-23/main/scala/org/apache/gluten/sql/shims/delta23/Delta23ShimProvider.scala
@@ -14,19 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims.delta23
-package org.apache.gluten.component
+import org.apache.gluten.sql.shims.{DeltaShimProvider, DeltaShims}
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
+class Delta23ShimProvider extends DeltaShimProvider {
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
+ override def createShim: DeltaShims = {
+ new Delta23Shims()
}
+
}
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta-23/main/scala/org/apache/gluten/sql/shims/delta23/Delta23Shims.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta-23/main/scala/org/apache/gluten/sql/shims/delta23/Delta23Shims.scala
index 1e9f0f9d77..3528363a6e 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta-23/main/scala/org/apache/gluten/sql/shims/delta23/Delta23Shims.scala
@@ -14,19 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims.delta23
-package org.apache.gluten.component
+import org.apache.gluten.sql.shims.DeltaShims
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
-
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
- }
-}
+class Delta23Shims extends DeltaShims {}
diff --git
a/backends-clickhouse/src-delta-23/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
b/backends-clickhouse/src-delta-23/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
index 9e76a282a5..d2966eafee 100644
---
a/backends-clickhouse/src-delta-23/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
+++
b/backends-clickhouse/src-delta-23/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
@@ -29,12 +29,14 @@ import
org.apache.spark.sql.execution.datasources.{HadoopFsRelation, PartitionDi
import
org.apache.spark.sql.execution.datasources.clickhouse.utils.MergeTreePartsPartitionsUtil
import org.apache.spark.sql.execution.datasources.mergetree.StorageMeta
import
org.apache.spark.sql.execution.datasources.v2.clickhouse.source.DeltaMergeTreeFileFormat
+import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.collection.BitSet
+
import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.types.StructType
import java.{util => ju}
+
import scala.collection.JavaConverters._
@SuppressWarnings(Array("io.github.zhztheplayer.scalawarts.InheritFromCaseClass"))
diff --git
a/backends-clickhouse/src-delta-32/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
b/backends-clickhouse/src-delta-32/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
new file mode 100644
index 0000000000..170775d428
--- /dev/null
+++
b/backends-clickhouse/src-delta-32/main/resources/META-INF/services/org.apache.gluten.sql.shims.DeltaShimProvider
@@ -0,0 +1 @@
+org.apache.gluten.sql.shims.delta32.Delta32ShimProvider
\ No newline at end of file
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta-32/main/scala/org/apache/gluten/sql/shims/delta32/Delta32ShimProvider.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta-32/main/scala/org/apache/gluten/sql/shims/delta32/Delta32ShimProvider.scala
index 1e9f0f9d77..de4e9682bd 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta-32/main/scala/org/apache/gluten/sql/shims/delta32/Delta32ShimProvider.scala
@@ -14,19 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims.delta32
-package org.apache.gluten.component
+import org.apache.gluten.sql.shims.{DeltaShimProvider, DeltaShims}
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
+class Delta32ShimProvider extends DeltaShimProvider {
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
+ override def createShim: DeltaShims = {
+ new Delta32Shims()
}
+
}
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta-32/main/scala/org/apache/gluten/sql/shims/delta32/Delta32Shims.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta-32/main/scala/org/apache/gluten/sql/shims/delta32/Delta32Shims.scala
index 1e9f0f9d77..77160f49ce 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta-32/main/scala/org/apache/gluten/sql/shims/delta32/Delta32Shims.scala
@@ -14,19 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims.delta32
-package org.apache.gluten.component
+import org.apache.gluten.execution.GlutenPlan
+import org.apache.gluten.sql.shims.DeltaShims
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.perf.DeltaOptimizedWriterTransformer
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
+class Delta32Shims extends DeltaShims {
+ override def supportDeltaOptimizedWriterExec(plan: SparkPlan): Boolean =
+ DeltaOptimizedWriterTransformer.support(plan)
+
+ override def offloadDeltaOptimizedWriterExec(plan: SparkPlan): GlutenPlan = {
+ DeltaOptimizedWriterTransformer.from(plan)
}
}
diff --git
a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala
b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala
index f559be7997..611d011ee7 100644
---
a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala
+++
b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala
@@ -267,7 +267,6 @@ class ClickhouseOptimisticTransaction(
// TODO: val checkInvariants = DeltaInvariantCheckerExec(empty2NullPlan,
constraints)
val checkInvariants = empty2NullPlan
- // TODO: DeltaOptimizedWriterExec
// No need to plan optimized write if the write command is OPTIMIZE,
which aims to produce
// evenly-balanced data files already.
val physicalPlan =
diff --git
a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
index 91cf31ba7c..2c7a57625b 100644
---
a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
+++
b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/catalog/ClickHouseTableV2.scala
@@ -33,11 +33,13 @@ import
org.apache.spark.sql.execution.datasources.clickhouse.utils.MergeTreePart
import org.apache.spark.sql.execution.datasources.mergetree.StorageMeta
import
org.apache.spark.sql.execution.datasources.v2.clickhouse.source.DeltaMergeTreeFileFormat
import
org.apache.spark.sql.execution.datasources.v2.clickhouse.utils.CHDataSourceUtils
+import org.apache.spark.sql.types.StructType
import org.apache.spark.util.collection.BitSet
+
import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.types.StructType
import java.{util => ju}
+
import scala.collection.JavaConverters._
@SuppressWarnings(Array("io.github.zhztheplayer.scalawarts.InheritFromCaseClass"))
diff --git
a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala
b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala
index 4d6040bae8..871df856eb 100644
---
a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala
+++
b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution
import org.apache.gluten.backendsapi.clickhouse.RuntimeSettings
import org.apache.gluten.exception.GlutenNotSupportException
import org.apache.gluten.vectorized.NativeExpressionEvaluator
-import org.apache.hadoop.fs.Path
+
import org.apache.spark.internal.Logging
import org.apache.spark.internal.io.FileCommitProtocol
import org.apache.spark.sql.catalyst.InternalRow
@@ -30,6 +30,8 @@ import org.apache.spark.sql.delta.stats.{DeltaFileStatistics,
DeltaJobStatistics
import org.apache.spark.sql.execution.datasources.{ExecutedWriteSummary,
WriteJobDescription, WriteTaskResult}
import org.apache.spark.util.Utils
+import org.apache.hadoop.fs.Path
+
import scala.collection.mutable.ArrayBuffer
case class DeltaFileCommitInfo(committer: FileDelayedCommitProtocol)
diff --git
a/backends-clickhouse/src-delta-32/test/scala/org/apache/spark/sql/execution/datasources/DeltaV1WritesSuite.scala
b/backends-clickhouse/src-delta-32/test/scala/org/apache/spark/sql/execution/datasources/DeltaV1WritesSuite.scala
index 6db41f56c5..906b743549 100644
---
a/backends-clickhouse/src-delta-32/test/scala/org/apache/spark/sql/execution/datasources/DeltaV1WritesSuite.scala
+++
b/backends-clickhouse/src-delta-32/test/scala/org/apache/spark/sql/execution/datasources/DeltaV1WritesSuite.scala
@@ -18,9 +18,12 @@ package org.apache.spark.sql.execution.datasources
import org.apache.gluten.config.GlutenConfig
import
org.apache.gluten.execution.{GlutenClickHouseWholeStageTransformerSuite,
GlutenPlan, SortExecTransformer}
+
import org.apache.spark.SparkConf
-import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.delta.{DeltaLog, DeltaOptions}
import org.apache.spark.sql.execution.{SortExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
class DeltaV1WritesSuite extends GlutenClickHouseWholeStageTransformerSuite {
@@ -29,6 +32,14 @@ class DeltaV1WritesSuite extends
GlutenClickHouseWholeStageTransformerSuite {
override protected def sparkConf: SparkConf = {
super.sparkConf
.set(GlutenConfig.NATIVE_WRITER_ENABLED.key, "true")
+ .set(
+ "spark.sql.catalog.spark_catalog",
+
"org.apache.spark.sql.execution.datasources.v2.clickhouse.ClickHouseSparkCatalog")
+ .set("spark.databricks.delta.maxSnapshotLineageLength", "20")
+ .set("spark.databricks.delta.snapshotPartitions", "1")
+ .set("spark.databricks.delta.properties.defaults.checkpointInterval",
"5")
+ .set("spark.databricks.delta.stalenessLimit", "3600000")
+ .set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
}
override def beforeAll(): Unit = {
@@ -97,4 +108,44 @@ class DeltaV1WritesSuite extends
GlutenClickHouseWholeStageTransformerSuite {
check(df.orderBy("j", "k").queryExecution.sparkPlan)
}
+ protected def checkResult(df: DataFrame, numFileCheck: Long => Boolean, dir:
String): Unit = {
+ val (_, snapshot) = DeltaLog.forTableWithSnapshot(spark, dir)
+ val files = snapshot.numOfFiles
+ assert(numFileCheck(files), s"file check failed: received $files")
+
+ checkAnswer(
+ spark.read.format("delta").load(dir),
+ df
+ )
+ }
+
+ test("test optimize write") {
+ withTempDir {
+ dir =>
+ val df = sql("select * from t0").toDF()
+ df.write
+ .format("delta")
+ .option(DeltaOptions.OPTIMIZE_WRITE_OPTION, "true")
+ .save(dir.getPath)
+ checkResult(df, numFileCheck = _ === 1, dir.getPath)
+ }
+ }
+
+ test("optimize write - partitioned write") {
+ withTempDir {
+ dir =>
+ val df = spark
+ .range(0, 100, 1, 4)
+ .withColumn("part", 'id % 5)
+
+ df.write
+ .partitionBy("part")
+ .option(DeltaOptions.OPTIMIZE_WRITE_OPTION, "true")
+ .format("delta")
+ .save(dir.getPath)
+
+ checkResult(df, numFileCheck = _ <= 5, dir.getPath)
+ }
+ }
+
}
diff --git
a/backends-clickhouse/src-delta/main/resources/META-INF/gluten-components/org.apache.gluten.component.CHDeltaComponent
b/backends-clickhouse/src-delta/main/resources/META-INF/gluten-components/org.apache.gluten.component.CHDeltaComponent
new file mode 100644
index 0000000000..e69de29bb2
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/component/CHDeltaComponent.scala
similarity index 67%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta/main/scala/org/apache/gluten/component/CHDeltaComponent.scala
index 1e9f0f9d77..48a4c88518 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/component/CHDeltaComponent.scala
@@ -14,19 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.gluten.component
import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
+import org.apache.gluten.execution.OffloadDeltaNode
+import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
+import org.apache.gluten.extension.columnar.validator.Validators
import org.apache.gluten.extension.injector.Injector
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
+class CHDeltaComponent extends Component {
+ override def name(): String = "ch-delta"
override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
+ Component.BuildInfo("CHDelta", "N/A", "N/A", "N/A")
override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
+ val legacy = injector.gluten.legacy
+ legacy.injectTransform {
+ c =>
+ val offload = Seq(OffloadDeltaNode())
+ HeuristicTransform.Simple(Validators.newValidator(c.glutenConf,
offload), offload)
+ }
}
}
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/execution/OffloadDeltaNode.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta/main/scala/org/apache/gluten/execution/OffloadDeltaNode.scala
index 1e9f0f9d77..bfde89733d 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/execution/OffloadDeltaNode.scala
@@ -14,19 +14,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.execution
-package org.apache.gluten.component
+import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
+import org.apache.gluten.sql.shims.DeltaShimLoader
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
+import org.apache.spark.sql.execution.SparkPlan
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
+case class OffloadDeltaNode() extends OffloadSingleNode {
+ override def offload(plan: SparkPlan): SparkPlan = plan match {
+ case plan if
DeltaShimLoader.getDeltaShims.supportDeltaOptimizedWriterExec(plan) =>
+ DeltaShimLoader.getDeltaShims.offloadDeltaOptimizedWriterExec(plan)
+ case other => other
}
}
diff --git
a/backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShimLoader.scala
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShimLoader.scala
new file mode 100644
index 0000000000..50634b97e9
--- /dev/null
+++
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShimLoader.scala
@@ -0,0 +1,56 @@
+/*
+ * 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.sql.shims
+
+import org.apache.spark.internal.Logging
+
+import java.util.ServiceLoader
+
+import scala.collection.JavaConverters._
+
+object DeltaShimLoader extends Logging {
+ private var deltaShims: DeltaShims = null
+
+ def getDeltaShims: DeltaShims = {
+ if (deltaShims == null) {
+ val provider = getDeltaShimProvider
+ deltaShims = provider.createShim
+ }
+ deltaShims
+ }
+
+ private def loadDeltaShimProvider(): DeltaShimProvider = {
+ // Load and filter the providers based on version
+ val shimProviders =
+ ServiceLoader.load(classOf[DeltaShimProvider]).asScala
+ if (shimProviders.size > 1) {
+ throw new IllegalStateException(s"More than one DeltaShimProvider found:
$shimProviders")
+ }
+
+ val shimProvider = shimProviders.headOption match {
+ case Some(shimProvider) => shimProvider
+ case None =>
+ throw new IllegalStateException(s"No Delta Shim Provider.")
+ }
+ logInfo(s"Using Shim provider: $shimProviders")
+ shimProvider
+ }
+
+ private def getDeltaShimProvider: DeltaShimProvider = {
+ loadDeltaShimProvider()
+ }
+}
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShimProvider.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShimProvider.scala
index 1e9f0f9d77..92900ecf71 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShimProvider.scala
@@ -14,19 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims
-package org.apache.gluten.component
+import org.apache.spark.internal.Logging
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
-
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
- }
+/** Provider interface for matching and retrieving the Shims of a specific
Spark version */
+trait DeltaShimProvider extends Logging {
+ def createShim: DeltaShims
}
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShims.scala
similarity index 57%
copy from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
copy to
backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShims.scala
index 1e9f0f9d77..032054c412 100644
---
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
+++
b/backends-clickhouse/src-delta/main/scala/org/apache/gluten/sql/shims/DeltaShims.scala
@@ -14,19 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.gluten.sql.shims
-package org.apache.gluten.component
+import org.apache.gluten.execution.GlutenPlan
-import org.apache.gluten.backendsapi.clickhouse.CHBackend
-import org.apache.gluten.execution.OffloadKafkaScan
-import org.apache.gluten.extension.injector.Injector
+import org.apache.spark.sql.execution.SparkPlan
-class CHKafkaComponent extends Component {
- override def name(): String = "clickhouse-kafka"
- override def buildInfo(): Component.BuildInfo =
- Component.BuildInfo("ClickHouseKafka", "N/A", "N/A", "N/A")
- override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend]
:: Nil
- override def injectRules(injector: Injector): Unit = {
- OffloadKafkaScan.inject(injector)
+sealed abstract class ShimDescriptor
+
+trait DeltaShims {
+ def supportDeltaOptimizedWriterExec(plan: SparkPlan): Boolean = false
+
+ def offloadDeltaOptimizedWriterExec(plan: SparkPlan): GlutenPlan = {
+ throw new UnsupportedOperationException(
+ s"Can't transform ColumnarDeltaOptimizedWriterExec from
${plan.getClass.getSimpleName}")
}
}
diff --git
a/backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
b/backends-clickhouse/src-kafka/main/scala/org/apache/gluten/component/CHKafkaComponent.scala
similarity index 100%
rename from
backends-clickhouse/src-kafka/main/scala/org.apache.gluten.component/CHKafkaComponent.scala
rename to
backends-clickhouse/src-kafka/main/scala/org/apache/gluten/component/CHKafkaComponent.scala
diff --git a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h
b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h
index a48a903b4c..0d3baf0457 100644
--- a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h
+++ b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h
@@ -204,6 +204,8 @@ public:
void collectStats(const std::deque<PartWithStats> & parts, const
std::string & partition_dir) const
{
+ if (parts.empty())
+ return;
const std::string & partition = partition_dir.empty() ?
WriteStatsBase::NO_PARTITION_ID : partition_dir;
size_t columnSize = parts[0].delta_stats->min.size();
assert(columns_.size() == stats_column_start + columnSize * 3);
diff --git a/gluten-delta/pom.xml b/gluten-delta/pom.xml
index ef30a4f563..a78f91ea28 100755
--- a/gluten-delta/pom.xml
+++ b/gluten-delta/pom.xml
@@ -66,6 +66,12 @@
<version>${spark.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
diff --git
a/gluten-delta/src-delta-32/main/scala/org/apache/spark/sql/perf/DeltaOptimizedWriterTransformer.scala
b/gluten-delta/src-delta-32/main/scala/org/apache/spark/sql/perf/DeltaOptimizedWriterTransformer.scala
new file mode 100644
index 0000000000..9f601d8466
--- /dev/null
+++
b/gluten-delta/src-delta-32/main/scala/org/apache/spark/sql/perf/DeltaOptimizedWriterTransformer.scala
@@ -0,0 +1,386 @@
+/*
+ * 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.perf
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.execution.GlutenPlan
+import org.apache.gluten.extension.columnar.transition.Convention
+
+import org.apache.spark._
+import org.apache.spark.internal.config
+import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.rdd.RDD
+import org.apache.spark.serializer.SerializerManager
+import org.apache.spark.shuffle.{FetchFailedException, ShuffleReader,
ShuffleReadMetricsReporter}
+import org.apache.spark.shuffle.sort.{ColumnarShuffleHandle,
ColumnarShuffleManager}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning,
Partitioning}
+import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog}
+import org.apache.spark.sql.delta.metering.DeltaLogging
+import org.apache.spark.sql.delta.perf.{DeltaOptimizedWriterExec,
OptimizedWriterBlocks}
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.execution._
+import org.apache.spark.sql.execution.exchange.ENSURE_REQUIREMENTS
+import org.apache.spark.sql.execution.metric.{SQLMetric,
SQLShuffleWriteMetricsReporter}
+import org.apache.spark.sql.metric.SQLColumnarShuffleReadMetricsReporter
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.storage.{BlockId, BlockManagerId,
ShuffleBlockFetcherIterator, ShuffleBlockId}
+import org.apache.spark.util.ThreadUtils
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.concurrent.duration.Duration
+
+/**
+ * An execution node which shuffles data to a target output of
`DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS`
+ * blocks, hash partitioned on the table partition columns. We group all
blocks by their
+ * reducer_id's and bin-pack into `DELTA_OPTIMIZE_WRITE_BIN_SIZE` bins. Then
we launch a Spark task
+ * per bin to write out a single file for each bin.
+ *
+ * @param child
+ * The execution plan
+ * @param partitionColumns
+ * The partition columns of the table. Used for hash partitioning the write
+ * @param deltaLog
+ * The DeltaLog for the table. Used for logging only
+ */
+
+case class DeltaOptimizedWriterTransformer(
+ child: SparkPlan,
+ partitionColumns: Seq[String],
+ @transient deltaLog: DeltaLog
+) extends UnaryExecNode
+ with GlutenPlan
+ with DeltaLogging {
+
+ lazy val useSortBasedShuffle: Boolean =
+
BackendsApiManager.getSparkPlanExecApiInstance.useSortBasedShuffle(outputPartitioning,
output)
+
+ override def output: Seq[Attribute] = child.output
+
+ private lazy val readMetrics =
+
SQLColumnarShuffleReadMetricsReporter.createShuffleReadMetrics(sparkContext)
+
+ private[sql] lazy val writeMetrics =
+ SQLShuffleWriteMetricsReporter.createShuffleWriteMetrics(sparkContext)
+
+ // Note: "metrics" is made transient to avoid sending driver-side metrics to
tasks.
+ @transient override lazy val metrics =
+ BackendsApiManager.getMetricsApiInstance
+ .genColumnarShuffleExchangeMetrics(
+ sparkContext,
+ useSortBasedShuffle) ++ readMetrics ++ writeMetrics
+
+ @transient lazy val inputColumnarRDD: RDD[ColumnarBatch] =
child.executeColumnar()
+
+ private lazy val childNumPartitions = inputColumnarRDD.getNumPartitions
+
+ private lazy val numPartitions: Int = {
+ if (childNumPartitions > 0) {
+ val targetShuffleBlocks =
getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS)
+ math.min(
+ math.max(targetShuffleBlocks / childNumPartitions, 1),
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS))
+ } else {
+ childNumPartitions
+ }
+ }
+
+ @transient private var cachedShuffleRDD: ShuffledColumnarBatchRDD = _
+
+ @transient override def outputPartitioning: Partitioning = {
+ val resolver =
org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw
DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ saltedPartitioning
+ }
+
+ @transient private lazy val mapTracker = SparkEnv.get.mapOutputTracker
+
+ private def getShuffleRDD: ShuffledColumnarBatchRDD = {
+ if (cachedShuffleRDD == null) {
+ val resolver =
org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+ val saltedPartitioning = HashPartitioning(
+ partitionColumns.map(
+ p =>
+ output
+ .find(o => resolver(p, o.name))
+ .getOrElse(throw
DeltaErrors.failedFindPartitionColumnInOutputPlan(p))),
+ numPartitions)
+ val shuffledRDD =
+ ColumnarShuffleExchangeExec(saltedPartitioning, child,
ENSURE_REQUIREMENTS, null)
+ .executeColumnar()
+ .asInstanceOf[ShuffledColumnarBatchRDD]
+ cachedShuffleRDD = shuffledRDD
+ }
+ cachedShuffleRDD
+ }
+
+ private def computeBins(): Array[List[(BlockManagerId, ArrayBuffer[(BlockId,
Long, Int)])]] = {
+ // Get all shuffle information
+ val shuffleStats = getShuffleStats()
+
+ // Group by blockId instead of block manager
+ val blockInfo = shuffleStats.flatMap {
+ case (bmId, blocks) =>
+ blocks.map {
+ case (blockId, size, index) =>
+ (blockId, (bmId, size, index))
+ }
+ }.toMap
+
+ val maxBinSize =
+
ByteUnit.BYTE.convertFrom(getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE),
ByteUnit.MiB)
+
+ val bins = shuffleStats.toSeq
+ .flatMap(_._2)
+ .groupBy(_._1.asInstanceOf[ShuffleBlockId].reduceId)
+ .flatMap {
+ case (_, blocks) =>
+ BinPackingUtils.binPackBySize[(BlockId, Long, Int), BlockId](
+ blocks,
+ _._2, // size
+ _._1, // blockId
+ maxBinSize)
+ }
+
+ bins
+ .map {
+ bin =>
+ var binSize = 0L
+ val blockLocations =
+ new mutable.HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long,
Int)]]()
+ for (blockId <- bin) {
+ val (bmId, size, index) = blockInfo(blockId)
+ binSize += size
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId,
Long, Int)]())
+ blocksAtBM.append((blockId, size, index))
+ }
+ (binSize, blockLocations.toList)
+ }
+ .toArray
+ .sortBy(_._1)(Ordering[Long].reverse) // submit largest blocks first
+ .map(_._2)
+ }
+
+ private def getShuffleStats(): Array[(BlockManagerId,
collection.Seq[(BlockId, Long, Int)])] = {
+ val dep = getShuffleRDD.dependency
+
+ // Gets the shuffle output stats
+ def getStats =
+ mapTracker.getMapSizesByExecutorId(dep.shuffleId, 0, Int.MaxValue, 0,
numPartitions).toArray
+
+ // Executes the shuffle map stage in case we are missing output stats
+ def awaitShuffleMapStage(): Unit = {
+ assert(dep != null, "Shuffle dependency should not be null")
+ // hack to materialize the shuffle files in a fault tolerant way
+ ThreadUtils.awaitResult(sparkContext.submitMapStage(dep), Duration.Inf)
+ }
+
+ try {
+ val res = getStats
+ if (res.isEmpty) awaitShuffleMapStage()
+ getStats
+ } catch {
+ case e: FetchFailedException =>
+ logWarning("Failed to fetch shuffle blocks for the optimized writer.
Retrying", e)
+ awaitShuffleMapStage()
+ getStats
+ }
+ }
+
+ override protected def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ if (childNumPartitions <= 1) return inputColumnarRDD
+ val shuffledRDD = getShuffleRDD
+ val partitions = computeBins()
+
+ recordDeltaEvent(
+ deltaLog,
+ "delta.optimizeWrite.planned",
+ data = Map(
+ "originalPartitions" -> childNumPartitions,
+ "outputPartitions" -> partitions.length,
+ "shufflePartitions" -> numPartitions,
+ "numShuffleBlocks" ->
getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_SHUFFLE_BLOCKS),
+ "binSize" -> getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_BIN_SIZE),
+ "maxShufflePartitions" ->
+ getConf(DeltaSQLConf.DELTA_OPTIMIZE_WRITE_MAX_SHUFFLE_PARTITIONS)
+ )
+ )
+ new ColumnarDeltaOptimizedWriterRDD(
+ sparkContext,
+ shuffledRDD.dependency,
+ readMetrics,
+ new OptimizedWriterBlocks(partitions))
+ }
+
+ private def getConf[T](entry: ConfigEntry[T]): T = {
+ conf.getConf(entry)
+ }
+
+ override protected def withNewChildInternal(
+ newChild: SparkPlan): DeltaOptimizedWriterTransformer =
+ copy(child = newChild)
+
+ override def batchType(): Convention.BatchType =
BackendsApiManager.getSettings.primaryBatchType
+
+ override def rowType0(): Convention.RowType = Convention.RowType.None
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ child.execute()
+ }
+}
+
+private class ColumnarDeltaOptimizedWriterRDD(
+ @transient sparkContext: SparkContext,
+ var dep: ShuffleDependency[Int, ColumnarBatch, ColumnarBatch],
+ metrics: Map[String, SQLMetric],
+ @transient blocks: OptimizedWriterBlocks
+) extends RDD[ColumnarBatch](sparkContext, Seq(dep))
+ with DeltaLogging {
+
+ override protected def getPartitions: Array[Partition] =
Array.tabulate(blocks.bins.length) {
+ i => ColumnarShuffleBlockRDDPartition(i, blocks.bins(i))
+ }
+
+ override def compute(split: Partition, context: TaskContext):
Iterator[ColumnarBatch] = {
+ val tempMetrics = context.taskMetrics().createTempShuffleReadMetrics()
+ val sqlMetricsReporter = new
SQLColumnarShuffleReadMetricsReporter(tempMetrics, metrics)
+
+ val blocks = if (context.stageAttemptNumber() > 0) {
+ // We lost shuffle blocks, so we need to now get new manager addresses
+ val executorTracker = SparkEnv.get.mapOutputTracker
+ val oldBlockLocations =
split.asInstanceOf[ColumnarShuffleBlockRDDPartition].blocks
+
+ // assumes we bin-pack by reducerId
+ val reducerId =
oldBlockLocations.head._2.head._1.asInstanceOf[ShuffleBlockId].reduceId
+ // Get block addresses
+ val newLocations = executorTracker
+ .getMapSizesByExecutorId(dep.shuffleId, reducerId)
+ .flatMap {
+ case (bmId, newBlocks) =>
+ newBlocks.map(blockInfo => (blockInfo._3, (bmId, blockInfo)))
+ }
+ .toMap
+
+ val blockLocations = new mutable.HashMap[BlockManagerId,
ArrayBuffer[(BlockId, Long, Int)]]()
+ oldBlockLocations.foreach {
+ case (_, oldBlocks) =>
+ oldBlocks.foreach {
+ oldBlock =>
+ val (bmId, blockInfo) = newLocations(oldBlock._3)
+ val blocksAtBM =
+ blockLocations.getOrElseUpdate(bmId, new ArrayBuffer[(BlockId,
Long, Int)]())
+ blocksAtBM.append(blockInfo)
+ }
+ }
+
+ blockLocations.iterator
+ } else {
+ split.asInstanceOf[ColumnarShuffleBlockRDDPartition].blocks.iterator
+ }
+
+ val serializerManager =
+ if (dep.shuffleHandle.isInstanceOf[ColumnarShuffleHandle[_, _]]) {
+ ColumnarShuffleManager.bypassDecompressionSerializerManger
+ } else {
+ SparkEnv.get.serializerManager
+ }
+
+ val reader =
+ new OptimizedWriterShuffleReader(dep, context, blocks,
sqlMetricsReporter, serializerManager)
+ reader.read().map(_._2)
+ }
+}
+
+private case class ColumnarShuffleBlockRDDPartition(
+ index: Int,
+ blocks: List[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])])
+ extends Partition
+
+/** A simplified implementation of the `BlockStoreShuffleReader` for reading
shuffle blocks. */
+private class OptimizedWriterShuffleReader(
+ dep: ShuffleDependency[Int, _, ColumnarBatch],
+ context: TaskContext,
+ blocks: Iterator[(BlockManagerId, ArrayBuffer[(BlockId, Long, Int)])],
+ readMetrics: ShuffleReadMetricsReporter,
+ serializerManager: SerializerManager)
+ extends ShuffleReader[Int, ColumnarBatch] {
+
+ /** Read the combined key-values for this reduce task */
+ override def read(): Iterator[Product2[Int, ColumnarBatch]] = {
+
+ val wrappedStreams = new ShuffleBlockFetcherIterator(
+ context,
+ SparkEnv.get.blockManager.blockStoreClient,
+ SparkEnv.get.blockManager,
+ SparkEnv.get.mapOutputTracker,
+ blocks,
+ serializerManager.wrapStream,
+ // Note: we use getSizeAsMb when no suffix is provided for backwards
compatibility
+ SparkEnv.get.conf.getSizeAsMb("spark.reducer.maxSizeInFlight", "48m") *
1024 * 1024,
+ SparkEnv.get.conf.getInt("spark.reducer.maxReqsInFlight", Int.MaxValue),
+ SparkEnv.get.conf.get(config.REDUCER_MAX_BLOCKS_IN_FLIGHT_PER_ADDRESS),
+ SparkEnv.get.conf.get(config.MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM),
+ SparkEnv.get.conf.get(config.SHUFFLE_MAX_ATTEMPTS_ON_NETTY_OOM),
+ SparkEnv.get.conf.getBoolean("spark.shuffle.detectCorrupt", true),
+
SparkEnv.get.conf.getBoolean("spark.shuffle.detectCorrupt.useExtraMemory",
false),
+ SparkEnv.get.conf.getBoolean("spark.shuffle.checksum.enabled", true),
+ SparkEnv.get.conf.get("spark.shuffle.checksum.algorithm", "ADLER32"),
+ readMetrics,
+ false
+ )
+
+ val serializerInstance = dep.serializer.newInstance()
+
+ // Create a key/value iterator for each stream
+ val recordIter = wrappedStreams
+ .flatMap {
+ case (_, wrappedStream) =>
+ // Note: the asKeyValueIterator below wraps a key/value iterator
inside of a
+ // NextIterator. The NextIterator makes sure that close() is called
on the
+ // underlying InputStream when all records have been read.
+
serializerInstance.deserializeStream(wrappedStream).asKeyValueIterator
+ }
+ .asInstanceOf[Iterator[Product2[Int, ColumnarBatch]]]
+
+ new InterruptibleIterator[Product2[Int, ColumnarBatch]](context,
recordIter)
+ }
+}
+
+object DeltaOptimizedWriterTransformer {
+ def support(plan: SparkPlan): Boolean =
plan.isInstanceOf[DeltaOptimizedWriterExec]
+
+ def from(plan: SparkPlan): DeltaOptimizedWriterTransformer = {
+ plan match {
+ case p: DeltaOptimizedWriterExec =>
+ DeltaOptimizedWriterTransformer(p.child, p.partitionColumns,
p.deltaLog)
+ case _ =>
+ throw new UnsupportedOperationException(
+ s"Can't transform ColumnarDeltaOptimizedWriterExec from
${plan.getClass.getSimpleName}")
+ }
+ }
+}
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala
b/gluten-substrait/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala
index 18e8b985b8..798dfcb3a4 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/shuffle/sort/ColumnarShuffleManager.scala
@@ -186,7 +186,7 @@ object ColumnarShuffleManager extends Logging {
executorComponents
}
- private def bypassDecompressionSerializerManger =
+ def bypassDecompressionSerializerManger: SerializerManager =
new SerializerManager(
SparkEnv.get.serializer,
SparkEnv.get.conf,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
