This is an automated email from the ASF dual-hosted git repository.
zhangzc 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 ed2657420 [GLUTEN-6499][CH] Support soft affinity for mergetree (#6545)
ed2657420 is described below
commit ed26574201c8db8e51623743b0e3599bea8a6006
Author: Shuai li <[email protected]>
AuthorDate: Tue Jul 23 13:06:52 2024 +0800
[GLUTEN-6499][CH] Support soft affinity for mergetree (#6545)
[CH] Support soft affinity for mergetree
---
.../org/apache/spark/affinity/CHAffinity.scala | 6 +
.../utils/MergeTreePartsPartitionsUtil.scala | 119 ++++++++++++++++----
.../org/apache/gluten/affinity/CHUTAffinity.scala | 40 +++++++
...lutenClickhouseMergetreeSoftAffinitySuite.scala | 123 +++++++++++++++++++++
4 files changed, 267 insertions(+), 21 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/spark/affinity/CHAffinity.scala
b/backends-clickhouse/src/main/scala/org/apache/spark/affinity/CHAffinity.scala
index 9f0c2d7c3..0cba09e19 100644
---
a/backends-clickhouse/src/main/scala/org/apache/spark/affinity/CHAffinity.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/spark/affinity/CHAffinity.scala
@@ -35,6 +35,12 @@ abstract class MixedAffinity(manager: AffinityManager)
extends Affinity(manager)
getHostLocations(filePartition.relativeTablePath + "/" +
filePartition.partList(0).name)
}
+ def getNativeMergeTreePartLocations(
+ partName: String,
+ relativeTablePath: String): Array[String] = {
+ getHostLocations(relativeTablePath + "/" + partName)
+ }
+
def getHostLocations(filePath: String): Array[String] = {
if (manager.usingSoftAffinity) {
internalGetHostLocations(filePath)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/datasources/utils/MergeTreePartsPartitionsUtil.scala
b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/datasources/utils/MergeTreePartsPartitionsUtil.scala
index 228dc9feb..64e41778c 100644
---
a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/datasources/utils/MergeTreePartsPartitionsUtil.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/datasources/utils/MergeTreePartsPartitionsUtil.scala
@@ -19,11 +19,13 @@ package org.apache.spark.sql.execution.datasources.utils
import org.apache.gluten.backendsapi.clickhouse.CHBackendSettings
import org.apache.gluten.execution.{GlutenMergeTreePartition,
MergeTreePartRange, MergeTreePartSplit}
import org.apache.gluten.expression.{ConverterUtils, ExpressionConverter}
+import org.apache.gluten.softaffinity.SoftAffinityManager
import org.apache.gluten.substrait.`type`.ColumnTypeNode
import org.apache.gluten.substrait.SubstraitContext
import org.apache.gluten.substrait.extensions.ExtensionBuilder
import org.apache.gluten.substrait.rel.{ExtensionTableBuilder, RelBuilder}
+import org.apache.spark.affinity.CHAffinity
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.catalog.BucketSpec
@@ -45,6 +47,7 @@ import com.google.protobuf.{Any, StringValue}
import io.substrait.proto.Plan
import java.lang.{Long => JLong}
+import java.util
import java.util.{ArrayList => JArrayList}
import scala.collection.JavaConverters._
@@ -231,13 +234,53 @@ object MergeTreePartsPartitionsUtil extends Logging {
size * part.size / part.marks)
}
}
- .sortBy(_.bytesOnDisk)(implicitly[Ordering[Long]].reverse)
- var currentSize = 0L
- val currentFiles = new ArrayBuffer[MergeTreePartSplit]
+ val openCostInBytes = sparkSession.sessionState.conf.filesOpenCostInBytes
+ val (partNameWithLocation, locationDistinct) =
+ calculatedLocationForSoftAffinity(splitFiles, relativeTablePath)
+
+ genInputPartitionSeqBySplitFiles(
+ engine,
+ database,
+ tableName,
+ snapshotId,
+ relativeTablePath,
+ absoluteTablePath,
+ tableSchemaJson,
+ partitions,
+ table,
+ clickhouseTableConfigs,
+ splitFiles,
+ openCostInBytes,
+ maxSplitBytes,
+ partNameWithLocation,
+ locationDistinct
+ )
+ }
+
+ def genInputPartitionSeqBySplitFiles(
+ engine: String,
+ database: String,
+ tableName: String,
+ snapshotId: String,
+ relativeTablePath: String,
+ absoluteTablePath: String,
+ tableSchemaJson: String,
+ partitions: ArrayBuffer[InputPartition],
+ table: ClickHouseTableV2,
+ clickhouseTableConfigs: Map[String, String],
+ splitFiles: Seq[MergeTreePartSplit],
+ openCostInBytes: Long,
+ maxSplitBytes: Long,
+ partNameWithLocation: util.HashMap[String, String],
+ locationDistinct: util.HashSet[String]): Unit = {
+
+ val currentSizeByLocation = new util.HashMap[String, Long]
+ val currentFilesByLocation = new util.HashMap[String,
ArrayBuffer[MergeTreePartSplit]]
/** Close the current partition and move to the next. */
- def closePartition(): Unit = {
+ def closePartition(location: String): Unit = {
+ val currentFiles: ArrayBuffer[MergeTreePartSplit] =
currentFilesByLocation.get(location)
if (currentFiles.nonEmpty) {
val newPartition = GlutenMergeTreePartition(
partitions.size,
@@ -260,23 +303,35 @@ object MergeTreePartsPartitionsUtil extends Logging {
partitions += newPartition
}
currentFiles.clear()
- currentSize = 0
+ currentSizeByLocation.put(location, 0)
}
// generate `Seq[InputPartition]` by file size
- val openCostInBytes = sparkSession.sessionState.conf.filesOpenCostInBytes
- // val maxSplitBytes =
sparkSession.sessionState.conf.filesMaxPartitionBytes
// Assign files to partitions using "Next Fit Decreasing"
+ locationDistinct.forEach(
+ location => {
+ currentSizeByLocation.put(location, 0L)
+ currentFilesByLocation.put(location, new
ArrayBuffer[MergeTreePartSplit])
+ })
+
splitFiles.foreach {
parts =>
- if ((currentSize + parts.bytesOnDisk > maxSplitBytes)) {
- closePartition()
+ {
+ val location = partNameWithLocation.get(parts.name)
+ var currentSize = currentSizeByLocation.get(location)
+ val currentFiles = currentFilesByLocation.get(location)
+
+ if (currentSize + parts.bytesOnDisk > maxSplitBytes) {
+ closePartition(location)
+ currentSize = 0L
+ }
+ // Add the given file to the current partition.
+ currentSizeByLocation.put(location, currentSize + parts.bytesOnDisk
+ openCostInBytes)
+ currentFiles += parts
}
- // Add the given file to the current partition.
- currentSize += parts.bytesOnDisk + openCostInBytes
- currentFiles += parts
}
- closePartition()
+
+ locationDistinct.forEach(closePartition)
}
/** Generate bucket partition */
@@ -397,7 +452,17 @@ object MergeTreePartsPartitionsUtil extends Logging {
}
}
- def getMergeTreePartRange(
+ private def useDriverFilter(filterExprs: Seq[Expression], sparkSession:
SparkSession): Boolean = {
+ val enableDriverFilterKey =
s"${CHBackendSettings.getBackendConfigPrefix}.runtime_settings" +
+ s".enabled_driver_filter_mergetree_index"
+
+ // When using soft affinity, disable driver filter
+ filterExprs.nonEmpty && sparkSession.sessionState.conf.getConfString(
+ enableDriverFilterKey,
+ "false") == "true" && !SoftAffinityManager.usingSoftAffinity
+ }
+
+ private def getMergeTreePartRange(
selectPartsFiles: Seq[AddMergeTreeParts],
snapshotId: String,
database: String,
@@ -410,14 +475,8 @@ object MergeTreePartsPartitionsUtil extends Logging {
filterExprs: Seq[Expression],
output: Seq[Attribute],
sparkSession: SparkSession): Seq[MergeTreePartRange] = {
- val enableDriverFilter =
s"${CHBackendSettings.getBackendConfigPrefix}.runtime_settings" +
- s".enabled_driver_filter_mergetree_index"
- if (
- filterExprs.nonEmpty && sparkSession.sessionState.conf.getConfString(
- enableDriverFilter,
- "false") == "true"
- ) {
+ if (useDriverFilter(filterExprs, sparkSession)) {
val size_per_mark = selectPartsFiles.map(part => (part.size,
part.marks)).unzip match {
case (l1, l2) => l1.sum / l2.sum
}
@@ -535,6 +594,24 @@ object MergeTreePartsPartitionsUtil extends Logging {
}
}
+ private def calculatedLocationForSoftAffinity(
+ splits: Seq[MergeTreePartSplit],
+ relativeTablePath: String): (util.HashMap[String, String],
util.HashSet[String]) = {
+ val partNameWithLocation = new util.HashMap[String, String]()
+ val locationDistinct = new util.HashSet[String]()
+
+ splits.foreach(
+ part => {
+ if (!partNameWithLocation.containsKey(part.name)) {
+ val locations =
CHAffinity.getNativeMergeTreePartLocations(part.name, relativeTablePath)
+ val localtionKey = locations.sorted.mkString(",")
+ locationDistinct.add(localtionKey)
+ partNameWithLocation.put(part.name, localtionKey)
+ }
+ })
+ (partNameWithLocation, locationDistinct)
+ }
+
def getMaxSplitBytes(
sparkSession: SparkSession,
selectedRanges: Seq[MergeTreePartRange]): Long = {
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/affinity/CHUTAffinity.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/affinity/CHUTAffinity.scala
new file mode 100644
index 000000000..c7d77e550
--- /dev/null
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/affinity/CHUTAffinity.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.affinity
+
+import org.apache.gluten.GlutenConfig
+import org.apache.gluten.backendsapi.clickhouse.CHBackendSettings
+import org.apache.gluten.softaffinity.AffinityManager
+
+import org.apache.spark.affinity.MixedAffinity
+
+/** Get the locations by SoftAffinityManager */
+object CHUTAffinity extends MixedAffinity(CHUTSoftAffinityManager) {
+ override lazy val affinityMode: String = CHBackendSettings.affinityMode
+}
+
+object CHUTSoftAffinityManager extends AffinityManager {
+ override lazy val usingSoftAffinity: Boolean = true
+
+ override lazy val minOnTargetHosts: Int =
+ GlutenConfig.GLUTEN_SOFT_AFFINITY_MIN_TARGET_HOSTS_DEFAULT_VALUE
+
+ override lazy val detectDuplicateReading = true
+
+ override lazy val maxDuplicateReadingRecords =
+
GlutenConfig.GLUTEN_SOFT_AFFINITY_MAX_DUPLICATE_READING_RECORDS_DEFAULT_VALUE
+}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickhouseMergetreeSoftAffinitySuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickhouseMergetreeSoftAffinitySuite.scala
new file mode 100644
index 000000000..d5620f5df
--- /dev/null
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickhouseMergetreeSoftAffinitySuite.scala
@@ -0,0 +1,123 @@
+/*
+ * 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.gluten.affinity.{CHUTAffinity, CHUTSoftAffinityManager}
+
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.delta.catalog.ClickHouseTableV2
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import
org.apache.spark.sql.execution.datasources.utils.MergeTreePartsPartitionsUtil
+
+import org.apache.hadoop.fs.Path
+
+import java.util
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+class GlutenClickhouseMergetreeSoftAffinitySuite
+ extends GlutenClickHouseTPCHAbstractSuite
+ with AdaptiveSparkPlanHelper {
+
+ override protected val tablesPath: String = basePath + "/tpch-data"
+ override protected val tpchQueries: String = rootPath +
"queries/tpch-queries-ch"
+ override protected val queriesResults: String = rootPath +
"mergetree-queries-output"
+
+ override protected def createTPCHNotNullTables(): Unit = {
+ createNotNullTPCHTablesInParquet(tablesPath)
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ assertResult(0)(CHUTSoftAffinityManager.nodesExecutorsMap.size)
+ CHUTSoftAffinityManager.handleExecutorAdded(("1", "host-1"))
+ CHUTSoftAffinityManager.handleExecutorAdded(("2", "host-2"))
+ CHUTSoftAffinityManager.handleExecutorAdded(("3", "host-3"))
+ }
+
+ override def afterAll(): Unit = {
+ super.afterAll()
+ CHUTSoftAffinityManager.handleExecutorRemoved("1")
+ CHUTSoftAffinityManager.handleExecutorRemoved("2")
+ CHUTSoftAffinityManager.handleExecutorRemoved("3")
+ assertResult(0)(CHUTSoftAffinityManager.nodesExecutorsMap.size)
+ }
+
+ test("Soft Affinity Scheduler with duplicate reading detection") {
+
+ val partitions: ArrayBuffer[InputPartition] = new
ArrayBuffer[InputPartition]()
+ var splitFiles: Seq[MergeTreePartSplit] = Seq()
+ val relativeTablePath = "tmp/"
+
+ for (i <- 1 to 10) {
+ splitFiles = splitFiles :+ MergeTreePartSplit(i.toString, i.toString,
i.toString, i, 30L, 40L)
+ }
+
+ val (partNameWithLocation, locationDistinct) =
+ calculatedLocationForSoftAffinity(splitFiles, relativeTablePath)
+
+ MergeTreePartsPartitionsUtil.genInputPartitionSeqBySplitFiles(
+ "mergetree",
+ "test",
+ "test_table",
+ "123",
+ relativeTablePath,
+ "/tmp",
+ "",
+ partitions,
+ new ClickHouseTableV2(spark, new Path("/")),
+ mutable.Map[String, String]().toMap,
+ splitFiles,
+ 1,
+ 1000,
+ partNameWithLocation,
+ locationDistinct
+ )
+
+ assertResult(3)(partitions.size)
+
+ for (partition <- partitions) {
+ val names =
+ partition
+ .asInstanceOf[GlutenMergeTreePartition]
+ .partList
+ .map(_.name.toInt)
+ .sorted
+ .mkString(",")
+ assert(names == "1,4,7,10" | names == "2,5,8" || names == "3,6,9")
+ }
+ }
+
+ def calculatedLocationForSoftAffinity(
+ splits: Seq[MergeTreePartSplit],
+ relativeTablePath: String): (util.HashMap[String, String],
util.HashSet[String]) = {
+ val partNameWithLocation = new util.HashMap[String, String]()
+ val locationDistinct = new util.HashSet[String]()
+
+ splits.foreach(
+ part => {
+ if (!partNameWithLocation.containsKey(part.name)) {
+ val locations =
CHUTAffinity.getNativeMergeTreePartLocations(part.name, relativeTablePath)
+ val localtionKey = locations.sorted.mkString(",")
+ locationDistinct.add(localtionKey)
+ partNameWithLocation.put(part.name, localtionKey)
+ }
+ })
+ (partNameWithLocation, locationDistinct)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
