From aa16e01265b0eedfd33ed3faf3144c176d3974cf Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 27 Oct 2021 16:56:33 -0700
Subject: [PATCH 01/15] Fix serialization of Params with set data type

---
 airflow/serialization/serialized_objects.py | 25 +++++++++++++--------
 1 file changed, 16 insertions(+), 9 deletions(-)

diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index f63c20943a475..8cc1adbe84dae 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -325,7 +325,7 @@ def _serialize(cls, var: Any) -> Any:  # Unfortunately there is no support for r
         elif isinstance(var, TaskGroup):
             return SerializedTaskGroup.serialize_task_group(var)
         elif isinstance(var, Param):
-            return cls._encode(var.dump(), type_=DAT.PARAM)
+            return cls._encode(cls._serialize_param(var), type_=DAT.PARAM)
         else:
             log.debug('Cast type %s to str in serialization.', type(var))
             return str(var)
@@ -368,9 +368,7 @@ def _deserialize(cls, encoded_var: Any) -> Any:
         elif type_ == DAT.TUPLE:
             return tuple(cls._deserialize(v) for v in var)
         elif type_ == DAT.PARAM:
-            param_class = import_string(var['_type'])
-            del var['_type']
-            return param_class(**var)
+            return cls._deserialize_param(var)
         else:
             raise TypeError(f'Invalid type {type_!s} in deserialization.')
 
@@ -409,6 +407,18 @@ def _value_is_hardcoded_default(cls, attrname: str, value: Any, instance: Any) -
             return True
         return False
 
+    @classmethod
+    def _serialize_param(cls, param: Param):
+        d = param.dump()
+        d['value'] = cls._serialize(d['value'])
+        return d
+
+    @classmethod
+    def _deserialize_param(cls, param: Dict):
+        param_class = import_string(param['__class'])
+        param['value'] = cls._deserialize(param['value'])
+        return param_class(default=param['value'], description=param['description'])
+
     @classmethod
     def _serialize_params_dict(cls, params: ParamsDict):
         """Serialize Params dict for a DAG/Task"""
@@ -416,9 +426,7 @@ def _serialize_params_dict(cls, params: ParamsDict):
         for k, v in params.items():
             # TODO: As of now, we would allow serialization of params which are of type Param only
             if f'{v.__module__}.{v.__class__.__name__}' == 'airflow.models.param.Param':
-                kwargs = v.dump()
-                kwargs['default'] = kwargs.pop('value')
-                serialized_params[k] = kwargs
+                serialized_params[k] = cls._serialize_param(v)
             else:
                 raise ValueError('Params to a DAG or a Task can be only of type airflow.models.param.Param')
         return serialized_params
@@ -429,8 +437,7 @@ def _deserialize_params_dict(cls, encoded_params: Dict) -> ParamsDict:
         op_params = {}
         for k, v in encoded_params.items():
             if isinstance(v, dict) and "__class" in v:
-                param_class = import_string(v['__class'])
-                op_params[k] = param_class(**v)
+                op_params[k] = cls._deserialize_param(v)
             else:
                 # Old style params, upgrade it
                 op_params[k] = Param(v)

From 2830ccabc37d18844e4418c0a92585ccc534ce73 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Thu, 28 Oct 2021 09:47:03 -0700
Subject: [PATCH 02/15] restore backward compat

---
 airflow/serialization/serialized_objects.py | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index 8cc1adbe84dae..d3859889b732f 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -409,15 +409,15 @@ def _value_is_hardcoded_default(cls, attrname: str, value: Any, instance: Any) -
 
     @classmethod
     def _serialize_param(cls, param: Param):
-        d = param.dump()
-        d['value'] = cls._serialize(d['value'])
-        return d
+        param_dict = param.dump()
+        param_dict['default'] = cls._serialize(param_dict.pop('value'))
+        return param_dict
 
     @classmethod
-    def _deserialize_param(cls, param: Dict):
-        param_class = import_string(param['__class'])
-        param['value'] = cls._deserialize(param['value'])
-        return param_class(default=param['value'], description=param['description'])
+    def _deserialize_param(cls, param_dict: Dict):
+        param_class = import_string(param_dict['__class'])
+        param_dict['default'] = cls._deserialize(param_dict['default'])
+        return param_class(**param_dict)
 
     @classmethod
     def _serialize_params_dict(cls, params: ParamsDict):

From 97fddf206ee1f6a69bf92e57fcacb76be848fa08 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Thu, 28 Oct 2021 10:08:28 -0700
Subject: [PATCH 03/15] add test case for when params are sets

---
 tests/serialization/test_dag_serialization.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index afba96a5bf270..2ef092dc1b3aa 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -21,6 +21,7 @@
 import copy
 import importlib
 import importlib.util
+import json
 import multiprocessing
 import os
 from datetime import datetime, timedelta
@@ -724,6 +725,7 @@ def test_roundtrip_relativedelta(self, val, expected):
         [
             (None, {}),
             ({"param_1": "value_1"}, {"param_1": "value_1"}),
+            ({"param_1": {1, 2, 3}}, {"param_1": {1, 2, 3}}),
         ],
     )
     def test_dag_params_roundtrip(self, val, expected_val):
@@ -734,6 +736,10 @@ def test_dag_params_roundtrip(self, val, expected_val):
         BaseOperator(task_id='simple_task', dag=dag, start_date=datetime(2019, 8, 1))
 
         serialized_dag = SerializedDAG.to_dict(dag)
+
+        # serialized dag dict must be json serializable
+        json.dumps(serialized_dag)
+
         assert "params" in serialized_dag["dag"]
 
         deserialized_dag = SerializedDAG.from_dict(serialized_dag)

From f98768f5b275dae299d9064f80367eb4b73c915a Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Thu, 28 Oct 2021 13:25:27 -0700
Subject: [PATCH 04/15] serialization should cover all attrs besides __class

---
 airflow/models/param.py                     |  3 ++-
 airflow/serialization/serialized_objects.py | 11 ++++++-----
 2 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/airflow/models/param.py b/airflow/models/param.py
index 1ae01dc427273..c649d28b15238 100644
--- a/airflow/models/param.py
+++ b/airflow/models/param.py
@@ -49,6 +49,7 @@ class Param:
     """
 
     __NO_VALUE_SENTINEL = NoValueSentinel()
+    CLASS_IDENTIFIER = '__class'
 
     def __init__(self, default: Any = __NO_VALUE_SENTINEL, description: str = None, **kwargs):
         self.value = default
@@ -90,7 +91,7 @@ def resolve(self, value: Optional[Any] = __NO_VALUE_SENTINEL, suppress_exception
 
     def dump(self) -> dict:
         """Dump the Param as a dictionary"""
-        out_dict = {'__class': f'{self.__module__}.{self.__class__.__name__}'}
+        out_dict = {self.CLASS_IDENTIFIER: f'{self.__module__}.{self.__class__.__name__}'}
         out_dict.update(self.__dict__)
         return out_dict
 
diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index d3859889b732f..2386d99a64b40 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -410,14 +410,15 @@ def _value_is_hardcoded_default(cls, attrname: str, value: Any, instance: Any) -
     @classmethod
     def _serialize_param(cls, param: Param):
         param_dict = param.dump()
-        param_dict['default'] = cls._serialize(param_dict.pop('value'))
-        return param_dict
+        param_class_name = param_dict.pop(Param.CLASS_IDENTIFIER)
+        param_kwargs = dict(default=param_dict.pop('value'), **param_dict)
+        return {Param.CLASS_IDENTIFIER: param_class_name, **cls._serialize(param_kwargs)}
 
     @classmethod
     def _deserialize_param(cls, param_dict: Dict):
-        param_class = import_string(param_dict['__class'])
-        param_dict['default'] = cls._deserialize(param_dict['default'])
-        return param_class(**param_dict)
+        param_class = import_string(param_dict.pop(Param.CLASS_IDENTIFIER))
+        param_kwargs = cls._deserialize(param_dict)
+        return param_class(**param_kwargs)
 
     @classmethod
     def _serialize_params_dict(cls, params: ParamsDict):

From 0fc6c0efa97f26718e7cd3670e3903fab370667d Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Thu, 28 Oct 2021 21:09:02 -0700
Subject: [PATCH 05/15] handle params serialized in 2.2.0, when param attrs
 were not run through serializer

---
 airflow/serialization/serialized_objects.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index 2386d99a64b40..ebc473390b08a 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -417,7 +417,10 @@ def _serialize_param(cls, param: Param):
     @classmethod
     def _deserialize_param(cls, param_dict: Dict):
         param_class = import_string(param_dict.pop(Param.CLASS_IDENTIFIER))
-        param_kwargs = cls._deserialize(param_dict)
+        try:
+            param_kwargs = cls._deserialize(param_dict)
+        except KeyError:
+            param_kwargs = param_dict
         return param_class(**param_kwargs)
 
     @classmethod

From 33ac787f89943fe7b2b19c566abdaa3b01758256 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Fri, 29 Oct 2021 07:43:49 -0700
Subject: [PATCH 06/15] be more explicit about ser and deser; add tests

---
 airflow/serialization/serialized_objects.py   |  36 ++--
 tests/serialization/test_dag_serialization.py | 167 ++++++++++++------
 2 files changed, 132 insertions(+), 71 deletions(-)

diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index ebc473390b08a..f973d56529ac6 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -55,7 +55,6 @@
 except ImportError:
     HAS_KUBERNETES = False
 
-
 if TYPE_CHECKING:
     from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
 
@@ -409,19 +408,30 @@ def _value_is_hardcoded_default(cls, attrname: str, value: Any, instance: Any) -
 
     @classmethod
     def _serialize_param(cls, param: Param):
-        param_dict = param.dump()
-        param_class_name = param_dict.pop(Param.CLASS_IDENTIFIER)
-        param_kwargs = dict(default=param_dict.pop('value'), **param_dict)
-        return {Param.CLASS_IDENTIFIER: param_class_name, **cls._serialize(param_kwargs)}
+        return dict(
+            __class=f"{param.__module__}.{param.__class__.__name__}",
+            default=cls._serialize(param.value),
+            description=cls._serialize(param.description),
+            schema=cls._serialize(param.schema),
+        )
 
     @classmethod
     def _deserialize_param(cls, param_dict: Dict):
-        param_class = import_string(param_dict.pop(Param.CLASS_IDENTIFIER))
-        try:
-            param_kwargs = cls._deserialize(param_dict)
-        except KeyError:
-            param_kwargs = param_dict
-        return param_class(**param_kwargs)
+        class_name = param_dict['__class']
+        class_ = import_string(class_name)  # type: Type[Param]
+        attrs = ('default', 'description', 'schema')
+        kwargs = {}
+        for attr in attrs:
+            if attr not in param_dict:
+                continue
+            val = param_dict[attr]
+            is_serialized = isinstance(val, dict) and '__type' in val
+            if is_serialized:
+                deserialized_val = cls._deserialize(param_dict[attr])
+                kwargs[attr] = deserialized_val
+            else:
+                kwargs[attr] = val
+        return class_(**kwargs)
 
     @classmethod
     def _serialize_params_dict(cls, params: ParamsDict):
@@ -437,13 +447,13 @@ def _serialize_params_dict(cls, params: ParamsDict):
 
     @classmethod
     def _deserialize_params_dict(cls, encoded_params: Dict) -> ParamsDict:
-        """Deserialize a DAGs Params dict"""
+        """Deserialize a DAG's Params dict"""
         op_params = {}
         for k, v in encoded_params.items():
             if isinstance(v, dict) and "__class" in v:
                 op_params[k] = cls._deserialize_param(v)
             else:
-                # Old style params, upgrade it
+                # Old style params, convert it
                 op_params[k] = Param(v)
 
         return ParamsDict(op_params)
diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index 2ef092dc1b3aa..99e208db48996 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -735,10 +735,9 @@ def test_dag_params_roundtrip(self, val, expected_val):
         dag = DAG(dag_id='simple_dag', params=val)
         BaseOperator(task_id='simple_task', dag=dag, start_date=datetime(2019, 8, 1))
 
-        serialized_dag = SerializedDAG.to_dict(dag)
+        serialized_dag_json = SerializedDAG.to_json(dag)
 
-        # serialized dag dict must be json serializable
-        json.dumps(serialized_dag)
+        serialized_dag = json.loads(serialized_dag_json)
 
         assert "params" in serialized_dag["dag"]
 
@@ -770,14 +769,37 @@ def __init__(self, path: str):
             params={'path': S3Param('s3://my_bucket/my_path')},
         )
 
-        with pytest.raises(SerializationError):
-            SerializedDAG.to_dict(dag)
+    @pytest.mark.parametrize(
+        'param',
+        [
+            Param('my value', description='hello', schema={'type': 'string'}),
+            Param('my value', description='hello'),
+            Param(None, description=None),
+        ]
+    )
+    def test_full_param_roundtrip(self, param):
+        """
+        Test to make sure that only native Param objects are being passed as dag or task params
+        """
+
+        dag = DAG(dag_id='simple_dag', params={'my_param':param})
+        serialized_json = SerializedDAG.to_json(dag)
+        serialized = json.loads(serialized_json)
+        SerializedDAG.validate_schema(serialized)
+        dag = SerializedDAG.from_dict(serialized)
+
+        assert dag.params["my_param"] == param.value
+        observed_param = dict.get(dag.params, 'my_param')
+        assert isinstance(observed_param, Param)
+        assert observed_param.description == param.description
+        assert observed_param.schema == param.schema
 
     @pytest.mark.parametrize(
         "val, expected_val",
         [
             (None, {}),
             ({"param_1": "value_1"}, {"param_1": "value_1"}),
+            ({"param_1": {1, 2, 3}}, {"param_1": {1, 2, 3}}),
         ],
     )
     def test_task_params_roundtrip(self, val, expected_val):
@@ -1073,56 +1095,56 @@ def test_no_new_fields_added_to_base_operator(self):
         base_operator = BaseOperator(task_id="10")
         fields = base_operator.__dict__
         assert {
-            '_BaseOperator__instantiated': True,
-            '_dag': None,
-            '_downstream_task_ids': set(),
-            '_inlets': [],
-            '_log': base_operator.log,
-            '_outlets': [],
-            '_upstream_task_ids': set(),
-            '_pre_execute_hook': None,
-            '_post_execute_hook': None,
-            'depends_on_past': False,
-            'do_xcom_push': True,
-            'doc': None,
-            'doc_json': None,
-            'doc_md': None,
-            'doc_rst': None,
-            'doc_yaml': None,
-            'email': None,
-            'email_on_failure': True,
-            'email_on_retry': True,
-            'end_date': None,
-            'execution_timeout': None,
-            'executor_config': {},
-            'inlets': [],
-            'label': '10',
-            'max_active_tis_per_dag': None,
-            'max_retry_delay': None,
-            'on_execute_callback': None,
-            'on_failure_callback': None,
-            'on_retry_callback': None,
-            'on_success_callback': None,
-            'outlets': [],
-            'owner': 'airflow',
-            'params': {},
-            'pool': 'default_pool',
-            'pool_slots': 1,
-            'priority_weight': 1,
-            'queue': 'default',
-            'resources': None,
-            'retries': 0,
-            'retry_delay': timedelta(0, 300),
-            'retry_exponential_backoff': False,
-            'run_as_user': None,
-            'sla': None,
-            'start_date': None,
-            'subdag': None,
-            'task_id': '10',
-            'trigger_rule': 'all_success',
-            'wait_for_downstream': False,
-            'weight_rule': 'downstream',
-        } == fields, """
+                   '_BaseOperator__instantiated': True,
+                   '_dag': None,
+                   '_downstream_task_ids': set(),
+                   '_inlets': [],
+                   '_log': base_operator.log,
+                   '_outlets': [],
+                   '_upstream_task_ids': set(),
+                   '_pre_execute_hook': None,
+                   '_post_execute_hook': None,
+                   'depends_on_past': False,
+                   'do_xcom_push': True,
+                   'doc': None,
+                   'doc_json': None,
+                   'doc_md': None,
+                   'doc_rst': None,
+                   'doc_yaml': None,
+                   'email': None,
+                   'email_on_failure': True,
+                   'email_on_retry': True,
+                   'end_date': None,
+                   'execution_timeout': None,
+                   'executor_config': {},
+                   'inlets': [],
+                   'label': '10',
+                   'max_active_tis_per_dag': None,
+                   'max_retry_delay': None,
+                   'on_execute_callback': None,
+                   'on_failure_callback': None,
+                   'on_retry_callback': None,
+                   'on_success_callback': None,
+                   'outlets': [],
+                   'owner': 'airflow',
+                   'params': {},
+                   'pool': 'default_pool',
+                   'pool_slots': 1,
+                   'priority_weight': 1,
+                   'queue': 'default',
+                   'resources': None,
+                   'retries': 0,
+                   'retry_delay': timedelta(0, 300),
+                   'retry_exponential_backoff': False,
+                   'run_as_user': None,
+                   'sla': None,
+                   'start_date': None,
+                   'subdag': None,
+                   'task_id': '10',
+                   'trigger_rule': 'all_success',
+                   'wait_for_downstream': False,
+                   'weight_rule': 'downstream',
+               } == fields, """
 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
 
      ACTION NEEDED! PLEASE READ THIS CAREFULLY AND CORRECT TESTS CAREFULLY
@@ -1439,11 +1461,12 @@ def test_serialized_objects_are_sorted(self, object_to_serialized, expected_outp
         assert serialized_obj == expected_output
 
     def test_params_upgrade(self):
+        """when pre-2.2.0 param (i.e. primitive) is deserialized we convert to Param"""
         serialized = {
             "__version": 1,
             "dag": {
                 "_dag_id": "simple_dag",
-                "fileloc": __file__,
+                "fileloc": '__file__',
                 "tasks": [],
                 "timezone": "UTC",
                 "params": {"none": None, "str": "str", "dict": {"a": "b"}},
@@ -1456,12 +1479,15 @@ def test_params_upgrade(self):
         assert isinstance(dict.__getitem__(dag.params, "none"), Param)
         assert dag.params["str"] == "str"
 
-    def test_params_serialize_default(self):
+    def test_params_serialize_default_2_2_0(self):
+        """In 2.0.0, param ``default`` was assumed to be json-serializable objects and were not run though
+        the standard serializer function.  In 2.2.2 we serialize param ``default``.  We keep this
+        test only to ensure that params stored in 2.2.0 can still be parsed correctly."""
         serialized = {
             "__version": 1,
             "dag": {
                 "_dag_id": "simple_dag",
-                "fileloc": __file__,
+                "fileloc": '/path/to/file.py',
                 "tasks": [],
                 "timezone": "UTC",
                 "params": {"str": {"__class": "airflow.models.param.Param", "default": "str"}},
@@ -1473,6 +1499,31 @@ def test_params_serialize_default(self):
         assert isinstance(dict.__getitem__(dag.params, "str"), Param)
         assert dag.params["str"] == "str"
 
+    def test_params_serialize_default(self):
+        """In 2.0.0, param ``default`` was assumed to be json-serializable objects and were not run though
+        the standard serializer function.  In 2.2.2 we serialize param ``default``.  We keep this
+        test only to ensure that params stored in 2.2.0 can still be parsed correctly."""
+        serialized = {
+            "__version": 1,
+            "dag": {
+                "_dag_id": "simple_dag",
+                "fileloc": '/path/to/file.py',
+                "tasks": [],
+                "timezone": "UTC",
+                "params": {"my_param": {"default": "a string value", "description": "hello",
+                                        "schema": {"__var": {"type": "string"}, "__type": "dict"},
+                                        "__class": "airflow.models.param.Param"}},
+            },
+        }
+        SerializedDAG.validate_schema(serialized)
+        dag = SerializedDAG.from_dict(serialized)
+
+        assert dag.params["my_param"] == "a string value"
+        param = dict.get(dag.params, 'my_param')
+        assert isinstance(param, Param)
+        assert param.description == 'hello'
+        assert param.schema == {'type': 'string'}
+
 
 def test_kubernetes_optional():
     """Serialisation / deserialisation continues to work without kubernetes installed"""

From fc7f4f59fb1d076def53870324a3bf7f5258c168 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Fri, 29 Oct 2021 08:07:54 -0700
Subject: [PATCH 07/15] update param schema

---
 airflow/serialization/schema.json             |  19 ++-
 tests/serialization/test_dag_serialization.py | 118 +++++++++---------
 2 files changed, 78 insertions(+), 59 deletions(-)

diff --git a/airflow/serialization/schema.json b/airflow/serialization/schema.json
index b4a64b4e79459..a7f0b360b26d3 100644
--- a/airflow/serialization/schema.json
+++ b/airflow/serialization/schema.json
@@ -135,6 +135,23 @@
       "type": "array",
       "additionalProperties": { "$ref": "#/definitions/operator" }
     },
+    "params": {
+      "type": "array",
+      "additionalProperties": { "$ref": "#/definitions/param" }
+    },
+    "param": {
+      "$comment": "A param for a dag / operator",
+      "type": "object",
+      "required": [
+        "__class",
+        "default"
+          ],
+          "properties": {
+        "__class": { "type": "string" },
+        "default": { "$ref": "#/definitions/dict" },
+        "description": { "type": "string" },
+        "schema": { "$ref": "#/definitions/dict" }
+          }},
     "operator": {
       "$comment": "A task/operator in a DAG",
       "type": "object",
@@ -166,7 +183,7 @@
         "retry_delay": { "$ref": "#/definitions/timedelta" },
         "retry_exponential_backoff": { "type": "boolean" },
         "max_retry_delay": { "$ref": "#/definitions/timedelta" },
-        "params": { "$ref": "#/definitions/dict" },
+        "params": { "$ref": "#/definitions/params" },
         "priority_weight": { "type": "number" },
         "weight_rule": { "type": "string" },
         "executor_config": { "$ref": "#/definitions/dict" },
diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index 99e208db48996..debf835f8decd 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -775,14 +775,14 @@ def __init__(self, path: str):
             Param('my value', description='hello', schema={'type': 'string'}),
             Param('my value', description='hello'),
             Param(None, description=None),
-        ]
+        ],
     )
     def test_full_param_roundtrip(self, param):
         """
         Test to make sure that only native Param objects are being passed as dag or task params
         """
 
-        dag = DAG(dag_id='simple_dag', params={'my_param':param})
+        dag = DAG(dag_id='simple_dag', params={'my_param': param})
         serialized_json = SerializedDAG.to_json(dag)
         serialized = json.loads(serialized_json)
         SerializedDAG.validate_schema(serialized)
@@ -1095,56 +1095,56 @@ def test_no_new_fields_added_to_base_operator(self):
         base_operator = BaseOperator(task_id="10")
         fields = base_operator.__dict__
         assert {
-                   '_BaseOperator__instantiated': True,
-                   '_dag': None,
-                   '_downstream_task_ids': set(),
-                   '_inlets': [],
-                   '_log': base_operator.log,
-                   '_outlets': [],
-                   '_upstream_task_ids': set(),
-                   '_pre_execute_hook': None,
-                   '_post_execute_hook': None,
-                   'depends_on_past': False,
-                   'do_xcom_push': True,
-                   'doc': None,
-                   'doc_json': None,
-                   'doc_md': None,
-                   'doc_rst': None,
-                   'doc_yaml': None,
-                   'email': None,
-                   'email_on_failure': True,
-                   'email_on_retry': True,
-                   'end_date': None,
-                   'execution_timeout': None,
-                   'executor_config': {},
-                   'inlets': [],
-                   'label': '10',
-                   'max_active_tis_per_dag': None,
-                   'max_retry_delay': None,
-                   'on_execute_callback': None,
-                   'on_failure_callback': None,
-                   'on_retry_callback': None,
-                   'on_success_callback': None,
-                   'outlets': [],
-                   'owner': 'airflow',
-                   'params': {},
-                   'pool': 'default_pool',
-                   'pool_slots': 1,
-                   'priority_weight': 1,
-                   'queue': 'default',
-                   'resources': None,
-                   'retries': 0,
-                   'retry_delay': timedelta(0, 300),
-                   'retry_exponential_backoff': False,
-                   'run_as_user': None,
-                   'sla': None,
-                   'start_date': None,
-                   'subdag': None,
-                   'task_id': '10',
-                   'trigger_rule': 'all_success',
-                   'wait_for_downstream': False,
-                   'weight_rule': 'downstream',
-               } == fields, """
+            '_BaseOperator__instantiated': True,
+            '_dag': None,
+            '_downstream_task_ids': set(),
+            '_inlets': [],
+            '_log': base_operator.log,
+            '_outlets': [],
+            '_upstream_task_ids': set(),
+            '_pre_execute_hook': None,
+            '_post_execute_hook': None,
+            'depends_on_past': False,
+            'do_xcom_push': True,
+            'doc': None,
+            'doc_json': None,
+            'doc_md': None,
+            'doc_rst': None,
+            'doc_yaml': None,
+            'email': None,
+            'email_on_failure': True,
+            'email_on_retry': True,
+            'end_date': None,
+            'execution_timeout': None,
+            'executor_config': {},
+            'inlets': [],
+            'label': '10',
+            'max_active_tis_per_dag': None,
+            'max_retry_delay': None,
+            'on_execute_callback': None,
+            'on_failure_callback': None,
+            'on_retry_callback': None,
+            'on_success_callback': None,
+            'outlets': [],
+            'owner': 'airflow',
+            'params': {},
+            'pool': 'default_pool',
+            'pool_slots': 1,
+            'priority_weight': 1,
+            'queue': 'default',
+            'resources': None,
+            'retries': 0,
+            'retry_delay': timedelta(0, 300),
+            'retry_exponential_backoff': False,
+            'run_as_user': None,
+            'sla': None,
+            'start_date': None,
+            'subdag': None,
+            'task_id': '10',
+            'trigger_rule': 'all_success',
+            'wait_for_downstream': False,
+            'weight_rule': 'downstream',
+        } == fields, """
 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
 
      ACTION NEEDED! PLEASE READ THIS CAREFULLY AND CORRECT TESTS CAREFULLY
@@ -1500,9 +1500,6 @@ def test_params_serialize_default_2_2_0(self):
         assert dag.params["str"] == "str"
 
     def test_params_serialize_default(self):
-        """In 2.0.0, param ``default`` was assumed to be json-serializable objects and were not run though
-        the standard serializer function.  In 2.2.2 we serialize param ``default``.  We keep this
-        test only to ensure that params stored in 2.2.0 can still be parsed correctly."""
         serialized = {
             "__version": 1,
             "dag": {
@@ -1510,9 +1507,14 @@ def test_params_serialize_default(self):
                 "fileloc": '/path/to/file.py',
                 "tasks": [],
                 "timezone": "UTC",
-                "params": {"my_param": {"default": "a string value", "description": "hello",
-                                        "schema": {"__var": {"type": "string"}, "__type": "dict"},
-                                        "__class": "airflow.models.param.Param"}},
+                "params": {
+                    "my_param": {
+                        "default": "a string value",
+                        "description": "hello",
+                        "schema": {"__var": {"type": "string"}, "__type": "dict"},
+                        "__class": "airflow.models.param.Param",
+                    }
+                },
             },
         }
         SerializedDAG.validate_schema(serialized)

From 6fa83fd33e1054406870c373cd67fb3a6928d12d Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Fri, 29 Oct 2021 08:15:20 -0700
Subject: [PATCH 08/15] add comment

---
 airflow/serialization/serialized_objects.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index f973d56529ac6..c1bc7734b5d16 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -417,6 +417,11 @@ def _serialize_param(cls, param: Param):
 
     @classmethod
     def _deserialize_param(cls, param_dict: Dict):
+        """
+        In 2.2.0, Param attrs were assumed to be json-serializable and were not run through
+        this class's ``_serialize`` method.  So before running through ``_deserialize``,
+        we first verify that it's necessary to do.
+        """
         class_name = param_dict['__class']
         class_ = import_string(class_name)  # type: Type[Param]
         attrs = ('default', 'description', 'schema')

From 01eacb425c8c93126a1e18a22723454cee99f4a4 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 3 Nov 2021 13:44:25 -0700
Subject: [PATCH 09/15] update schema: dag

---
 airflow/serialization/schema.json | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/airflow/serialization/schema.json b/airflow/serialization/schema.json
index a7f0b360b26d3..f5c7cdce28555 100644
--- a/airflow/serialization/schema.json
+++ b/airflow/serialization/schema.json
@@ -78,7 +78,7 @@
     "dag": {
       "type": "object",
       "properties": {
-        "params": { "$ref": "#/definitions/dict" },
+        "params": { "$ref": "#/definitions/params" },
         "_dag_id": { "type": "string" },
         "tasks": {  "$ref": "#/definitions/tasks" },
         "timezone": { "$ref": "#/definitions/timezone" },
@@ -145,13 +145,14 @@
       "required": [
         "__class",
         "default"
-          ],
-          "properties": {
+      ],
+      "properties": {
         "__class": { "type": "string" },
         "default": { "$ref": "#/definitions/dict" },
         "description": { "type": "string" },
         "schema": { "$ref": "#/definitions/dict" }
-          }},
+      }
+    },
     "operator": {
       "$comment": "A task/operator in a DAG",
       "type": "object",

From ec839e43be5c628ad281410e3f537a0c24aae17d Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 3 Nov 2021 20:00:48 -0700
Subject: [PATCH 10/15] fix schema

---
 airflow/serialization/schema.json | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/airflow/serialization/schema.json b/airflow/serialization/schema.json
index f5c7cdce28555..d9e7f8f43e05a 100644
--- a/airflow/serialization/schema.json
+++ b/airflow/serialization/schema.json
@@ -78,7 +78,7 @@
     "dag": {
       "type": "object",
       "properties": {
-        "params": { "$ref": "#/definitions/params" },
+        "params": { "$ref": "#/definitions/params_dict" },
         "_dag_id": { "type": "string" },
         "tasks": {  "$ref": "#/definitions/tasks" },
         "timezone": { "$ref": "#/definitions/timezone" },
@@ -135,9 +135,9 @@
       "type": "array",
       "additionalProperties": { "$ref": "#/definitions/operator" }
     },
-    "params": {
-      "type": "array",
-      "additionalProperties": { "$ref": "#/definitions/param" }
+    "params_dict": {
+      "type": "object",
+      "additionalProperties": {"anyOf": [ { "$ref": "#/definitions/param" }, { "type":  "object"}]}
     },
     "param": {
       "$comment": "A param for a dag / operator",
@@ -184,7 +184,7 @@
         "retry_delay": { "$ref": "#/definitions/timedelta" },
         "retry_exponential_backoff": { "type": "boolean" },
         "max_retry_delay": { "$ref": "#/definitions/timedelta" },
-        "params": { "$ref": "#/definitions/params" },
+        "params": { "$ref": "#/definitions/params_dict" },
         "priority_weight": { "type": "number" },
         "weight_rule": { "type": "string" },
         "executor_config": { "$ref": "#/definitions/dict" },

From 10b14580362c0fc7d025ad223d8e5c4be3a388e3 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 3 Nov 2021 21:07:23 -0700
Subject: [PATCH 11/15] validate json serializable in param init

---
 airflow/models/param.py           | 31 +++++++++++++++++++++++++------
 airflow/serialization/schema.json |  6 +++---
 2 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/airflow/models/param.py b/airflow/models/param.py
index c649d28b15238..6304729358601 100644
--- a/airflow/models/param.py
+++ b/airflow/models/param.py
@@ -14,7 +14,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
+import json
+import warnings
 from typing import Any, Dict, Optional
 
 import jsonschema
@@ -52,16 +53,34 @@ class Param:
     CLASS_IDENTIFIER = '__class'
 
     def __init__(self, default: Any = __NO_VALUE_SENTINEL, description: str = None, **kwargs):
+
         self.value = default
         self.description = description
         self.schema = kwargs.pop('schema') if 'schema' in kwargs else kwargs
 
-        # If we have a value, validate it once. May raise ValueError.
         if self.has_value:
-            try:
-                jsonschema.validate(self.value, self.schema, format_checker=FormatChecker())
-            except ValidationError as err:
-                raise ValueError(err)
+            self._validate(self.value, self.schema)
+
+    @staticmethod
+    def _validate(value, schema):
+        """
+        1. Check that value is json-serializable; if not, warn.  In future release we will require
+        the value to be json-serializable.
+        2. Validate ``value`` against ``schema``
+        """
+        try:
+            json.dumps(value)
+        except TypeError:
+            warnings.warn(
+                "The use of non-json-serializable params is deprecated and will be removed in a"
+                " future release",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        try:
+            jsonschema.validate(value, schema, format_checker=FormatChecker())
+        except ValidationError as err:
+            raise ValueError(err)
 
     def resolve(self, value: Optional[Any] = __NO_VALUE_SENTINEL, suppress_exception: bool = False) -> Any:
         """
diff --git a/airflow/serialization/schema.json b/airflow/serialization/schema.json
index d9e7f8f43e05a..6d25c1ee3972e 100644
--- a/airflow/serialization/schema.json
+++ b/airflow/serialization/schema.json
@@ -137,7 +137,7 @@
     },
     "params_dict": {
       "type": "object",
-      "additionalProperties": {"anyOf": [ { "$ref": "#/definitions/param" }, { "type":  "object"}]}
+      "additionalProperties": {"$ref": "#/definitions/param" }
     },
     "param": {
       "$comment": "A param for a dag / operator",
@@ -148,8 +148,8 @@
       ],
       "properties": {
         "__class": { "type": "string" },
-        "default": { "$ref": "#/definitions/dict" },
-        "description": { "type": "string" },
+        "default": {},
+        "description": {"anyOf": [{"type":"string"}, {"type":"null"}]},
         "schema": { "$ref": "#/definitions/dict" }
       }
     },

From a9b56ef4fb6d6de4c2300075bde787d9a376495e Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 3 Nov 2021 22:15:17 -0700
Subject: [PATCH 12/15] update schema to reflect old params structure

---
 airflow/serialization/schema.json | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/airflow/serialization/schema.json b/airflow/serialization/schema.json
index 6d25c1ee3972e..1e5b0109665ae 100644
--- a/airflow/serialization/schema.json
+++ b/airflow/serialization/schema.json
@@ -78,7 +78,7 @@
     "dag": {
       "type": "object",
       "properties": {
-        "params": { "$ref": "#/definitions/params_dict" },
+        "params": {"anyOf": [{ "$ref": "#/definitions/params_dict" }, {"$ref": "#/definitions/dict"}]},
         "_dag_id": { "type": "string" },
         "tasks": {  "$ref": "#/definitions/tasks" },
         "timezone": { "$ref": "#/definitions/timezone" },
@@ -184,7 +184,7 @@
         "retry_delay": { "$ref": "#/definitions/timedelta" },
         "retry_exponential_backoff": { "type": "boolean" },
         "max_retry_delay": { "$ref": "#/definitions/timedelta" },
-        "params": { "$ref": "#/definitions/params_dict" },
+        "params": {"anyOf": [{ "$ref": "#/definitions/params_dict" }, {"$ref": "#/definitions/dict"}]},
         "priority_weight": { "type": "number" },
         "weight_rule": { "type": "string" },
         "executor_config": { "$ref": "#/definitions/dict" },

From d4cc8256f23b72e1991de37c19126e6f4c380792 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 3 Nov 2021 23:27:52 -0700
Subject: [PATCH 13/15] don't validate pre-2.2.0 schema against current schema

---
 airflow/serialization/schema.json             | 4 ++--
 tests/serialization/test_dag_serialization.py | 1 -
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/airflow/serialization/schema.json b/airflow/serialization/schema.json
index 1e5b0109665ae..6d25c1ee3972e 100644
--- a/airflow/serialization/schema.json
+++ b/airflow/serialization/schema.json
@@ -78,7 +78,7 @@
     "dag": {
       "type": "object",
       "properties": {
-        "params": {"anyOf": [{ "$ref": "#/definitions/params_dict" }, {"$ref": "#/definitions/dict"}]},
+        "params": { "$ref": "#/definitions/params_dict" },
         "_dag_id": { "type": "string" },
         "tasks": {  "$ref": "#/definitions/tasks" },
         "timezone": { "$ref": "#/definitions/timezone" },
@@ -184,7 +184,7 @@
         "retry_delay": { "$ref": "#/definitions/timedelta" },
         "retry_exponential_backoff": { "type": "boolean" },
         "max_retry_delay": { "$ref": "#/definitions/timedelta" },
-        "params": {"anyOf": [{ "$ref": "#/definitions/params_dict" }, {"$ref": "#/definitions/dict"}]},
+        "params": { "$ref": "#/definitions/params_dict" },
         "priority_weight": { "type": "number" },
         "weight_rule": { "type": "string" },
         "executor_config": { "$ref": "#/definitions/dict" },
diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index debf835f8decd..b2e5350d112c2 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -1472,7 +1472,6 @@ def test_params_upgrade(self):
                 "params": {"none": None, "str": "str", "dict": {"a": "b"}},
             },
         }
-        SerializedDAG.validate_schema(serialized)
         dag = SerializedDAG.from_dict(serialized)
 
         assert dag.params["none"] is None

From 838f2dfcb2138bf3f870ba46183d751f2c4b5b29 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Thu, 4 Nov 2021 20:50:26 -0700
Subject: [PATCH 14/15] fix path

---
 tests/serialization/test_dag_serialization.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index b2e5350d112c2..6fec7f9bc745c 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -1466,7 +1466,7 @@ def test_params_upgrade(self):
             "__version": 1,
             "dag": {
                 "_dag_id": "simple_dag",
-                "fileloc": '__file__',
+                "fileloc": '/path/to/file.py',
                 "tasks": [],
                 "timezone": "UTC",
                 "params": {"none": None, "str": "str", "dict": {"a": "b"}},

From b097396f8b6b72535cd8b2c70f7fd2d2215d2dbb Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Thu, 4 Nov 2021 20:52:37 -0700
Subject: [PATCH 15/15] revert json validation

---
 airflow/models/param.py | 30 +++++-------------------------
 1 file changed, 5 insertions(+), 25 deletions(-)

diff --git a/airflow/models/param.py b/airflow/models/param.py
index 6304729358601..53ac79a0bb0ec 100644
--- a/airflow/models/param.py
+++ b/airflow/models/param.py
@@ -14,8 +14,6 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-import json
-import warnings
 from typing import Any, Dict, Optional
 
 import jsonschema
@@ -53,34 +51,16 @@ class Param:
     CLASS_IDENTIFIER = '__class'
 
     def __init__(self, default: Any = __NO_VALUE_SENTINEL, description: str = None, **kwargs):
-
         self.value = default
         self.description = description
         self.schema = kwargs.pop('schema') if 'schema' in kwargs else kwargs
 
+        # If we have a value, validate it once. May raise ValueError.
         if self.has_value:
-            self._validate(self.value, self.schema)
-
-    @staticmethod
-    def _validate(value, schema):
-        """
-        1. Check that value is json-serializable; if not, warn.  In future release we will require
-        the value to be json-serializable.
-        2. Validate ``value`` against ``schema``
-        """
-        try:
-            json.dumps(value)
-        except TypeError:
-            warnings.warn(
-                "The use of non-json-serializable params is deprecated and will be removed in a"
-                " future release",
-                DeprecationWarning,
-                stacklevel=2,
-            )
-        try:
-            jsonschema.validate(value, schema, format_checker=FormatChecker())
-        except ValidationError as err:
-            raise ValueError(err)
+            try:
+                jsonschema.validate(self.value, self.schema, format_checker=FormatChecker())
+            except ValidationError as err:
+                raise ValueError(err)
 
     def resolve(self, value: Optional[Any] = __NO_VALUE_SENTINEL, suppress_exception: bool = False) -> Any:
         """