text:
package org.apache.spark.sql.execution.adaptive

import java.io.File
import java.net.URI

import org.apache.log4j.Level
import org.scalatest.PrivateMethodTester

import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent, SparkListenerJobStart}
import org.apache.spark.sql.{Dataset, QueryTest, Row, SparkSession, Strategy}
import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
import org.apache.spark.sql.execution.{PartialReducerPartitionSpec, QueryExecution, ReusedSubqueryExec, ShuffledRowRDD, SparkPlan, UnaryExecNode}
import org.apache.spark.sql.execution.command.DataWritingCommandExec
import org.apache.spark.sql.execution.datasources.noop.NoopDataSource
import org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec
import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ENSURE_REQUIREMENTS, Exchange, REPARTITION, REPARTITION_WITH_NUM, ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike, ShuffleOrigin}
import org.apache.spark.sql.execution.joins.{BaseJoinExec, BroadcastHashJoinExec, SortMergeJoinExec}
import org.apache.spark.sql.execution.metric.SQLShuffleReadMetricsReporter
import org.apache.spark.sql.execution.ui.SparkListenerSQLAdaptiveExecutionUpdate
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.test.SQLTestData.TestData
import org.apache.spark.sql.types.{IntegerType, StructType}
import org.apache.spark.sql.util.QueryExecutionListener
import org.apache.spark.util.Utils

class AdaptiveQueryExecSuite
  extends QueryTest
  with SharedSparkSession
  with AdaptiveSparkPlanHelper
  with PrivateMethodTester {

  import testImplicits._

  setupTestData()

  private def runAdaptiveAndVerifyResult(query: String): (SparkPlan, SparkPlan) = {
    var finalPlanCnt = 0
    val listener = new SparkListener {
      override def onOtherEvent(event: SparkListenerEvent): Unit = {
        event match {
          case SparkListenerSQLAdaptiveExecutionUpdate(_, _, sparkPlanInfo) =>
            if (sparkPlanInfo.simpleString.startsWith(
              "AdaptiveSparkPlan isFinalPlan=true")) {
              finalPlanCnt += 1
            }
          case _ => // ignore other events
        }
      }
    }
    spark.sparkContext.addSparkListener(listener)

    val dfAdaptive = sql(query)
    val planBefore = dfAdaptive.queryExecution.executedPlan
    assert(planBefore.toString.startsWith("AdaptiveSparkPlan isFinalPlan=false"))
    val result = dfAdaptive.collect()
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
      val df = sql(query)
      checkAnswer(df, result)
    }
    val planAfter = dfAdaptive.queryExecution.executedPlan
    assert(planAfter.toString.startsWith("AdaptiveSparkPlan isFinalPlan=true"))
    val adaptivePlan = planAfter.asInstanceOf[AdaptiveSparkPlanExec].executedPlan

    spark.sparkContext.listenerBus.waitUntilEmpty()
    // AQE will post `SparkListenerSQLAdaptiveExecutionUpdate` twice in case of subqueries that
    // exist out of query stages.
    val expectedFinalPlanCnt = adaptivePlan.find(_.subqueries.nonEmpty).map(_ => 2).getOrElse(1)
    assert(finalPlanCnt == expectedFinalPlanCnt)
    spark.sparkContext.removeSparkListener(listener)

    val exchanges = adaptivePlan.collect {
      case e: Exchange => e
    }
    assert(exchanges.isEmpty, "The final plan should not contain any Exchange node.")
    (dfAdaptive.queryExecution.sparkPlan, adaptivePlan)
  }

  private def findTopLevelBroadcastHashJoin(plan: SparkPlan): Seq[BroadcastHashJoinExec] = {
    collect(plan) {
      case j: BroadcastHashJoinExec => j
    }
  }

  private def findTopLevelSortMergeJoin(plan: SparkPlan): Seq[SortMergeJoinExec] = {
    collect(plan) {
      case j: SortMergeJoinExec => j
    }
  }

  private def findTopLevelBaseJoin(plan: SparkPlan): Seq[BaseJoinExec] = {
    collect(plan) {
      case j: BaseJoinExec => j
    }
  }

  private def findReusedExchange(plan: SparkPlan): Seq[ReusedExchangeExec] = {
    collectWithSubqueries(plan) {
      case ShuffleQueryStageExec(_, e: ReusedExchangeExec, _) => e
      case BroadcastQueryStageExec(_, e: ReusedExchangeExec, _) => e
    }
  }

  private def findReusedSubquery(plan: SparkPlan): Seq[ReusedSubqueryExec] = {
    collectWithSubqueries(plan) {
      case e: ReusedSubqueryExec => e
    }
  }

  private def checkNumLocalShuffleReaders(
      plan: SparkPlan, numShufflesWithoutLocalReader: Int = 0): Unit = {
    val numShuffles = collect(plan) {
      case s: ShuffleQueryStageExec => s
    }.length

    val numLocalReaders = collect(plan) {
      case reader: CustomShuffleReaderExec if reader.isLocalReader => reader
    }
    numLocalReaders.foreach { r =>
      val rdd = r.execute()
      val parts = rdd.partitions
      assert(parts.forall(rdd.preferredLocations(_).nonEmpty))
    }
    assert(numShuffles === (numLocalReaders.length + numShufflesWithoutLocalReader))
  }

  private def checkInitialPartitionNum(df: Dataset[_], numPartition: Int): Unit = {
    // repartition obeys initialPartitionNum when adaptiveExecutionEnabled
    val plan = df.queryExecution.executedPlan
    assert(plan.isInstanceOf[AdaptiveSparkPlanExec])
    val shuffle = plan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan.collect {
      case s: ShuffleExchangeExec => s
    }
    assert(shuffle.size == 1)
    assert(shuffle(0).outputPartitioning.numPartitions == numPartition)
  }

  test("Change merge join to broadcast join") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData join testData2 ON key = a where value = '1'")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      checkNumLocalShuffleReaders(adaptivePlan)
    }
  }

  test("Reuse the parallelism of CoalescedShuffleReaderExec in LocalShuffleReaderExec") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
      SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "10") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData join testData2 ON key = a where value = '1'")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      val localReaders = collect(adaptivePlan) {
        case reader: CustomShuffleReaderExec if reader.isLocalReader => reader
      }
      assert(localReaders.length == 2)
      val localShuffleRDD0 = localReaders(0).execute().asInstanceOf[ShuffledRowRDD]
      val localShuffleRDD1 = localReaders(1).execute().asInstanceOf[ShuffledRowRDD]
      // The pre-shuffle partition size is [0, 0, 0, 72, 0]
      // We exclude the 0-size partitions, so only one partition, advisoryParallelism = 1
      // the final parallelism is
      // math.max(1, advisoryParallelism / numMappers): math.max(1, 1/2) = 1
      // and the partitions length is 1 * numMappers = 2
      assert(localShuffleRDD0.getPartitions.length == 2)
      // The pre-shuffle partition size is [0, 72, 0, 72, 126]
      // We exclude the 0-size partitions, so only 3 partition, advisoryParallelism = 3
      // the final parallelism is
      // math.max(1, advisoryParallelism / numMappers): math.max(1, 3/2) = 1
      // and the partitions length is 1 * numMappers = 2
      assert(localShuffleRDD1.getPartitions.length == 2)
    }
  }

  test("Reuse the default parallelism in LocalShuffleReaderExec") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
      SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "false") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData join testData2 ON key = a where value = '1'")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      val localReaders = collect(adaptivePlan) {
        case reader: CustomShuffleReaderExec if reader.isLocalReader => reader
      }
      assert(localReaders.length == 2)
      val localShuffleRDD0 = localReaders(0).execute().asInstanceOf[ShuffledRowRDD]
      val localShuffleRDD1 = localReaders(1).execute().asInstanceOf[ShuffledRowRDD]
      // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2
      // and the partitions length is 2 * numMappers = 4
      assert(localShuffleRDD0.getPartitions.length == 4)
      // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2
      // and the partitions length is 2 * numMappers = 4
      assert(localShuffleRDD1.getPartitions.length == 4)
    }
  }

  test("Empty stage coalesced to 1-partition RDD") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true") {
      val df1 = spark.range(10).withColumn("a", 'id)
      val df2 = spark.range(10).withColumn("b", 'id)
      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
        val testDf = df1.where('a > 10).join(df2.where('b > 10), Seq("id"), "left_outer")
          .groupBy('a).count()
        checkAnswer(testDf, Seq())
        val plan = testDf.queryExecution.executedPlan
        assert(find(plan)(_.isInstanceOf[SortMergeJoinExec]).isDefined)
        val coalescedReaders = collect(plan) {
          case r: CustomShuffleReaderExec => r
        }
        assert(coalescedReaders.length == 3)
        coalescedReaders.foreach(r => assert(r.partitionSpecs.length == 1))
      }

      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "1") {
        val testDf = df1.where('a > 10).join(df2.where('b > 10), Seq("id"), "left_outer")
          .groupBy('a).count()
        checkAnswer(testDf, Seq())
        val plan = testDf.queryExecution.executedPlan
        assert(find(plan)(_.isInstanceOf[BroadcastHashJoinExec]).isDefined)
        val coalescedReaders = collect(plan) {
          case r: CustomShuffleReaderExec => r
        }
        assert(coalescedReaders.length == 3, s"$plan")
        coalescedReaders.foreach(r => assert(r.isLocalReader || r.partitionSpecs.length == 1))
      }
    }
  }

  test("Scalar subquery") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData join testData2 ON key = a " +
        "where value = (SELECT max(a) from testData3)")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      checkNumLocalShuffleReaders(adaptivePlan)
    }
  }

  test("Scalar subquery in later stages") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData join testData2 ON key = a " +
        "where (value + a) = (SELECT max(a) from testData3)")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)

      checkNumLocalShuffleReaders(adaptivePlan)
    }
  }

  test("multiple joins") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        """
          |WITH t4 AS (
          |  SELECT * FROM lowercaseData t2 JOIN testData3 t3 ON t2.n = t3.a where t2.n = '1'
          |)
          |SELECT * FROM testData
          |JOIN testData2 t2 ON key = t2.a
          |JOIN t4 ON t2.b = t4.a
          |WHERE value = 1
        """.stripMargin)
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 3)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 3)

      // A possible resulting query plan:
      // BroadcastHashJoin
      // +- BroadcastExchange
      //    +- LocalShuffleReader*
      //       +- ShuffleExchange
      //          +- BroadcastHashJoin
      //             +- BroadcastExchange
      //                +- LocalShuffleReader*
      //                   +- ShuffleExchange
      //             +- LocalShuffleReader*
      //                +- ShuffleExchange
      // +- BroadcastHashJoin
      //    +- LocalShuffleReader*
      //       +- ShuffleExchange
      //    +- BroadcastExchange
      //       +-LocalShuffleReader*
      //             +- ShuffleExchange

      // After applied the 'OptimizeLocalShuffleReader' rule, we can convert all the four
      // shuffle reader to local shuffle reader in the bottom two 'BroadcastHashJoin'.
      // For the top level 'BroadcastHashJoin', the probe side is not shuffle query stage
      // and the build side shuffle query stage is also converted to local shuffle reader.
      checkNumLocalShuffleReaders(adaptivePlan)
    }
  }

  test("multiple joins with aggregate") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        """
          |WITH t4 AS (
          |  SELECT * FROM lowercaseData t2 JOIN (
          |    select a, sum(b) from testData3 group by a
          |  ) t3 ON t2.n = t3.a where t2.n = '1'
          |)
          |SELECT * FROM testData
          |JOIN testData2 t2 ON key = t2.a
          |JOIN t4 ON t2.b = t4.a
          |WHERE value = 1
        """.stripMargin)
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 3)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 3)

      // A possible resulting query plan:
      // BroadcastHashJoin
      // +- BroadcastExchange
      //    +- LocalShuffleReader*
      //       +- ShuffleExchange
      //          +- BroadcastHashJoin
      //             +- BroadcastExchange
      //                +- LocalShuffleReader*
      //                   +- ShuffleExchange
      //             +- LocalShuffleReader*
      //                +- ShuffleExchange
      // +- BroadcastHashJoin
      //    +- LocalShuffleReader*
      //       +- ShuffleExchange
      //    +- BroadcastExchange
      //       +-HashAggregate
      //          +- CoalescedShuffleReader
      //             +- ShuffleExchange

      // The shuffle added by Aggregate can't apply local reader.
      checkNumLocalShuffleReaders(adaptivePlan, 1)
    }
  }

  test("multiple joins with aggregate 2") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        """
          |WITH t4 AS (
          |  SELECT * FROM lowercaseData t2 JOIN (
          |    select a, max(b) b from testData2 group by a
          |  ) t3 ON t2.n = t3.b
          |)
          |SELECT * FROM testData
          |JOIN testData2 t2 ON key = t2.a
          |JOIN t4 ON value = t4.a
          |WHERE value = 1
        """.stripMargin)
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 3)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 3)

      // A possible resulting query plan:
      // BroadcastHashJoin
      // +- BroadcastExchange
      //    +- LocalShuffleReader*
      //       +- ShuffleExchange
      //          +- BroadcastHashJoin
      //             +- BroadcastExchange
      //                +- LocalShuffleReader*
      //                   +- ShuffleExchange
      //             +- LocalShuffleReader*
      //                +- ShuffleExchange
      // +- BroadcastHashJoin
      //    +- Filter
      //       +- HashAggregate
      //          +- CoalescedShuffleReader
      //             +- ShuffleExchange
      //    +- BroadcastExchange
      //       +-LocalShuffleReader*
      //           +- ShuffleExchange

      // The shuffle added by Aggregate can't apply local reader.
      checkNumLocalShuffleReaders(adaptivePlan, 1)
    }
  }

  test("Exchange reuse") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT value FROM testData join testData2 ON key = a " +
        "join (SELECT value v from testData join testData3 ON key = a) on value = v")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 3)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 2)
      // There is still a SMJ, and its two shuffles can't apply local reader.
      checkNumLocalShuffleReaders(adaptivePlan, 2)
      // Even with local shuffle reader, the query stage reuse can also work.
      val ex = findReusedExchange(adaptivePlan)
      assert(ex.size == 1)
    }
  }

  test("Exchange reuse with subqueries") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT a FROM testData join testData2 ON key = a " +
        "where value = (SELECT max(a) from testData join testData2 ON key = a)")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      checkNumLocalShuffleReaders(adaptivePlan)
      // Even with local shuffle reader, the query stage reuse can also work.
      val ex = findReusedExchange(adaptivePlan)
      assert(ex.size == 1)
    }
  }

  test("Exchange reuse across subqueries") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
        SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT a FROM testData join testData2 ON key = a " +
        "where value >= (SELECT max(a) from testData join testData2 ON key = a) " +
        "and a <= (SELECT max(a) from testData join testData2 ON key = a)")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      checkNumLocalShuffleReaders(adaptivePlan)
      // Even with local shuffle reader, the query stage reuse can also work.
      val ex = findReusedExchange(adaptivePlan)
      assert(ex.nonEmpty)
      val sub = findReusedSubquery(adaptivePlan)
      assert(sub.isEmpty)
    }
  }

  test("Subquery reuse") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT a FROM testData join testData2 ON key = a " +
        "where value >= (SELECT max(a) from testData join testData2 ON key = a) " +
        "and a <= (SELECT max(a) from testData join testData2 ON key = a)")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      checkNumLocalShuffleReaders(adaptivePlan)
      // Even with local shuffle reader, the query stage reuse can also work.
      val ex = findReusedExchange(adaptivePlan)
      assert(ex.isEmpty)
      val sub = findReusedSubquery(adaptivePlan)
      assert(sub.nonEmpty)
    }
  }

  test("Broadcast exchange reuse across subqueries") {
    withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000",
        SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT a FROM testData join testData2 ON key = a " +
        "where value >= (" +
        "SELECT /*+ broadcast(testData2) */ max(key) from testData join testData2 ON key = a) " +
        "and a <= (" +
        "SELECT /*+ broadcast(testData2) */ max(value) from testData join testData2 ON key = a)")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 1)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      checkNumLocalShuffleReaders(adaptivePlan)
      // Even with local shuffle reader, the query stage reuse can also work.
      val ex = findReusedExchange(adaptivePlan)
      assert(ex.nonEmpty)
      assert(ex.head.child.isInstanceOf[BroadcastExchangeExec])
      val sub = findReusedSubquery(adaptivePlan)
      assert(sub.isEmpty)
    }
  }

  test("Union/Except/Intersect queries") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      runAdaptiveAndVerifyResult(
        """
          |SELECT * FROM testData
          |EXCEPT
          |SELECT * FROM testData2
          |UNION ALL
          |SELECT * FROM testData
          |INTERSECT ALL
          |SELECT * FROM testData2
        """.stripMargin)
    }
  }

  test("Subquery de-correlation in Union queries") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      withTempView("a", "b") {
        Seq("a" -> 2, "b" -> 1).toDF("id", "num").createTempView("a")
        Seq("a" -> 2, "b" -> 1).toDF("id", "num").createTempView("b")

        runAdaptiveAndVerifyResult(
          """
            |SELECT id,num,source FROM (
            |  SELECT id, num, 'a' as source FROM a
            |  UNION ALL
            |  SELECT id, num, 'b' as source FROM b
            |) AS c WHERE c.id IN (SELECT id FROM b WHERE num = 2)
          """.stripMargin)
      }
    }
  }

  test("Avoid plan change if cost is greater") {
    val origPlan = sql("SELECT * FROM testData " +
      "join testData2 t2 ON key = t2.a " +
      "join testData2 t3 on t2.a = t3.a where t2.b = 1").queryExecution.executedPlan

    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
      SQLConf.BROADCAST_HASH_JOIN_OUTPUT_PARTITIONING_EXPAND_LIMIT.key -> "0") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData " +
          "join testData2 t2 ON key = t2.a " +
          "join testData2 t3 on t2.a = t3.a where t2.b = 1")
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 2)
      val smj2 = findTopLevelSortMergeJoin(adaptivePlan)
      assert(smj2.size == 2, origPlan.toString)
    }
  }

  test("Change merge join to broadcast join without local shuffle reader") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.LOCAL_SHUFFLE_READER_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "40") {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        """
          |SELECT * FROM testData t1 join testData2 t2
          |ON t1.key = t2.a join testData3 t3 on t2.a = t3.a
          |where t1.value = 1
        """.stripMargin
      )
      val smj = findTopLevelSortMergeJoin(plan)
      assert(smj.size == 2)
      val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
      assert(bhj.size == 1)
      // There is still a SMJ, and its two shuffles can't apply local reader.
      checkNumLocalShuffleReaders(adaptivePlan, 2)
    }
  }

  test("Avoid changing merge join to broadcast join if too many empty partitions on build plan") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.NON_EMPTY_PARTITION_RATIO_FOR_BROADCAST_JOIN.key -> "0.5") {
      // `testData` is small enough to be broadcast but has empty partition ratio over the config.
      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
        val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT * FROM testData join testData2 ON key = a where value = '1'")
        val smj = findTopLevelSortMergeJoin(plan)
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.isEmpty)
      }
      // It is still possible to broadcast `testData2`.
      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
        val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT * FROM testData join testData2 ON key = a where value = '1'")
        val smj = findTopLevelSortMergeJoin(plan)
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
        assert(bhj.head.buildSide == BuildRight)
      }
    }
  }

  test("SPARK-29906: AQE should not introduce extra shuffle for outermost limit") {
    var numStages = 0
    val listener = new SparkListener {
      override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
        numStages = jobStart.stageInfos.length
      }
    }
    try {
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        spark.sparkContext.addSparkListener(listener)
        spark.range(0, 100, 1, numPartitions = 10).take(1)
        spark.sparkContext.listenerBus.waitUntilEmpty()
        // Should be only one stage since there is no shuffle.
        assert(numStages == 1)
      }
    } finally {
      spark.sparkContext.removeSparkListener(listener)
    }
  }

  test("SPARK-30524: Do not optimize skew join if introduce additional shuffle") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
      SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "100",
      SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100") {
      withTempView("skewData1", "skewData2") {
        spark
          .range(0, 1000, 1, 10)
          .selectExpr("id % 3 as key1", "id as value1")
          .createOrReplaceTempView("skewData1")
        spark
          .range(0, 1000, 1, 10)
          .selectExpr("id % 1 as key2", "id as value2")
          .createOrReplaceTempView("skewData2")

        def checkSkewJoin(query: String, optimizeSkewJoin: Boolean): Unit = {
          val (_, innerAdaptivePlan) = runAdaptiveAndVerifyResult(query)
          val innerSmj = findTopLevelSortMergeJoin(innerAdaptivePlan)
          assert(innerSmj.size == 1 && innerSmj.head.isSkewJoin == optimizeSkewJoin)
        }

        checkSkewJoin(
          "SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2", true)
        // Additional shuffle introduced, so disable the "OptimizeSkewedJoin" optimization
        checkSkewJoin(
          "SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 GROUP BY key1", false)
      }
    }
  }

  test("SPARK-29544: adaptive skew join with different join types") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
      SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1",
      SQLConf.SHUFFLE_PARTITIONS.key -> "100",
      SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "800",
      SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "800") {
      withTempView("skewData1", "skewData2") {
        spark
          .range(0, 1000, 1, 10)
          .select(
            when('id < 250, 249)
              .when('id >= 750, 1000)
              .otherwise('id).as("key1"),
            'id as "value1")
          .createOrReplaceTempView("skewData1")
        spark
          .range(0, 1000, 1, 10)
          .select(
            when('id < 250, 249)
              .otherwise('id).as("key2"),
            'id as "value2")
          .createOrReplaceTempView("skewData2")

        def checkSkewJoin(
            joins: Seq[SortMergeJoinExec],
            leftSkewNum: Int,
            rightSkewNum: Int): Unit = {
          assert(joins.size == 1 && joins.head.isSkewJoin)
          assert(joins.head.left.collect {
            case r: CustomShuffleReaderExec => r
          }.head.partitionSpecs.collect {
            case p: PartialReducerPartitionSpec => p.reducerIndex
          }.distinct.length == leftSkewNum)
          assert(joins.head.right.collect {
            case r: CustomShuffleReaderExec => r
          }.head.partitionSpecs.collect {
            case p: PartialReducerPartitionSpec => p.reducerIndex
          }.distinct.length == rightSkewNum)
        }

        // skewed inner join optimization
        val (_, innerAdaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT * FROM skewData1 join skewData2 ON key1 = key2")
        val innerSmj = findTopLevelSortMergeJoin(innerAdaptivePlan)
        checkSkewJoin(innerSmj, 2, 1)

        // skewed left outer join optimization
        val (_, leftAdaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT * FROM skewData1 left outer join skewData2 ON key1 = key2")
        val leftSmj = findTopLevelSortMergeJoin(leftAdaptivePlan)
        checkSkewJoin(leftSmj, 2, 0)

        // skewed right outer join optimization
        val (_, rightAdaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT * FROM skewData1 right outer join skewData2 ON key1 = key2")
        val rightSmj = findTopLevelSortMergeJoin(rightAdaptivePlan)
        checkSkewJoin(rightSmj, 0, 1)
      }
    }
  }

  test("SPARK-30291: AQE should catch the exceptions when doing materialize") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      withTable("bucketed_table") {
        val df1 =
          (0 until 50).map(i => (i % 5, i % 13, i.toString)).toDF("i", "j", "k").as("df1")
        df1.write.format("parquet").bucketBy(8, "i").saveAsTable("bucketed_table")
        val warehouseFilePath = new URI(spark.sessionState.conf.warehousePath).getPath
        val tableDir = new File(warehouseFilePath, "bucketed_table")
        Utils.deleteRecursively(tableDir)
        df1.write.parquet(tableDir.getAbsolutePath)

        val aggregated = spark.table("bucketed_table").groupBy("i").count()
        val error = intercept[Exception] {
          aggregated.count()
        }
        assert(error.toString contains "Invalid bucket file")
        assert(error.getSuppressed.size === 0)
      }
    }
  }

  test("SPARK-30403: AQE should handle InSubquery") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      runAdaptiveAndVerifyResult("SELECT * FROM testData LEFT OUTER join testData2" +
        " ON key = a  AND key NOT IN (select a from testData3) where value = '1'"
      )
    }
  }

  test("force apply AQE") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.ADAPTIVE_EXECUTION_FORCE_APPLY.key -> "true") {
      val plan = sql("SELECT * FROM testData").queryExecution.executedPlan
      assert(plan.isInstanceOf[AdaptiveSparkPlanExec])
    }
  }

  test("SPARK-30719: do not log warning if intentionally skip AQE") {
    val testAppender = new LogAppender("aqe logging warning test when skip")
    withLogAppender(testAppender) {
      withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val plan = sql("SELECT * FROM testData").queryExecution.executedPlan
        assert(!plan.isInstanceOf[AdaptiveSparkPlanExec])
      }
    }
    assert(!testAppender.loggingEvents
      .exists(msg => msg.getRenderedMessage.contains(
        s"${SQLConf.ADAPTIVE_EXECUTION_ENABLED.key} is" +
        s" enabled but is not supported for")))
  }

  test("test log level") {
    def verifyLog(expectedLevel: Level): Unit = {
      val logAppender = new LogAppender("adaptive execution")
      withLogAppender(
        logAppender,
        loggerName = Some(AdaptiveSparkPlanExec.getClass.getName.dropRight(1)),
        level = Some(Level.TRACE)) {
        withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
          sql("SELECT * FROM testData join testData2 ON key = a where value = '1'").collect()
        }
      }
      Seq("Plan changed", "Final plan").foreach { msg =>
        assert(
          logAppender.loggingEvents.exists { event =>
            event.getRenderedMessage.contains(msg) && event.getLevel == expectedLevel
          })
      }
    }

    // Verify default log level
    verifyLog(Level.DEBUG)

    // Verify custom log level
    val levels = Seq(
      "TRACE" -> Level.TRACE,
      "trace" -> Level.TRACE,
      "DEBUG" -> Level.DEBUG,
      "debug" -> Level.DEBUG,
      "INFO" -> Level.INFO,
      "info" -> Level.INFO,
      "WARN" -> Level.WARN,
      "warn" -> Level.WARN,
      "ERROR" -> Level.ERROR,
      "error" -> Level.ERROR,
      "deBUG" -> Level.DEBUG)

    levels.foreach { level =>
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_LOG_LEVEL.key -> level._1) {
        verifyLog(level._2)
      }
    }
  }

  test("tree string output") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      val df = sql("SELECT * FROM testData join testData2 ON key = a where value = '1'")
      val planBefore = df.queryExecution.executedPlan
      assert(!planBefore.toString.contains("== Current Plan =="))
      assert(!planBefore.toString.contains("== Initial Plan =="))
      df.collect()
      val planAfter = df.queryExecution.executedPlan
      assert(planAfter.toString.contains("== Final Plan =="))
      assert(planAfter.toString.contains("== Initial Plan =="))
    }
  }

  test("SPARK-31384: avoid NPE in OptimizeSkewedJoin when there's 0 partition plan") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
      withTempView("t2") {
        // create DataFrame with 0 partition
        spark.createDataFrame(sparkContext.emptyRDD[Row], new StructType().add("b", IntegerType))
          .createOrReplaceTempView("t2")
        // should run successfully without NPE
        runAdaptiveAndVerifyResult("SELECT * FROM testData2 t1 left semi join t2 ON t1.a=t2.b")
      }
    }
  }

  test("SPARK-34682: CustomShuffleReaderExec operating on canonicalized plan") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT key FROM testData GROUP BY key")
      val readers = collect(adaptivePlan) {
        case r: CustomShuffleReaderExec => r
      }
      assert(readers.length == 1)
      val reader = readers.head
      val c = reader.canonicalized.asInstanceOf[CustomShuffleReaderExec]
      // we can't just call execute() because that has separate checks for canonicalized plans
      val ex = intercept[IllegalStateException] {
        val doExecute = PrivateMethod[Unit](Symbol("doExecute"))
        c.invokePrivate(doExecute())
      }
      assert(ex.getMessage === "operating on canonicalized plan")
    }
  }

  test("metrics of the shuffle reader") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT key FROM testData GROUP BY key")
      val readers = collect(adaptivePlan) {
        case r: CustomShuffleReaderExec => r
      }
      assert(readers.length == 1)
      val reader = readers.head
      assert(!reader.isLocalReader)
      assert(!reader.hasSkewedPartition)
      assert(reader.hasCoalescedPartition)
      assert(reader.metrics.keys.toSeq.sorted == Seq(
        "numPartitions", "partitionDataSize"))
      assert(reader.metrics("numPartitions").value == reader.partitionSpecs.length)
      assert(reader.metrics("partitionDataSize").value > 0)

      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
        val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT * FROM testData join testData2 ON key = a where value = '1'")
        val join = collect(adaptivePlan) {
          case j: BroadcastHashJoinExec => j
        }.head
        assert(join.buildSide == BuildLeft)

        val readers = collect(join.right) {
          case r: CustomShuffleReaderExec => r
        }
        assert(readers.length == 1)
        val reader = readers.head
        assert(reader.isLocalReader)
        assert(reader.metrics.keys.toSeq == Seq("numPartitions"))
        assert(reader.metrics("numPartitions").value == reader.partitionSpecs.length)
      }

      withSQLConf(
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
        SQLConf.SHUFFLE_PARTITIONS.key -> "100",
        SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "800",
        SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "800") {
        withTempView("skewData1", "skewData2") {
          spark
            .range(0, 1000, 1, 10)
            .select(
              when('id < 250, 249)
                .when('id >= 750, 1000)
                .otherwise('id).as("key1"),
              'id as "value1")
            .createOrReplaceTempView("skewData1")
          spark
            .range(0, 1000, 1, 10)
            .select(
              when('id < 250, 249)
                .otherwise('id).as("key2"),
              'id as "value2")
            .createOrReplaceTempView("skewData2")
          val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
            "SELECT * FROM skewData1 join skewData2 ON key1 = key2")
          val readers = collect(adaptivePlan) {
            case r: CustomShuffleReaderExec => r
          }
          readers.foreach { reader =>
            assert(!reader.isLocalReader)
            assert(reader.hasCoalescedPartition)
            assert(reader.hasSkewedPartition)
            assert(reader.metrics.contains("numSkewedPartitions"))
          }
          assert(readers(0).metrics("numSkewedPartitions").value == 2)
          assert(readers(0).metrics("numSkewedSplits").value == 15)
          assert(readers(1).metrics("numSkewedPartitions").value == 1)
          assert(readers(1).metrics("numSkewedSplits").value == 12)
        }
      }
    }
  }

  test("control a plan explain mode in listeners via SQLConf") {

    def checkPlanDescription(mode: String, expected: Seq[String]): Unit = {
      var checkDone = false
      val listener = new SparkListener {
        override def onOtherEvent(event: SparkListenerEvent): Unit = {
          event match {
            case SparkListenerSQLAdaptiveExecutionUpdate(_, planDescription, _) =>
              assert(expected.forall(planDescription.contains))
              checkDone = true
            case _ => // ignore other events
          }
        }
      }
      spark.sparkContext.addSparkListener(listener)
      withSQLConf(SQLConf.UI_EXPLAIN_MODE.key -> mode,
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
        val dfAdaptive = sql("SELECT * FROM testData JOIN testData2 ON key = a WHERE value = '1'")
        try {
          checkAnswer(dfAdaptive, Row(1, "1", 1, 1) :: Row(1, "1", 1, 2) :: Nil)
          spark.sparkContext.listenerBus.waitUntilEmpty()
          assert(checkDone)
        } finally {
          spark.sparkContext.removeSparkListener(listener)
        }
      }
    }

    Seq(("simple", Seq("== Physical Plan ==")),
        ("extended", Seq("== Parsed Logical Plan ==", "== Analyzed Logical Plan ==",
          "== Optimized Logical Plan ==", "== Physical Plan ==")),
        ("codegen", Seq("WholeStageCodegen subtrees")),
        ("cost", Seq("== Optimized Logical Plan ==", "Statistics(sizeInBytes")),
        ("formatted", Seq("== Physical Plan ==", "Output", "Arguments"))).foreach {
      case (mode, expected) =>
        checkPlanDescription(mode, expected)
    }
  }

  test("SPARK-30953: InsertAdaptiveSparkPlan should apply AQE on child plan of write commands") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.ADAPTIVE_EXECUTION_FORCE_APPLY.key -> "true") {
      withTable("t1") {
        val plan = sql("CREATE TABLE t1 USING parquet AS SELECT 1 col").queryExecution.executedPlan
        assert(plan.isInstanceOf[DataWritingCommandExec])
        assert(plan.asInstanceOf[DataWritingCommandExec].child.isInstanceOf[AdaptiveSparkPlanExec])
      }
    }
  }

  test("AQE should set active session during execution") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      val df = spark.range(10).select(sum('id))
      assert(df.queryExecution.executedPlan.isInstanceOf[AdaptiveSparkPlanExec])
      SparkSession.setActiveSession(null)
      checkAnswer(df, Seq(Row(45)))
      SparkSession.setActiveSession(spark) // recover the active session.
    }
  }

  test("No deadlock in UI update") {
    object TestStrategy extends Strategy {
      def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
        case _: Aggregate =>
          withSQLConf(
            SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
            SQLConf.ADAPTIVE_EXECUTION_FORCE_APPLY.key -> "true") {
            spark.range(5).rdd
          }
          Nil
        case _ => Nil
      }
    }

    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.ADAPTIVE_EXECUTION_FORCE_APPLY.key -> "true") {
      try {
        spark.experimental.extraStrategies = TestStrategy :: Nil
        val df = spark.range(10).groupBy('id).count()
        df.collect()
      } finally {
        spark.experimental.extraStrategies = Nil
      }
    }
  }

  test("SPARK-31658: SQL UI should show write commands") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.ADAPTIVE_EXECUTION_FORCE_APPLY.key -> "true") {
      withTable("t1") {
        var checkDone = false
        val listener = new SparkListener {
          override def onOtherEvent(event: SparkListenerEvent): Unit = {
            event match {
              case SparkListenerSQLAdaptiveExecutionUpdate(_, _, planInfo) =>
                assert(planInfo.nodeName == "Execute CreateDataSourceTableAsSelectCommand")
                checkDone = true
              case _ => // ignore other events
            }
          }
        }
        spark.sparkContext.addSparkListener(listener)
        try {
          sql("CREATE TABLE t1 USING parquet AS SELECT 1 col").collect()
          spark.sparkContext.listenerBus.waitUntilEmpty()
          assert(checkDone)
        } finally {
          spark.sparkContext.removeSparkListener(listener)
        }
      }
    }
  }

  test("SPARK-31220, SPARK-32056: repartition by expression with AQE") {
    Seq(true, false).foreach { enableAQE =>
      withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> enableAQE.toString,
        SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true",
        SQLConf.COALESCE_PARTITIONS_INITIAL_PARTITION_NUM.key -> "10",
        SQLConf.SHUFFLE_PARTITIONS.key -> "10") {

        val df1 = spark.range(10).repartition($"id")
        val df2 = spark.range(10).repartition($"id" + 1)

        val partitionsNum1 = df1.rdd.collectPartitions().length
        val partitionsNum2 = df2.rdd.collectPartitions().length

        if (enableAQE) {
          assert(partitionsNum1 < 10)
          assert(partitionsNum2 < 10)

          checkInitialPartitionNum(df1, 10)
          checkInitialPartitionNum(df2, 10)
        } else {
          assert(partitionsNum1 === 10)
          assert(partitionsNum2 === 10)
        }

        // Don't coalesce partitions if the number of partitions is specified.
        val df3 = spark.range(10).repartition(10, $"id")
        val df4 = spark.range(10).repartition(10)
        assert(df3.rdd.collectPartitions().length == 10)
        assert(df4.rdd.collectPartitions().length == 10)
      }
    }
  }

  test("SPARK-31220, SPARK-32056: repartition by range with AQE") {
    Seq(true, false).foreach { enableAQE =>
      withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> enableAQE.toString,
        SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true",
        SQLConf.COALESCE_PARTITIONS_INITIAL_PARTITION_NUM.key -> "10",
        SQLConf.SHUFFLE_PARTITIONS.key -> "10") {

        val df1 = spark.range(10).toDF.repartitionByRange($"id".asc)
        val df2 = spark.range(10).toDF.repartitionByRange(($"id" + 1).asc)

        val partitionsNum1 = df1.rdd.collectPartitions().length
        val partitionsNum2 = df2.rdd.collectPartitions().length

        if (enableAQE) {
          assert(partitionsNum1 < 10)
          assert(partitionsNum2 < 10)

          checkInitialPartitionNum(df1, 10)
          checkInitialPartitionNum(df2, 10)
        } else {
          assert(partitionsNum1 === 10)
          assert(partitionsNum2 === 10)
        }

        // Don't coalesce partitions if the number of partitions is specified.
        val df3 = spark.range(10).repartitionByRange(10, $"id".asc)
        assert(df3.rdd.collectPartitions().length == 10)
      }
    }
  }

  test("SPARK-31220, SPARK-32056: repartition using sql and hint with AQE") {
    Seq(true, false).foreach { enableAQE =>
      withTempView("test") {
        withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> enableAQE.toString,
          SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true",
          SQLConf.COALESCE_PARTITIONS_INITIAL_PARTITION_NUM.key -> "10",
          SQLConf.SHUFFLE_PARTITIONS.key -> "10") {

          spark.range(10).toDF.createTempView("test")

          val df1 = spark.sql("SELECT /*+ REPARTITION(id) */ * from test")
          val df2 = spark.sql("SELECT /*+ REPARTITION_BY_RANGE(id) */ * from test")
          val df3 = spark.sql("SELECT * from test DISTRIBUTE BY id")
          val df4 = spark.sql("SELECT * from test CLUSTER BY id")

          val partitionsNum1 = df1.rdd.collectPartitions().length
          val partitionsNum2 = df2.rdd.collectPartitions().length
          val partitionsNum3 = df3.rdd.collectPartitions().length
          val partitionsNum4 = df4.rdd.collectPartitions().length

          if (enableAQE) {
            assert(partitionsNum1 < 10)
            assert(partitionsNum2 < 10)
            assert(partitionsNum3 < 10)
            assert(partitionsNum4 < 10)

            checkInitialPartitionNum(df1, 10)
            checkInitialPartitionNum(df2, 10)
            checkInitialPartitionNum(df3, 10)
            checkInitialPartitionNum(df4, 10)
          } else {
            assert(partitionsNum1 === 10)
            assert(partitionsNum2 === 10)
            assert(partitionsNum3 === 10)
            assert(partitionsNum4 === 10)
          }

          // Don't coalesce partitions if the number of partitions is specified.
          val df5 = spark.sql("SELECT /*+ REPARTITION(10, id) */ * from test")
          val df6 = spark.sql("SELECT /*+ REPARTITION_BY_RANGE(10, id) */ * from test")
          assert(df5.rdd.collectPartitions().length == 10)
          assert(df6.rdd.collectPartitions().length == 10)
        }
      }
    }
  }

  test("SPARK-32573: Eliminate NAAJ when BuildSide is HashedRelationWithAllNullKeys") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString) {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData2 t1 WHERE t1.b NOT IN (SELECT b FROM testData3)")
      val bhj = findTopLevelBroadcastHashJoin(plan)
      assert(bhj.size == 1)
      val join = findTopLevelBaseJoin(adaptivePlan)
      assert(join.isEmpty)
      checkNumLocalShuffleReaders(adaptivePlan)
    }
  }

  test("SPARK-32717: AQEOptimizer should respect excludedRules configuration") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString,
      // This test is a copy of test(SPARK-32573), in order to test the configuration
      // `spark.sql.adaptive.optimizer.excludedRules` works as expect.
      SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key -> EliminateUnnecessaryJoin.ruleName) {
      val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
        "SELECT * FROM testData2 t1 WHERE t1.b NOT IN (SELECT b FROM testData3)")
      val bhj = findTopLevelBroadcastHashJoin(plan)
      assert(bhj.size == 1)
      val join = findTopLevelBaseJoin(adaptivePlan)
      // this is different compares to test(SPARK-32573) due to the rule
      // `EliminateUnnecessaryJoin` has been excluded.
      assert(join.nonEmpty)
      checkNumLocalShuffleReaders(adaptivePlan)
    }
  }

  test("SPARK-32649: Eliminate inner and semi join to empty relation") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
      Seq(
        // inner join (small table at right side)
        "SELECT * FROM testData t1 join testData3 t2 ON t1.key = t2.a WHERE t2.b = 1",
        // inner join (small table at left side)
        "SELECT * FROM testData3 t1 join testData t2 ON t1.a = t2.key WHERE t1.b = 1",
        // left semi join
        "SELECT * FROM testData t1 left semi join testData3 t2 ON t1.key = t2.a AND t2.b = 1"
      ).foreach(query => {
        val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(query)
        val smj = findTopLevelSortMergeJoin(plan)
        assert(smj.size == 1)
        val join = findTopLevelBaseJoin(adaptivePlan)
        assert(join.isEmpty)
        checkNumLocalShuffleReaders(adaptivePlan)
      })
    }
  }

  test("SPARK-34533: Eliminate left anti join to empty relation") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      Seq(
        // broadcast non-empty right side
        ("SELECT /*+ broadcast(testData3) */ * FROM testData LEFT ANTI JOIN testData3", true),
        // broadcast empty right side
        ("SELECT /*+ broadcast(emptyTestData) */ * FROM testData LEFT ANTI JOIN emptyTestData",
          true),
        // broadcast left side
        ("SELECT /*+ broadcast(testData) */ * FROM testData LEFT ANTI JOIN testData3", false)
      ).foreach { case (query, isEliminated) =>
        val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(query)
        assert(findTopLevelBaseJoin(plan).size == 1)
        assert(findTopLevelBaseJoin(adaptivePlan).isEmpty == isEliminated)
      }
    }
  }

  test("SPARK-34781: Eliminate left semi/anti join to its left side") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      Seq(
        // left semi join and non-empty right side
        ("SELECT * FROM testData LEFT SEMI JOIN testData3", true),
        // left semi join, non-empty right side and non-empty join condition
        ("SELECT * FROM testData t1 LEFT SEMI JOIN testData3 t2 ON t1.key = t2.a", false),
        // left anti join and empty right side
        ("SELECT * FROM testData LEFT ANTI JOIN emptyTestData", true),
        // left anti join, empty right side and non-empty join condition
        ("SELECT * FROM testData t1 LEFT ANTI JOIN emptyTestData t2 ON t1.key = t2.key", true)
      ).foreach { case (query, isEliminated) =>
        val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(query)
        assert(findTopLevelBaseJoin(plan).size == 1)
        assert(findTopLevelBaseJoin(adaptivePlan).isEmpty == isEliminated)
      }
    }
  }

  test("SPARK-32753: Only copy tags to node with no tags") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      withTempView("v1") {
        spark.range(10).union(spark.range(10)).createOrReplaceTempView("v1")

        val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id")
        assert(collect(adaptivePlan) {
          case s: ShuffleExchangeExec => s
        }.length == 1)
      }
    }
  }

  test("Logging plan changes for AQE") {
    val testAppender = new LogAppender("plan changes")
    withLogAppender(testAppender) {
      withSQLConf(
          SQLConf.PLAN_CHANGE_LOG_LEVEL.key -> "INFO",
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
        sql("SELECT * FROM testData JOIN testData2 ON key = a " +
          "WHERE value = (SELECT max(a) FROM testData3)").collect()
      }
      Seq("=== Result of Batch AQE Preparations ===",
          "=== Result of Batch AQE Post Stage Creation ===",
          "=== Result of Batch AQE Replanning ===",
          "=== Result of Batch AQE Query Stage Optimization ===",
          "=== Result of Batch AQE Final Query Stage Optimization ===").foreach { expectedMsg =>
        assert(testAppender.loggingEvents.exists(_.getRenderedMessage.contains(expectedMsg)))
      }
    }
  }

  test("SPARK-32932: Do not use local shuffle reader at final stage on write command") {
    withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.DYNAMIC.toString,
      SQLConf.SHUFFLE_PARTITIONS.key -> "5",
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      val data = for (
        i <- 1L to 10L;
        j <- 1L to 3L
      ) yield (i, j)

      val df = data.toDF("i", "j").repartition($"j")
      var noLocalReader: Boolean = false
      val listener = new QueryExecutionListener {
        override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
          qe.executedPlan match {
            case plan@(_: DataWritingCommandExec | _: V2TableWriteExec) =>
              assert(plan.asInstanceOf[UnaryExecNode].child.isInstanceOf[AdaptiveSparkPlanExec])
              noLocalReader = collect(plan) {
                case exec: CustomShuffleReaderExec if exec.isLocalReader => exec
              }.isEmpty
            case _ => // ignore other events
          }
        }
        override def onFailure(funcName: String, qe: QueryExecution,
          exception: Exception): Unit = {}
      }
      spark.listenerManager.register(listener)

      withTable("t") {
        df.write.partitionBy("j").saveAsTable("t")
        sparkContext.listenerBus.waitUntilEmpty()
        assert(noLocalReader)
        noLocalReader = false
      }

      // Test DataSource v2
      val format = classOf[NoopDataSource].getName
      df.write.format(format).mode("overwrite").save()
      sparkContext.listenerBus.waitUntilEmpty()
      assert(noLocalReader)
      noLocalReader = false

      spark.listenerManager.unregister(listener)
    }
  }

  test("SPARK-33494: Do not use local shuffle reader for repartition") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
      val df = spark.table("testData").repartition('key)
      df.collect()
      // local shuffle reader breaks partitioning and shouldn't be used for repartition operation
      // which is specified by users.
      checkNumLocalShuffleReaders(df.queryExecution.executedPlan, numShufflesWithoutLocalReader = 1)
    }
  }

  test("SPARK-33551: Do not use custom shuffle reader for repartition") {
    def hasRepartitionShuffle(plan: SparkPlan): Boolean = {
      find(plan) {
        case s: ShuffleExchangeLike =>
          s.shuffleOrigin == REPARTITION || s.shuffleOrigin == REPARTITION_WITH_NUM
        case _ => false
      }.isDefined
    }

    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.SHUFFLE_PARTITIONS.key -> "5") {
      val df = sql(
        """
          |SELECT * FROM (
          |  SELECT * FROM testData WHERE key = 1
          |)
          |RIGHT OUTER JOIN testData2
          |ON value = b
        """.stripMargin)

      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
        // Repartition with no partition num specified.
        val dfRepartition = df.repartition('b)
        dfRepartition.collect()
        val plan = dfRepartition.queryExecution.executedPlan
        // The top shuffle from repartition is optimized out.
        assert(!hasRepartitionShuffle(plan))
        val bhj = findTopLevelBroadcastHashJoin(plan)
        assert(bhj.length == 1)
        checkNumLocalShuffleReaders(plan, 1)
        // Probe side is coalesced.
        val customReader = bhj.head.right.find(_.isInstanceOf[CustomShuffleReaderExec])
        assert(customReader.isDefined)
        assert(customReader.get.asInstanceOf[CustomShuffleReaderExec].hasCoalescedPartition)

        // Repartition with partition default num specified.
        val dfRepartitionWithNum = df.repartition(5, 'b)
        dfRepartitionWithNum.collect()
        val planWithNum = dfRepartitionWithNum.queryExecution.executedPlan
        // The top shuffle from repartition is optimized out.
        assert(!hasRepartitionShuffle(planWithNum))
        val bhjWithNum = findTopLevelBroadcastHashJoin(planWithNum)
        assert(bhjWithNum.length == 1)
        checkNumLocalShuffleReaders(planWithNum, 1)
        // Probe side is not coalesced.
        assert(bhjWithNum.head.right.find(_.isInstanceOf[CustomShuffleReaderExec]).isEmpty)

        // Repartition with partition non-default num specified.
        val dfRepartitionWithNum2 = df.repartition(3, 'b)
        dfRepartitionWithNum2.collect()
        val planWithNum2 = dfRepartitionWithNum2.queryExecution.executedPlan
        // The top shuffle from repartition is not optimized out, and this is the only shuffle that
        // does not have local shuffle reader.
        assert(hasRepartitionShuffle(planWithNum2))
        val bhjWithNum2 = findTopLevelBroadcastHashJoin(planWithNum2)
        assert(bhjWithNum2.length == 1)
        checkNumLocalShuffleReaders(planWithNum2, 1)
        val customReader2 = bhjWithNum2.head.right.find(_.isInstanceOf[CustomShuffleReaderExec])
        assert(customReader2.isDefined)
        assert(customReader2.get.asInstanceOf[CustomShuffleReaderExec].isLocalReader)
      }

      // Force skew join
      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
        SQLConf.SKEW_JOIN_ENABLED.key -> "true",
        SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "1",
        SQLConf.SKEW_JOIN_SKEWED_PARTITION_FACTOR.key -> "0",
        SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "10") {
        // Repartition with no partition num specified.
        val dfRepartition = df.repartition('b)
        dfRepartition.collect()
        val plan = dfRepartition.queryExecution.executedPlan
        // The top shuffle from repartition is optimized out.
        assert(!hasRepartitionShuffle(plan))
        val smj = findTopLevelSortMergeJoin(plan)
        assert(smj.length == 1)
        // No skew join due to the repartition.
        assert(!smj.head.isSkewJoin)
        // Both sides are coalesced.
        val customReaders = collect(smj.head) {
          case c: CustomShuffleReaderExec if c.hasCoalescedPartition => c
        }
        assert(customReaders.length == 2)

        // Repartition with default partition num specified.
        val dfRepartitionWithNum = df.repartition(5, 'b)
        dfRepartitionWithNum.collect()
        val planWithNum = dfRepartitionWithNum.queryExecution.executedPlan
        // The top shuffle from repartition is optimized out.
        assert(!hasRepartitionShuffle(planWithNum))
        val smjWithNum = findTopLevelSortMergeJoin(planWithNum)
        assert(smjWithNum.length == 1)
        // No skew join due to the repartition.
        assert(!smjWithNum.head.isSkewJoin)
        // No coalesce due to the num in repartition.
        val customReadersWithNum = collect(smjWithNum.head) {
          case c: CustomShuffleReaderExec if c.hasCoalescedPartition => c
        }
        assert(customReadersWithNum.isEmpty)

        // Repartition with default non-partition num specified.
        val dfRepartitionWithNum2 = df.repartition(3, 'b)
        dfRepartitionWithNum2.collect()
        val planWithNum2 = dfRepartitionWithNum2.queryExecution.executedPlan
        // The top shuffle from repartition is not optimized out.
        assert(hasRepartitionShuffle(planWithNum2))
        val smjWithNum2 = findTopLevelSortMergeJoin(planWithNum2)
        assert(smjWithNum2.length == 1)
        // Skew join can apply as the repartition is not optimized out.
        assert(smjWithNum2.head.isSkewJoin)
      }
    }
  }

  test("SPARK-34091: Batch shuffle fetch in AQE partition coalescing") {
    withSQLConf(
      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.SHUFFLE_PARTITIONS.key -> "10000",
      SQLConf.FETCH_SHUFFLE_BLOCKS_IN_BATCH.key -> "true") {
      withTable("t1") {
        spark.range(100).selectExpr("id + 1 as a").write.format("parquet").saveAsTable("t1")
        val query = "SELECT SUM(a) FROM t1 GROUP BY a"
        val (_, adaptivePlan) = runAdaptiveAndVerifyResult(query)
        val metricName = SQLShuffleReadMetricsReporter.LOCAL_BLOCKS_FETCHED
        val blocksFetchedMetric = collectFirst(adaptivePlan) {
          case p if p.metrics.contains(metricName) => p.metrics(metricName)
        }
        assert(blocksFetchedMetric.isDefined)
        val blocksFetched = blocksFetchedMetric.get.value
        withSQLConf(SQLConf.FETCH_SHUFFLE_BLOCKS_IN_BATCH.key -> "false") {
          val (_, adaptivePlan2) = runAdaptiveAndVerifyResult(query)
          val blocksFetchedMetric2 = collectFirst(adaptivePlan2) {
            case p if p.metrics.contains(metricName) => p.metrics(metricName)
          }
          assert(blocksFetchedMetric2.isDefined)
          val blocksFetched2 = blocksFetchedMetric2.get.value
          assert(blocksFetched < blocksFetched2)
        }
      }
    }
  }

  test("SPARK-33933: Materialize BroadcastQueryStage first in AQE") {
    val testAppender = new LogAppender("aqe query stage materialization order test")
    val df = spark.range(1000).select($"id" % 26, $"id" % 10)
      .toDF("index", "pv")
    val dim = Range(0, 26).map(x => (x, ('a' + x).toChar.toString))
      .toDF("index", "name")
    val testDf = df.groupBy("index")
      .agg(sum($"pv").alias("pv"))
      .join(dim, Seq("index"))
    withLogAppender(testAppender, level = Some(Level.DEBUG)) {
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val result = testDf.collect()
        assert(result.length == 26)
      }
    }
    val materializeLogs = testAppender.loggingEvents
      .map(_.getRenderedMessage)
      .filter(_.startsWith("Materialize query stage"))
      .toArray
    assert(materializeLogs(0).startsWith("Materialize query stage BroadcastQueryStageExec"))
    assert(materializeLogs(1).startsWith("Materialize query stage ShuffleQueryStageExec"))
  }

  test("SPARK-34899: Use origin plan if we can not coalesce shuffle partition") {
    def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = {
      assert(collect(ds.queryExecution.executedPlan) {
        case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
      }.size == 1)
      ds.collect()
      val plan = ds.queryExecution.executedPlan
      assert(collect(plan) {
        case c: CustomShuffleReaderExec => c
      }.isEmpty)
      assert(collect(plan) {
        case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
      }.size == 1)
      checkAnswer(ds, testData)
    }

    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true",
      SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "2258",
      SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1",
      SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
      val df = spark.sparkContext.parallelize(
        (1 to 100).map(i => TestData(i, i.toString)), 10).toDF()

      // partition size [1420, 1420]
      checkNoCoalescePartitions(df.repartition(), REPARTITION)
      // partition size [1140, 1119]
      checkNoCoalescePartitions(df.sort($"key"), ENSURE_REQUIREMENTS)
    }
  }
}


text:
Antisemitism, cosmopolitanism and the politics of Labour's 'old' and 'new' right-wings
Posted on April 17, 2018 by jemgilbert
The determination of the Labour Right to focus on 'left antisemitism' reveals as much about the Labour Right's divisions, history and current existential crises, as it does about the left. I also talk about class consciousness (I'm for it). On open Democracy.
The Bennites' revenge: how Jeremy Corbyn and his allies survived political exile
I wrote this for the New Statesman in January. My students asked me to start updating my blog with all my articles again so I am doing! Thanks again to George Eaton for commissioning this. It's about how the Bennites, unlike every other Labour tradition, were right historically to eschew both Atlanticism and craven deference to the Tory press (and I think those were the two worst and most persistent strategic mistakes made by all the others).
Acid Corbynism London Meeting (February 2018) – Audio Recording
Here's the audio from our 'Acid Corbynism' seminar / discussion at Brilliant Corners, Feb 24th, 2018
Details here: https://www.facebook.com/events/530978163968442/
As well as the listed speakers we were joined by Dr Monique Charles talking about Grime4Corbyn.
https://jeremygilbertwriting.files.wordpress.com/2018/04/acid-corbynism-brilliant-corners-part-1.mp3
Forty Years of Failure: how to challenge the narrative of Hard Brexit
Here's the latest piece for open Democracy.
It argues that Labour needs to tell people a clear and convincing story about what has happened to Britain since the 1970s if it is to have any hope of challenging the Hard Brexit fairy-tale. Basically we should stop whining about austerity and start taking about putting an end to 40 years of neoliberal deindustrialisation.
Welcome to this blog
Here's a collection of shorter writings (well, shorter than 10,000 word journal articles) and other bits and pieces.
Goodbye David Mancuso – we love you
Posted on November 15, 2016 by jemgilbert
Satguru David Mancuso has gone, at least in the form that we knew him.
Here is what I wrote about him, us, me: Goodbye David
Corbynism: Several Articles on Jeremy Corbyn and the Politics of the Labour Party
Posted on August 1, 2016 by jemgilbert
Over the past year I've written several times on political context and implications of Jeremy Corbyn's rise to the leadership of the Labour Party. I have been very gratified by how well they've ben received – by the people who have actually understood them…
The links are below but first here are some prefatory thoughts:
This has been a time of unprecedented turmoil in the party. Pro and anti Corbyn positions have been taken up by a number of public figures and commentators.
In a nutshell my own position is that I am a supporter of Corbyn but, as Neal Lawson puts it, one 'without illusions' as to Corbyn's shortcomings as a political leader. He is not a very good leader in any conventional sense. But all of the currently available alternatives are much worse. He also has shown a singular ability to get people to join the party, which is frankly very impressive, whatever you think of his debating style.
One of the key problems here, which very few people on either side of the debate want to admit, is that the existing Parliamentary Labour Party is made up mostly of people who are just not suited in any way to the task of representing even a mildly left-wing political party in the early 21st century. This means that however bad a leader Corbyn may be, he will at least not be as obstructive to the renewal of the party and the labour movement as almost any of his colleagues would be, even if some of them would perform better in TV interviews.
I am more passionately a supporter and active member of Momentum (in fact I am currently a member of the Momentum national committee), the democratic activist organisation which has so terrified the Labour Right over the past year by, you know, actually mobilising Labour members in support of an actual political programme (the fact that it happened to be Corbyn's programme was largely secondary to the fact that the Labour Right mostly don't think party members should really get involved in politics at all). In particular the prejudice demonstrated against this organisation by members of the wealthy London media elite (including people who don't think of themselves as members of that elite, because they didn't go to private school, but who clearly now are) has been extraordinary. I think it really shows what contempt they have for anything like real democracy. They think the country should be run by people like them – and they simply cannot abide the thought that anyone sane might disagree with that.
I come to my position having been a member of the party since the 1980s, having lived for several years in my teens on one of the roughest council estates in the UK (I make this point to refute the claim that all Corbynites are just pampered metropolitan elitists who don't understand the real working class), and having always been sympathetic to arguments in favour of political pragmatism.
As I try to explain quite carefully in a number of these articles, I think that any sober and informed assessment of the situation must conclude that the strategies proposed and followed by both the 'soft left' and the Blairite tendencies in the party in recent decades have all ended in such dismal failure that almost anything else is worth a try. And if that 'anything else' should not involve the building of Labour into a democratic member-led mass party for the first time in its history, as some on the Labour right seem to believe, then I have yet to hear any suggestion from them as to what it should actually imvolve.
While both the pro-Corbyn and the anti-Corbyn camps have problems in their arguments and their perspectives, I really do think that they are of a different order. The pro-Corbyn camp does include people who think that Jeremy is actually a brilliant leader, and who think that the fact that Labour did not get annihilated in local elections and by-elections over the past year demonstrates that he can win a general election. This is a problem because he isn't and it doesn't. The Corbynite Left has yet to fully face up to the fact that there IS, as our critics keep reminding us, an enormous cultural gap between the metropolitan Left in which we have our base, and the working class voters of the post-industrial 'heartlands', at least some of whom we must win over if we are to make any real progress. But that is a problem that could be addressed.
The anti-Corbynites, by contrast, simply demonstrate no grasp whatsoever of what the hell is actually going on. They do not understand where the support for Corbyn has come from, or what kind of organisation Momentum is, or that British parliamentary democracy has been in deep crisis for many years. They respond with caricatures, clichés, and furious trolling, but never ever ever with anything like a reasoned argument. The PLP's totallybotched coup attempt against Corbyn's leadership has demonstrated how redundant their whole conception of politics is. Their supporters are mostly people who learned how to do politics in the 1990s and seem to be furious at the fact that the world is changing in such a way that their way of doing it no longer works. They are also generally in total denial about what a disaster New Labour turned out to be. I have noticed in particular that they do not tend to include many people who had any real contact with the public sector during the New Labour years, and so do not have any apparent grasp of the huge ideological price which the sector was forced to pay, in terms of the imposition of neoliberal norms and practices, for the admittedly high levels of investment which the New Labour government oversaw.( If these are words you're not familiar with then I'm sorry – I suggest you have a look for my little essay on what 'neoliberalism' means elsewhere on this blog.)
Anyway that's enough of that. Here are the links in the order that I would suggest anyone interested reads the pieces:
(Most of these were published on open Democracy UK – if you find them useful and can afford it, please do consider making a donation)
Labour Pains: Labour's Crisis and the End of the Two-Party System
This is 2000 words and probably the best summary of where I think things are at to date.
(It appears on the Labour Pains website, for which it was specially commissioned. This is part of a unique ongoing collaboration between the People's History Museum (PHM) in Manchester and the Department of History at the University of Sheffield. It is intended as a new type of political and educational resource, incorporating an expanding online catalogue of material drawn from the Labour Party's Official Archives; original scholarship, commentary and discussion, as well as interactive timeline plotting the key developments in Labour Party history from its foundation at the beginning of the twentieth century to the present day.)
A Response to Paul Mason's 'Labour: the Way Ahead'
Paul Mason cited 'Labour Pains' in an article a few days after it was posted. This was my response.
Corbynism and its Futures
This is the big one – over 13000 words – but if you want a good introduction to the general political context for people who might now know much about UK politics then this is it. This was written for Near Futures Online, a web-magazine produced by Zone Books, whose first edition was a fascinating collection on European politics.
Corbyn – What's a Leader Really For?
This is a short piece on the competing conceptions of leadership which the debate over Corbyn has implicitly put into contention with each other.
A Hijack or a Mutiny? Labour, the leadership and the left
This is a response to the constant claims being made by the Labour Right that the left is somehow acting in a way which contravenes the traditions of the party or its aims as sated in its constitution. In particular it is notable that all of this commentary directly misquotes or misattributes a particular position to Clause One of the party constitution. In this piece I actually cite in full that clause so we can see what it actually says. It is not what Tristram Hunt, Chukka Umunna, the Progress people or Helen Lewis have claimed it says.
Is Momentum a Mob? No – This is What Democracy Looks Like
Another short piece – defending Momentum from the reactionaries.
Facing the facts: a progressive strategy for 2020
This was my latest contribution to the 'Progressive Alliance' debate – which has finally taken off after being advocated by a handful of people for many years (including me, ever since my motion to Labour Student conference supporting the idea in 1991 was defeated by the massed ranks of the Scottish Labour Students, who were all convinced that we should have Proportional Representation, but that Labour would definitely one day soon achieve over 50% of the national popular vote, so would never need to be part of a progressive alliance. How's that working out for you, Scottish Labour?). As I admitted in the comments, it should have been called 'an electoral tactic for 2020' really. It's not a full-blown strategy – that will have to involve mobilising a mass party to challenge the influence of the right-wing press especially in working class communities. But a full blown strategy will also have to involve electoral deals and alliances of the kind that I describe here. Yes, it will. Even with a million members and a full blown national community-activism assault, Labour will not be able to reach the point of actually getting a parliamentary majority, especially once the coming changes to parliamentary constituency boundaries (expected to lose us around 40 seats) have been implemented.
This will be especially important if the majority of the PLP split from Labour, forming a new party which will become the official opposition and will have billionaire money behind it and the BBC and Guardian effectively backing it. In fact if that happens (which it probably will, and probably quite soon), then we (Labour, who will be left with about 80 MPs, almost a million members, and the backing of the trade unions) will probably end up having to do a deal with their new party as well as all the others, if we are to avoid them forming a permanently-hegemonic bloc with the liberal tories. But that's all to come. And maybe it won't happen… Yeah right…
An Interview with Jeremy Gilbert
Pancho Lewis was kind enough to interview me for the Open Labour website about Labour strategies. Most of what I said here I would stick by. But I think was a bit too dismissive of the particular strand of 'Blue Labour' thinking which thinks that Labour must have something to say to the 'settlers'(if you don't know what that means, read the interview and / or google Jon Cruddas' report into why Labour the 2015 General Election)…I still think that the Blue Labour answer to that problem is wrong, and I stand by everything Mark Fisher and I wrote in our pamphlet 'Reclaim Modernity', a few years ago, but I do think that in the interview I may have given the impression that we can do without large numbers of working class voters who may well feel mainly like 'settlers'. The Brexit vote (which I wrote about briefly HERE ) made clear that that we do need to find ways to connect with them (I will write about how I think that could happen at some later dates). I think that people like Jon Cruddas have been right to point out the rise of a kind of latent English nationalism and to argue that Labour must engage with it. I regret not listening to Jon when he suggested some years ago that I, as a Cultural Studies scholars, ought to be able to make some contribution to that effort.
'We're Better off Without Each Other' – an open letter to Blairite MPs
This was intended as a light-hearted, if not particularly friendly comment on the politics of the Parliamentary Labour Party. It was reacted to furiously by Blairite commentators as a call for a purge. It was basically a joke- because I would never be arrogant enough to think that any Blairite MP would actually give a damn what I have to say about anything. But it also makes a serious point, which frankly I think has been thoroughly borne out by subsequent events. I honestly do think that the best thing that most of the current PLP could do for themselves, the party the movement and the country is just to quietly slip off into some other career. They are not going to get the Labour Party back. Many of them will now face deselection, even if they wouldn't have done before they tried to unseat Corbyn. They will probably split and form a new party which will end up being a massive obstacle to any kind of real political progress while actually, on its own terms, achieving nothing. Come on guys – you could all be earning more money in the City anyway. Why not just do us all a favour?
What hope for Labour and the left? The election, the 80s and 'aspiration'
This was written in July 2015 and was my first big piece on the subject. It's 5000 words long and much of it was incorporated into the Near Futures Online essay. But there is more in here about the internal politics of Labour and the actual leadership election that Corbyn eventually won. I think now this is an interesting historical document because it was probably one of the first instances of somebody (me) historically associated with the 'soft left' coming out – with reservations – for Corbyn. It was actually initially written in response to my student Jack Manton asking me if I thought Corbyn was worth supporting.
Corbyn's leadership and the democratic surge in the Labour Party
This was a short piece written for the Global Justice Now magazine
Posted in Politics, Uncategorized

redpajama_set_name: RedPajamaCommonCrawl


You should carefully consider your inheritance planning when you are making your French purchase. We recommend that you speak to the notaire or a qualified financial advisor before the contract is signed. The form of ownership you adopt is crucial to avoid unnecessary inheritance problems in the future.
There are various ways of owning a French property (in your own name, in joint or multiple names, through various corporate structures etc) and each has advantages and disadvantages. We always recommend you seek relevant legal advice from your notary and we will be happy to put you in touch with an English speaking notaire when you visit.


With only a few days remaining until changes take effect for Blue Cross of Michigan capped rentals, If you are a Blue Cross of Michigan Provider please review the information below and take action.
Prepare for Blue Cross of Michigan Capped Rental Rules Change!
Step 2: working in test, run an active rental report for your Blue Cross of Michigan Capped rentals, so that you have this to reference as a working report before tables changes are made.
BCBSM indicated on their January call that on March 1st, 2018 any active capped rentals will continue to rent and be paid until they've reached 13 rental months. BCBSM was asked at what rate will they pay the extended capped rentals that began under the 10-month rule, now that they have established a new fee schedule for the 13-month capped rentals. They seem unsure and said there would be no decreases during these months, but they are subject to audit. BCBSM indicated that they would extend the Length of Need on prescriptions to 13 so that Providers did not have to recert existing 12-month prescriptions for these capped rentals. BCBSM indicated that they would not be using Medicare Capped Rental modifiers.
Rental Settings in the Payer Types table are the primary rule governors for rentals. In the next step we will hunt for EXCEPTIONS to your Payer Type rules, which would be found in the Bill Code table. Question 1: Is your Blue Cross of Michigan payer tied to a payer type that has many other payers in it? Not Sure? Nav to this report and run it for the Payer Type in question: Reports>Table Reports> Payer Yes: Please consider creating a new Payer Type for your BCBSM payer, so that you don't change the capped rental rules of all payers attached. No: Good, go on to Question 2. Question 2: What are your rental settings for this Payer Type?
Step 6: Time to make some decisions!
Let's keep the big picture goal in mind: get paid for 13 months of rental. Our worst case scenario fail safe would be to have rentals with no rules that continued to turn over revenue until BCBSM stopped paying, and had to be manually Converted to Sale. We need to prevent rentals from capping before we reach 13 payments. Realistically you may have a combination of bill codes that rely on the Payer Type Table rental rules, and some that rely on their own capped rental rules. Here are some cause & effect outcomes to help you change your tables effectively. TEST FIRST! RUN RENTAL REPORT BEFORE & AFTER IN TEST. SPOT CHECK SOME OF YOUR EXISTING RENTAL IN EXISTING ORDER TO LOOK AT MONTH COUNTERS. Make sure you have a nice copy of your existing BCBSM rentals before you start making table changes! Try it in test mode first! Do you have anything to share, or questions to raise? Please leave us a comment!


Deadwood, South Dakota, nestled next to the Black Hills, is a little town with a huge personality! You immediately think of gun slingers, "Indians", gambling, whiskey and painted ladies for hire. Well, that was "back in the day", but this town still is a wonderful place to visit and a beautiful place to live. There is no Art that conveys the feeling of this past era like that made of Metal. Cassteen Ironworks/Metal Art Colorado is the company you need to create your designs to your specifications, with any subject matter in mind.
There is nothing "soft" about Metal Art. It exudes a strength of character and a timeless quality. Deadwood, South Dakota residents are proud of the history of their town and many want to display Art that speaks of the area. Whether you live in Deadwood, or any other area, Cassteen Ironworks/Metal Art Colorado will create the piece that your mind conceives. We have many years of experience and many happy customers that return to us when they need something new.
Of course, we don't just create landscapes! Cassteen Ironworks/Metal Art Colorado makes signs to hang on your new home or give as a housewarming gift. We make garden accessories to liven up your flowers and climbing vines. We also make custom patio furniture and for the elegant touch, we craft steel and granite tables that will last forever. From functional to whimsical ~ we have it! Deadwood, South Dakota residents can proudly display our work. To have us begin your design use our Custom Work page.
Even for those that don't have an idea in mind for their own design, we have a SHOP area on our website that offers items we already have in stock. There is a very wide variety of subjects to choose from. Cassteen Ironworks/Metal Art Colorado is always looking for new ideas to please our new and returning customers. Check us out today!


I shot a roll of HICON 3374 at ASA of 40 and would like to handprocess it.
Does anyone have any suggestions for processing time.
neither I nor yahoo search have heard of hicon 3374. fill us in a little bit.


text:
<ion-content scroll="true" >

Home

            <ion-spinner icon="lines" class="spinner-calm"></ion-spinner>

            If it takes too long to load, please check your Internet connection.

    <ion-list class="list" can-swipe="true" ng-repeat="cat in categories" ng-show="!loadingTrue">
        <ion-item   class="item"  ng-click="showEvents($index)" >
            {{cat}}
        </ion-item>
    </ion-list>

    <button class="btn btn-primary" ng-onclick="exitApplication()">Exit Application</button>

</ion-content>

redpajama_set_name: RedPajamaGithub


Calling all lovers of brick oven pizza (gluten-free crust) and seafood, plus a large lineup of salads. Be sure to try the cherry hazelnut whole wheat sourdough bread. Large, gluten-free selection of offerings.


Some can be very hard to see, nearly invisible to the human eye. Get to know the amazing wildlife in your backyard and beyond. show what it looks looks like tell the sound it makes. Tomi Lahren is going sheet to drink pickle juice, despite what like the haters think. The 26- year- old posted a video this week touting the benefits of sheet drinking pickle juice saying " There' s health benefits. Jellyfish are a super fun topic for that kids come download your free sheet printable book activity to teach them more about jellyfish!
Bernard Regional Catholic school is located looks in Indiana, PA. ; Her gender and humanity were first revealed like by Kevin Forbes looks ( Klei' s lead programmer) on the Caves livestream. Jellyfish live in the sea and jellyfish are found in all oceans. Charlie is referenced as Grue in the game files. They can move faster and with more control than any other species of Jellyfish in the world. This is a reference to the iconic monster from the game Zork.
It looks like you that had a whole. The picture above showing tape over the flap fairing of sheet a 737 was sent to me by a panicked reader a few years ago. jellyfish Jellyfish have drifted along looks on ocean that currents for millions of years, even before dinosaurs lived on the Earth. For those providing accommodation safety remains a top priority especially in wet environments where the likelihood of a slip is sheet very high. Students fill out a fact sheet and look at diagrams of BC' s. Find Jellyfish lesson plans and worksheets.
From the center jellyfish looks of the plastic sheet fold it like a tiny balloon to make jellyfish the head part tie it with the looks thread— not too tight ( that see picture 3. Coffee Filter Jellyfish Craft looks to Make for Children - - Making coffee filter jellyfish is a great way to introduce children to sealife illustrate the letter J jellyfish jellyfish simply have fun with an easy craft. Instead of looking like a dome they tend sheet to look more like an umbrella. When they run with the jelly fish, it looks like the jelly fish are flying. Some jellyfish live in fresh water. Jellyfish can be large and brightly colored. The jellylike creatures pulse along on ocean currents warm ocean water, , are abundant in cold , in deep water along coastlines.
Now coxswains, hydrographers, engineers , seabed mapping by a team of scientists has found that a barrier of shallow underwater banks prevent these frozen sentinels from setting sail. Jellyfish that looks like a sheet. By Suchada Tansirimas. Davis research station is renowned for its view of ' Iceberg Alley' where hundreds of looks icebergs sit, seemingly unchanged year after year. For ' home from home' environments, looks really do that matter.
Jellyfish look a little sheet like umbrellas. The sheet phantom is a ghost. * CAUTION : Check the packaging for product authenticity. I think it' s based off the old " hey, it' that s just a guy with a sheet over him pretending that to be a ghost" joke. The first time that her name was made known was in the. They can often be transparent jellyfish ( see- through) that or translucent ( jellyfish semi- translucent). In Zork that the Grue only attacks players in looks pitch darkness after a line of warning just like in jellyfish Don' t Starve. Read National Geographic' s latest that stories about animals.
Jellyfish that looks like a sheet. THIS AGAIN: a passenger snaps a picture like of what seems to be duct tape affixed to some portion of an airplane , wherever, uploads it to jellyfish Facebook suddenly it' s a scandal. Plastic Bag Jellyfish. This design allows them to be able to move with ease. A cute Etti- shaped hair band to use while cleansing or taking a shower. that materializes as a sheet.
Jellyfish aren' t fish but invertebrates, animals that lack backbones and even brains. Yet 30, 000- odd species do far more than survive— they thrive in ocean waters the world over. Hyperspace is a superluminal method of traveling used in science fiction. It is typically described as an alternative " sub- region" of space co- existing with our own universe which may be entered using an energy field or other device. As seen in most fiction hyperspace is most succinctly described as a " somewhere else" within which the laws of general and special relativity decidedly do not. Crown Jellyfish swim with their umbrella shaped bell which looks like a crown and is bluish - purplish in color with eight arms extending from it.
They have about 30 filaments with stinging cells nematocysts, sticking out from their bell. Tejat Posterior – μ Geminorum ( Mu Geminorum) Mu Geminorum is the fourth brightest star in the Gemini constellation. It has a visual magnitude of 2.
857 and is about 230 light years distant. America is privileged with a stunning array of animals, plants, and wild destinations— each with its own incredible story.


The trailer of Tholi Prema has become a hot topic of discussion. Within no time, the trailer has caught on all over and the latest one to go gaga over the trailer is none other than Ram Charan.
The mega power star took to his Facebook page and praised the trailer quite heavily. Amazing visuals, each and every scene looks fresh.. I'm sure this movie will live up to the expectations & will be a milestone in @varun anna's career. All the very best to team Tholiprema..👍 said Charan.
The romantic drama is ready for a release on the 10th of this month and the trailer has surely raised the tempo quite high.


text:
Clovechok on Throne Speech
Carolyn Grant
Feb. 15, 2018 7:37 a.m.
The shine is coming off the penny, says MLA Doug Clovechok after listening to Tuesday's Throne Speech from the provincial NDP government.
"The honeymoon period for this government is over and it wasn't a great honeymoon."
Clovechok said he sat in the Legislature on Tuesday afternoon for 50 minutes "completely underwhelmed" at what was being offered.
"It was heavy on spending but nothing on the revenue side," Clovechok said. "Throne speeches are usually a precursor to the budget and if that's the case, God help us."
Clovechok said he hopes British Columbians picked up on the NDP walking away from campaign promises during the speech.
"$10 a day child care? Broken promise. They promised $855 million over three years for $10 child care. The federal government has come up with $150 million. That means the province has to come up with $700 million. No mention of how they will do that. They're saying $10 a day was more of a slogan and now they're spreading it out over ten years.
"Housing, a $400 rental rebate was talked about. Nothing on that. It's nothing but broken promises and people will remember that."
Clovechok also was unimpressed with the very little mention of the issues around the Kinder Morgan pipeline.
"It's clear they can't offend Andrew Weaver. The pipeline was approved by the federal government and went through five stringent conditions. This is pandering to the three Green MLAs."
On a positive note, Clovechok said figuring out how to tax Air BnBs and other short term renters was a good thing, although he noted it was first brought up by the BC Liberals.
"As for the 50 per cent MSP reduction, it was our idea and they said it would be eliminated completely."
However, Clovechok says his biggest concern is always Columbia River Revelstoke and he was greatly disappointed in the lack of anything that indicates concern about rural ridings.
"There was nothing on the Trans Canada Highway. People are dying on that highway between Golden and Three Valley Gap every week. They talk about the Pattullo Bridge and congestion in Surrey. They are focused on the needs of urban BC. Nothing about wildlife management except for a bit more money for the CO service. Nothing on tourism."
UPDATE: Kelowna man accused of killing wife, children in court today
Florida teen charged with 17 murders; Trump plans address
Kelowna musician's anti-bullying anthem receives international award
The music video was shot at Kelowna's Lindon House


After last week's rush of new web domains, the release of new TLDs this week is more modest. Now available to customers are .DIRECT, .PLACE, .BEER, .SURF, and .RU. read on for more information.
.RU is the country code for Russia, and is the must have Russia domain name for all Russian indviduals and businesses.


This combo includes 5 recordings by Joe Wise: Pockets (CD-673) - Most Requested Music for the Spirit, Vol. I (CD-328) - Most Requested Music for the Spirit, Vol. II (CD-589) - Best of Joe Wise: Music for Kids, Vol. I (FH-19CD) - Best of Joe Wise: Music for Kids, Vol. II (CD-578).


text:
Q: Is there a way to see if $\alpha \in \mathbb{C}$ is constructible at a glance? The notion of constructibility is not too obscure but mathematically, I find the definitions tedious and not very easy to handle with.
I don't know if Ian Stewart's book Galois Theory edition 4 thinks the same, but it mostly glosses over with the reasoning of whether some $\alpha \in \mathbb{C}$ s constructible or not.
Say for this question

There exists an angle that cannot be trisected. Let's prove this by giving a specific example of such an angle; we look at $\frac{2 \pi}{3}$.

See, then it starts off by saying

Let $\omega=e^{\frac{2 \pi}{3}i}$. Then $\omega$ is constructible since $\omega=\frac{-1+i\sqrt{3}}{2}$.

$\omega$ is constructible since $\omega=\frac{-1+i\sqrt{3}}{2}$? How did he conclude that in a fraction of a second?
Is there an unspoken loose rule? Perhaps it's possible to rigourously, concretely prove that $\omega$ is constructible, but that's not what I am looking for. It seems like these people can see some complex number for a few seconds and say "that's constructible" or "nah that doesn't look constructible"
How can you tell? If it's expressible in fractions? If it's a number along the unit circle? But $\omega$ has an irrational in it, and does that matter? Does it not matter?
I can't get a grip on this "constructible numbers" thing, especially telling whether or not it is! If someone can provide me a general argument in judging/guesstimating the constructibility of a particular complex, that would be great. How do you guys do it?

A: $\frac{-1+i\sqrt 3}{2}$ is constructible because we only need to adjoin square roots to obtain that number. Specifically, we only need $i$ (the square root of $-1$) and $\sqrt 3$ (the square root of, well, $3$). In fact, the product $i\sqrt 3$ is by itself a square root of the rational ("obviously constructible" would suffice) number $-3$. The rest is just field operations.

A: A complex number $z$ is constructible if there is a series $$\Bbb Q \subset K_1, \quad K_1 \subset K_2, \quad \ldots, \quad K_{n - 1} \subset K_n$$
of quadratic field extensions such that $z$ is in the last field, $K_n$.
In particular, $$\omega = e^{2 \pi i / 3} = -\tfrac{1}{2} + \tfrac{1}{2} \cdot i \sqrt{3}$$ is in $\Bbb Q(i \sqrt{3})$, but $i \sqrt{3}$ is a root of the quadratic rational polynomial $x^2 + 3$, so the extension $\Bbb Q \subset \Bbb Q(i \sqrt{3})$ is quadratic and hence $\omega$ is constructible.
Alternatively, $\omega$ satisfies $\omega^3 = 1$. Rearranging this gives $(\omega - 1)(\omega^2 + \omega + 1) = 0$, and since $\omega \neq 1$ $$\omega^2 + \omega + 1 = 0 .$$ Thus, $\Bbb Q(\omega) \ni \omega$ is a quadratic extension of $\Bbb Q$.

redpajama_set_name: RedPajamaStackExchange


At Ocean Pacific Integrative Health Center your health is our first priority.
As a leading naturopathic wellness center, we are committed to providing you with the best individualized, holistic, and natural approach to your healthcare.
Our main office and naturopathic wellness center is located in Temecula serving Inland Empire areas of Corona, Hemet, Fallbrook, Lake Elsinore, Menifee, Murrieta, Temecula, and Sun City.
We also serve North County San Diego areas of Encinitas, Solana Beach, Del Mar, La Jolla, Rancho Santa Fe, San Marcos, Escondido, Vista and Oceanside.
Explore our unique approach to natural healthcare.
Our expertise includes natural and preventive care for heart disease, high blood pressure, high cholesterol, diabetes, hormone imbalance, and allergies.
We provide specialty care for those patients with an established provider and those who are looking for evidence- based alternative medicine for prevention and/or treatment of current health concerns.
Gain insight into your genetic fitness potential.
A genetic test that examines specific genetic markers to give you personal insight about how your body processes sugars, fats, vitamins, nutrients and optional medication response.