From 6ff3ffd1ced7d0f9acfdaaa3773da4482bba8587 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Fri, 31 Oct 2025 15:02:53 -0400 Subject: [PATCH 1/4] Fix spammy warning message (and its formatting) --- sdks/python/apache_beam/transforms/core.py | 23 +++++++++++-------- .../apache_beam/transforms/core_test.py | 14 +---------- 2 files changed, 14 insertions(+), 23 deletions(-) diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py index 7ba8aa128c24..1c57f89d6f5d 100644 --- a/sdks/python/apache_beam/transforms/core.py +++ b/sdks/python/apache_beam/transforms/core.py @@ -1507,25 +1507,28 @@ def _check_fn_use_yield_and_return(fn): source_code = _get_function_body_without_inners(fn) has_yield = False has_return = False - return_none_warning = ( - "No iterator is returned by the process method in %s.", - fn.__self__.__class__) + has_return_none = False for line in source_code.split("\n"): lstripped_line = line.lstrip() if lstripped_line.startswith("yield ") or lstripped_line.startswith( "yield("): has_yield = True - if lstripped_line.startswith("return ") or lstripped_line.startswith( + elif lstripped_line.rstrip() == "return": + has_return_none = True + elif lstripped_line.startswith("return ") or lstripped_line.startswith( "return("): - has_return = True - if lstripped_line.startswith( - "return None") or lstripped_line.rstrip() == "return": - _LOGGER.warning(return_none_warning) + if lstripped_line.startswith("return None"): + has_return_none = True + else: + has_return = True if has_yield and has_return: return True - if not has_yield and not has_return: - _LOGGER.warning(return_none_warning) + if not has_yield and not has_return and has_return_none: + _LOGGER.warning( + "Process method returned None (element won't be emitted): %s. Check if intended.", + fn.__self__.__class__) + print(has_yield, has_return, has_return_none) return False except Exception as e: diff --git a/sdks/python/apache_beam/transforms/core_test.py b/sdks/python/apache_beam/transforms/core_test.py index 0d680c969c9b..203f28ddb6f5 100644 --- a/sdks/python/apache_beam/transforms/core_test.py +++ b/sdks/python/apache_beam/transforms/core_test.py @@ -40,7 +40,7 @@ from apache_beam.typehints import row_type from apache_beam.typehints import typehints -RETURN_NONE_PARTIAL_WARNING = "No iterator is returned" +RETURN_NONE_PARTIAL_WARNING = "Process method returned None" class TestDoFn1(beam.DoFn): @@ -114,12 +114,6 @@ def process(self, element): return None -class TestDoFn11(beam.DoFn): - """test process returning None (no return and no yield)""" - def process(self, element): - pass - - class TestDoFn12(beam.DoFn): """test process returning None (return statement without a value)""" def process(self, element): @@ -191,12 +185,6 @@ def test_dofn_with_explicit_return_none(self): assert RETURN_NONE_PARTIAL_WARNING in self._caplog.text assert str(TestDoFn10) in self._caplog.text - def test_dofn_with_implicit_return_none_missing_return_and_yield(self): - with self._caplog.at_level(logging.WARNING): - beam.ParDo(TestDoFn11()) - assert RETURN_NONE_PARTIAL_WARNING in self._caplog.text - assert str(TestDoFn11) in self._caplog.text - def test_dofn_with_implicit_return_none_return_without_value(self): with self._caplog.at_level(logging.WARNING): beam.ParDo(TestDoFn12()) From bf3d24db5d002b3628b4e6918ef2ac667fbab3fe Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Fri, 31 Oct 2025 16:05:06 -0400 Subject: [PATCH 2/4] address comments - make raw return a valid return --- sdks/python/apache_beam/transforms/core.py | 7 ++++--- sdks/python/apache_beam/transforms/core_test.py | 15 ++++++++------- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py index 1c57f89d6f5d..8e9f1805c75e 100644 --- a/sdks/python/apache_beam/transforms/core.py +++ b/sdks/python/apache_beam/transforms/core.py @@ -1514,19 +1514,20 @@ def _check_fn_use_yield_and_return(fn): "yield("): has_yield = True elif lstripped_line.rstrip() == "return": - has_return_none = True + has_return = True elif lstripped_line.startswith("return ") or lstripped_line.startswith( "return("): if lstripped_line.startswith("return None"): has_return_none = True else: has_return = True - if has_yield and has_return: + if has_yield and (has_return or has_return_none): return True if not has_yield and not has_return and has_return_none: _LOGGER.warning( - "Process method returned None (element won't be emitted): %s. Check if intended.", + "Process method returned None (element won't be emitted): %s." + " Check if intended.", fn.__self__.__class__) print(has_yield, has_return, has_return_none) diff --git a/sdks/python/apache_beam/transforms/core_test.py b/sdks/python/apache_beam/transforms/core_test.py index 203f28ddb6f5..0894cc559ef7 100644 --- a/sdks/python/apache_beam/transforms/core_test.py +++ b/sdks/python/apache_beam/transforms/core_test.py @@ -114,10 +114,12 @@ def process(self, element): return None -class TestDoFn12(beam.DoFn): - """test process returning None (return statement without a value)""" +class TestDoFn11(beam.DoFn): + """test process returning None in a filter pattern""" def process(self, element): - return + if element == 0: + return + return element class TestDoFnStateful(beam.DoFn): @@ -185,11 +187,10 @@ def test_dofn_with_explicit_return_none(self): assert RETURN_NONE_PARTIAL_WARNING in self._caplog.text assert str(TestDoFn10) in self._caplog.text - def test_dofn_with_implicit_return_none_return_without_value(self): + def test_dofn_with_implicit_return_none_and_value(self): with self._caplog.at_level(logging.WARNING): - beam.ParDo(TestDoFn12()) - assert RETURN_NONE_PARTIAL_WARNING in self._caplog.text - assert str(TestDoFn12) in self._caplog.text + beam.ParDo(TestDoFn11()) + assert RETURN_NONE_PARTIAL_WARNING not in self._caplog.text class PartitionTest(unittest.TestCase): From 703b9e672af2fc69a8954a24f2376ef954812799 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Mon, 3 Nov 2025 11:14:29 -0500 Subject: [PATCH 3/4] address comments --- sdks/python/apache_beam/transforms/core.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py index 8e9f1805c75e..5dbf4d8272a2 100644 --- a/sdks/python/apache_beam/transforms/core.py +++ b/sdks/python/apache_beam/transforms/core.py @@ -1517,19 +1517,19 @@ def _check_fn_use_yield_and_return(fn): has_return = True elif lstripped_line.startswith("return ") or lstripped_line.startswith( "return("): - if lstripped_line.startswith("return None"): + if lstripped_line.rstrip() == "return None" or lstripped_line.rstrip( + ) == "return(None)": has_return_none = True else: has_return = True if has_yield and (has_return or has_return_none): return True - if not has_yield and not has_return and has_return_none: + if has_return_none: _LOGGER.warning( "Process method returned None (element won't be emitted): %s." " Check if intended.", fn.__self__.__class__) - print(has_yield, has_return, has_return_none) return False except Exception as e: From 0feb9ebe64166d08d5501cf06e7555fadfca3080 Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Mon, 3 Nov 2025 11:20:57 -0500 Subject: [PATCH 4/4] add back tests --- sdks/python/apache_beam/transforms/core.py | 5 ++--- sdks/python/apache_beam/transforms/core_test.py | 13 ++++++++++++- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py index 5dbf4d8272a2..12b546da53d9 100644 --- a/sdks/python/apache_beam/transforms/core.py +++ b/sdks/python/apache_beam/transforms/core.py @@ -1520,9 +1520,8 @@ def _check_fn_use_yield_and_return(fn): if lstripped_line.rstrip() == "return None" or lstripped_line.rstrip( ) == "return(None)": has_return_none = True - else: - has_return = True - if has_yield and (has_return or has_return_none): + has_return = True + if has_yield and has_return: return True if has_return_none: diff --git a/sdks/python/apache_beam/transforms/core_test.py b/sdks/python/apache_beam/transforms/core_test.py index 0894cc559ef7..80ab6a88afb4 100644 --- a/sdks/python/apache_beam/transforms/core_test.py +++ b/sdks/python/apache_beam/transforms/core_test.py @@ -115,6 +115,12 @@ def process(self, element): class TestDoFn11(beam.DoFn): + """test process returning None (no return and no yield)""" + def process(self, element): + pass + + +class TestDoFn12(beam.DoFn): """test process returning None in a filter pattern""" def process(self, element): if element == 0: @@ -187,11 +193,16 @@ def test_dofn_with_explicit_return_none(self): assert RETURN_NONE_PARTIAL_WARNING in self._caplog.text assert str(TestDoFn10) in self._caplog.text - def test_dofn_with_implicit_return_none_and_value(self): + def test_dofn_with_implicit_return_none_missing_return_and_yield(self): with self._caplog.at_level(logging.WARNING): beam.ParDo(TestDoFn11()) assert RETURN_NONE_PARTIAL_WARNING not in self._caplog.text + def test_dofn_with_implicit_return_none_and_value(self): + with self._caplog.at_level(logging.WARNING): + beam.ParDo(TestDoFn12()) + assert RETURN_NONE_PARTIAL_WARNING not in self._caplog.text + class PartitionTest(unittest.TestCase): def test_partition_with_bools(self):