feat(flink): add read_***() for Flink backend

[mfatihaktas]

Adds the following functions in ibis/backends/flink/__init__.py:

• register()
• read_file()
• read_parquet()
• read_csv()
• read_json()

Addition of these functions clears several test functions in test_param.py and test_register.py.

Edit: Removed register() per this comment.

[cpcloud]

Let's avoid adding register for now. It's a bit crufty and from a time when we thought we might be able to unify all the read_* methods under a single read(...) function.

[mfatihaktas]

Let's avoid adding register for now. It's a bit crufty and from a time when we thought we might be able to unify all the read_* methods under a single read(...) function.

Thanks for the note, removed register().

[jcrist]

Does flink support natively loading directly from files? If so, I think we should use that.

If it doesn't, I'm not sure how I feel about automatically loading files using pandas and forwarding them that way. We do something similar for local in-memory backends like duckdb/pandas where file paths are unambiguously local. But for a potentially distributed system like flink, automatically using a local file reader may have unexpected behavior with path names, and also may be inefficient.

If we do decide to handle file reading manually using an in-memory reader here, we should use the ones provided by pyarrow directly, not pandas. These have builtin support for directory datasets, and better match the behavior of other backends.

[mfatihaktas]

Thanks, I agree that reading the file is not ideal while creating the table. Flink actually supports creating tables with filesystem connector. This however requires specifying the schema, as create_table() also requires:

def create_table(
        self,
        name: str,
        obj: pd.DataFrame | pa.Table | ir.Table | None = None,
        *,
        schema: sch.Schema | None = None,
        database: str | None = None,
        catalog: str | None = None,
        tbl_properties: dict | None = None,
        watermark: Watermark | None = None,
        temp: bool = False,
        overwrite: bool = False,
    ) -> ir.Table:
        ...

               if obj is None and schema is None:
                   raise exc.IbisError("`schema` or `obj` is required")
    ...

So I think we have two options:

1. Add a required argument schema for read_***(). This would deviate from the interface existing for other backends.
2. "Read" the schema from the file with pyarrow and feed it into create_table() with filesystem connector. This has the same issue you raised in accessing the files in a distributed system.

Which one do you think makes more sense? We could also implement both where the user can specify the schema, and if not specified we could construct the schema from the file and raise an error in case of a failure in accessing the file.

[cpcloud]

@mfatihaktas Apologies again for the churn, can you reopen this PR against main?

[mfatihaktas]

@mfatihaktas Apologies again for the churn, can you reopen this PR against main?

Reopened: #7908