Writing skipping commits

[adriangb]

I'm interested in writing what I'll call sparse commit ids.

The idea is to get control over concurrency so I can have multiple concurrent writers without commit contention on an append-only table.

My plan is to use a Postgres sequence (or similar from Redis, etc.) to get monotonically increasing commit ids numbers. Each writer asks for a commit id that it now owns then goes and writes out that commit. I'd space them out by 10 (so writers would get commit ids 10, 20, 30, etc. approximately). So they'll be some missing commits, either because of the spacing, because a writer claimed a commit and then failed to write it, or because writer for commit 20 gets there before writer for commit 10. Somewhat to my surprise, readers (or at least delta-rs) do not seem to care if there are missing commits.

This is a all a bit abstract, so here's an example with a home made writer:

import shutil
from dataclasses import dataclass
from io import BytesIO
from pathlib import Path
from time import time
from typing import Any, Iterable
from uuid import uuid4

import polars as pl
from deltalake import DeltaTable
from pydantic_core import to_json


def get_min(x: pl.Series) -> Any:
    try:
        return x.min()
    except pl.exceptions.InvalidOperationError:
        return None


def get_max(x: pl.Series) -> Any:
    try:
        return x.max()
    except pl.exceptions.InvalidOperationError:
        return None



def write_dataframe(
    df: pl.DataFrame,
    commit: int,
    table: Path,
) -> None:
    commit_lines: list[bytes] = []
    min_values = {col: get_min(df[col]) for col in df.columns}
    max_values = {col: get_max(df[col]) for col in df.columns}
    min_values = {col: v for col, v in min_values.items() if v is not None}
    max_values = {col: v for col, v in max_values.items() if v is not None}
    stats = {
        'numRecords': df.height,
        'minValues': min_values,
        'maxValues': max_values,
        'nullCount': {col: df[col].null_count() for col in df.columns},
    }
    out = BytesIO()
    df.write_parquet(out)
    path = f'{commit}-{uuid4()}-0.parquet'
    operation = {
        'add': {
            'path': path,
            'size': out.tell(),
            'modificationTime': round(time()),
            'stats': to_json(stats).decode(),
            'dataChange': True,
            'tags': None,
            'deletionVector': None,
            'baseRowId': None,
            'defaultRwoCommitVersion': None,
            'clusteringProvider': None,
        }
    }
    commit_lines.append(to_json(operation))
    file_path = table / path
    file_path.parent.mkdir(parents=True, exist_ok=True)
    out.seek(0)
    file_path.write_bytes(out.getbuffer())

    commit_info = {
        'commitInfo': {
            'timestamp': round(time()),
            'operation': 'WRITE',
            'operationParameters': {
                'mode': 'append',
            },
            'clientVersion': 'test.0.0.0',
        }
    }
    commit_lines.append(to_json(commit_info))
    commit_data = b'\n'.join(commit_lines)
    commit_path = table / '_delta_log' / f'{commit:020d}.json'
    commit_path.write_bytes(commit_data)



def generate_dataframe(commit: int) -> pl.DataFrame:
    return pl.DataFrame(
        {
            'part': [1, 2] * 1000,
            'commit': [commit] * 2000,
            'x': range(2000)
        }
    )


@dataclass
class Test:
    latest_commit: int
    compaction_run_at: int
    compaction_run_against: int
    checkpoint_commit_run_at: int
    checkpoint_commit_run_against: int


def make_commit_sequence() -> Iterable[int]:
    return iter(range(1, 50, 10))


def run_test(num: int, test: Test, test_path_dir: Path) -> pl.DataFrame:
    test_path = test_path_dir / str(num)
    shutil.rmtree(test_path, ignore_errors=True)
    test_path.mkdir(parents=True, exist_ok=True)
    table = DeltaTable.create(test_path, schema=schema)

    for commit in make_commit_sequence():
        df = generate_dataframe(commit)
        write_dataframe(df, commit, test_path)

        if commit == test.compaction_run_at:
            table.load_as_version(test.compaction_run_against)
            metrics = table.optimize.compact(target_size=1024 * 1024 * 10)
            if len(table.files()) > 1:
                assert metrics['numFilesAdded'] >= 0

        if commit == test.checkpoint_commit_run_at:
            table.load_as_version(test.compaction_run_against)
            table.create_checkpoint()


    df = pl.scan_delta(str(test_path))
    return df.collect()



def worker(num: int, test: Test, expected: pl.DataFrame, total: int, test_path_dir: Path) -> None:
    print(f'Running test {num} of {total}')
    print(test)
    df = run_test(num, test, test_path_dir)
    df.sort(df.columns).equals(expected.sort(expected.columns))



if __name__ == '__main__':

    tests: list[Test] = []

    latest_commit = 0

    for latest_commit in make_commit_sequence():
        for compaction_run_at in range(1, latest_commit + 1, 10):
            for compaction_run_against in range(1, latest_commit + 1, 10):
                for checkpoint_commit_run_at in range(1, compaction_run_at + 1, 10):
                    for checkpoint_commit_run_against in range(1, checkpoint_commit_run_at + 1, 10):
                        tests.append(
                            Test(
                                latest_commit=latest_commit,
                                compaction_run_at=compaction_run_at,
                                compaction_run_against=compaction_run_against,
                                checkpoint_commit_run_at=checkpoint_commit_run_at,
                                checkpoint_commit_run_against=checkpoint_commit_run_against,
                            )
                        )


    LAST_COMMIT = latest_commit

    print(f'Total tests: {len(tests)}')

    test_path_dir = Path('./data')

    shutil.rmtree(test_path_dir, ignore_errors=True)

    schema = generate_dataframe(0).to_arrow().schema

    expected = run_test(
        0,
        Test(
            latest_commit=LAST_COMMIT,
            compaction_run_at=LAST_COMMIT,
            compaction_run_against=LAST_COMMIT,
            checkpoint_commit_run_at=LAST_COMMIT,
            checkpoint_commit_run_against=LAST_COMMIT,
        ),
        test_path_dir,
    )
    for num, test in enumerate(tests):
        worker(num, test, expected, len(tests), test_path_dir)

This example seeks to prove that with this approach you always end up with the same table no matter in what order you run the operations. Realistically things wouldn't be happening this out of order, something maybe realistic would be:

sequenceDiagram
    participant Writer 1
    participant Writer 2
    participant Writer 3
    participant S3
    participant Postgres
    participant Maintenance
    Writer 1-->>Postgres: Get commit 10
    Writer 2-->>Postgres: Get Commit 20
    Writer 2->>S3: Write 2.parquet
    Writer 2->>S3: Write 20.json
    Writer 1->>S3: Write 1.parquet
    Writer 1->>S3: Write 10.json
    Writer 3-->>Postgres: Get commit
    Writer 3->>S3: Write 3.parquet
    Maintenance->>S3: Compaction @ 20.json (add 12.parquet, remove 1.parquet and 2.parquet)
    Maintenance->>S3: 21.json
    Writer 3->>S3: Write 30.json
    Writer 2->>S3: 20.checkpoint.parquet
    Writer 1->>S3: 10.checkpoint.parquet

Loading

The protocol explicitly states that checkpoints can be run at any time. The one caveat with this setup is that all commits older than the checkpoint commit must have been committed or they'll be lost forever. I'm okay with that, I can just wait 10s or so before writing a checkpoint.

Compaction is a bit trickier: you can't have two compactions running concurrently on different commits, otherwise they might delete the same files. Since compaction can be a long running operation and isn't happening that frequently I'm okay running it out of band and ensuring only 1 compaction per table runs at a time.

[ion-elgreco]

I think you came with a similar idea which is going to land in the protocol: delta-io/delta#2598, you might jump in that discussion since this is something that we decide on the protocol level and not implementation

[adriangb]

Makes sense. I had seen that issue but haven't read the attached doc, I will read and try to grok. Either way, even if that lands it will take a while so I would probably still go with my approach and upgrade later.

Are there any issues you can think of with the commit skipping / backfilling?

[ion-elgreco]

Missing commits won't work I believe, you can't validate the table state whether it's correct or not. It might be missing because someone deleted a commit, that we won't know now anymore.

The discussion in that RFC issue is I think the right place to discuss this though, it might already solve your problem, or you might be able to bring your inputs to influence the design.

The moment it's in the protocol, it's at everyone's leisure to implement it

[adriangb]

Missing commits won't work I believe, you can't validate the table state whether it's correct or not. It might be missing because someone deleted a commit, that we won't know now anymore.

But does any tooling actually do this? delta-rs seems not to 😄

[ion-elgreco]

We don't as of now :), but there are still many things we don't properly do yet, especially around configuration handling.

Delta-Spark might do this though, so you can check their conflict resolution implementation