You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/04/26 22:53:57 UTC
(spark) branch master updated: [SPARK-56614][SQL][CONNECT] Add config for strict DataFrame column resolution
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new fb3092a41488 [SPARK-56614][SQL][CONNECT] Add config for strict DataFrame column resolution
fb3092a41488 is described below
commit fb3092a4148857a8386353782817de01f10ec036
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Mon Apr 27 07:53:46 2026 +0900
[SPARK-56614][SQL][CONNECT] Add config for strict DataFrame column resolution
### What changes were proposed in this pull request?
Add a new internal SQL config `spark.sql.analyzer.strictDataFrameColumnResolution` (default `true`) that controls how `UnresolvedAttribute`s carrying a `PLAN_ID_TAG` (Spark Connect DataFrame columns) are resolved in `ColumnResolutionHelper`:
- When `true` (default): strict plan-id-based resolution only; no name-based fallback.
- When `false`: if plan-id-based resolution does not resolve the tagged attribute, also try name-based resolution as a fallback.
Additional test hygiene in this PR:
- Move `test_self_join`, `test_self_join_II`, `test_self_join_III`, `test_self_join_IV`, `test_drop_notexistent_col`, and `test_select_join_keys` from `test_dataframe.py` to `ColumnTestsMixin` in `test_column.py`; they exercise DataFrame column resolution and belong with the Column tests.
- In `test_parity_column.py`, add `ColumnParityTestsWithNonStrictDFColResolution` which re-runs the Column parity suite with the new config set to `false`. Both parity classes pin the config value via `setUpClass`/`tearDownClass` and include a `test_df_col_resolution_mode` sanity check.
- In `test_connect_column.py`, add `test_select_column_replaced_by_withcolumn`: under `strictDataFrameColumnResolution=false` the plan-id-tagged reference to a `withColumn`-shadowed column resolves via name fallback; under `strictDataFrameColumnResolution=true` the same query fails with `CANNOT_RESOLVE_DATAFRAME_COLUMN`.
### Why are the changes needed?
Keep Spark Connect DataFrame column references strictly anchored to their plan of origin by default, while giving users an escape hatch to opt into the more permissive name-based fallback path.
### Does this PR introduce _any_ user-facing change?
A new internal config is added. Default behavior is equivalent to current Apache Spark master for tagged attributes; setting `spark.sql.analyzer.strictDataFrameColumnResolution=false` opts into a slightly more permissive path where tagged attributes can also be resolved by name.
### How was this patch tested?
- `build/sbt 'catalyst/testOnly *AnalysisSuite'`
- `build/sbt 'sql/testOnly *SparkSessionExtensionSuite *ColumnNodeToExpressionConverterSuite *ResolverGuardSuite *PlanResolutionSuite *DataFrameSuite *DataFrameColumnarSuite'`
- `python/run-tests --testnames "pyspark.sql.tests.test_column,pyspark.sql.tests.connect.test_parity_column,pyspark.sql.tests.connect.test_connect_column SparkConnectColumnTests.test_select_column_replaced_by_withcolumn"`
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic), claude-opus-4-7
Closes #55531 from zhengruifeng/spark-dev-4.
Authored-by: Ruifeng Zheng <ru...@apache.org>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../sql/tests/connect/test_connect_column.py | 22 ++++++++
.../sql/tests/connect/test_parity_column.py | 42 ++++++++++++++++
python/pyspark/sql/tests/test_column.py | 56 +++++++++++++++++++++
python/pyspark/sql/tests/test_dataframe.py | 58 ----------------------
.../catalyst/analysis/ColumnResolutionHelper.scala | 8 ++-
.../org/apache/spark/sql/internal/SQLConf.scala | 12 +++++
6 files changed, 138 insertions(+), 60 deletions(-)
diff --git a/python/pyspark/sql/tests/connect/test_connect_column.py b/python/pyspark/sql/tests/connect/test_connect_column.py
index ef2965fde3f4..6fa6c4686c52 100644
--- a/python/pyspark/sql/tests/connect/test_connect_column.py
+++ b/python/pyspark/sql/tests/connect/test_connect_column.py
@@ -145,6 +145,28 @@ class SparkConnectColumnTests(ReusedMixedTestCase, PandasOnSparkTestUtils):
},
)
+ def test_select_column_replaced_by_withcolumn(self):
+ from pyspark.errors.exceptions.connect import AnalysisException
+
+ # Selecting the original DataFrame's column after `withColumn` replaces it
+ # with a cast: the plan-id-tagged reference must resolve to the overwritten
+ # alias rather than the inner column. With non-strict DataFrame column
+ # resolution, the analyzer falls back to name-based resolution for the
+ # tagged attribute and the query succeeds.
+ with self.connect_conf({"spark.sql.analyzer.strictDataFrameColumnResolution": False}):
+ df = self.connect.sql("SELECT 123 AS c")
+ df.withColumn("c", CF.col("c").cast("string")).select(df["c"]).collect()
+
+ # Under strict DataFrame column resolution (the default), the tagged
+ # reference cannot be resolved: the resolved attribute from the original
+ # plan is filtered out at the shadowing `withColumn` Project, and
+ # name-based fallback is disabled, so analysis fails with
+ # CANNOT_RESOLVE_DATAFRAME_COLUMN.
+ with self.connect_conf({"spark.sql.analyzer.strictDataFrameColumnResolution": True}):
+ df = self.connect.sql("SELECT 123 AS c")
+ with self.assertRaisesRegex(AnalysisException, "CANNOT_RESOLVE_DATAFRAME_COLUMN"):
+ df.withColumn("c", CF.col("c").cast("string")).select(df["c"]).collect()
+
def test_column_with_null(self):
# SPARK-41751: test isNull, isNotNull, eqNullSafe
diff --git a/python/pyspark/sql/tests/connect/test_parity_column.py b/python/pyspark/sql/tests/connect/test_parity_column.py
index 3c531fd48ea5..3903bb57a375 100644
--- a/python/pyspark/sql/tests/connect/test_parity_column.py
+++ b/python/pyspark/sql/tests/connect/test_parity_column.py
@@ -22,10 +22,52 @@ from pyspark.testing.connectutils import ReusedConnectTestCase
class ColumnParityTests(ColumnTestsMixin, ReusedConnectTestCase):
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls.spark.conf.set("spark.sql.analyzer.strictDataFrameColumnResolution", "true")
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ cls.spark.conf.unset("spark.sql.analyzer.strictDataFrameColumnResolution")
+ finally:
+ super().tearDownClass()
+
@unittest.skip("Requires JVM access.")
def test_validate_column_types(self):
super().test_validate_column_types()
+ def test_df_col_resolution_mode(self):
+ self.assertEqual(
+ self.spark.conf.get("spark.sql.analyzer.strictDataFrameColumnResolution"),
+ "true",
+ )
+
+
+class ColumnParityTestsWithNonStrictDFColResolution(ColumnParityTests):
+ """Re-run the Column parity tests with
+ `spark.sql.analyzer.strictDataFrameColumnResolution=false` to exercise the
+ name-based fallback path for tagged UnresolvedAttributes."""
+
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls.spark.conf.set("spark.sql.analyzer.strictDataFrameColumnResolution", "false")
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ cls.spark.conf.unset("spark.sql.analyzer.strictDataFrameColumnResolution")
+ finally:
+ super().tearDownClass()
+
+ def test_df_col_resolution_mode(self):
+ self.assertEqual(
+ self.spark.conf.get("spark.sql.analyzer.strictDataFrameColumnResolution"),
+ "false",
+ )
+
if __name__ == "__main__":
from pyspark.testing import main
diff --git a/python/pyspark/sql/tests/test_column.py b/python/pyspark/sql/tests/test_column.py
index 70b25aff9eb3..06359854c6d5 100644
--- a/python/pyspark/sql/tests/test_column.py
+++ b/python/pyspark/sql/tests/test_column.py
@@ -522,6 +522,62 @@ class ColumnTestsMixin:
self.assertEqual(result[1][0], 40)
self.assertEqual(result[2][0], 60)
+ def test_self_join(self):
+ df1 = self.spark.range(10).withColumn("a", sf.lit(0))
+ df2 = df1.withColumnRenamed("a", "b")
+ df = df1.join(df2, df1["a"] == df2["b"])
+ self.assertTrue(df.count() == 100)
+ df = df2.join(df1, df2["b"] == df1["a"])
+ self.assertTrue(df.count() == 100)
+
+ def test_self_join_II(self):
+ df = self.spark.createDataFrame([(1, 2), (3, 4)], schema=["a", "b"])
+ df2 = df.select(df.a.alias("aa"), df.b)
+ df3 = df2.join(df, df2.b == df.b)
+ self.assertTrue(df3.columns, ["aa", "b", "a", "b"])
+ self.assertTrue(df3.count() == 2)
+
+ def test_self_join_III(self):
+ df1 = self.spark.range(10).withColumn("value", sf.lit(1))
+ df2 = df1.union(df1)
+ df3 = df1.join(df2, df1.id == df2.id, "left")
+ self.assertTrue(df3.columns, ["id", "value", "id", "value"])
+ self.assertTrue(df3.count() == 20)
+
+ def test_self_join_IV(self):
+ df1 = self.spark.range(10).withColumn("value", sf.lit(1))
+ df2 = df1.withColumn("value", sf.lit(2)).union(df1.withColumn("value", sf.lit(3)))
+ df3 = df1.join(df2, df1.id == df2.id, "right")
+ self.assertTrue(df3.columns, ["id", "value", "id", "value"])
+ self.assertTrue(df3.count() == 20)
+
+ def test_select_join_keys(self):
+ df1 = self.spark.range(10).withColumn("v1", sf.lit(1))
+ df2 = self.spark.range(10).withColumn("v2", sf.lit(2))
+ for how in ["inner", "left", "right", "full", "cross"]:
+ self.assertTrue(df1.join(df2, "id", how).select(df1["id"]).count() >= 0, how)
+ self.assertTrue(df1.join(df2, "id", how).select(df2["id"]).count() >= 0, how)
+
+ def test_drop_notexistent_col(self):
+ df1 = self.spark.createDataFrame(
+ [("a", "b", "c")],
+ schema="colA string, colB string, colC string",
+ )
+ df2 = self.spark.createDataFrame(
+ [("c", "d", "e")],
+ schema="colC string, colD string, colE string",
+ )
+ df3 = df1.join(df2, df1["colC"] == df2["colC"]).withColumn(
+ "colB",
+ sf.when(
+ df1["colB"] == "b", sf.concat(df1["colB"].cast("string"), sf.lit("x"))
+ ).otherwise(df1["colB"]),
+ )
+ df4 = df3.drop(df1["colB"])
+
+ self.assertEqual(df4.columns, ["colA", "colB", "colC", "colC", "colD", "colE"])
+ self.assertEqual(df4.count(), 1)
+
class ColumnTests(ColumnTestsMixin, ReusedSQLTestCase):
pass
diff --git a/python/pyspark/sql/tests/test_dataframe.py b/python/pyspark/sql/tests/test_dataframe.py
index c426e5a88d59..23a5bb1d99c1 100644
--- a/python/pyspark/sql/tests/test_dataframe.py
+++ b/python/pyspark/sql/tests/test_dataframe.py
@@ -35,8 +35,6 @@ from pyspark.sql.functions import (
to_date,
array,
explode,
- when,
- concat,
)
from pyspark.sql.types import (
StringType,
@@ -136,42 +134,6 @@ class DataFrameTestsMixin:
self.assertEqual(df3.select(count(df3["*"])).columns, ["count(1)"])
self.assertEqual(df3.select(count(col("*"))).columns, ["count(1)"])
- def test_self_join(self):
- df1 = self.spark.range(10).withColumn("a", lit(0))
- df2 = df1.withColumnRenamed("a", "b")
- df = df1.join(df2, df1["a"] == df2["b"])
- self.assertTrue(df.count() == 100)
- df = df2.join(df1, df2["b"] == df1["a"])
- self.assertTrue(df.count() == 100)
-
- def test_self_join_II(self):
- df = self.spark.createDataFrame([(1, 2), (3, 4)], schema=["a", "b"])
- df2 = df.select(df.a.alias("aa"), df.b)
- df3 = df2.join(df, df2.b == df.b)
- self.assertTrue(df3.columns, ["aa", "b", "a", "b"])
- self.assertTrue(df3.count() == 2)
-
- def test_self_join_III(self):
- df1 = self.spark.range(10).withColumn("value", lit(1))
- df2 = df1.union(df1)
- df3 = df1.join(df2, df1.id == df2.id, "left")
- self.assertTrue(df3.columns, ["id", "value", "id", "value"])
- self.assertTrue(df3.count() == 20)
-
- def test_self_join_IV(self):
- df1 = self.spark.range(10).withColumn("value", lit(1))
- df2 = df1.withColumn("value", lit(2)).union(df1.withColumn("value", lit(3)))
- df3 = df1.join(df2, df1.id == df2.id, "right")
- self.assertTrue(df3.columns, ["id", "value", "id", "value"])
- self.assertTrue(df3.count() == 20)
-
- def test_select_join_keys(self):
- df1 = self.spark.range(10).withColumn("v1", lit(1))
- df2 = self.spark.range(10).withColumn("v2", lit(2))
- for how in ["inner", "left", "right", "full", "cross"]:
- self.assertTrue(df1.join(df2, "id", how).select(df1["id"]).count() >= 0, how)
- self.assertTrue(df1.join(df2, "id", how).select(df2["id"]).count() >= 0, how)
-
def test_lateral_column_alias(self):
df1 = self.spark.range(10).select(
(col("id") + lit(1)).alias("x"), (col("x") + lit(1)).alias("y")
@@ -213,26 +175,6 @@ class DataFrameTestsMixin:
self.assertEqual(df.drop(col("name"), col("age")).columns, ["active"])
self.assertEqual(df.drop(col("name"), col("age"), col("random")).columns, ["active"])
- def test_drop_notexistent_col(self):
- df1 = self.spark.createDataFrame(
- [("a", "b", "c")],
- schema="colA string, colB string, colC string",
- )
- df2 = self.spark.createDataFrame(
- [("c", "d", "e")],
- schema="colC string, colD string, colE string",
- )
- df3 = df1.join(df2, df1["colC"] == df2["colC"]).withColumn(
- "colB",
- when(df1["colB"] == "b", concat(df1["colB"].cast("string"), lit("x"))).otherwise(
- df1["colB"]
- ),
- )
- df4 = df3.drop(df1["colB"])
-
- self.assertEqual(df4.columns, ["colA", "colB", "colC", "colC", "colD", "colE"])
- self.assertEqual(df4.count(), 1)
-
def test_drop_col_from_different_dataframe(self):
df1 = self.spark.range(10)
df2 = df1.withColumn("v0", lit(0))
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala
index 49792bf4d4bc..93d71642ac9f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala
@@ -140,8 +140,12 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
}
matched(ordinal)
- case u @ UnresolvedAttribute(nameParts) if !u.containsTag(LogicalPlan.PLAN_ID_TAG) =>
- // UnresolvedAttribute with PLAN_ID_TAG should be resolved in resolveDataFrameColumn
+ case u @ UnresolvedAttribute(nameParts)
+ if !u.containsTag(LogicalPlan.PLAN_ID_TAG) ||
+ !conf.getConf(SQLConf.STRICT_DATAFRAME_COLUMN_RESOLUTION) =>
+ // UnresolvedAttribute with PLAN_ID_TAG should be resolved in resolveDataFrameColumn.
+ // When strict DataFrame column resolution is disabled, we also allow name-based
+ // resolution as a fallback for tagged attributes.
val result = withPosition(u) {
resolveColumnByName(nameParts)
.orElse(LiteralFunctionResolution.resolve(nameParts))
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 8a5dd860f18b..99a5339950a5 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -303,6 +303,18 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STRICT_DATAFRAME_COLUMN_RESOLUTION =
+ buildConf("spark.sql.analyzer.strictDataFrameColumnResolution")
+ .internal()
+ .version("4.2.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .doc(
+ "When true (default), enforce strict resolution of Spark Connect DataFrame columns " +
+ "(UnresolvedAttribute carrying a plan id tag) via plan-id-based resolution. When " +
+ "false, also try name-based resolution as a fallback for tagged attributes.")
+ .booleanConf
+ .createWithDefault(true)
+
val BLOCK_CREATE_TEMP_TABLE_USING_PROVIDER =
buildConf("spark.sql.legacy.blockCreateTempTableUsingProvider")
.doc("If enabled, we fail legacy CREATE TEMPORARY TABLE ... USING provider during parsing.")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org