kartothek 4.0 update_dataset_from_ddf corrupts datasets if a different table name is used

[peter-hoffmann-by]

Problem description

Kartothek 4.0 breaks updating existing datasets that use a different table name then the default.

Expected Behaviour

the table name keyword is respected in updating datasets.

Example code

import dask
dask.config.set(scheduler='synchronous')
import dask.dataframe as dd

ddf = dask.datasets.timeseries()
ddf = ddf.reset_index()
print(ddf.head())
ddf['date'] = ddf.timestamp.dt.date
ddf = ddf.loc[:10]
store_url = f"hfs://testdata"

store = get_store_from_url(store_url)
delayed = update_dataset_from_ddf(ddf, store, "testdata", table='predictions', partition_on=['date'], shuffle=True, num_buckets=5)
res = delayed.compute()

Output:

$ tree testdata/testdata 
testdata/testdata 
└── table 
    ├── _common_metadata 
    ├── date=2000-01-01 
    │   ├── 414bb87213014db0b5d34a68e4a03edb.parquet 
    │   ├── 76ab5551c3dc4929b2676b04f2f972ac.parquet
    │   ├── b0614814e3814aecb76705a0f8e5b238.parquet 
    │   ├── eb430028a14644ac845c036b619fff47.parquet 
    │   └── f886ae72fa824827ad850f4af941b8b0.parquet 

Expected Output:

testdata/testdata 
  └── predictions 

The example above is just a minimal example to show that the the table name is not used in update_dataset_from_ddf. The corruption of dataset occurs if you update an existing (pre 4.0 kartothek dataset) with the new release and you suddenly have two table name in your dataset ('predictions' and 'table').

import dask
import pandas as pd
import dask.dataframe as dd
from datetime import date
from storefact import get_store_from_url
from kartothek.io.dask.dataframe import update_dataset_from_ddf, read_dataset_as_ddf

dask.config.set(scheduler='synchronous')
store_url = f"hfs://testdata"
store = get_store_from_url(store_url)


def create():
    df = pd.DataFrame({"date": [date(2021,1,x) for x in range(1,6)], 'value': range(5)})
    ddf = dask.dataframe.from_pandas(df, npartitions=1)
    delayed = update_dataset_from_ddf(ddf, store, dataset_uuid, table='predictions', partition_on=['date'])
    res = delayed.compute()

def update():
    df = pd.DataFrame({"date": [date(2021,1,x) for x in range(6,10)], 'value': range(5)})
    ddf = dask.dataframe.from_pandas(df, npartitions=1)
    delayed = update_dataset_from_ddf(ddf, store, dataset_uuid, table='predictions', partition_on=['date'])
    res = delayed.compute()

def validate():
    dataset_uuid = "testdata"
    ddf = read_dataset_as_ddf(dataset_uuid, store, "predictions")
    df = ddf.compute()
    print(df)

if __name__ == '__main__':
    #first run with kartothek3
    create()

    #then run with kartothek4
    #update()
    #validate()

raises the following exception

raceback (most recent call last):
  File "test_kartothek_bug.py", line 48, in <module>
    validate()
  File "test_kartothek_bug.py", line 42, in validate
    ddf = read_dataset_as_ddf(dataset_uuid, store, "predictions")
  File "<decorator-gen-7>", line 2, in read_dataset_as_ddf
  File "python3.8/site-packages/kartothek/io_components/utils.py", line 277, in normalize_args
    return _wrapper(*args, **kwargs)
  File "python3.8/site-packages/kartothek/io_components/utils.py", line 275, in _wrapper
    return function(*args, **kwargs)
  File "python3.8/site-packages/kartothek/io/dask/dataframe.py", line 113, in read_dataset_as_ddf
    delayed_partitions = read_dataset_as_delayed(
  File "python3.8/site-packages/kartothek/io/dask/delayed.py", line 239, in read_dataset_as_delayed
    mps = read_dataset_as_delayed_metapartitions(
  File "<decorator-gen-5>", line 2, in read_dataset_as_delayed_metapartitions
  File "python3.8/site-packages/kartothek/io_components/utils.py", line 277, in normalize_args
    return _wrapper(*args, **kwargs)
  File "python3.8/site-packages/kartothek/io_components/utils.py", line 275, in _wrapper
    return function(*args, **kwargs)
  File "/Users/phoffmann/code/macenv/venv/lib/python3.8/site-packages/kartothek/io/dask/delayed.py", line 217, in read_dataset_as_delayed_metapartitions
    return list(mps)
  File "python3.8/site-packages/kartothek/io_components/read.py", line 102, in dispatch_metapartitions_from_factory
    yield MetaPartition.from_partition(
  File "python3.8/site-packages/kartothek/io_components/metapartition.py", line 426, in from_partition
    file=partition.files[table_name],
KeyError: 'predictions'

Dataset layout

$ tree testdata/testdata
testdata/testdata
├── predictions
│   ├── _common_metadata
│   ├── date=2021-01-01
│   │   └── 160182cebd724a8899f1656bdc6d7628.parquet
│   ├── date=2021-01-02
│   │   └── 2859042309f3498f84efdcf3626a8c93.parquet
│   ├── date=2021-01-03
│   │   └── abca32a940e044fc962da03b1fe55bd5.parquet
│   ├── date=2021-01-04
│   │   └── e133c297ef6c44b4bdcabb92af6567a5.parquet
│   └── date=2021-01-05
│       └── d80e42f70fe2425aa623cc342fb9b1e9.parquet
└── table
    ├── date=2021-01-06
    │   └── e9c0d16f75f14454bfdba8377c34df74.parquet
    ├── date=2021-01-07
    │   └── 25c8c53f0d79410290c9b5427a07da09.parquet
    ├── date=2021-01-08
    │   └── 05c191f27e444244b8275296d1302c32.parquet
    ├── date=2021-01-09
    │   └── 42e38bcd6ab54bae83cffe2dd1548e26.parquet
    └── date=2021-01-10
        └── 40f8305134b74b5f85b356417b027959.parquet

Used versions

kartothek==4.0.0

[peter-hoffmann-by]

The bug is in Line https://github.com/JDASoftwareGroup/kartothek/blob/master/kartothek/io_components/metapartition.py#L468 which drops the table_name silently and falls back to the default. Constructor should add

        table_name=metapartition.table_name

[stephan-hesselmann-by]

Fixed by #451