TableDataset

TableDataset is used to load and save data to tables using the Ibis framework.

kedro_datasets.ibis.TableDataset

TableDataset(
    *,
    table_name,
    database=None,
    connection=None,
    credentials=None,
    load_args=None,
    save_args=None,
    metadata=None
)

Bases: ConnectionMixin, AbstractDataset[Table, Table]

TableDataset loads/saves data from/to Ibis table expressions.

Examples:

Using the YAML API:

cars:
  type: ibis.TableDataset
  table_name: cars
  connection:
    backend: duckdb
    database: company.db
  save_args:
    materialized: table
    mode: append

boats:
  type: ibis.TableDataset
  table_name: boats
  connection:
    backend: duckdb
    database: company.db
  save_args:
    materialized: table
    mode: upsert
    on: id # The 'on' argument is only accepted for upserts.

motorbikes:
  type: ibis.TableDataset
  table_name: motorbikes
  connection:
    backend: duckdb
    database: company.db
  save_args:
    materialized: view
    mode: overwrite

Using the Python API:

>>> import ibis
>>> from kedro_datasets.ibis import TableDataset
>>>
>>> data = ibis.memtable({"col1": [1, 2], "col2": [4, 5], "col3": [5, 6]})
>>>
>>> dataset = TableDataset(
...     table_name="test",
...     connection={"backend": "duckdb", "database": tmp_path / "file.db"},
...     save_args={"materialized": "table"},
... )
>>> dataset.save(data)
>>> reloaded = dataset.load()
>>> assert data.execute().equals(reloaded.execute())

TableDataset connects to the Ibis backend object constructed from the connection configuration. The backend key provided in the config can be any of the supported backends. The remaining dictionary entries will be passed as arguments to the underlying connect() method (e.g. ibis.duckdb.connect()).

The dataset establishes a connection to the relevant table for the execution backend. Therefore, Ibis doesn't fetch data on load; all compute is deferred until materialization, when the expression is saved. In practice, this happens when another TableDataset instance is saved, after running code defined across one more more nodes.

Parameters:

• table_name (str) –

  The name of the table or view to read or create.

• database (str | None, default: None ) –

  The name of the database to read the table or view from or create the table or view in. If not passed, then the current database is used. Provide a tuple of strings (e.g. ("catalog", "database")) or a dotted string path (e.g. "catalog.database") to reference a table or view in a multi-level table hierarchy.

• connection (dict[str, Any] | None, default: None ) –

  Configuration for connecting to an Ibis backend. If not provided, connect to DuckDB in in-memory mode.

• credentials (dict[str, Any] | None, default: None ) –

  Credentials or additional configuration used to connect (e.g. user, password, token, account). If given, these values override the base connection configuration.

• load_args (dict[str, Any] | None, default: None ) –

  Additional arguments passed to the Ibis backend's table method.

• save_args (dict[str, Any] | None, default: None ) –

  Additional arguments passed to the Ibis backend's create_{materialized} method. By default, ir.Table objects are materialized as views. To save a table using a different materialization strategy, supply a value for materialized in save_args. The mode parameter controls the behavior when saving data: - "overwrite": Overwrite existing data in the table. - "append": Append contents of the new data to the existing table (does not overwrite). - "error" or "errorifexists": Throw an exception if the table already exists. - "ignore": Silently ignore the operation if the table already exists. - "upsert": Update existing rows and insert new rows.

• metadata (dict[str, Any] | None, default: None ) –

  Any arbitrary metadata. This is ignored by Kedro, but may be consumed by users or external plugins.

Source code in kedro_datasets/ibis/table_dataset.py

def __init__(  # noqa: PLR0913
    self,
    *,
    table_name: str,
    database: str | None = None,
    connection: dict[str, Any] | None = None,
    credentials: dict[str, Any] | None = None,
    load_args: dict[str, Any] | None = None,
    save_args: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Creates a new ``TableDataset`` pointing to a table.

    ``TableDataset`` connects to the Ibis backend object constructed
    from the connection configuration. The `backend` key provided in
    the config can be any of the
    [supported backends](https://ibis-project.org/install). The
    remaining dictionary entries will be passed as arguments to the
    underlying ``connect()`` method (e.g.
    [ibis.duckdb.connect()](https://ibis-project.org/backends/duckdb#ibis.duckdb.connect)).

    The dataset establishes a connection to the relevant table for the execution
    backend. Therefore, Ibis doesn't fetch data on load; all compute
    is deferred until materialization, when the expression is saved.
    In practice, this happens when another ``TableDataset`` instance
    is saved, after running code defined across one more more nodes.

    Args:
        table_name: The name of the table or view to read or create.
        database: The name of the database to read the table or view
            from or create the table or view in. If not passed, then
            the current database is used. Provide a tuple of strings
            (e.g. `("catalog", "database")`) or a dotted string path
            (e.g. `"catalog.database"`) to reference a table or view
            in a multi-level table hierarchy.
        connection: Configuration for connecting to an Ibis backend.
            If not provided, connect to DuckDB in in-memory mode.
        credentials: Credentials or additional configuration used to
            connect (e.g. user, password, token, account). If given,
            these values override the base connection configuration.
        load_args: Additional arguments passed to the Ibis backend's
            `table` method.
        save_args: Additional arguments passed to the Ibis backend's
            `create_{materialized}` method. By default, ``ir.Table``
            objects are materialized as views. To save a table using
            a different materialization strategy, supply a value for
            `materialized` in `save_args`. The `mode` parameter controls
            the behavior when saving data:
            - _"overwrite"_: Overwrite existing data in the table.
            - _"append"_: Append contents of the new data to the existing table (does not overwrite).
            - _"error"_ or _"errorifexists"_: Throw an exception if the table already exists.
            - _"ignore"_: Silently ignore the operation if the table already exists.
            - _"upsert"_: Update existing rows and insert new rows.
        metadata: Any arbitrary metadata. This is ignored by Kedro,
            but may be consumed by users or external plugins.
    """

    self._table_name = table_name
    self._database = database
    _connection_config = connection or self.DEFAULT_CONNECTION_CONFIG
    _credentials = deepcopy(credentials) or {}
    self._connection_config = {**_connection_config, **_credentials}
    self.metadata = metadata

    # Set load and save arguments, overwriting defaults if provided.
    self._load_args = deepcopy(self.DEFAULT_LOAD_ARGS)
    if load_args is not None:
        self._load_args.update(load_args)
    if database is not None:
        self._load_args["database"] = database

    self._save_args = deepcopy(self.DEFAULT_SAVE_ARGS)
    if save_args is not None:
        self._save_args.update(save_args)
    if database is not None:
        self._save_args["database"] = database

    self._materialized = self._save_args.pop("materialized")

    # Handle mode/overwrite conflict.
    if save_args and "mode" in save_args and "overwrite" in self._save_args:
        raise ValueError("Cannot specify both 'mode' and deprecated 'overwrite'.")

    # Map legacy overwrite if present.
    if "overwrite" in self._save_args:
        warn(
            "'overwrite' is deprecated and will be removed in a future release. "
            "Please use 'mode' instead.",
            KedroDeprecationWarning,
            stacklevel=2,
        )
        legacy = self._save_args.pop("overwrite")
        # Remove any lingering 'mode' key from defaults to avoid
        # leaking into writer kwargs.
        del self._save_args["mode"]
        mode = "overwrite" if legacy else "error"
    else:
        mode = self._save_args.pop("mode")

    self._mode = SaveMode(mode)

DEFAULT_CONNECTION_CONFIG class-attribute

DEFAULT_CONNECTION_CONFIG = {
    "backend": "duckdb",
    "database": ":memory:",
}

DEFAULT_LOAD_ARGS class-attribute

DEFAULT_LOAD_ARGS = {}

DEFAULT_SAVE_ARGS class-attribute

DEFAULT_SAVE_ARGS = {
    "materialized": "view",
    "mode": "overwrite",
}

_CONNECTION_GROUP class-attribute

_CONNECTION_GROUP = 'ibis'

_connection_config instance-attribute

_connection_config = {
    None: _connection_config,
    None: _credentials,
}

_database instance-attribute

_database = database

_load_args instance-attribute

_load_args = deepcopy(DEFAULT_LOAD_ARGS)

_materialized instance-attribute

_materialized = pop('materialized')

_mode instance-attribute

_mode = SaveMode(mode)

_save_args instance-attribute

_save_args = deepcopy(DEFAULT_SAVE_ARGS)

_table_name instance-attribute

_table_name = table_name

connection property

connection

The Backend instance for the connection configuration.

metadata instance-attribute

metadata = metadata

_connect

_connect()

Source code in kedro_datasets/ibis/table_dataset.py

def _connect(self) -> BaseBackend:
    import ibis  # noqa: PLC0415

    config = deepcopy(self._connection_config)
    backend = getattr(ibis, config.pop("backend"))
    return backend.connect(**config)

_describe

_describe()

Source code in kedro_datasets/ibis/table_dataset.py

def _describe(self) -> dict[str, Any]:
    load_args = deepcopy(self._load_args)
    save_args = deepcopy(self._save_args)
    load_args.pop("database", None)
    save_args.pop("database", None)
    return {
        "table_name": self._table_name,
        "database": self._database,
        "backend": self._connection_config["backend"],
        "load_args": load_args,
        "save_args": save_args,
        "materialized": self._materialized,
        "mode": self._mode,
    }

_exists

_exists()

Source code in kedro_datasets/ibis/table_dataset.py

def _exists(self) -> bool:
    return (
        self._table_name is not None
        and self._table_name in self.connection.list_tables(database=self._database)
    )

load

load()

Source code in kedro_datasets/ibis/table_dataset.py

def load(self) -> ir.Table:
    return self.connection.table(self._table_name, **self._load_args)

save

save(data)

Source code in kedro_datasets/ibis/table_dataset.py

def save(self, data: ir.Table) -> None:
    writer = getattr(self.connection, f"create_{self._materialized}")
    if self._mode in {"append", "upsert"}:
        if not self._exists():
            save_args = self._save_args
            if self._mode == "upsert" and "on" in save_args:
                # The 'on' argument is only accepted for upserts, so
                # we remove it if we're falling back to creating the
                # database object.
                save_args = {k: v for k, v in save_args.items() if k != "on"}

            writer(self._table_name, data, overwrite=False, **save_args)
        elif hasattr(
            self.connection,
            method_name := "insert" if self._mode == "append" else "upsert",
        ):
            method = getattr(self.connection, method_name)
            method(self._table_name, data, **self._save_args)
        else:
            raise DatasetError(
                f"The {self.connection.name} backend for Ibis does "
                f"not support {method_name}s."
            )
    elif self._mode == "overwrite":
        writer(self._table_name, data, overwrite=True, **self._save_args)
    elif self._mode in {"error", "errorifexists"}:
        writer(self._table_name, data, overwrite=False, **self._save_args)
    elif self._mode == "ignore" and not self._exists():
        writer(self._table_name, data, overwrite=False, **self._save_args)