def METHOD_NAME(
    self, context: OutputContext, table_slice: TableSlice, obj: pd.DataFrame, connection
):
    """Stores the pandas DataFrame in duckdb."""
    connection.execute(
        f"create table if not exists {table_slice.schema}.{table_slice.table} as select * from"
        " obj;"
    )
    if not connection.fetchall():
        # table was not created, therefore already exists. Insert the data
        connection.execute(
            f"insert into {table_slice.schema}.{table_slice.table} select * from obj"
        )
    context.add_output_metadata(
        {
            "row_count": obj.shape[0],
            "dataframe_columns": MetadataValue.table_schema(
                TableSchema(
                    columns=[
                        TableColumn(name=name, type=str(dtype))  # type: ignore  # (bad stubs)
                        for name, dtype in obj.dtypes.items()
                    ]
                )
            ),
        }
    )