def METHOD_NAME(
    self, config: Mapping[str, Any], configured_catalog: ConfiguredAirbyteCatalog, input_messages: Iterable[AirbyteMessage]
) -> Iterable[AirbyteMessage]:
    """
    Reads the input stream of messages, config, and catalog to write data to the destination.
    This method returns an iterable (typically a generator of AirbyteMessages via yield) containing state messages received
    in the input message stream. Outputting a state message means that every AirbyteRecordMessage which came before it has been
    successfully persisted to the destination. This is used to ensure fault tolerance in the case that a sync fails before fully completing,
    then the source is given the last state message output from this method as the starting point of the next sync.
    :param config: dict of JSON configuration matching the configuration declared in spec.json
    :param configured_catalog: The Configured Catalog describing the schema of the data being received and how it should be persisted in the
                                destination
    :param input_messages: The stream of input messages received from the source
    :return: Iterable of AirbyteStateMessages wrapped in AirbyteMessage structs
    """
    streams = {s.stream.name for s in configured_catalog.streams}
    with establish_connection(config) as connection:
        writer = create_firebolt_wirter(connection, config, logger)
        for configured_stream in configured_catalog.streams:
            if configured_stream.destination_sync_mode == DestinationSyncMode.overwrite:
                writer.delete_table(configured_stream.stream.name)
                logger.info(f"Stream {configured_stream.stream.name} is wiped.")
            writer.create_raw_table(configured_stream.stream.name)
        for message in input_messages:
            if message.type == Type.STATE:
                yield message
            elif message.type == Type.RECORD:
                data = message.record.data
                stream = message.record.stream
                # Skip unselected streams
                if stream not in streams:
                    logger.debug(f"Stream {stream} was not present in configured streams, skipping")
                    continue
                writer.queue_write_data(stream, str(uuid4()), datetime.now(), json.dumps(data))
        # Flush any leftover messages
        writer.flush()