[FLINK-40418][python] Support attribute-based column access on DataFrame

This closes #29164.

Author: beetle0915

flink-python/docs/reference/pyflink.dataframe/dataframe.rst

@@ -25,13 +25,19 @@ Transformation methods return new DataFrames and support fluent chaining. They b
 plans lazily without starting a Flink job; execution is triggered by an action such as
 ``DataFrame.collect`` or ``DataFrame.to_pandas``.
 
+Columns can also be referenced as attributes, such as ``df.name``, when their names are valid
+Python identifiers, do not start with an underscore, are not keywords, and do not conflict with
+existing DataFrame attributes. Use bracket access for other names, such as ``df["select"]``,
+``df["_name"]``, or ``df["first name"]``.
+
 Example::
 
     >>> import pyflink.dataframe as pf
     >>> df = pf.from_dict({"id": [1, 2], "name": ["a", "b"]})
     >>> result = df.select("id", "name") \
     ...            .with_column("id_doubled", pf.col("id") * 2) \
     ...            .filter(pf.col("id") > 0)
+    >>> names = df.select(df.name)
 
 DataFrame
 ---------
@@ -69,6 +75,7 @@ Transformations
     DataFrame.offset
     DataFrame.head
     DataFrame.__getitem__
+    DataFrame.__getattr__
 
 Set Operations
 --------------

flink-python/pyflink/dataframe/dataframe.py

@@ -17,6 +17,7 @@
 ################################################################################
 
 import datetime
+import keyword
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -1230,6 +1231,52 @@ def __getitem__(
             return self.filter(key)
         raise TypeError("key must be a string, list, tuple, or Expression")
 
+    @PublicEvolving()
+    def __getattr__(self, name: str) -> Expression:
+        """
+        Return a column expression for an attribute name.
+
+        The name must be a valid Python identifier, must not start with an underscore, must not be
+        a Python keyword, and must identify an existing column. Existing DataFrame attributes take
+        precedence over columns. Use ``df["column name"]`` for names that cannot be accessed as
+        attributes, or ``df["select"]`` for columns that conflict with existing attributes.
+
+        This method resolves the schema without executing a Flink job.
+
+        :param name: Name of the referenced column.
+        :return: An expression referencing the column.
+        :raises AttributeError: If the name is invalid, private, or does not identify an existing
+            column.
+
+        Example::
+
+            >>> import pyflink.dataframe as pf
+            >>> df = pf.from_records([{"id": 1, "name": "Alice"}])
+            >>> selected = df.select(df.name)
+            >>> filtered = df.filter(df.id > 0)
+
+        .. versionadded:: 2.4.0
+        """
+        if (
+            name.startswith("_")
+            or not name.isidentifier()
+            or keyword.iskeyword(name)
+            or any(name in cls.__dict__ for cls in type(self).__mro__)
+        ):
+            raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
+
+        # Avoid re-entering __getattr__ if the underlying table has not been initialized.
+        try:
+            table = object.__getattribute__(self, "_table")
+        except AttributeError:
+            raise AttributeError(
+                f"'{type(self).__name__}' object has no attribute '{name}'"
+            ) from None
+
+        if name not in table.get_resolved_schema().get_column_names():
+            raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
+        return table_col(name)
+
     # ======================== Composition ========================
 
     @PublicEvolving()

flink-python/pyflink/dataframe/tests/test_dataframe.py

@@ -1131,6 +1131,140 @@ def test_getitem_rejects_unsupported_key(self):
             self.dataframe[42]
 
 
+class DataFrameGetAttrTests(PyFlinkDataFrameUTTestCase):
+    def setUp(self):
+        super().setUp()
+        self.dataframe = pf.from_records(
+            [(1, "Alice"), (2, "Bob")], schema=["id", "name"]
+        )
+
+    def test_getattr_returns_column_expression(self):
+        self.assertIsInstance(self.dataframe.name, Expression)
+        self.assertEqual(str(self.dataframe.name), str(self.dataframe["name"]))
+        self.assert_dataframe_schema(
+            self.dataframe.select(self.dataframe.name), ["name"]
+        )
+
+    def test_getattr_composes_with_filter_and_select(self):
+        result = self.dataframe.filter(lambda df: df.id > 1).select(
+            self.dataframe.name, (self.dataframe.id + 1).alias("next_id")
+        )
+        self.assert_dataframe_schema(result, ["name", "next_id"])
+
+    def test_missing_column_raises_attribute_error(self):
+        with self.assertRaisesRegex(AttributeError, "missing"):
+            self.dataframe.missing
+        self.assertFalse(hasattr(self.dataframe, "missing"))
+        default = object()
+        self.assertIs(getattr(self.dataframe, "missing", default), default)
+        self.assertTrue(hasattr(self.dataframe, "name"))
+
+    def test_existing_attributes_take_precedence_over_columns(self):
+        names = ["select", "filter", "columns", "schema", "_table", "__class__"]
+        dataframe = pf.from_records([tuple(range(len(names)))], schema=names)
+        self.assertIs(dataframe.select.__func__, pf.DataFrame.select)
+        self.assertIs(dataframe.filter.__func__, pf.DataFrame.filter)
+        self.assertEqual(dataframe.columns, names)
+        self.assertIsInstance(dataframe.schema, TableSchema)
+        self.assertIs(dataframe._table, dataframe.to_table())
+        self.assertIs(dataframe.__class__, pf.DataFrame)
+        for name in names:
+            with self.subTest(name=name):
+                self.assert_dataframe_schema(dataframe.select(dataframe[name]), [name])
+
+    def test_instance_attributes_take_precedence_over_columns(self):
+        marker = object()
+        self.dataframe.name = marker
+        self.assertIs(self.dataframe.name, marker)
+        self.assertIsInstance(self.dataframe["name"], Expression)
+
+    def test_invalid_identifiers_and_keywords_are_not_attributes(self):
+        for name in ("first name", "first-name", "1name", "class", "None"):
+            with self.subTest(name=name):
+                dataframe = pf.from_records([(1,)], schema=[name])
+                with self.assertRaises(AttributeError):
+                    getattr(dataframe, name)
+                self.assertIsInstance(dataframe[name], Expression)
+
+    def test_valid_identifiers_include_unicode_and_digits(self):
+        for name in ("name_2", "\u540d\u5b57", "match"):
+            with self.subTest(name=name):
+                dataframe = pf.from_records([(1,)], schema=[name])
+                self.assert_dataframe_schema(dataframe.select(getattr(dataframe, name)), [name])
+
+    def test_private_names_require_bracket_access(self):
+        names = ["_name", "__dataframe__", "__deepcopy__", "_repr_html_"]
+        dataframe = pf.from_records([tuple(range(len(names)))], schema=names)
+
+        for name in names:
+            with self.subTest(name=name):
+                with self.assertRaises(AttributeError):
+                    getattr(dataframe, name)
+                self.assertFalse(hasattr(dataframe, name))
+                self.assert_dataframe_schema(dataframe.select(dataframe[name]), [name])
+
+    def test_getattr_uses_the_transformed_schema(self):
+        renamed = self.dataframe.rename_columns({"name": "label"})
+        self.assertIsInstance(renamed.label, Expression)
+        self.assertFalse(hasattr(renamed, "name"))
+        projected = self.dataframe.select("id")
+        self.assertFalse(hasattr(projected, "name"))
+        self.assertIsInstance(self.dataframe.name, Expression)
+
+
+class DataFrameGetAttrValidationTests(unittest.TestCase):
+    def test_invalid_names_do_not_resolve_schema(self):
+        table = Mock()
+        for name in (
+            "",
+            "first name",
+            "class",
+            "_name",
+            "__dataframe__",
+            "__deepcopy__",
+            "_repr_html_",
+        ):
+            with self.subTest(name=name):
+                with self.assertRaises(AttributeError):
+                    getattr(pf.DataFrame(table), name)
+        table.get_resolved_schema.assert_not_called()
+
+    def test_failing_class_descriptor_does_not_fall_back_to_column(self):
+        class DerivedDataFrame(pf.DataFrame):
+            @property
+            def name(self):
+                raise AttributeError("descriptor unavailable")
+
+        table = Mock()
+        table.get_resolved_schema.return_value.get_column_names.return_value = ["name"]
+
+        with self.assertRaises(AttributeError):
+            DerivedDataFrame(table).name
+        table.get_resolved_schema.assert_not_called()
+
+    def test_uninitialized_dataframe_does_not_recurse(self):
+        dataframe = object.__new__(pf.DataFrame)
+        for name in ("_table", "name", "__setstate__"):
+            with self.subTest(name=name):
+                with self.assertRaises(AttributeError):
+                    getattr(dataframe, name)
+
+    def test_column_access_does_not_execute_a_job(self):
+        table = Mock()
+        table.get_resolved_schema.return_value.get_column_names.return_value = ["name"]
+        expression = object()
+        with patch("pyflink.dataframe.dataframe.table_col", return_value=expression) as col:
+            self.assertIs(pf.DataFrame(table).name, expression)
+            col.assert_called_once_with("name")
+        table.execute.assert_not_called()
+
+    def test_schema_errors_are_not_hidden(self):
+        table = Mock()
+        table.get_resolved_schema.side_effect = RuntimeError("schema unavailable")
+        with self.assertRaisesRegex(RuntimeError, "schema unavailable"):
+            pf.DataFrame(table).name
+
+
 class DataFrameLiteralTests(PyFlinkDataFrameUTTestCase):
     def setUp(self):
         super().setUp()
@@ -2304,6 +2438,15 @@ def setUp(self):
         self.addCleanup(pf.set_table_environment, previous_environment)
         self.t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
 
+    def test_attribute_column_access_executes(self):
+        dataframe = pf.from_table(self.t_env.sql_query(
+            "SELECT * FROM (VALUES (1, 'Alice'), (2, 'Bob')) AS T(id, name)"
+        ))
+        result = dataframe.filter(lambda df: df.id > 1).select(
+            dataframe.name, (dataframe.id + 10).alias("next_id")
+        )
+        self.assertEqual(result.collect(), [Row("Bob", 12)])
+
     def _ordered_dataframe(self):
         table = self.t_env.sql_query(
             "SELECT * FROM (VALUES (3, 'C'), (1, 'A'), (4, 'D'), (2, 'B')) "