Python 3.10 support

.github/actions/setup-default-test-properties/test-properties.json

@@ -1,12 +1,12 @@
 {
   "PythonTestProperties": {
-    "ALL_SUPPORTED_VERSIONS": ["3.7", "3.8", "3.9"],
+    "ALL_SUPPORTED_VERSIONS": ["3.7", "3.8", "3.9", "3.10"],
     "LOWEST_SUPPORTED": ["3.7"],
-    "HIGHEST_SUPPORTED": ["3.9"],
-    "ESSENTIAL_VERSIONS": ["3.7", "3.9"],
-    "CROSS_LANGUAGE_VALIDATES_RUNNER_PYTHON_VERSIONS": ["3.7", "3.9"],
-    "CROSS_LANGUAGE_VALIDATES_RUNNER_DATAFLOW_USING_SQL_PYTHON_VERSIONS": ["3.9"],
-    "VALIDATES_CONTAINER_DATAFLOW_PYTHON_VERSIONS": ["3.7", "3.8", "3.9"],
+    "HIGHEST_SUPPORTED": ["3.10"],
+    "ESSENTIAL_VERSIONS": ["3.7", "3.10"],
+    "CROSS_LANGUAGE_VALIDATES_RUNNER_PYTHON_VERSIONS": ["3.7", "3.10"],
+    "CROSS_LANGUAGE_VALIDATES_RUNNER_DATAFLOW_USING_SQL_PYTHON_VERSIONS": ["3.10"],
+    "VALIDATES_CONTAINER_DATAFLOW_PYTHON_VERSIONS": ["3.7", "3.8", "3.9", "3.10"],
     "LOAD_TEST_PYTHON_VERSION": "3.7",
     "CHICAGO_TAXI_EXAMPLE_FLINK_PYTHON_VERSION": "3.7",
     "DEFAULT_INTERPRETER": "python3.8",
@@ -20,4 +20,4 @@
   "GoTestProperties": {
     "SUPPORTED_VERSIONS": ["1.18"]
   }
-}
+}

.github/workflows/build_wheels.yml

@@ -212,13 +212,13 @@ jobs:
     strategy:
       matrix:
         os_python: [
-          {"os": "ubuntu-latest", "python": "cp37-* cp38-* cp39-*"},
-          {"os": "macos-latest", "python": "cp37-* cp38-* cp39-*"},
-          {"os": "windows-latest", "python": "cp37-* cp38-* cp39-*"},
+          {"os": "ubuntu-latest", "python": "cp37-* cp38-* cp39-* cp310-*"},
+          {"os": "macos-latest", "python": "cp37-* cp38-* cp39-* cp310-*"},
+          {"os": "windows-latest", "python": "cp37-* cp38-* cp39-* cp310-*"},
         ]
         arch: [auto]
         include:
-          - os_python: {"os": "ubuntu-latest", "python": "cp37-* cp38-* cp39-*"}
+          - os_python: {"os": "ubuntu-latest", "python": "cp37-* cp38-* cp39-* cp310-*"}
             arch: aarch64
     steps:
     - name: Download python source distribution from artifacts

.github/workflows/python_tests.yml

@@ -104,6 +104,7 @@ jobs:
           {"py_ver": "3.7", "tox_env": "py37"},
           {"py_ver": "3.8", "tox_env": "py38"},
           {"py_ver": "3.9", "tox_env": "py39"},
+          {"py_ver": "3.10", "tox_env": "py310" },
         ]
     steps:
       - name: Checkout code
@@ -139,7 +140,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-latest, windows-latest]
-        python: [3.7, 3.8, 3.9]
+        python: ["3.7", "3.8", "3.9", "3.10"]
     steps:
       - name: Checkout code
         uses: actions/checkout@v3
@@ -167,7 +168,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-latest, windows-latest]
-        python: [3.7, 3.8, 3.9]
+        python: ["3.7", "3.8", "3.9", "3.10"]
     steps:
       - name: Checkout code
         uses: actions/checkout@v3

.test-infra/jenkins/PythonTestProperties.groovy

@@ -22,7 +22,8 @@ class PythonTestProperties {
   final static List<String> ALL_SUPPORTED_VERSIONS = [
     '3.7',
     '3.8',
-    '3.9'
+    '3.9',
+    '3.10'
   ]
   final static List<String> SUPPORTED_CONTAINER_TASKS = ALL_SUPPORTED_VERSIONS.collect {
     "py${it.replace('.', '')}"

.test-infra/jenkins/job_PostCommit_Python_ValidatesRunner_Dataflow.groovy

@@ -38,6 +38,7 @@ PostcommitJobBuilder.postCommitJob('beam_PostCommit_Py_VR_Dataflow', 'Run Python
           rootBuildScriptDir(commonJobProperties.checkoutDir)
           tasks(':sdks:python:test-suites:dataflow:validatesRunnerBatchTests')
           tasks(':sdks:python:test-suites:dataflow:validatesRunnerStreamingTests')
+          switches('-PdisableRunnerV2')
           commonJobProperties.setGradleSwitches(delegate)
         }
       }

.test-infra/jenkins/job_PreCommit_Portable_Python.groovy

@@ -48,7 +48,7 @@ builder.build {
     }
     gradle {
       rootBuildScriptDir(commonJobProperties.checkoutDir)
-      tasks(':sdks:python:test-suites:portable:py39:preCommitPy39')
+      tasks(':sdks:python:test-suites:portable:py310:preCommitPy310')
       commonJobProperties.setGradleSwitches(delegate)
     }
   }

CHANGES.md

@@ -53,8 +53,8 @@
 
 ## Highlights
 
-* New highly anticipated feature X added to Python SDK ([#X](https://github.com/apache/beam/issues/X)).
-* New highly anticipated feature Y added to Java SDK ([#Y](https://github.com/apache/beam/issues/Y)).
+* Python 3.10 support in Apache Beam ([#21458](https://github.com/apache/beam/issues/21458)).
+
 
 ## I/Os
 

build.gradle.kts

@@ -306,6 +306,7 @@ tasks.register("pythonPreCommit") {
   dependsOn(":sdks:python:test-suites:tox:py37:preCommitPy37")
   dependsOn(":sdks:python:test-suites:tox:py38:preCommitPy38")
   dependsOn(":sdks:python:test-suites:tox:py39:preCommitPy39")
+  dependsOn(":sdks:python:test-suites:tox:py310:preCommitPy310")
   dependsOn(":sdks:python:test-suites:dataflow:preCommitIT")
   dependsOn(":sdks:python:test-suites:dataflow:preCommitIT_V2")
 }
@@ -318,6 +319,7 @@ tasks.register("pythonDockerBuildPreCommit") {
   dependsOn(":sdks:python:container:py37:docker")
   dependsOn(":sdks:python:container:py38:docker")
   dependsOn(":sdks:python:container:py39:docker")
+  dependsOn(":sdks:python:container:py310:docker")
 }
 
 tasks.register("pythonLintPreCommit") {
@@ -356,7 +358,14 @@ tasks.register("python39PostCommit") {
   dependsOn(":sdks:python:test-suites:direct:py39:postCommitIT")
   dependsOn(":sdks:python:test-suites:direct:py39:hdfsIntegrationTest")
   dependsOn(":sdks:python:test-suites:portable:py39:postCommitPy39")
-  dependsOn(":sdks:python:test-suites:direct:py39:inferencePostCommitIT")
+}
+
+tasks.register("python310PostCommit") {
+  dependsOn(":sdks:python:test-suites:dataflow:py310:postCommitIT")
+  dependsOn(":sdks:python:test-suites:direct:py310:postCommitIT")
+  dependsOn(":sdks:python:test-suites:direct:py310:hdfsIntegrationTest")
+  dependsOn(":sdks:python:test-suites:portable:py310:postCommitPy310")
+  dependsOn(":sdks:python:test-suites:direct:py310:inferencePostCommitIT")
 }
 
 task("python37SickbayPostCommit") {
@@ -373,13 +382,14 @@ task("python39SickbayPostCommit") {
 
 tasks.register("portablePythonPreCommit") {
   dependsOn(":sdks:python:test-suites:portable:py37:preCommitPy37")
-  dependsOn(":sdks:python:test-suites:portable:py39:preCommitPy39")
+  dependsOn(":sdks:python:test-suites:portable:py310:preCommitPy310")
 }
 
 tasks.register("pythonSparkPostCommit") {
   dependsOn(":sdks:python:test-suites:portable:py37:sparkValidatesRunner")
   dependsOn(":sdks:python:test-suites:portable:py38:sparkValidatesRunner")
   dependsOn(":sdks:python:test-suites:portable:py39:sparkValidatesRunner")
+  dependsOn(":sdks:python:test-suites:portable:py310:sparkValidatesRunner")
 }
 
 tasks.register("websitePreCommit") {

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -2672,6 +2672,7 @@ class BeamModulePlugin implements Plugin<Project> {
             ':sdks:python:container:py37:docker',
             ':sdks:python:container:py38:docker',
             ':sdks:python:container:py39:docker',
+            ':sdks:python:container:py310:docker',
           ]
           doLast {
             // TODO: Figure out GCS credentials and use real GCS input and output.

sdks/python/apache_beam/__init__.py

@@ -75,7 +75,7 @@
 import warnings
 
 if sys.version_info.major == 3:
-  if sys.version_info.minor <= 6 or sys.version_info.minor >= 10:
+  if sys.version_info.minor <= 6 or sys.version_info.minor >= 11:
     warnings.warn(
         'This version of Apache Beam has not been sufficiently tested on '
         'Python %s.%s. You may encounter bugs or missing features.' %

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -81,7 +81,7 @@
 
 _LOGGER = logging.getLogger(__name__)
 
-_PYTHON_VERSIONS_SUPPORTED_BY_DATAFLOW = ['3.6', '3.7', '3.8', '3.9']
+_PYTHON_VERSIONS_SUPPORTED_BY_DATAFLOW = ['3.6', '3.7', '3.8', '3.9', '3.10']
 
 
 class Step(object):

sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py

@@ -1032,7 +1032,7 @@ def test_interpreter_version_check_passes_py38(self):
 
   @mock.patch(
       'apache_beam.runners.dataflow.internal.apiclient.sys.version_info',
-      (3, 10, 0))
+      (3, 11, 0))
   @mock.patch(
       'apache_beam.runners.dataflow.internal.apiclient.'
       'beam_version.__version__',

sdks/python/apache_beam/runners/interactive/sql/utils.py

@@ -115,8 +115,7 @@ def pformat_namedtuple(schema: NamedTuple) -> str:
   return '{}({})'.format(
       schema.__name__,
       ', '.join([
-          '{}: {}'.format(k, v.__name__ if hasattr(v, '__name__') else repr(v))
-          for k,
+          '{}: {}'.format(k, repr(v)) for k,
           v in schema.__annotations__.items()
       ]))
 

sdks/python/apache_beam/runners/interactive/sql/utils_test.py

@@ -77,7 +77,7 @@ def test_replace_single_pcoll_token(self):
 
   def test_pformat_namedtuple(self):
     actual = pformat_namedtuple(ANamedTuple)
-    self.assertEqual('ANamedTuple(a: int, b: str)', actual)
+    self.assertEqual("ANamedTuple(a: <class 'int'>, b: <class 'str'>)", actual)
 
   def test_pformat_namedtuple_with_unnamed_fields(self):
     actual = pformat_namedtuple(OptionalUnionType)

sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner_test.py

@@ -191,7 +191,8 @@ def test_batch_rebatch_pardos(self):
     # - The output batch type of the producer
     # - The input batch type of the consumer
     with self.assertWarnsRegex(InefficientExecutionWarning,
-                               r'ListPlusOneDoFn.*NumpyArray.*List\[int64\]'):
+                               (r'ListPlusOneDoFn.*NumpyArray.*List\[<class '
+                                r'\'numpy.int64\'>\]')):
       with self.create_pipeline() as p:
         res = (
             p

sdks/python/apache_beam/transforms/batch_dofn_test.py

@@ -240,7 +240,8 @@ def test_mismatched_batch_producer_raises(self):
     with self.assertRaisesRegex(
         TypeError,
         (r'(?ms)MismatchedBatchProducingDoFn.*'
-         r'process: List\[int\].*process_batch: List\[float\]')):
+         r'process: List\[<class \'int\'>\].*process_batch: '
+         r'List\[<class \'float\'>\]')):
       _ = pc | beam.ParDo(MismatchedBatchProducingDoFn())
 
   def test_mismatched_element_producer_raises(self):

sdks/python/apache_beam/transforms/ptransform_test.py

@@ -1584,7 +1584,7 @@ def test_group_by_does_not_type_check(self):
         e.exception.args[0],
         "Input type hint violation at T: "
         "expected Tuple[TypeVariable[K], TypeVariable[V]], "
-        "got Iterable[int]")
+        "got Iterable[<class 'int'>]")
 
   def test_pipeline_checking_pardo_insufficient_type_information(self):
     self.p._options.view_as(TypeOptions).type_check_strictness = 'ALL_REQUIRED'
@@ -1705,9 +1705,9 @@ def is_even_as_key(a):
     self.assertStartswith(
         e.exception.args[0],
         "Runtime type violation detected within ParDo(IsEven): "
-        "Tuple[bool, int] hint type-constraint violated. "
+        "Tuple[<class 'bool'>, <class 'int'>] hint type-constraint violated. "
         "The type of element #0 in the passed tuple is incorrect. "
-        "Expected an instance of type bool, "
+        "Expected an instance of type <class 'bool'>, "
         "instead received an instance of type int.")
 
   def test_pipeline_checking_satisfied_run_time_checking_satisfied(self):
@@ -1771,10 +1771,10 @@ def test_pipeline_runtime_checking_violation_composite_type_input(self):
         e.exception.args[0],
         "Runtime type violation detected within ParDo(Add): "
         "Type-hint for argument: 'x_y' violated: "
-        "Tuple[int, int] hint type-constraint violated. "
+        "Tuple[<class 'int'>, <class 'int'>] hint type-constraint violated. "
         "The type of element #1 in the passed tuple is incorrect. "
-        "Expected an instance of type int, instead received an instance "
-        "of type float.")
+        "Expected an instance of type <class 'int'>, instead received an "
+        "instance of type float.")
 
   def test_pipeline_runtime_checking_violation_simple_type_output(self):
     self.p._options.view_as(TypeOptions).runtime_type_check = True
@@ -2097,8 +2097,9 @@ def test_mean_globally_pipeline_checking_violated(self):
 
     expected_msg = \
       "Type hint violation for 'CombinePerKey': " \
-      "requires Tuple[TypeVariable[K], Union[float, float64, int, int64]] " \
-      "but got Tuple[None, str] for element"
+      "requires Tuple[TypeVariable[K], Union[<class 'float'>, <class 'int'>, " \
+      "<class 'numpy.float64'>, <class 'numpy.int64'>]] " \
+      "but got Tuple[None, <class 'str'>] for element"
 
     self.assertStartswith(e.exception.args[0], expected_msg)
 
@@ -2163,8 +2164,9 @@ def test_mean_per_key_pipeline_checking_violated(self):
 
     expected_msg = \
       "Type hint violation for 'CombinePerKey(MeanCombineFn)': " \
-      "requires Tuple[TypeVariable[K], Union[float, float64, int, int64]] " \
-      "but got Tuple[str, str] for element"
+      "requires Tuple[TypeVariable[K], Union[<class 'float'>, <class 'int'>, " \
+      "<class 'numpy.float64'>, <class 'numpy.int64'>]] " \
+      "but got Tuple[<class 'str'>, <class 'str'>] for element"
 
     self.assertStartswith(e.exception.args[0], expected_msg)
 
@@ -2203,8 +2205,10 @@ def test_mean_per_key_runtime_checking_violated(self):
       "Runtime type violation detected within " \
       "OddMean/CombinePerKey(MeanCombineFn): " \
       "Type-hint for argument: 'element' violated: " \
-      "Union[float, float64, int, int64] type-constraint violated. " \
-      "Expected an instance of one of: ('float', 'float64', 'int', 'int64'), " \
+      "Union[<class 'float'>, <class 'int'>, <class 'numpy.float64'>, <class " \
+      "'numpy.int64'>] type-constraint violated. " \
+      "Expected an instance of one of: (\"<class 'float'>\", \"<class " \
+      "'int'>\", \"<class 'numpy.float64'>\", \"<class 'numpy.int64'>\"), " \
       "received str instead"
 
     self.assertStartswith(e.exception.args[0], expected_msg)
@@ -2570,9 +2574,9 @@ def test_inferred_bad_kv_type(self):
 
     self.assertStartswith(
         e.exception.args[0],
-        'Input type hint violation at GroupByKey: '
-        'expected Tuple[TypeVariable[K], TypeVariable[V]], '
-        'got Tuple[str, int, float]')
+        "Input type hint violation at GroupByKey: "
+        "expected Tuple[TypeVariable[K], TypeVariable[V]], "
+        "got Tuple[<class 'str'>, <class 'int'>, <class 'float'>]")
 
   def test_type_inference_command_line_flag_toggle(self):
     self.p._options.view_as(TypeOptions).pipeline_type_check = False

sdks/python/apache_beam/typehints/native_type_compatibility.py

@@ -192,6 +192,12 @@ def convert_to_beam_type(typ):
     # TODO(https://github.com/apache/beam/issues/19954): Currently unhandled.
     _LOGGER.info('Converting string literal type hint to Any: "%s"', typ)
     return typehints.Any
+  elif sys.version_info >= (3, 10) and isinstance(typ, typing.NewType):  # pylint: disable=isinstance-second-argument-not-valid-type
+    # Special case for NewType, where, since Python 3.10, NewType is now a class
+    # rather than a function.
+    # TODO(https://github.com/apache/beam/issues/20076): Currently unhandled.
+    _LOGGER.info('Converting NewType type hint to Any: "%s"', typ)
+    return typehints.Any
   elif getattr(typ, '__module__', None) != 'typing':
     # Only translate types from the typing module.
     return typ

sdks/python/apache_beam/typehints/opcodes.py

@@ -421,6 +421,12 @@ def delete_fast(state, arg):
   state.vars[arg] = Any  # really an error
 
 
+# bpo-43683 Adds GEN_START in Python 3.10, but removed in Python 3.11
+# https://github.com/python/cpython/pull/25138
+def gen_start(state, arg):
+  assert len(state.stack) == 0
+
+
 def load_closure(state, arg):
   state.stack.append(state.get_closure(arg))
 

sdks/python/apache_beam/typehints/row_type.py

@@ -171,8 +171,7 @@ def __hash__(self):
 
   def __repr__(self):
     return 'Row(%s)' % ', '.join(
-        '%s=%s' % (name, typehints._unified_repr(t)) for name,
-        t in self._fields)
+        '%s=%s' % (name, repr(t)) for name, t in self._fields)
 
   def get_type_for(self, name):
     return dict(self._fields)[name]

sdks/python/apache_beam/typehints/sharded_key_type.py

@@ -61,10 +61,8 @@ def type_check(self, instance):
       raise typehints.CompositeTypeHintError(
           "%s type-constraint violated. The type of key in 'ShardedKey' "
           "is incorrect. Expected an instance of type '%s', "
-          "instead received an instance of type '%s'." % (
-              repr(self),
-              typehints._unified_repr(self.key_type),
-              instance.key.__class__.__name__))
+          "instead received an instance of type '%s'." %
+          (repr(self), repr(self.key_type), instance.key.__class__.__name__))
 
   def match_type_variables(self, concrete_type):
     if isinstance(concrete_type, ShardedKeyTypeConstraint):
@@ -80,7 +78,7 @@ def __hash__(self):
     return hash(self.key_type)
 
   def __repr__(self):
-    return 'ShardedKey[%s]' % typehints._unified_repr(self.key_type)
+    return 'ShardedKey[%s]' % repr(self.key_type)
 
 
 ShardedKeyType = ShardedKeyTypeConstraint

sdks/python/apache_beam/typehints/sharded_key_type_test.py

@@ -37,7 +37,7 @@ def test_compatibility(self):
 
   def test_repr(self):
     constraint = ShardedKeyType[int]
-    self.assertEqual('ShardedKey[int]', repr(constraint))
+    self.assertEqual('ShardedKey[<class \'int\'>]', repr(constraint))
 
   def test_type_check_not_sharded_key(self):
     constraint = ShardedKeyType[int]
@@ -55,9 +55,9 @@ def test_type_check_invalid_key_type(self):
     with self.assertRaises((TypeError, TypeError)) as e:
       constraint.type_check(obj)
     self.assertEqual(
-        "ShardedKey[int] type-constraint violated. The type of key in "
-        "'ShardedKey' is incorrect. Expected an instance of type 'int', "
-        "instead received an instance of type 'str'.",
+        "ShardedKey[<class \'int\'>] type-constraint violated. The type of key "
+        "in 'ShardedKey' is incorrect. Expected an instance of type \'<class "
+        "\'int\'>\', instead received an instance of type 'str'.",
         e.exception.args[0])
 
   def test_type_check_valid_simple_type(self):