Refactor parquet dataloader

[zyaoj]

What does this PR do? Please describe:
The first attempt to extract and migrate generic parquet dataloader from MERES to fairseq2.

• Excalidraw board for brainstorming
• Sibling PR in MERES
• Draft RFC as a semi-official design doc

Does your PR introduce any breaking changes? If yes, please list them:
N/A

Check list:

• Was the content of this PR discussed and approved via a GitHub issue? (no need for typos or documentation improvements)
• Did you read the contributor guideline?
• Did you make sure that your PR does only one thing instead of bundling different changes together?
• Did you make sure to update the documentation with your changes? (if necessary)
• Did you write any new necessary tests?
• Did you verify new and existing tests pass locally with your changes?
• Did you update the CHANGELOG? (no need for typos, documentation, or minor internal changes)

[avidale]

Is it possible to apply filters that use non-partition columns (e.g. some numeric scores computed within a dataset)?

[zyaoj]

I could think of some simple example like:

import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.compute as pc
import tempfile
from pathlib import Path

table = pa.Table.from_pydict(
    {
        "col1": [1, 2, 3, 4, 5],
        "col2": ["a", "b", "c", "d", "e"],
        "col3": [1.1, 2.2, 3.3, 4.4, 5.5],
    }
)

# Create a temporary directory and file
with tempfile.TemporaryDirectory() as temp_dir:
    file_path = Path(temp_dir) / "test.parquet"

    # Write the parquet file
    pq.write_table(table, file_path)

    filters = pq.filters_to_expression(eval("pc.field('col1') == pc.scalar(1)"))  # the way we do the filter conversion in this PR
    dataset = pq.ParquetDataset(
        file_path, filters=filters, filesystem=None
    )
    df = dataset.read().to_pandas()

Haven't tried more advanced lambda functions that dynamically computes some values yet.

[artemru]

yes, it's possible to pass this filter expression to FragmentLoadingConfig(filters ="(pc.field('col1') == 1) & (pc.field('col2') > 1)") which will apply the filter in memory just after loading the table. This filters should be expressed in terms of the original dataset name space (not after renaming).
https://github.com/facebookresearch/fairseq2/pull/867/files#diff-5965d7b9782e8b1d6ee5af20dd9dd95e4e26d6f4bfc6262645f8bbb2fe5bb239R147
you can pass them as string (to do python eval) or directly as pa.dataset.Expression

[artemru]

so, we offer both partition filters (at fragment scheduling time) and any type of row filters (after downloading)

[artemru]

but SafeFragment will do an auto retry in case of lost connection!