) -> Dict[str, Any]:
        assert isinstance(input, dict)
        return {
            **input,
            **self.mapper.invoke(input, config, **kwargs),
        }
[docs]    async def ainvoke(
        self,
        input: Dict[str, Any],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        assert isinstance(input, dict)
        return {
            **input,
            **await self.mapper.ainvoke(input, config, **kwargs),
        }
[docs]    def transform(
        self,
        input: Iterator[Dict[str, Any]],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Iterator[Dict[str, Any]]:
        # collect mapper keys
        mapper_keys = set(self.mapper.steps.keys())
        # create two streams, one for the map and one for the passthrough
        for_passthrough, for_map = safetee(input, 2, lock=threading.Lock())
        # create map output stream
        map_output = self.mapper.transform(for_map, config, **kwargs)
        # get executor to start map output stream in background
        with get_executor_for_config(config or {}) as executor:
            # start map output stream
            first_map_chunk_future = executor.submit(next, map_output)  # type: ignore
            # consume passthrough stream
            for chunk in for_passthrough:
                assert isinstance(chunk, dict)
                # remove mapper keys from passthrough chunk, to be overwritten by map
                filtered = AddableDict(
                    {k: v for k, v in chunk.items() if k not in mapper_keys}
                )
                if filtered:
                    yield filtered