Incorrect handling of reused exchanges in `CometExecRule`

[andygrove]

Describe the bug

The following match arms in CometExecRule will never replac the exchange with a Comet exchange, because the conversion only happens if all children are native, and the child in this case is ReusedExchangeExec .

This issue became apparent after the recent refactoring work.

      case s @ BroadcastQueryStageExec(
            _,
            ReusedExchangeExec(_, _: CometBroadcastExchangeExec),
            _) =>
        convertToCometIfAllChildrenAreNative(s, CometExchangeSink).getOrElse(s)

      case s @ ShuffleQueryStageExec(_, ReusedExchangeExec(_, _: CometShuffleExchangeExec), _) =>
        convertToCometIfAllChildrenAreNative(s, CometExchangeSink).getOrElse(s)

Steps to reproduce

No response

Expected behavior

No response

Additional context

No response

[andygrove]

It looks like this issue has existed since the 0.1 release:

case s @ ShuffleQueryStageExec(
              _,
              ReusedExchangeExec(_, _: CometShuffleExchangeExec),
              _) =>
          val newOp = transform1(s)
          newOp match {
            case Some(nativeOp) =>
              CometSinkPlaceHolder(nativeOp, s, s)
            case None =>
              s
          }


      def transform1(op: SparkPlan): Option[Operator] = {
        if (op.children.forall(_.isInstanceOf[CometNativeExec])) {
          QueryPlanSerde.operator2Proto(
            op,
            op.children.map(_.asInstanceOf[CometNativeExec].nativeOp): _*)
        } else {
          withInfo(
            op,
            s"${op.nodeName} is not native because the following children are not native " +
              s"${explainChildNotNative(op)}")
          None
        }
      }

[andygrove]

Nevermind, the logic is confusing but not incorrect. The reason is that op.children.forall(_.isInstanceOf[CometNativeExec]) will always return true because query stages are leaf nodes and have no children. The code is confusing though, so I will create a PR to address this.