Add dynamic_streams_name only for full_resolve command

Author: lazebnyi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1427,13 +1427,6 @@ definitions:
             - "$ref": "#/definitions/LegacyToPerPartitionStateMigration"
             - "$ref": "#/definitions/CustomStateMigration"
         default: []
-      dynamic_stream_name:
-        title: Dynamic Stream Name
-        description: The dynamic stream name that create current stream, if None is static stream.
-        type: string
-        default: None
-        example:
-          - "Tables"
       $parameters:
         type: object
         additional_properties: true

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -119,7 +119,7 @@ def message_repository(self) -> MessageRepository:
 
     @property
     def dynamic_streams(self) -> List[Dict[str, Any]]:
-        return self._dynamic_stream_configs(manifest=self._source_config, config=self._config)
+        return self._dynamic_stream_configs(manifest=self._source_config, config=self._config, with_dynamic_stream_name=True)
 
     @property
     def connection_checker(self) -> ConnectionChecker:
@@ -353,7 +353,7 @@ def _stream_configs(self, manifest: Mapping[str, Any]) -> List[Dict[str, Any]]:
         return stream_configs
 
     def _dynamic_stream_configs(
-        self, manifest: Mapping[str, Any], config: Mapping[str, Any]
+        self, manifest: Mapping[str, Any], config: Mapping[str, Any], with_dynamic_stream_name: Optional[bool] = None
     ) -> List[Dict[str, Any]]:
         dynamic_stream_definitions: List[Dict[str, Any]] = manifest.get("dynamic_streams", [])
         dynamic_stream_configs: List[Dict[str, Any]] = []
@@ -398,9 +398,10 @@ def _dynamic_stream_configs(
                 # Ensure that each stream is created with a unique name
                 name = dynamic_stream.get("name")
 
-                dynamic_stream["dynamic_stream_name"] = dynamic_definition.get(
-                    "name", f"dynamic_stream_{dynamic_definition_index}"
-                )
+                if with_dynamic_stream_name:
+                    dynamic_stream["dynamic_stream_name"] = dynamic_definition.get(
+                        "name", f"dynamic_stream_{dynamic_definition_index}"
+                    )
 
                 if not isinstance(name, str):
                     raise ValueError(

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2049,12 +2049,6 @@ class Config:
         description="Array of state migrations to be applied on the input state",
         title="State Migrations",
     )
-    dynamic_stream_name: Optional[str] = Field(
-        "None",
-        description="The dynamic stream name that create current stream, if None is static stream.",
-        example=["Tables"],
-        title="Dynamic Stream Name",
-    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")