diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index 6519e4d5f..cee29cff7 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py @@ -186,8 +186,9 @@ def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]: # Resolve references in the manifest resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest) # Propagate types and parameters throughout the manifest - propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters( - "", resolved_manifest, {} + # Resolver output is freshly built and contains no shared containers. + propagated_manifest = ManifestComponentTransformer()._propagate_in_place( + "", dict(resolved_manifest), {} ) return propagated_manifest diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 24803a72b..f0c9a6d98 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -311,8 +311,9 @@ def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]: # Resolve references in the manifest resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest) # Propagate types and parameters throughout the manifest - propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters( - "", resolved_manifest, {} + # Resolver output is freshly built and contains no shared containers. + propagated_manifest = ManifestComponentTransformer()._propagate_in_place( + "", dict(resolved_manifest), {} ) return propagated_manifest diff --git a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py index 97c63c69b..bf9193d0c 100644 --- a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py +++ b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py @@ -87,6 +87,17 @@ class ManifestComponentTransformer: + @staticmethod + def _copy_tree(value: Any) -> Any: + """Copy containers without preserving shared dict/list identities.""" + if isinstance(value, dict): + return { + key: ManifestComponentTransformer._copy_tree(item) for key, item in value.items() + } + if isinstance(value, list): + return [ManifestComponentTransformer._copy_tree(item) for item in value] + return copy.deepcopy(value) + def propagate_types_and_parameters( self, parent_field_identifier: str, @@ -96,8 +107,8 @@ def propagate_types_and_parameters( ) -> Dict[str, Any]: """ Recursively transforms the specified declarative component and subcomponents to propagate parameters and insert the - default component type if it was not already present. The resulting transformed components are a deep copy of the input - components, not an in-place transformation. + default component type if it was not already present. The input is copied once and then transformed in place; the + resulting transformed components never share containers with the input. :param declarative_component: The current component that is having type and parameters added :param parent_field_identifier: The name of the field of the current component coming from the parent component @@ -105,7 +116,22 @@ def propagate_types_and_parameters( :param use_parent_parameters: If set, parent parameters will be used as the source of truth when key names are the same :return: A deep copy of the transformed component with types and parameters persisted to it """ - propagated_component = dict(copy.deepcopy(declarative_component)) + return self._propagate_in_place( + parent_field_identifier, + self._copy_tree(dict(declarative_component)), + self._copy_tree(dict(parent_parameters)), + use_parent_parameters, + ) + + def _propagate_in_place( + self, + parent_field_identifier: str, + declarative_component: Dict[str, Any], + parent_parameters: Mapping[str, Any], + use_parent_parameters: Optional[bool] = None, + ) -> Dict[str, Any]: + """Mutate a caller-owned input tree containing no shared dict/list objects.""" + propagated_component = declarative_component if "type" not in propagated_component: # If the component has class_name we assume that this is a reference to a custom component. This is a slight change to # existing behavior because we originally allowed for either class or type to be specified. After the pydantic migration, @@ -120,7 +146,7 @@ def propagate_types_and_parameters( # Combines parameters defined at the current level with parameters from parent components. Parameters at the current # level take precedence - current_parameters = dict(copy.deepcopy(parent_parameters)) + current_parameters = dict(parent_parameters) component_parameters = propagated_component.pop(PARAMETERS_STR, {}) current_parameters = ( {**component_parameters, **current_parameters} @@ -151,17 +177,18 @@ def propagate_types_and_parameters( # Parameters should be applied to the current component fields with the existing field taking precedence over parameters if # both exist + written_from_parameters = set() for parameter_key, parameter_value in current_parameters.items(): - propagated_component[parameter_key] = ( - propagated_component.get(parameter_key) or parameter_value - ) + if not propagated_component.get(parameter_key): + propagated_component[parameter_key] = copy.deepcopy(parameter_value) + written_from_parameters.add(parameter_key) for field_name, field_value in propagated_component.items(): if isinstance(field_value, dict): # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle excluded_parameter = current_parameters.pop(field_name, None) parent_type_field_identifier = f"{propagated_component.get('type')}.{field_name}" - propagated_component[field_name] = self.propagate_types_and_parameters( + propagated_component[field_name] = self._propagate_in_place( parent_type_field_identifier, field_value, current_parameters, @@ -177,17 +204,19 @@ def propagate_types_and_parameters( parent_type_field_identifier = ( f"{propagated_component.get('type')}.{field_name}" ) - field_value[i] = self.propagate_types_and_parameters( + field_value[i] = self._propagate_in_place( parent_type_field_identifier, element, current_parameters, use_parent_parameters=use_parent_parameters, ) if excluded_parameter: - current_parameters[field_name] = excluded_parameter + current_parameters[field_name] = ( + field_value if field_name in written_from_parameters else excluded_parameter + ) if current_parameters: - propagated_component[PARAMETERS_STR] = current_parameters + propagated_component[PARAMETERS_STR] = copy.deepcopy(current_parameters) return propagated_component @staticmethod @@ -213,7 +242,7 @@ def _process_nested_components( ) -> Dict[str, Any]: for field_name, field_value in propagated_component.items(): if isinstance(field_value, dict) and field_value.get("type"): - nested_component_with_parameters = self.propagate_types_and_parameters( + nested_component_with_parameters = self._propagate_in_place( parent_field_identifier, field_value, current_parameters, diff --git a/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py b/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py index 0294171d9..ba48c5020 100644 --- a/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py +++ b/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py @@ -19,6 +19,14 @@ _FIELDS_HOLDING_CONFIG_VALUES = frozenset({"config_overrides"}) +def _copy_without_sharing(value: Any) -> Any: + if isinstance(value, dict): + return {key: _copy_without_sharing(item) for key, item in value.items()} + if isinstance(value, list): + return [_copy_without_sharing(item) for item in value] + return deepcopy(value) + + class ManifestReferenceResolver: """ An incoming manifest can contain references to values previously defined. @@ -115,7 +123,7 @@ def preprocess_manifest(self, manifest: Mapping[str, Any]) -> Dict[str, Any]: def _evaluate_node(self, node: Any, manifest: Mapping[str, Any], visited: Set[Any]) -> Any: if isinstance(node, dict): evaluated_dict = { - k: deepcopy(v) + k: _copy_without_sharing(v) if k in _FIELDS_HOLDING_CONFIG_VALUES else self._evaluate_node(v, manifest, visited) for k, v in node.items() diff --git a/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py b/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py index 7f5977d56..60751b983 100644 --- a/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py +++ b/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py @@ -2,12 +2,309 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import copy + import pytest from airbyte_cdk.sources.declarative.parsers.manifest_component_transformer import ( ManifestComponentTransformer, ) +LIST_PARAMETER_CASE = ( + { + "$parameters": { + "name": "users", + "transformations": [{"fields": [], "type": "AddFields"}], + }, + "streams": [{"transformations": [], "retriever": {}}], + "type": "DeclarativeSource", + }, + {}, + None, + { + "$parameters": { + "name": "users", + "transformations": [ + { + "$parameters": {"name": "users"}, + "fields": [], + "name": "users", + "type": "AddFields", + } + ], + }, + "name": "users", + "streams": [ + { + "$parameters": { + "name": "users", + "transformations": [ + { + "$parameters": {"name": "users"}, + "fields": [], + "name": "users", + "type": "AddFields", + } + ], + }, + "name": "users", + "retriever": { + "$parameters": { + "name": "users", + "transformations": [ + { + "$parameters": {"name": "users"}, + "fields": [], + "name": "users", + "type": "AddFields", + } + ], + }, + "name": "users", + "transformations": [ + { + "$parameters": {"name": "users"}, + "fields": [], + "name": "users", + "type": "AddFields", + } + ], + "type": "SimpleRetriever", + }, + "transformations": [ + { + "$parameters": {"name": "users"}, + "fields": [], + "name": "users", + "type": "AddFields", + } + ], + "type": "DeclarativeStream", + } + ], + "transformations": [ + { + "$parameters": {"name": "users"}, + "fields": [], + "name": "users", + "type": "AddFields", + } + ], + "type": "DeclarativeSource", + }, +) + +PROPAGATION_CASES = [ + pytest.param( + { + "type": "DeclarativeStream", + "$parameters": {"primary_key": ["param_pk"]}, + "primary_key": ["own_pk"], + "retriever": {}, + }, + {}, + None, + { + "$parameters": {"primary_key": ["param_pk"]}, + "primary_key": ["own_pk"], + "retriever": { + "$parameters": {"primary_key": ["param_pk"]}, + "primary_key": ["param_pk"], + "type": "SimpleRetriever", + }, + "type": "DeclarativeStream", + }, + id="list-param-existing-truthy-field", + ), + pytest.param( + { + "$parameters": {"limit": 10, "name": "root"}, + "streams": [ + { + "$parameters": {"name": "users"}, + "retriever": {"$parameters": {"limit": 20}, "requester": {}}, + } + ], + "type": "DeclarativeSource", + }, + {}, + None, + { + "$parameters": {"limit": 10, "name": "root"}, + "limit": 10, + "name": "root", + "streams": [ + { + "$parameters": {"limit": 10, "name": "users"}, + "limit": 10, + "name": "users", + "retriever": { + "$parameters": {"limit": 20, "name": "users"}, + "limit": 20, + "name": "users", + "requester": { + "$parameters": {"limit": 20, "name": "users"}, + "limit": 20, + "name": "users", + "type": "HttpRequester", + }, + "type": "SimpleRetriever", + }, + "type": "DeclarativeStream", + } + ], + "type": "DeclarativeSource", + }, + id="parameters-multiple-levels", + ), + pytest.param( + {"$parameters": {"name": "child"}, "retriever": {}, "type": "DeclarativeStream"}, + {"name": "parent"}, + True, + { + "$parameters": {"name": "parent"}, + "name": "parent", + "retriever": { + "$parameters": {"name": "parent"}, + "name": "parent", + "type": "SimpleRetriever", + }, + "type": "DeclarativeStream", + }, + id="parent-parameters-win", + ), + pytest.param( + { + "$parameters": {"discard": 1}, + "properties": {"type": {"type": "string"}}, + "type": "object", + }, + {}, + None, + {"properties": {"type": {"type": "string"}}, "type": "object"}, + id="json-schema-object", + ), + pytest.param( + { + "$parameters": {"discard": 1}, + "properties": {"type": {"type": "string"}}, + "type": ["null", "object"], + }, + {}, + None, + {"properties": {"type": {"type": "string"}}, "type": ["null", "object"]}, + id="json-schema-union", + ), + pytest.param( + { + "$parameters": {"name": "users"}, + "request_parameters": {"query": {"type": "QueryProperties"}}, + "type": "HttpRequester", + }, + {}, + None, + { + "$parameters": {"name": "users"}, + "name": "users", + "request_parameters": { + "query": { + "$parameters": {"name": "users"}, + "name": "users", + "type": "QueryProperties", + } + }, + "type": "HttpRequester", + }, + id="untyped-nested-component", + ), + pytest.param(*LIST_PARAMETER_CASE, id="list-param-falsy-field"), + pytest.param({}, {}, None, {}, id="empty"), +] + + +@pytest.mark.parametrize( + "component, parameters, use_parent_parameters, expected", PROPAGATION_CASES +) +def test_propagation_preserves_original_semantics( + component, parameters, use_parent_parameters, expected +): + original_component = copy.deepcopy(component) + original_parameters = copy.deepcopy(parameters) + transformer = ManifestComponentTransformer() + result = transformer.propagate_types_and_parameters( + "", component, parameters, use_parent_parameters + ) + assert result == expected + assert component == original_component + assert parameters == original_parameters + assert ( + transformer._propagate_in_place( + "", copy.deepcopy(component), copy.deepcopy(parameters), use_parent_parameters + ) + == result + ) + + +def test_list_parameter_fills_missing_stream_field(): + component, parameters, use_parent_parameters, expected = copy.deepcopy(LIST_PARAMETER_CASE) + del component["streams"][0]["transformations"] + original = copy.deepcopy(component) + assert ( + ManifestComponentTransformer().propagate_types_and_parameters( + "", component, parameters, use_parent_parameters + ) + == expected + ) + assert component == original + + +def test_propagate_in_place_reuses_the_input_tree(): + stream = {"type": "DeclarativeStream", "retriever": {"requester": {}}} + tree = {"type": "DeclarativeSource", "streams": [stream]} + retriever = stream["retriever"] + requester = retriever["requester"] + out = ManifestComponentTransformer()._propagate_in_place("", tree, {}) + assert out is tree + assert out["streams"][0] is stream + assert out["streams"][0]["retriever"] is retriever + assert out["streams"][0]["retriever"]["requester"] is requester + + +def _container_ids(value): + if isinstance(value, (dict, list)): + yield id(value) + for child in value.values() if isinstance(value, dict) else value: + yield from _container_ids(child) + + +def test_propagation_breaks_shared_input_containers(): + shared = {"type": "DeclarativeStream", "retriever": {}} + component = { + "type": "DeclarativeSource", + "$parameters": {"options": {"values": [1]}}, + "streams": [shared, shared], + } + parameters = {"extra": [2]} + original = copy.deepcopy(component) + output = ManifestComponentTransformer().propagate_types_and_parameters( + "", component, parameters + ) + ids = list(_container_ids(output)) + assert len(ids) == len(set(ids)) + assert not set(ids).intersection(_container_ids(component)) + assert not set(ids).intersection(_container_ids(parameters)) + output["streams"][0]["$parameters"]["options"]["values"].append(3) + assert output["streams"][1]["$parameters"]["options"]["values"] == [1] + assert component == original + + +def test_propagation_invalid_parameters_preserves_input(): + component = {"type": "DeclarativeSource", "streams": [{"$parameters": 1}]} + original = copy.deepcopy(component) + with pytest.raises(TypeError): + ManifestComponentTransformer().propagate_types_and_parameters("", component, {}) + assert component == original + @pytest.mark.parametrize( "component, expected_component", diff --git a/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py b/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py index 29afa0e70..bb74e9c33 100644 --- a/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py +++ b/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py @@ -3,6 +3,7 @@ # import pytest +import yaml from airbyte_cdk.sources.declarative.parsers.custom_exceptions import ( CircularReferenceException, @@ -16,6 +17,25 @@ resolver = ManifestReferenceResolver() +def test_config_overrides_break_yaml_aliases(): + manifest = yaml.safe_load(""" +check: + type: CheckStream + stream_names: [s] + config_overrides: + a: &x {type: oauth, $parameters: {region: eu}} + b: {nested: *x} +""") + overrides = ManifestReferenceResolver().preprocess_manifest(manifest)["check"][ + "config_overrides" + ] + assert overrides == manifest["check"]["config_overrides"] + assert overrides["a"] == overrides["b"]["nested"] + assert overrides["a"] is not overrides["b"]["nested"] + assert overrides["a"]["$parameters"] is not overrides["b"]["nested"]["$parameters"] + assert overrides["a"] is not manifest["check"]["config_overrides"]["a"] + + def test_refer(): content = {"limit": 50, "limit_ref": "#/limit"} config = resolver.preprocess_manifest(content)