[C++][Python][Parquet] Implement Content-Defined Chunking for the Parquet writer

[kszucs]

Rationale for this change

I have been working on to improve Parquet's deduplication efficiency for content-addressable storages. These system generally use some kind of a CDC algorithm which are better suited for uncompressed row-major formats. Although thanks to Parquet's unique features I was able to reach good deduplication results by consistently chunking data pages by maintaining a gearhash based chunker for each column.

A purpose built evaluation tool is available at https://github.com/kszucs/de

Deduplication efficiency for all revisions of openfoodfacts/product-database/food.parquet

❯ GIT_LFS_SKIP_SMUDGE=1 git clone https://huggingface.co/datasets/openfoodfacts/product-database.git ~/Datasets/product-database
❯ de revisions -d /tmp/food ~/Datasets/product-database/food.parquet
food.parquet has 32 revisions
Checking out 2e19b51
Checking out 1f84d31
Checking out d31d108
Checking out 9cd809c
Checking out 41e5f38
Checking out 9a30ddd
...

❯ de stats /tmp/food 
Estimating deduplication for Parquet
Estimating deduplication for CDC Snappy
Estimating deduplication for CDC ZSTD
┏━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃            ┃            ┃            ┃             ┃             ┃ Compressed ┃
┃            ┃            ┃            ┃  Compressed ┃ Deduplicat… ┃ Deduplica… ┃
┃ Title      ┃ Total Size ┃ Chunk Size ┃  Chunk Size ┃       Ratio ┃      Ratio ┃
┡━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Parquet    │  182.6 GiB │  148.0 GiB │   140.5 GiB │         81% │        77% │
│ CDC Snappy │  178.3 GiB │   75.5 GiB │    73.0 GiB │         42% │        41% │
│ CDC ZSTD   │  109.6 GiB │   55.9 GiB │    55.6 GiB │         51% │        51% │
└────────────┴────────────┴────────────┴─────────────┴─────────────┴────────────┘

What changes are included in this PR?

Are these changes tested?

Not yet.

Are there any user-facing changes?

There is a new parquet writer property called content_defined_chunking which is subject to renaming.

[github-actions]

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose

Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename the pull request title in the following format?

GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

See also:

• Other pull requests
• Contribution Guidelines - Contributing Overview

[kszucs]

Skipping bytes until the min size is reached speeds up the boundary detection quite a lot.

[kszucs]

These defaults values are subject to change, especially because the default maximum page size is 1Mi.

[mapleFU]

Is cdc a part of the parquet spec? Or is it a poc?

[kszucs]

Is cdc a part of the parquet spec? Or is it a poc?

It is not. You can think of it as an implementation specific feature similar to the existing options to specify how record batches and pages are being split.

[kszucs]

Need to find a better way to embed these constants, I used the following script to generate this:

import hashlib
import sys


def gear(n: int, seed: int):
    value = bytes([seed] * 64 + [n] * 64)
    hasher = hashlib.md5(value)
    return hasher.hexdigest()[:16]


def print_table(seed: int, length=256, comma=True):
    table = [gear(n, seed=seed) for n in range(length)]
    print(f"{{ // seed = {seed}")
    for i in range(0, length, 4):
        print("  ", end="")
        values = [f"0x{value}" for value in table[i:i + 4]]
        values = ", ".join(values)
        print(f"  {values}", end=",\n" if i < length - 4 else "\n")
    print(" }", end=", " if comma else "")



if __name__ == "__main__":
    print("{")
    n = int(sys.argv[1])
    for seed in range(n):
        print_table(seed, comma=seed < n)
    print("}")

[kszucs]

Maybe we should expose min and max chunk size again, these factors try to normalize the chunk size.