This is an automated email from the ASF dual-hosted git repository.
tvalentyn pushed a commit to branch revert-15969-combine-type-hints
in repository https://gitbox.apache.org/repos/asf/beam.git
commit d71203a01307008114238a8b4506e67ae24a5ad7
Author: tvalentyn <tvalen...@users.noreply.github.com>
AuthorDate: Mon Nov 29 15:19:04 2021 -0800
Revert "Key-inferable type hints for CombinePerKey."
---
sdks/python/apache_beam/pipeline.py | 4 +---
sdks/python/apache_beam/transforms/core.py | 24 ++++++++++------------
.../apache_beam/transforms/ptransform_test.py | 13 ------------
3 files changed, 12 insertions(+), 29 deletions(-)
diff --git a/sdks/python/apache_beam/pipeline.py
b/sdks/python/apache_beam/pipeline.py
index a512a18..25c02d4 100644
--- a/sdks/python/apache_beam/pipeline.py
+++ b/sdks/python/apache_beam/pipeline.py
@@ -892,9 +892,7 @@ class Pipeline(object):
pcoll.element_type)
if (isinstance(output.element_type,
typehints.TupleHint.TupleConstraint) and
- len(output.element_type.tuple_types) == 2 and
- pcoll.element_type.tuple_types[0] ==
- output.element_type.tuple_types[0]):
+ len(output.element_type.tuple_types) == 2):
output.requires_deterministic_key_coder = (
deterministic_key_coders and transform_node.full_label)
for side_input in transform_node.transform.side_inputs:
diff --git a/sdks/python/apache_beam/transforms/core.py
b/sdks/python/apache_beam/transforms/core.py
index bd999f8..c71c00c 100644
--- a/sdks/python/apache_beam/transforms/core.py
+++ b/sdks/python/apache_beam/transforms/core.py
@@ -2306,20 +2306,18 @@ class CombinePerKey(PTransformWithSideInputs):
self.fn, *args, **kwargs)
def default_type_hints(self):
- result = self.fn.get_type_hints()
- k = typehints.TypeVariable('K')
- if result.input_types:
- args, kwargs = result.input_types
- args = (typehints.Tuple[k, args[0]], ) + args[1:]
- result = result.with_input_types(*args, **kwargs)
- else:
- result = result.with_input_types(typehints.Tuple[k, typehints.Any])
- if result.output_types:
- main_output_type = result.simple_output_type('')
- result = result.with_output_types(typehints.Tuple[k, main_output_type])
+ hints = self.fn.get_type_hints()
+ if hints.input_types:
+ K = typehints.TypeVariable('K')
+ args, kwargs = hints.input_types
+ args = (typehints.Tuple[K, args[0]], ) + args[1:]
+ hints = hints.with_input_types(*args, **kwargs)
else:
- result = result.with_output_types(typehints.Tuple[k, typehints.Any])
- return result
+ K = typehints.Any
+ if hints.output_types:
+ main_output_type = hints.simple_output_type('')
+ hints = hints.with_output_types(typehints.Tuple[K, main_output_type])
+ return hints
def to_runner_api_parameter(
self,
diff --git a/sdks/python/apache_beam/transforms/ptransform_test.py
b/sdks/python/apache_beam/transforms/ptransform_test.py
index 6a139eb..ac4f632 100644
--- a/sdks/python/apache_beam/transforms/ptransform_test.py
+++ b/sdks/python/apache_beam/transforms/ptransform_test.py
@@ -1839,19 +1839,6 @@ class PTransformTypeCheckTestCase(TypeHintTestCase):
assert_that(d, equal_to([6]))
self.p.run()
- def test_combine_properly_pipeline_type_checks_without_decorator(self):
- def sum_ints(ints):
- return sum(ints)
-
- d = (
- self.p
- | beam.Create([1, 2, 3])
- | beam.Map(lambda x: ('key', x))
- | beam.CombinePerKey(sum_ints))
-
- self.assertEqual(typehints.Tuple[str, typehints.Any], d.element_type)
- self.p.run()
-
def test_combine_func_type_hint_does_not_take_iterable_using_decorator(self):
@with_output_types(int)
@with_input_types(a=int)
