This is an automated email from the ASF dual-hosted git repository. taiyang-li pushed a commit to branch fake_add_bolt_backend in repository https://gitbox.apache.org/repos/asf/gluten.git
commit 5ae169adc978041f66848457c04cbd8ff37aa7ea Author: liyang.127 <[email protected]> AuthorDate: Mon Jun 29 21:49:45 2026 +0800 [GLUTEN][CORE] Extract shared Celeborn factory bases Move the duplicated Celeborn serializer and shuffle writer factory scaffolding into backends-core so backend modules only keep the backend-specific class name and writer construction logic. --- .../CHCelebornColumnarBatchSerializerFactory.scala | 10 +++----- .../CHCelebornColumnarShuffleWriterFactory.scala | 5 ++-- backends-core/pom.xml | 30 ++++++++++++++++++++++ ...redCelebornColumnarBatchSerializerFactory.scala | 6 ++--- ...haredCelebornColumnarShuffleWriterFactory.scala | 20 ++++++++------- ...loxCelebornColumnarBatchSerializerFactory.scala | 10 +++----- ...VeloxCelebornColumnarShuffleWriterFactory.scala | 5 ++-- 7 files changed, 54 insertions(+), 32 deletions(-) diff --git a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarBatchSerializerFactory.scala b/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarBatchSerializerFactory.scala index 8eb40c0c7f..11b11eb0bc 100644 --- a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarBatchSerializerFactory.scala +++ b/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarBatchSerializerFactory.scala @@ -16,10 +16,6 @@ */ package org.apache.spark.shuffle -import org.apache.spark.shuffle.gluten.celeborn.CelebornColumnarBatchSerializerFactory - -class CHCelebornColumnarBatchSerializerFactory extends CelebornColumnarBatchSerializerFactory { - - override def columnarBatchSerializerClass(): String = - "org.apache.spark.shuffle.CHCelebornColumnarBatchSerializer" -} +class CHCelebornColumnarBatchSerializerFactory + extends SharedCelebornColumnarBatchSerializerFactory( + "org.apache.spark.shuffle.CHCelebornColumnarBatchSerializer") diff --git a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriterFactory.scala b/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriterFactory.scala index d8393b0932..d81d1cd21f 100644 --- a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriterFactory.scala +++ b/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriterFactory.scala @@ -18,14 +18,13 @@ package org.apache.spark.shuffle import org.apache.spark.TaskContext import org.apache.spark.shuffle.celeborn.CelebornShuffleHandle -import org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleWriterFactory import org.apache.celeborn.client.ShuffleClient import org.apache.celeborn.common.CelebornConf -class CHCelebornColumnarShuffleWriterFactory extends CelebornShuffleWriterFactory { +class CHCelebornColumnarShuffleWriterFactory extends SharedCelebornColumnarShuffleWriterFactory { - override def createShuffleWriterInstance[K, V]( + override protected def createBackendShuffleWriter[K, V]( shuffleId: Int, handle: CelebornShuffleHandle[K, V, V], context: TaskContext, diff --git a/backends-core/pom.xml b/backends-core/pom.xml index 113cbed416..8903c9a45a 100644 --- a/backends-core/pom.xml +++ b/backends-core/pom.xml @@ -94,6 +94,36 @@ </build> <profiles> + <profile> + <id>celeborn</id> + <activation> + <activeByDefault>false</activeByDefault> + </activation> + <dependencies> + <dependency> + <groupId>org.apache.gluten</groupId> + <artifactId>gluten-celeborn</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.celeborn</groupId> + <artifactId>celeborn-client-spark-${spark.major.version}-shaded_${scala.binary.version}</artifactId> + <version>${celeborn.version}</version> + <scope>provided</scope> + <exclusions> + <exclusion> + <groupId>org.apache.celeborn</groupId> + <artifactId>celeborn-client-spark-${spark.major.version}_${scala.binary.version}</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.celeborn</groupId> + <artifactId>celeborn-spark-${spark.major.version}-columnar-shuffle_${scala.binary.version}</artifactId> + </exclusion> + </exclusions> + </dependency> + </dependencies> + </profile> <profile> <id>iceberg</id> <activation> diff --git a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarBatchSerializerFactory.scala b/backends-core/src-celeborn/main/scala/org/apache/spark/shuffle/SharedCelebornColumnarBatchSerializerFactory.scala similarity index 80% copy from backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarBatchSerializerFactory.scala copy to backends-core/src-celeborn/main/scala/org/apache/spark/shuffle/SharedCelebornColumnarBatchSerializerFactory.scala index 8eb40c0c7f..d11f2d3d42 100644 --- a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarBatchSerializerFactory.scala +++ b/backends-core/src-celeborn/main/scala/org/apache/spark/shuffle/SharedCelebornColumnarBatchSerializerFactory.scala @@ -18,8 +18,8 @@ package org.apache.spark.shuffle import org.apache.spark.shuffle.gluten.celeborn.CelebornColumnarBatchSerializerFactory -class CHCelebornColumnarBatchSerializerFactory extends CelebornColumnarBatchSerializerFactory { +abstract class SharedCelebornColumnarBatchSerializerFactory(serializerClassName: String) + extends CelebornColumnarBatchSerializerFactory { - override def columnarBatchSerializerClass(): String = - "org.apache.spark.shuffle.CHCelebornColumnarBatchSerializer" + final override def columnarBatchSerializerClass(): String = serializerClassName } diff --git a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriterFactory.scala b/backends-core/src-celeborn/main/scala/org/apache/spark/shuffle/SharedCelebornColumnarShuffleWriterFactory.scala similarity index 71% copy from backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriterFactory.scala copy to backends-core/src-celeborn/main/scala/org/apache/spark/shuffle/SharedCelebornColumnarShuffleWriterFactory.scala index d8393b0932..d76bdca5c3 100644 --- a/backends-clickhouse/src-celeborn/main/scala/org/apache/spark/shuffle/CHCelebornColumnarShuffleWriterFactory.scala +++ b/backends-core/src-celeborn/main/scala/org/apache/spark/shuffle/SharedCelebornColumnarShuffleWriterFactory.scala @@ -23,21 +23,23 @@ import org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleWriterFactory import org.apache.celeborn.client.ShuffleClient import org.apache.celeborn.common.CelebornConf -class CHCelebornColumnarShuffleWriterFactory extends CelebornShuffleWriterFactory { +abstract class SharedCelebornColumnarShuffleWriterFactory extends CelebornShuffleWriterFactory { - override def createShuffleWriterInstance[K, V]( + protected def createBackendShuffleWriter[K, V]( + shuffleId: Int, + handle: CelebornShuffleHandle[K, V, V], + context: TaskContext, + celebornConf: CelebornConf, + client: ShuffleClient, + writeMetrics: ShuffleWriteMetricsReporter): ShuffleWriter[K, V] + + final override def createShuffleWriterInstance[K, V]( shuffleId: Int, handle: CelebornShuffleHandle[K, V, V], context: TaskContext, celebornConf: CelebornConf, client: ShuffleClient, writeMetrics: ShuffleWriteMetricsReporter): ShuffleWriter[K, V] = { - new CHCelebornColumnarShuffleWriter[K, V]( - shuffleId, - handle, - context, - celebornConf, - client, - writeMetrics) + createBackendShuffleWriter(shuffleId, handle, context, celebornConf, client, writeMetrics) } } diff --git a/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarBatchSerializerFactory.scala b/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarBatchSerializerFactory.scala index f763d392ac..726e192310 100644 --- a/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarBatchSerializerFactory.scala +++ b/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarBatchSerializerFactory.scala @@ -16,10 +16,6 @@ */ package org.apache.spark.shuffle -import org.apache.spark.shuffle.gluten.celeborn.CelebornColumnarBatchSerializerFactory - -class VeloxCelebornColumnarBatchSerializerFactory extends CelebornColumnarBatchSerializerFactory { - - override def columnarBatchSerializerClass(): String = - "org.apache.spark.shuffle.CelebornColumnarBatchSerializer" -} +class VeloxCelebornColumnarBatchSerializerFactory + extends SharedCelebornColumnarBatchSerializerFactory( + "org.apache.spark.shuffle.CelebornColumnarBatchSerializer") diff --git a/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarShuffleWriterFactory.scala b/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarShuffleWriterFactory.scala index e09d601ab9..7b975dc6b1 100644 --- a/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarShuffleWriterFactory.scala +++ b/backends-velox/src-celeborn/main/scala/org/apache/spark/shuffle/VeloxCelebornColumnarShuffleWriterFactory.scala @@ -18,14 +18,13 @@ package org.apache.spark.shuffle import org.apache.spark.TaskContext import org.apache.spark.shuffle.celeborn.CelebornShuffleHandle -import org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleWriterFactory import org.apache.celeborn.client.ShuffleClient import org.apache.celeborn.common.CelebornConf -class VeloxCelebornColumnarShuffleWriterFactory extends CelebornShuffleWriterFactory { +class VeloxCelebornColumnarShuffleWriterFactory extends SharedCelebornColumnarShuffleWriterFactory { - override def createShuffleWriterInstance[K, V]( + override protected def createBackendShuffleWriter[K, V]( shuffleId: Int, handle: CelebornShuffleHandle[K, V, V], context: TaskContext, --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
