From 732b0c81d783055c119660c98233545f6adf1662 Mon Sep 17 00:00:00 2001 From: Thang Long VU Date: Wed, 8 Jul 2026 16:10:57 +0000 Subject: [PATCH 1/2] [CONNECT] Default refreshPhaseEnabled from spark.api.mode Derive the default of `QueryExecution.refreshPhaseEnabled` from the session's `spark.api.mode` config instead of hardcoding `true`: the refresh phase stays on in Classic and is off in Connect. The `QueryExecution` refresh phase (`tableVersionsRefreshed`, backed by `V2TableRefreshUtil.refresh`) reloads every versioned DSv2 relation from the catalog at execution time to account for the delay between analysis and the subsequent phases. Spark Connect re-resolves and re-analyzes each plan on every request, so by the time a plan reaches execution the analyzed plan already references the latest table state, making the reload a redundant `catalog.loadTable` round-trip. Rather than threading `refreshPhaseEnabled = false` through every Connect-side `QueryExecution` / `Dataset.ofRows` / `executePlan` construction site, this makes the flag's default a function of `spark.api.mode` (keyed off `SparkSessionBuilder.API_MODE_KEY`). All the existing construction sites that do not pass the flag explicitly (including `SessionState.executePlan` via `createQueryExecution`, and `QueryExecution.create` / `runCommand`) then pick up the mode-appropriate default with no per-call-site changes. Co-authored-by: Isaac --- .../spark/sql/execution/QueryExecution.scala | 22 ++++++++++++--- .../sql/execution/QueryExecutionSuite.scala | 27 ++++++++++++++++++- 2 files changed, 44 insertions(+), 5 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala index e37588faf8fb2..501e98c5590cc 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala @@ -18,6 +18,7 @@ package org.apache.spark.sql.execution import java.io.{BufferedWriter, OutputStreamWriter} +import java.util.Locale import java.util.UUID import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong} import javax.annotation.concurrent.GuardedBy @@ -29,7 +30,7 @@ import org.apache.hadoop.fs.Path import org.apache.spark.{SparkContext, SparkException} import org.apache.spark.internal.LogKeys.EXTENDED_EXPLAIN_GENERATOR import org.apache.spark.rdd.RDD -import org.apache.spark.sql.{AnalysisException, ExtendedExplainGenerator, Row} +import org.apache.spark.sql.{AnalysisException, ExtendedExplainGenerator, Row, SparkSessionBuilder} import org.apache.spark.sql.catalyst.{InternalRow, QueryPlanningTracker} import org.apache.spark.sql.catalyst.analysis.{Analyzer, LazyExpression, NameParameterizedQuery, UnsupportedOperationChecker} import org.apache.spark.sql.catalyst.expressions.codegen.ByteCodeStats @@ -70,7 +71,7 @@ class QueryExecution( val tracker: QueryPlanningTracker = new QueryPlanningTracker, val mode: CommandExecutionMode.Value = CommandExecutionMode.ALL, val shuffleCleanupModeOpt: Option[ShuffleCleanupMode] = None, - val refreshPhaseEnabled: Boolean = true, + val refreshPhaseEnabled: Boolean = QueryExecution.refreshPhaseEnabledDefault(sparkSession), val queryId: UUID = UUIDv7Generator.generate(), // When a transaction is active, callers creating nested QueryExecution instances MUST pass // the enclosing QueryExecution's analyzer here to propagate the transaction context. @@ -731,10 +732,23 @@ object QueryExecution { private def nextExecutionId: Long = _nextExecutionId.getAndIncrement + // The refresh phase reloads versioned DSv2 relations from the catalog so that the plan reflects + // the latest table state at execution time, accounting for the delay between analysis and the + // subsequent phases. Spark Connect re-resolves and re-analyzes every plan per request, so by the + // time a plan reaches execution it already references the latest table state and the refresh is a + // redundant catalog round-trip. It is therefore disabled by default in Connect and enabled by + // default in Classic, keyed off the session's [[SparkSessionBuilder.API_MODE_KEY]]. + private[sql] def refreshPhaseEnabledDefault(sparkSession: SparkSession): Boolean = { + val apiMode = sparkSession.sessionState.conf + .getConfString(SparkSessionBuilder.API_MODE_KEY, SparkSessionBuilder.API_MODE_CLASSIC) + .trim.toLowerCase(Locale.ROOT) + apiMode != SparkSessionBuilder.API_MODE_CONNECT + } + private[execution] def create( sparkSession: SparkSession, logical: LogicalPlan, - refreshPhaseEnabled: Boolean = true): QueryExecution = { + refreshPhaseEnabled: Boolean = refreshPhaseEnabledDefault(sparkSession)): QueryExecution = { new QueryExecution( sparkSession, logical, @@ -932,7 +946,7 @@ object QueryExecution { sparkSession: SparkSession, command: LogicalPlan, name: String, - refreshPhaseEnabled: Boolean = true, + refreshPhaseEnabled: Boolean = refreshPhaseEnabledDefault(sparkSession), mode: CommandExecutionMode.Value = CommandExecutionMode.SKIP, shuffleCleanupModeOpt: Option[ShuffleCleanupMode] = None, analyzerOpt: Option[Analyzer] = None) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala index f7afdb5e6e537..be54023719543 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala @@ -22,7 +22,7 @@ import scala.io.Source import scala.util.Try import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent, SparkListenerJobStart} -import org.apache.spark.sql.{AnalysisException, ExtendedExplainGenerator, FastOperator, SaveMode} +import org.apache.spark.sql.{AnalysisException, ExtendedExplainGenerator, FastOperator, SaveMode, SparkSessionBuilder} import org.apache.spark.sql.catalyst.{QueryPlanningTracker, QueryPlanningTrackerCallback, TableIdentifier} import org.apache.spark.sql.catalyst.analysis.{CurrentNamespace, UnresolvedFunction, UnresolvedRelation} import org.apache.spark.sql.catalyst.expressions.{Alias, UnsafeRow} @@ -585,6 +585,31 @@ class QueryExecutionSuite extends SharedSparkSession { } } + test("refreshPhaseEnabled defaults to true in Classic and false in Connect") { + // Not set: defaults to Classic, so the refresh phase is on. + assert(QueryExecution.refreshPhaseEnabledDefault(spark)) + assert(spark.sessionState.executePlan(OneRowRelation()).refreshPhaseEnabled) + + withSQLConf(SparkSessionBuilder.API_MODE_KEY -> SparkSessionBuilder.API_MODE_CONNECT) { + assert(!QueryExecution.refreshPhaseEnabledDefault(spark)) + assert(!spark.sessionState.executePlan(OneRowRelation()).refreshPhaseEnabled) + } + + withSQLConf(SparkSessionBuilder.API_MODE_KEY -> SparkSessionBuilder.API_MODE_CLASSIC) { + assert(QueryExecution.refreshPhaseEnabledDefault(spark)) + assert(spark.sessionState.executePlan(OneRowRelation()).refreshPhaseEnabled) + } + } + + test("refreshPhaseEnabled default is derived from spark.api.mode case-insensitively") { + withSQLConf(SparkSessionBuilder.API_MODE_KEY -> "CONNECT") { + assert(!QueryExecution.refreshPhaseEnabledDefault(spark)) + } + withSQLConf(SparkSessionBuilder.API_MODE_KEY -> " Connect ") { + assert(!QueryExecution.refreshPhaseEnabledDefault(spark)) + } + } + case class MockCallbackEagerCommand( var trackerAnalyzed: QueryPlanningTracker = null, var trackerReadyForExecution: QueryPlanningTracker = null) From f33c3a2cd4cc61b58251782083dc74b31beac140 Mon Sep 17 00:00:00 2001 From: Thang Long VU Date: Wed, 8 Jul 2026 20:47:56 +0000 Subject: [PATCH 2/2] [CONNECT] Fix compile: resolve refreshPhaseEnabled default in constructor body A default argument expression cannot reference another parameter in the same parameter list, so `refreshPhaseEnabled: Boolean = refreshPhaseEnabledDefault( sparkSession)` failed to compile ("not found: value sparkSession"). Make the primary constructor take a private `refreshPhaseEnabledOpt: Option[Boolean]` (default None) and resolve the public `refreshPhaseEnabled` field in the class body, where `sparkSession` is in scope: honor an explicit value, otherwise fall back to the mode-based default (enabled in Classic, disabled in Connect). QueryExecution.create and runCommand keep a plain Boolean parameter and forward it via refreshPhaseEnabledOpt. Co-authored-by: Isaac --- .../spark/sql/execution/QueryExecution.scala | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala index 501e98c5590cc..6d2dd9421492b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala @@ -71,7 +71,11 @@ class QueryExecution( val tracker: QueryPlanningTracker = new QueryPlanningTracker, val mode: CommandExecutionMode.Value = CommandExecutionMode.ALL, val shuffleCleanupModeOpt: Option[ShuffleCleanupMode] = None, - val refreshPhaseEnabled: Boolean = QueryExecution.refreshPhaseEnabledDefault(sparkSession), + // Whether the refresh phase runs. When left unset it defaults based on the session's + // `spark.api.mode`: enabled in Classic and disabled in Connect. A default value that + // references `sparkSession` cannot be expressed directly on this parameter, so callers that + // want the mode-based default omit it and it is resolved in [[refreshPhaseEnabled]] below. + refreshPhaseEnabledOpt: Option[Boolean] = None, val queryId: UUID = UUIDv7Generator.generate(), // When a transaction is active, callers creating nested QueryExecution instances MUST pass // the enclosing QueryExecution's analyzer here to propagate the transaction context. @@ -79,6 +83,11 @@ class QueryExecution( // of the transaction and will load tables outside the transaction's catalog scope. val analyzerOpt: Option[Analyzer] = None) extends LookupCatalog { + // Resolve the refresh phase flag: honor an explicit value if provided, otherwise fall back to + // the mode-based default (enabled in Classic, disabled in Connect). + val refreshPhaseEnabled: Boolean = + refreshPhaseEnabledOpt.getOrElse(QueryExecution.refreshPhaseEnabledDefault(sparkSession)) + val id: Long = QueryExecution.nextExecutionId // Used by SQLExecution to determine whether to use the existing queryId or generate a new one. @@ -748,13 +757,13 @@ object QueryExecution { private[execution] def create( sparkSession: SparkSession, logical: LogicalPlan, - refreshPhaseEnabled: Boolean = refreshPhaseEnabledDefault(sparkSession)): QueryExecution = { + refreshPhaseEnabled: Boolean = true): QueryExecution = { new QueryExecution( sparkSession, logical, mode = CommandExecutionMode.ALL, shuffleCleanupModeOpt = Some(determineShuffleCleanupMode(sparkSession.sessionState.conf)), - refreshPhaseEnabled = refreshPhaseEnabled) + refreshPhaseEnabledOpt = Some(refreshPhaseEnabled)) } /** @@ -946,7 +955,7 @@ object QueryExecution { sparkSession: SparkSession, command: LogicalPlan, name: String, - refreshPhaseEnabled: Boolean = refreshPhaseEnabledDefault(sparkSession), + refreshPhaseEnabled: Boolean = true, mode: CommandExecutionMode.Value = CommandExecutionMode.SKIP, shuffleCleanupModeOpt: Option[ShuffleCleanupMode] = None, analyzerOpt: Option[Analyzer] = None) @@ -956,7 +965,7 @@ object QueryExecution { command, mode = mode, shuffleCleanupModeOpt = shuffleCleanupModeOpt, - refreshPhaseEnabled = refreshPhaseEnabled, + refreshPhaseEnabledOpt = Some(refreshPhaseEnabled), analyzerOpt = analyzerOpt) val result = QueryExecution.withInternalError(s"Executed $name failed.") { SQLExecution.withNewExecutionId(qe, Some(name)) {