def METHOD_NAME(self) -> Flow:
    df = self.sap_to_df_task.bind(
        query=self.query,
        sep=self.sep,
        func=self.func,
        rfc_total_col_width_character_limit=self.rfc_total_col_width_character_limit,
        flow=self,
    )
    df_mapped = cast_df_to_str.bind(df, flow=self)
    df_with_metadata = add_ingestion_metadata_task.bind(df_mapped, flow=self)
    parquet = df_to_parquet.bind(
        df=df_with_metadata,
        path=self.local_file_path,
        if_exists=self.if_exists,
        flow=self,
    )
    table = self.create_duckdb_table_task.bind(
        path=self.local_file_path,
        schema=self.schema,
        table=self.table,
        if_exists=self.if_exists,
        if_empty=self.if_empty,
        flow=self,
    )
    table.set_upstream(parquet, flow=self)
    if self.update_kv == True:
        set_new_kv.bind(
            kv_name=self.name,
            df=df,
            filter_column=self.filter_column,
            flow=self,
        )
        set_new_kv.set_upstream(table, flow=self)