<commit_before>/*
 * 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.catalyst.analysis

import scala.collection.mutable.ArrayBuffer

import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess

object DeduplicateRelations extends Rule[LogicalPlan] {
  override def apply(plan: LogicalPlan): LogicalPlan = {
    renewDuplicatedRelations(Nil, plan)._1.resolveOperatorsUpWithPruning(
      AlwaysProcess.fn, ruleId) {
      case p: LogicalPlan if !p.childrenResolved => p
      // To resolve duplicate expression IDs for Join.
      case j @ Join(left, right, _, _, _) if !j.duplicateResolved =>
        j.copy(right = dedupRight(left, right))
      // intersect/except will be rewritten to join at the beginning of optimizer. Here we need to
      // deduplicate the right side plan, so that we won't produce an invalid self-join later.
      case i @ Intersect(left, right, _) if !i.duplicateResolved =>
        i.copy(right = dedupRight(left, right))
      case e @ Except(left, right, _) if !e.duplicateResolved =>
        e.copy(right = dedupRight(left, right))
      // Only after we finish by-name resolution for Union
      case u: Union if !u.byName && !u.duplicateResolved =>
        // Use projection-based de-duplication for Union to avoid breaking the checkpoint sharing
        // feature in streaming.
        val newChildren = u.children.foldRight(Seq.empty[LogicalPlan]) { (head, tail) =>
          head +: tail.map {
            case child if head.outputSet.intersect(child.outputSet).isEmpty =>
              child
            case child =>
              val projectList = child.output.map { attr =>
                Alias(attr, attr.name)()
              }
              Project(projectList, child)
          }
        }
        u.copy(children = newChildren)
      case m @ MergeIntoTable(targetTable, sourceTable, _, _, _) if !m.duplicateResolved =>
        m.copy(sourceTable = dedupRight(targetTable, sourceTable))
    }
  }

  /**
   * Deduplicate any duplicated relations of a LogicalPlan
   * @param existingRelations the known unique relations for a LogicalPlan
   * @param plan the LogicalPlan that requires the deduplication
   * @return (the new LogicalPlan which already deduplicate all duplicated relations (if any),
   *         all relations of the new LogicalPlan )
   */
  private def renewDuplicatedRelations(
      existingRelations: Seq[MultiInstanceRelation],
      plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
    case p: LogicalPlan if p.isStreaming => (plan, Nil)

    case m: MultiInstanceRelation =>
      if (isDuplicated(existingRelations, m)) {
        val newNode = m.newInstance()
        newNode.copyTagsFrom(m)
        (newNode, Nil)
      } else {
        (m, Seq(m))
      }

    case plan: LogicalPlan =>
      val relations = ArrayBuffer.empty[MultiInstanceRelation]
      val newPlan = if (plan.children.nonEmpty) {
        val newChildren = ArrayBuffer.empty[LogicalPlan]
        for (c <- plan.children) {
          val (renewed, collected) = renewDuplicatedRelations(existingRelations ++ relations, c)
          newChildren += renewed
          relations ++= collected
        }

        if (plan.childrenResolved) {
          val attrMap = AttributeMap(
            plan
              .children
              .flatMap(_.output).zip(newChildren.flatMap(_.output))
              .filter { case (a1, a2) => a1.exprId != a2.exprId }
          )
          plan.withNewChildren(newChildren.toSeq).rewriteAttrs(attrMap)
        } else {
          plan.withNewChildren(newChildren.toSeq)
        }
      } else {
        plan
      }

      val planWithNewSubquery = newPlan.transformExpressions {
        case subquery: SubqueryExpression =>
          val (renewed, collected) = renewDuplicatedRelations(
            existingRelations ++ relations, subquery.plan)
          relations ++= collected
          subquery.withNewPlan(renewed)
      }
      (planWithNewSubquery, relations.toSeq)
  }

  private def isDuplicated(
      existingRelations: Seq[MultiInstanceRelation],
      relation: MultiInstanceRelation): Boolean = {
    existingRelations.exists { er =>
      er.asInstanceOf[LogicalPlan].outputSet
        .intersect(relation.asInstanceOf[LogicalPlan].outputSet).nonEmpty
    }
  }

  /**
   * Generate a new logical plan for the right child with different expression IDs
   * for all conflicting attributes.<commit_msg>[SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations<commit_after>/*
 * 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.catalyst.analysis

import scala.collection.mutable

import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess

/**
 * A helper class used to detect duplicate relations fast in `DeduplicateRelations`
 */
case class ReferenceEqualPlanWrapper(plan: LogicalPlan) {
  private val _hashCode = System.identityHashCode(plan)
  override def hashCode(): Int = _hashCode
  override def equals(obj: Any): Boolean = obj match {
    case wrapper: ReferenceEqualPlanWrapper =>
      plan.eq(wrapper.plan)
    case _ =>
      false
  }
}

object DeduplicateRelations extends Rule[LogicalPlan] {
  override def apply(plan: LogicalPlan): LogicalPlan = {
    renewDuplicatedRelations(mutable.HashSet.empty, plan)._1.resolveOperatorsUpWithPruning(
      AlwaysProcess.fn, ruleId) {
      case p: LogicalPlan if !p.childrenResolved => p
      // To resolve duplicate expression IDs for Join.
      case j @ Join(left, right, _, _, _) if !j.duplicateResolved =>
        j.copy(right = dedupRight(left, right))
      // intersect/except will be rewritten to join at the beginning of optimizer. Here we need to
      // deduplicate the right side plan, so that we won't produce an invalid self-join later.
      case i @ Intersect(left, right, _) if !i.duplicateResolved =>
        i.copy(right = dedupRight(left, right))
      case e @ Except(left, right, _) if !e.duplicateResolved =>
        e.copy(right = dedupRight(left, right))
      // Only after we finish by-name resolution for Union
      case u: Union if !u.byName && !u.duplicateResolved =>
        // Use projection-based de-duplication for Union to avoid breaking the checkpoint sharing
        // feature in streaming.
        val newChildren = u.children.foldRight(Seq.empty[LogicalPlan]) { (head, tail) =>
          head +: tail.map {
            case child if head.outputSet.intersect(child.outputSet).isEmpty =>
              child
            case child =>
              val projectList = child.output.map { attr =>
                Alias(attr, attr.name)()
              }
              Project(projectList, child)
          }
        }
        u.copy(children = newChildren)
      case m @ MergeIntoTable(targetTable, sourceTable, _, _, _) if !m.duplicateResolved =>
        m.copy(sourceTable = dedupRight(targetTable, sourceTable))
    }
  }

  /**
   * Deduplicate any duplicated relations of a LogicalPlan
   * @param existingRelations the known unique relations for a LogicalPlan
   * @param plan the LogicalPlan that requires the deduplication
   * @return (the new LogicalPlan which already deduplicate all duplicated relations (if any),
   *         all relations of the new LogicalPlan, whether the plan is changed or not)
   */
  private def renewDuplicatedRelations(
      existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
      plan: LogicalPlan)
    : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
    case p: LogicalPlan if p.isStreaming => (plan, mutable.HashSet.empty, false)

    case m: MultiInstanceRelation =>
      val planWrapper = ReferenceEqualPlanWrapper(m)
      if (existingRelations.contains(planWrapper)) {
        val newNode = m.newInstance()
        newNode.copyTagsFrom(m)
        (newNode, mutable.HashSet.empty, true)
      } else {
        val mWrapper = new mutable.HashSet[ReferenceEqualPlanWrapper]()
        mWrapper.add(planWrapper)
        (m, mWrapper, false)
      }

    case plan: LogicalPlan =>
      val relations = new mutable.HashSet[ReferenceEqualPlanWrapper]()
      var planChanged = false
      val newPlan = if (plan.children.nonEmpty) {
        val newChildren = mutable.ArrayBuffer.empty[LogicalPlan]
        for (c <- plan.children) {
          val (renewed, collected, changed) =
            renewDuplicatedRelations(existingRelations ++ relations, c)
          newChildren += renewed
          relations ++= collected
          if (changed) {
            planChanged = true
          }
        }

        if (planChanged) {
          if (plan.childrenResolved) {
            val planWithNewChildren = plan.withNewChildren(newChildren.toSeq)
            val attrMap = AttributeMap(
              plan
                .children
                .flatMap(_.output).zip(newChildren.flatMap(_.output))
                .filter { case (a1, a2) => a1.exprId != a2.exprId }
            )
            if (attrMap.isEmpty) {
              planWithNewChildren
            } else {
              planWithNewChildren.rewriteAttrs(attrMap)
            }
          } else {
            plan.withNewChildren(newChildren.toSeq)
          }
        } else {
          plan
        }
      } else {
        plan
      }

      val planWithNewSubquery = newPlan.transformExpressions {
        case subquery: SubqueryExpression =>
          val (renewed, collected, changed) = renewDuplicatedRelations(
            existingRelations ++ relations, subquery.plan)
          relations ++= collected
          if (changed) planChanged = true
          subquery.withNewPlan(renewed)
      }
      (planWithNewSubquery, relations, planChanged)
  }

  /**
   * Generate a new logical plan for the right child with different expression IDs
   * for all conflicting attributes.
{'fraction_non_alphanumeric': 0.08697872340425532, 'fraction_numerical': 0.0019574468085106385, 'mean_word_length': 2.7411652340019104, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5568136', 'n_tokens_mistral': 3237, 'n_tokens_neox': 3127, 'n_words': 952}