ziting-openai commented on code in PR #5537: URL: https://github.com/apache/datafusion-comet/pull/5537#discussion_r3887888314
########## spark/src/test/scala/org/apache/spark/sql/comet/execution/shuffle/CometCelebornShufflePlanningSuite.scala: ########## @@ -0,0 +1,722 @@ +/* + * 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.comet.execution.shuffle + +import java.util.concurrent.atomic.AtomicInteger + +import org.apache.spark.{ShuffleDependency, SparkConf, SparkEnv} +import org.apache.spark.shuffle.ShuffleHandle +import org.apache.spark.shuffle.sort.SortShuffleManager +import org.apache.spark.sql.{CometTestBase, DataFrame, Row} +import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, AttributeReference, SortOrder} +import org.apache.spark.sql.catalyst.expressions.aggregate.{Final, Partial, PartialMerge} +import org.apache.spark.sql.catalyst.plans.logical.LocalRelation +import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, RangePartitioning, RoundRobinPartitioning, SinglePartition} +import org.apache.spark.sql.comet.{CometCollectLimitExec, CometHashAggregateExec, CometLocalTableScanExec, CometNativeExec, CometScanWrapper, CometSortExec, CometSparkToColumnarExec, CometTakeOrderedAndProjectExec} +import org.apache.spark.sql.execution.{CollectLimitExec, ColumnarToRowTransition, LocalTableScanExec, SortExec, SparkPlan, TakeOrderedAndProjectExec} +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec +import org.apache.spark.sql.execution.aggregate.BaseAggregateExec +import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec +import org.apache.spark.sql.functions.{col, expr} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{ArrayType, IntegerType, LongType, ObjectType} + +import org.apache.comet.{CometConf, CometExplainInfo} +import org.apache.comet.CometSparkSessionExtensions.{isCometShuffleEnabled, isCometShuffleManagerEnabled} +import org.apache.comet.rules.{CometExecRule, RevertNativeForTransitionHeavyStages} +import org.apache.comet.serde.{Compatible, OperatorOuterClass} + +/** + * Plans against the actual composite manager without requiring the optional Celeborn client. Only + * Spark-fallback queries execute: the local test backend must never receive a native dependency. + * Native plan assertions below are not transport or live-cluster integration tests. + */ +class CometCelebornShufflePlanningSuite extends CometTestBase { + + override protected val shuffleManager: String = + classOf[CometCelebornPlanningTestShuffleManager].getName + + override protected def sparkConf: SparkConf = + super.sparkConf + .set(CometConf.COMET_SHUFFLE_MODE.key, "auto") + .set(CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key, "false") + .set("spark.io.encryption.enabled", "false") + .set("spark.celeborn.client.spark.stageRerun.enabled", "true") + + private def manager: CometCelebornPlanningTestShuffleManager = + SparkEnv.get.shuffleManager.asInstanceOf[CometCelebornPlanningTestShuffleManager] + + private def nativeChild( + attributes: Seq[Attribute] = Seq(AttributeReference("value", LongType)())) + : CometNativeExec = { + val original = sparkLeaf(attributes) + CometScanWrapper(OperatorOuterClass.Operator.getDefaultInstance, original) + } + + // Let Spark supply its version-specific LocalTableScanExec constructor arguments. + private def sparkLeaf(attributes: Seq[Attribute] = Seq(AttributeReference("value", LongType)())) + : LocalTableScanExec = + spark.sessionState.planner + .plan(LocalRelation(attributes)) + .next() + .asInstanceOf[LocalTableScanExec] + + private def reasons(plan: SparkPlan): Set[String] = + plan.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String]) + + private def cometExchanges(plan: SparkPlan): Seq[CometShuffleExchangeExec] = + collect(plan) { case exchange: CometShuffleExchangeExec => exchange } + + private def assertSparkExchange(plan: SparkPlan): Unit = { + assert(cometExchanges(plan).isEmpty, s"unexpected Comet shuffle:\n$plan") + assert( + collect(plan) { case exchange: ShuffleExchangeExec => exchange }.nonEmpty, + s"expected a Spark shuffle:\n$plan") + } + + private def input: DataFrame = + spark.range(0, 32, 1, 4).selectExpr("id + 1 AS value") + + private def collectionAggregatePlan: SparkPlan = + input + .selectExpr("value % 3 AS grouping_key", "value") + .groupBy("grouping_key") + .agg(expr("collect_list(value)")) + .queryExecution + .executedPlan + + private def assertNativeExecutionLoaded(): Unit = { + val plan = input.queryExecution.executedPlan + assert(collect(plan) { case op: CometNativeExec => op }.nonEmpty, s"$plan") + } + + private def specialOperators(child: SparkPlan): (CollectLimitExec, TakeOrderedAndProjectExec) = + ( + CollectLimitExec(3, child), + TakeOrderedAndProjectExec( + 3, + Seq(SortOrder(child.output.head, Ascending)), + child.output, + child)) + + private def assertSpecialSupport(expected: Boolean): Unit = { + val (limit, topK) = specialOperators(nativeChild()) + assert(CometCollectLimitExec.getSupportLevel(limit).isInstanceOf[Compatible] == expected) + assert( + CometTakeOrderedAndProjectExec.getSupportLevel(topK).isInstanceOf[Compatible] == expected) + val rule = CometExecRule(spark) + assert(rule(limit).isInstanceOf[CometCollectLimitExec] == expected) + assert(rule(topK).isInstanceOf[CometTakeOrderedAndProjectExec] == expected) + } + + // Spark core configuration cannot be changed through RuntimeConfig. Deliberately modify the + // session's copy only, proving it cannot override the manager already owned by SparkEnv. + private def withSessionConfOverride(settings: (String, String)*)(f: => Unit): Unit = { + val conf = spark.sessionState.conf + val previous = settings.map { case (key, _) => key -> Option(conf.getConfString(key, null)) } + try { + settings.foreach { case (key, value) => conf.setConfString(key, value) } + f + } finally { + previous.foreach { + case (key, Some(value)) => conf.setConfString(key, value) + case (key, None) => conf.unsetConf(key) + } + } + } + + test("the actual composite manager loads Comet and default auto preserves Spark shuffle") { + val conf = spark.sessionState.conf + assert(isCometShuffleManagerEnabled(conf)) + assertNativeExecutionLoaded() + assert(CometConf.COMET_SHUFFLE_MODE.get(conf) == "auto") + assert(!isCometShuffleEnabled(conf)) + assertSpecialSupport(expected = false) + } + + test("native opt-in, execution, and shuffle flags gate exchanges and special producers") { + for { + mode <- Seq("auto", "jvm", "native") + nativeExecution <- Seq(false, true) + shuffleEnabled <- Seq(false, true) + } { + withSQLConf( + CometConf.COMET_SHUFFLE_MODE.key -> mode, + CometConf.COMET_EXEC_ENABLED.key -> nativeExecution.toString, + CometConf.COMET_SHUFFLE_ENABLED.key -> shuffleEnabled.toString) { + val expected = mode == "native" && nativeExecution && shuffleEnabled + assert(isCometShuffleEnabled(spark.sessionState.conf) == expected) + val exchange = ShuffleExchangeExec(SinglePartition, nativeChild()) + assert( + CometShuffleExchangeExec.shuffleSupported(exchange).contains(CometNativeShuffle) == + expected) + assertSpecialSupport(expected) + } + } + } + + test("supported native partitionings never select JVM Comet shuffle") { + withSQLConf( + CometConf.COMET_SHUFFLE_MODE.key -> "native", + CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key -> "true", + CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_ENABLED.key -> "true") { + val child = nativeChild() + val partitionings = Seq( + SinglePartition, + HashPartitioning(child.output, 2), + RangePartitioning(Seq(SortOrder(child.output.head, Ascending)), 2), + RoundRobinPartitioning(2)) + partitionings.foreach { partitioning => + val exchange = ShuffleExchangeExec(partitioning, child) + assert(CometShuffleExchangeExec.shuffleSupported(exchange).contains(CometNativeShuffle)) + assert(reasons(exchange).isEmpty) + } + } + } + + test("unsupported native partitioning falls back without trying Comet columnar shuffle") { + withSQLConf( + CometConf.COMET_SHUFFLE_MODE.key -> "native", + CometConf.COMET_SHUFFLE_NATIVE_HASH_PARTITIONING_ENABLED.key -> "false", + CometConf.COMET_SHUFFLE_NATIVE_RANGE_PARTITIONING_ENABLED.key -> "false", + CometConf.COMET_SHUFFLE_NATIVE_ROUND_ROBIN_PARTITIONING_ENABLED.key -> "false") { + val child = nativeChild() + val partitionings = Seq( + HashPartitioning(child.output, 2), + RangePartitioning(Seq(SortOrder(child.output.head, Ascending)), 2), + RoundRobinPartitioning(2)) + partitionings.foreach { partitioning => + val exchange = ShuffleExchangeExec(partitioning, child) + assert(CometShuffleExchangeExec.shuffleSupported(exchange).isEmpty) + assert(reasons(exchange).exists(_.contains("disabled"))) + assert(reasons(exchange).exists(_.contains("columnar shuffle"))) + } + } + } + + test("unsupported output and hash-key types retain the Spark exchange") { + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + val objectChild = + nativeChild(Seq(AttributeReference("value", ObjectType(classOf[String]))())) + val objectExchange = ShuffleExchangeExec(SinglePartition, objectChild) + assert(CometShuffleExchangeExec.shuffleSupported(objectExchange).isEmpty) + assert(reasons(objectExchange).exists(_.contains("unsupported shuffle data type"))) + + val arrayChild = nativeChild(Seq(AttributeReference("value", ArrayType(IntegerType))())) + val arrayExchange = ShuffleExchangeExec(HashPartitioning(arrayChild.output, 2), arrayChild) + assert(CometShuffleExchangeExec.shuffleSupported(arrayExchange).isEmpty) + assert(reasons(arrayExchange).exists(_.contains("unsupported hash partitioning data type"))) + } + } + + test("Spark and already-unwrapped Comet children do not enter the native createExec branch") { + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + val leaf = sparkLeaf() + val children = Seq( + leaf, + CometSparkToColumnarExec(leaf), + CometLocalTableScanExec(leaf, Nil, leaf.output), + CometCollectLimitExec(CollectLimitExec(3, leaf), 3, 0, leaf)) + children.foreach { child => + val exchange = ShuffleExchangeExec(SinglePartition, child) + assert(CometShuffleExchangeExec.shuffleSupported(exchange).isEmpty) + assert(reasons(exchange).exists(_.contains("Comet"))) + assert(CometExecRule(spark)(exchange).isInstanceOf[ShuffleExchangeExec]) + } + } + } + + test("fallback is sticky after AQE reshapes the child or native mode becomes available") { + val child = nativeChild() + val exchange = ShuffleExchangeExec(SinglePartition, child) + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "auto") { + assert(CometShuffleExchangeExec.shuffleSupported(exchange).isEmpty) + assert(reasons(exchange).nonEmpty) + } + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + val reshaped = + exchange.withNewChildren(Seq(nativeChild())).asInstanceOf[ShuffleExchangeExec] + assert(CometShuffleExchangeExec.shuffleSupported(reshaped).isEmpty) + assert(CometExecRule(spark)(reshaped).isInstanceOf[ShuffleExchangeExec]) + assert( + CometShuffleExchangeExec + .shuffleSupported(ShuffleExchangeExec(SinglePartition, child)) + .contains(CometNativeShuffle)) + } + } + + test("a session manager override cannot enable unsupported JVM Comet shuffle") { + withSessionConfOverride("spark.shuffle.manager" -> classOf[CometShuffleManager].getName) { + Seq("auto", "jvm").foreach { mode => + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> mode) { + assert(!isCometShuffleEnabled(spark.sessionState.conf)) + assert( + CometShuffleExchangeExec + .shuffleSupported(ShuffleExchangeExec(SinglePartition, nativeChild())) + .isEmpty) + assertSpecialSupport(expected = false) + } + } + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + assert(isCometShuffleEnabled(spark.sessionState.conf)) + } + } + } + + test("runtime encryption and disabled stage reruns cannot be masked by session overrides") { + val runtimeConfigurations = Seq( + new SparkConf(false).set("spark.io.encryption.enabled", "true"), + new SparkConf(false).set("spark.celeborn.client.spark.stageRerun.enabled", "false")) + runtimeConfigurations.foreach { runtimeConf => + val support = CometCelebornPlanningTestShuffleManager.planningSupport(runtimeConf) + manager.withPlanningSupport(support) { + withSessionConfOverride( + "spark.io.encryption.enabled" -> "false", + "spark.celeborn.client.spark.stageRerun.enabled" -> "true") { + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + assertNativeExecutionLoaded() + assert(!isCometShuffleEnabled(spark.sessionState.conf)) + assert( + CometShuffleExchangeExec + .shuffleSupported(ShuffleExchangeExec(SinglePartition, nativeChild())) + .isEmpty) + assertSpecialSupport(expected = false) + } + } + } + } + } + + test("forced local fallback protects special operators as well as ordinary exchanges") { + manager.withPlanningSupport(CelebornNativeShufflePlanningSupport(fallbackPolicy = "ALWAYS")) { + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + assert(!isCometShuffleEnabled(spark.sessionState.conf)) + assert( + CometShuffleExchangeExec + .shuffleSupported(ShuffleExchangeExec(SinglePartition, nativeChild())) + .isEmpty) + assertSpecialSupport(expected = false) + } + } + } + + test("fallback partition thresholds use the exchange's reducer count") { + manager.withPlanningSupport( + CelebornNativeShufflePlanningSupport(fallbackPartitionThreshold = 2L)) { + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + val child = nativeChild() + assert(isCometShuffleEnabled(spark.sessionState.conf)) + assertSpecialSupport(expected = true) + assert( + CometShuffleExchangeExec + .shuffleSupported(ShuffleExchangeExec(SinglePartition, child)) + .contains(CometNativeShuffle)) + val exchange = ShuffleExchangeExec(HashPartitioning(child.output, 2), child) + assert(CometShuffleExchangeExec.shuffleSupported(exchange).isEmpty) + assert(reasons(exchange).nonEmpty) + } + } + } + + test("single-partition thresholds also block CollectLimit and TakeOrdered") { + manager.withPlanningSupport( + CelebornNativeShufflePlanningSupport(fallbackPartitionThreshold = 1L)) { + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + assert(!isCometShuffleEnabled(spark.sessionState.conf)) + assertSpecialSupport(expected = false) + } + } + manager.withPlanningSupport( + CelebornNativeShufflePlanningSupport( + fallbackPolicy = "NEVER", + fallbackPartitionThreshold = 1L)) { + withSQLConf(CometConf.COMET_SHUFFLE_MODE.key -> "native") { + assert(isCometShuffleEnabled(spark.sessionState.conf)) + assertSpecialSupport(expected = true) + } + } + } + + for (skipShuffle <- Seq(false, true)) { + test( + s"aggregate fallback restores native ancestors and survives re-entry: skip=$skipShuffle") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_SHUFFLE_MODE.key -> "native") { + val partial = collect(collectionAggregatePlan) { + case aggregate: CometHashAggregateExec if aggregate.modes == Seq(Partial) => aggregate + }.head + val rule = CometExecRule(spark) + val sorted = + rule(SortExec(Seq(SortOrder(partial.output.head, Ascending)), false, partial)) + assert(sorted.isInstanceOf[CometSortExec], s"$sorted") + + val exchange = ShuffleExchangeExec(HashPartitioning(Seq(sorted.output.head), 4), sorted) + if (skipShuffle) { + exchange.setTagValue(CometExecRule.SKIP_COMET_SHUFFLE_TAG, ()) + } + var restored: SparkPlan = exchange + manager.withPlanningSupport( + CelebornNativeShufflePlanningSupport(fallbackPartitionThreshold = 2L)) { + restored = rule(exchange) + assertSparkExchange(restored) + assert(restored.children.head.isInstanceOf[SortExec], s"$restored") + assert(collect(restored) { case aggregate: CometHashAggregateExec => + aggregate + }.isEmpty) + val sparkPartial = collect(restored) { case aggregate: BaseAggregateExec => + aggregate + }.head + assert(sparkPartial.getTagValue(CometExecRule.COMET_UNSAFE_PARTIAL).isDefined) + } + + // Even if the runtime policy now permits native shuffle, AQE must not reconvert the + // producer after its exchange committed to Spark's intermediate buffer representation. + val reentered = rule(restored) + assertSparkExchange(reentered) + assert(reentered.children.head.isInstanceOf[SortExec], s"$reentered") + assert(collect(reentered) { case aggregate: CometHashAggregateExec => aggregate }.isEmpty) + } + } + } + + test("an outer fallback exchange preserves completed native aggregate stages") { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_SHUFFLE_MODE.key -> "native") { + val finalAggregate = collect(collectionAggregatePlan) { + case aggregate: CometHashAggregateExec if aggregate.modes == Seq(Final) => aggregate + }.head + val originalAggregates = collect(finalAggregate) { case aggregate: CometHashAggregateExec => + aggregate + } + assert(originalAggregates.exists(_.modes == Seq(Partial))) + + manager.withPlanningSupport( + CelebornNativeShufflePlanningSupport(fallbackPartitionThreshold = 2L)) { + val exchange = + ShuffleExchangeExec( + HashPartitioning(Seq(finalAggregate.output.head), 4), + finalAggregate) + val restored = CometExecRule(spark)(exchange) + assert(restored.isInstanceOf[ShuffleExchangeExec]) + val retainedAggregates = collect(restored) { case aggregate: CometHashAggregateExec => + aggregate + } + assert( + retainedAggregates.map(aggregate => (aggregate.modes, aggregate.output)) == + originalAggregates.map(aggregate => (aggregate.modes, aggregate.output)), + s"$restored") + assert( + retainedAggregates.forall( + _.originalPlan.getTagValue(CometExecRule.COMET_UNSAFE_PARTIAL).isEmpty)) + } + } + } + + test("transition reversion preserves native aggregate buffers across a Celeborn exchange") { + manager.withPlanningSupport(CelebornNativeShufflePlanningSupport()) { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false", + CometConf.COMET_SHUFFLE_MODE.key -> "native", + CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key -> "false") { + val query = spark + .range(0, 256, 1, 4) + .selectExpr("id % 4 AS grouping_key", "CAST(id AS DOUBLE) AS value") + .groupBy("grouping_key") + .agg(expr("percentile(value, 0.5)").as("percentile_value")) + val nativePlan = query.queryExecution.executedPlan + + assert(cometExchanges(nativePlan).nonEmpty, s"$nativePlan") + assert( + collect(nativePlan) { + case aggregate: CometHashAggregateExec if aggregate.modes == Seq(Final) => aggregate + }.nonEmpty, + s"expected a native final aggregate before transition reversion:\n$nativePlan") + assert( + collect(nativePlan) { case transition: ColumnarToRowTransition => transition }.nonEmpty, + s"test requires a result-stage transition:\n$nativePlan") + + val reverted = withSQLConf( Review Comment: [P2] Avoid using withSQLConf's result as a SparkPlan on legacy Spark `reverted` is inferred as `Unit` on the Spark 3.4/3.5 Scala 2.12 profiles, so `collect(reverted)` at line 464 does not compile. Both current-head CI jobs fail in `scala-test-compile-first` with `found: Unit; required: org.apache.spark.sql.execution.SparkPlan` ([Spark 3.4 log](https://github.com/apache/datafusion-comet/actions/runs/33279103988/job/99171326170), [Spark 3.5 log](https://github.com/apache/datafusion-comet/actions/runs/33279103988/job/99171326204)). Please declare the result as a `SparkPlan` outside `withSQLConf` and assign it inside the block, so the regression works with the older helper signature. Apply the same adjustment to the new producer test in `RevertNativeForTransitionHeavyStagesSuite`, which also captures this helper's return value. _Posted by Codex on behalf of ziting-openai using the spark-pr-review-memo skill._ ########## spark/src/main/scala/org/apache/comet/rules/RevertNativeForTransitionHeavyStages.scala: ########## @@ -105,6 +114,29 @@ case class RevertNativeForTransitionHeavyStages(session: SparkSession) case _ => false } + private def hasUnsafeMixedAggregateAtStageBoundary(stagePlan: SparkPlan): Boolean = { + def reachesBoundaryBeforeAggregate(plan: SparkPlan): Boolean = plan match { + case _ if isStageBoundary(plan) => true + case _: CometHashAggregateExec => false + case _ => plan.children.exists(reachesBoundaryBeforeAggregate) + } + + def visit(plan: SparkPlan): Boolean = plan match { + case _ if isStageBoundary(plan) => false + case aggregate: CometHashAggregateExec + if !QueryPlanSerde.allAggsSupportMixedExecution(aggregate.aggregateExpressions) => + val producesBuffer = + aggregate.modes.exists(mode => mode == Partial || mode == PartialMerge) + val consumesAcrossBoundary = + aggregate.modes.exists(mode => mode == Final || mode == PartialMerge) && + reachesBoundaryBeforeAggregate(aggregate.child) + producesBuffer || consumesAcrossBoundary Review Comment: [P2] Keep traversing after a Final that does not itself cross a boundary This branch returns `false` without visiting the aggregate's children. With native Celeborn shuffle, AQE off, and `spark.comet.exec.transitionRevert.enabled=true` / `maxTransitions=0`, a nested query such as `SELECT k, percentile(p, CAST(0.5 AS DOUBLE)) FROM (SELECT k, percentile(v, CAST(0.5 AS DOUBLE)) AS p FROM t GROUP BY k) q GROUP BY k` on supported partitioned input can have `Final_outer -> Partial_outer -> Final_inner -> exchange -> Partial_inner`: the outer aggregation keeps the existing partitioning on `k`, so it needs no additional exchange. At `Final_outer`, both local checks are false because `reachesBoundaryBeforeAggregate` stops at `Partial_outer`; `Final_inner` is never examined. `revertToSpark` then restores the inner consumer to Spark while retaining the native exchange/producer, so Spark's percentile merge reads Comet's array state as its binary buffer and can fail. Please also recurse into children when these local checks are false, and add a nested-aggregate reg ression with whole-stage codegen disabled. This is source-traced, not an executed reproduction. _Posted by Codex on behalf of ziting-openai using the spark-pr-review-memo skill._ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
