[SPARK-52751][PYTHON][CONNECT] Don't eagerly validate column name in dataframe['col_name']

python/docs/source/migration_guide/pyspark_upgrade.rst

@@ -22,8 +22,8 @@ Upgrading PySpark
 Upgrading from PySpark 4.0 to 4.1
 ---------------------------------
 
+* In Spark 4.1, ``DataFrame['name']`` on Spark Connect Python Client no longer eagerly validate the column name. To restore the legacy behavior, set ``PYSPARK_VALIDATE_COLUMN_NAME_LEGACY`` environment variable to ``1``.
 * In Spark 4.1, Arrow-optimized Python UDF supports UDT input / output instead of falling back to the regular UDF. To restore the legacy behavior, set ``spark.sql.execution.pythonUDF.arrow.legacy.fallbackOnUDT`` to ``true``.
-
 * In Spark 4.1, unnecessary conversion to pandas instances is removed when ``spark.sql.execution.pythonUDTF.arrow.enabled`` is enabled. As a result, the type coercion changes when the produced output has a schema different from the specified schema. To restore the previous behavior, enable ``spark.sql.legacy.execution.pythonUDTF.pandas.conversion.enabled``.
 
 

python/pyspark/sql/connect/dataframe.py

@@ -44,6 +44,7 @@
 )
 
 import copy
+import os
 import sys
 import random
 import pyarrow as pa
@@ -1740,7 +1741,9 @@ def __getitem__(
                 # }
 
                 # validate the column name
-                if not hasattr(self._session, "is_mock_session"):
+                if os.environ.get("PYSPARK_VALIDATE_COLUMN_NAME_LEGACY") == "1" and not hasattr(
+                    self._session, "is_mock_session"
+                ):
                     from pyspark.sql.connect.types import verify_col_name
 
                     # Try best to verify the column name with cached schema

python/pyspark/sql/tests/test_column.py

@@ -133,8 +133,8 @@ def test_access_column(self):
         self.assertTrue(isinstance(df["key"], Column))
         self.assertTrue(isinstance(df[0], Column))
         self.assertRaises(IndexError, lambda: df[2])
-        self.assertRaises(AnalysisException, lambda: df["bad_key"])
         self.assertRaises(TypeError, lambda: df[{}])
+        self.assertRaises(AnalysisException, lambda: df.select(df["bad_key"]).schema)
 
     def test_column_name_with_non_ascii(self):
         columnName = "数量"