[SPARK-39448] Add ReplaceCTERefWithRepartition into nonExcludableRules list

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.3.0, 3.4.0
• Fix Version/s: 3.3.1, 3.4.0
• Component/s: SQL
• Assignee: Yuming Wang
• Reporter: Yuming Wang
• Created: 11/Jun/22 13:26
• Updated: 14/Jun/22 07:44
• Resolved: 14/Jun/22 07:43

Description

A unit test to test excluded rules:

/*
 * 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

import org.apache.spark.internal.config.Tests.IS_TESTING
import org.apache.spark.launcher.SparkLauncher
import org.apache.spark.sql.catalyst.util.resourceToString
import org.apache.spark.sql.internal.SQLConf

// scalastyle:off println
object ExcludedRulesTestSuite extends TPCDSBase {
  override val spark = SparkSession.builder()
    .appName(this.getClass.getSimpleName)
    .master("local[40]")
    .config(SparkLauncher.DRIVER_MEMORY, "4g")
    .config(SQLConf.SHUFFLE_PARTITIONS.key, 2)
    .getOrCreate()

  override def injectStats: Boolean = true

  createTables()

  private lazy val excludableRules = spark.sessionState.optimizer.batches
    .flatMap(_.rules.map(_.ruleName))
    .distinct
    .filterNot(spark.sessionState.optimizer.nonExcludableRules.contains)

  def main(args: Array[String]): Unit = {
    System.setProperty(IS_TESTING.key, "true")

    tpcdsQueries.foreach { name =>
      val queryString = resourceToString(s"tpcds/$name.sql",
        classLoader = Thread.currentThread().getContextClassLoader)
      excludableRules.foreach { rule =>

        println(name + ": " + rule)
        try {
          withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> rule) {
            sql(queryString).collect()
          }
        } catch {
          case e: Exception =>
            println("Exception: " + e.getMessage)
        }
      }
    }

    tpcdsQueriesV2_7_0.foreach { name =>
      val queryString = resourceToString(s"tpcds-v2.7.0/$name.sql",
        classLoader = Thread.currentThread().getContextClassLoader)
      excludableRules.foreach { rule =>

        println("tpcds-v2.7.0 " + name + ": " + rule)
        try {
          withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> rule) {
            sql(queryString).collect()
          }
        } catch {
          case e: Exception =>
            println("Exception: " + e.getMessage)
        }
      }
    }

    modifiedTPCDSQueries.foreach { name =>
      val queryString = resourceToString(s"tpcds-modifiedQueries/$name.sql",
        classLoader = Thread.currentThread().getContextClassLoader)
      excludableRules.foreach { rule =>

        println("tpcds-modifiedQueries " + name + ": " + rule)
        try {
          withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> rule) {
            sql(queryString).collect()
          }
        } catch {
          case e: Exception =>
            println("Exception: " + e.getMessage)
        }
      }
    }

    // scalastyle:on println
  }
}

Issue Links

• links to: [Github] Pull Request #36847 (wangyum)
• links to: [Github] Pull Request #36847 (wangyum)