Delay Iceberg manifest initialization until after pruning

[Kuinox]

Fixes #995.
This is something I hacked together with codex.
It postpone the InitializeFiles and disable GetCardinality, GetStatistics until it is init.
I don't know what are the implications, but thanks to this patch, I can query my big dataset. There is still some limitation, if the partition I filter against doesn't filter out enough manifest, it still try to do an allocation too big.

I do understand that this PR is maybe disabling features in some scenarios, if there is no better easy fix, we could put this fix when the metastore reach a certain size threshold, with some warning.

[Tmonster]

Thanks for the reported issue and the PR.

Can you try to create a reproducer with just duckdb as well?
Some tips that might work,

You can reduce the memory limit of DuckDB with

set memory_limit='1GB`;

And you can run a number of inserts to build up metadata with

loop i 1 10000

statement ok
INSERT INTO my_table SELECT * FROM ...

endloop

If it is easier to create the table in pyiceberg, that is fine too.

If you can create a reproducer, reviewing this PR and verifying it actually fixes the problem

[Tishj]

I have mixed feelings about this PR, I understand that the filter pushdown produces a new IcebergMultiFileList with these additional filters pushed in, and it's kind of opaque whether any of the other methods are called on the unfiltered multi file list, but this leaves me worried that the InitializeFiles might never be called in some circumstances

[Kuinox]

@Tmonster it doesn't seems to respect the memory limit setting.

I think I managed to reproduce it when creating metadata pyiceberg.

CREATE OR REPLACE TABLE payload AS
SELECT
    file_id::INTEGER AS value,
    file_id::INTEGER AS file_id
FROM range(20000) AS t(file_id);

COPY payload TO 'iceberg_table/data/date=2024-01-01/group_id=1' (
    FORMAT PARQUET,
    PARTITION_BY (file_id),
    WRITE_PARTITION_COLUMNS false,
    OVERWRITE_OR_IGNORE
);

COPY (SELECT 999999::INTEGER AS value) TO 'iceberg_table/data/date=2099-01-01/group_id=999999/rare.parquet' (FORMAT PARQUET);

DROP TABLE payload;

to create an initial dataset
then

from pathlib import Path
import shutil

from pyiceberg.catalog import load_catalog
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.transforms import IdentityTransform
from pyiceberg.types import DateType, IntegerType, NestedField


TABLE = Path("iceberg_table")
CATALOG = Path("catalog.db")
RARE = TABLE / "data" / "date=2099-01-01" / "group_id=999999" / "rare.parquet"

COMMON_MANIFESTS = 499
FILES_PER_MANIFEST = 20_000


if CATALOG.exists():
    CATALOG.unlink()
if (TABLE / "metadata").exists():
    shutil.rmtree(TABLE / "metadata")

catalog = load_catalog(
    "local",
    **{
        "type": "sql",
        "uri": f"sqlite:///{CATALOG.resolve()}",
        "warehouse": str(TABLE),
    },
)
catalog.create_namespace_if_not_exists("repro")

table = catalog.create_table(
    ("repro", "t"),
    location=str(TABLE),
    schema=Schema(
        NestedField(1, "value", IntegerType()),
        NestedField(2, "date", DateType()),
        NestedField(3, "group_id", IntegerType()),
    ),
    partition_spec=PartitionSpec(
        PartitionField(2, 1000, IdentityTransform(), "date"),
        PartitionField(3, 1001, IdentityTransform(), "group_id"),
    ),
)

common_files = sorted((TABLE / "data" / "date=2024-01-01" / "group_id=1").glob("file_id=*/*.parquet"))
if len(common_files) != FILES_PER_MANIFEST:
    raise RuntimeError(f"Expected {FILES_PER_MANIFEST} common parquet files, found {len(common_files)}")
common = [str(path) for path in common_files]
rare = str(RARE)

for manifest_id in range(COMMON_MANIFESTS):
    print(f"adding common manifest {manifest_id + 1}/{COMMON_MANIFESTS}")
    files = common[:FILES_PER_MANIFEST]
    table.add_files(files, check_duplicate_files=False)

print("adding rare manifest")
table.add_files([rare], check_duplicate_files=False)

To create the metadata. It's extremely slow, 50 manifest should be enough to go beyond 1GB of ram usage.
I've put it at 500 in hope to trigger the OOM.

I then run

set memory_limit='500MB';
SET unsafe_enable_version_guessing = true;
SELECT * FROM iceberg_scan('iceberg_table') WHERE "date" = DATE '2099-01-01' AND group_id = 999999;

I'll update when my script finish and I manage to trigger the OOM.

[Kuinox]

So it doesn't OOM with only this, but it manage to fill the RAM and be extremely slow, despite the set memory_limit='500MB';.
Is this enough for you ?

[Tmonster]

Hi @Kuinox ,

Yes, this should be enough to verify on our end. We will take a look

[Tishj]

Can you adjust the test_table_information_requests.test test?
Seems like this results in fewer requests being made in certain cases, which is an unexpected but welcome side-effect 👍

[Kuinox]

Before my patch, DESCRIBE did:

table information request
table metadata JSON request
manifest list Avro request

After patch, DESCRIBE does:

table information request
table metadata JSON request

Given the change it's expected.

Again I want to press that I understand if this patch doesn't looks like the correct direction to the maintainers, if you think this cannot be shipped, tell me what kind of change you want so this kind of datasets can be queried.

[Kuinox]

I have mixed feelings about this PR, I understand that the filter pushdown produces a new IcebergMultiFileList with these additional filters pushed in, and it's kind of opaque whether any of the other methods are called on the unfiltered multi file list, but this leaves me worried that the InitializeFiles might never be called in some circumstances

So there is a tests failing exactly because of that.
How do you want to continue ?

[Tishj]

Could you check if #1066 solves your problem?

[Kuinox]

It still load all the metadata with this PR.

[Tmonster]

Hi @Kuinox, can you rebase & retarget main with this? We are going to limit the PRs to v1.5 now

[Kuinox]

@Tmonster should be good

[Tishj]

I noticed that the equality delete optimizer is breaking your expectations:

 0x000060c00045a540) at iceberg_optimizer.cpp:48:47
   45                           continue;
   46                   }
   47                   auto &iceberg_list = mfbd.file_list->Cast<IcebergMultiFileList>();
-> 48                   auto delete_manifest_entries = iceberg_list.GetDeleteManifestEntries();

This ends up calling initialize (and also happens for an EXPLAIN sadly)

[Kuinox]

I'm totally open to redo a PR for a cleaner way to handle this problem, this was the easy path I could see without understanding more of the architecture.
Any idea how you want this solved ?

[Tishj]

I don't, otherwise I would have mentioned it
The job of that optimizer is to make sure columns we need for an equality delete don't get optimized out. So it has to aggressively scan the list, just in case

I can't think of a reliable way to identify if a scan will have an equality delete without scanning the manifest list+all manifest files