add option to not include parent slice in child stream slice partitions

Author: aldogonzalez8

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3881,6 +3881,11 @@ definitions:
         type: array
         items:
           "$ref": "#/definitions/ParentStreamConfig"
+      include_parent_slice:
+        title: Include Parent Slice
+        description: If False, the parent stream slice will not be included in the child stream slice.
+        type: boolean
+        default: true
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2867,6 +2867,10 @@ class SubstreamPartitionRouter(BaseModel):
         description="Specifies which parent streams are being iterated over and how parent records should be used to partition the child stream data set.",
         title="Parent Stream Configs",
     )
+    include_parent_slice: Optional[bool] = Field(True,
+                                                 description="If False, the parent stream slice will not be included in the child stream slice.",
+                                                 title="Include Parent Slice",
+                                                 )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

@@ -89,6 +89,7 @@ class SubstreamPartitionRouter(PartitionRouter):
     parent_stream_configs: List[ParentStreamConfig]
     config: Config
     parameters: InitVar[Mapping[str, Any]]
+    include_parent_slice: Optional[bool] = True
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         if not self.parent_stream_configs:
@@ -221,11 +222,15 @@ def stream_slices(self) -> Iterable[StreamSlice]:
                             **extracted_extra_fields,
                         }
 
-                    yield StreamSlice(
-                        partition={
+                    slice_partition = {
                             partition_field: partition_value,
-                            "parent_slice": parent_partition or {},
-                        },
+                        }
+
+                    if self.include_parent_slice:
+                        slice_partition["parent_slice"] = parent_partition or {}
+
+                    yield StreamSlice(
+                        partition=slice_partition,
                         cursor_slice={},
                         extra_fields=extracted_extra_fields,
                     )