Conversation
…e API Generated-by: OpenAI Codex (GPT-6)
| rolling_policy_file_size: str = "128mb", | ||
| rolling_policy_rollover_interval: str = "30min", | ||
| rolling_policy_check_interval: Optional[str] = None, | ||
| partition_commit_trigger: str = "process-time", |
There was a problem hiding this comment.
Partition commit only works for partitioned sinks, but the writer APIs do not expose partition columns and the descriptor never calls partitioned_by(...). Please add complete partition_by support.
There was a problem hiding this comment.
Thanks for catching this. Added partition_by to both readers and writers in b104961, with the partition keys passed to TableDescriptor.partitioned_by(...). I also added partitioned round-trip/overwrite tests and streaming partition-commit tests, including a non-UTC case.
| # ======================== I/O ======================== | ||
|
|
||
| @PublicEvolving() | ||
| def write_parquet( |
There was a problem hiding this comment.
There are many options defined in the filesystem connector, I think we are not limited to the parameters defined in the FLIP-591.
Do you think it makes sense to make the following changes:
- Evaluate the options defined in the filesystem connector and parquet / json format and make sure frequently used options are defined here.
- Introduce parameters connector_options and format_options for the options which are not defined here. It could also avoid functionality break if users introduce new features in the filesystem connector or parquet/json format and forget to update this API.
There was a problem hiding this comment.
Agreed. I went through the connector and format options and added parameters for sink parallelism, inactivity rolling, JSON parsing/timestamps, and Parquet UTC conversion. All four helpers now also accept connector_options and format_options for options without a dedicated parameter.
| path: str, | ||
| *, | ||
| mode: str = "overwrite", | ||
| compression: str = "SNAPPY", |
There was a problem hiding this comment.
JSON provides a generic format_options escape hatch. Could we provide equivalent format_options for Parquet and consider a generic sink_options escape hatch for advanced upstream options? This avoids adding a new public argument for every supported connector option.
There was a problem hiding this comment.
Parquet now accepts format_options too. I used connector_options for advanced sink options to keep the naming consistent across readers and writers. Defaults are applied after user options, so format_options={"compression": "GZIP"} is preserved rather than overwritten by the SNAPPY default.
|
|
||
| :param path: File or directory URI supported by Flink's filesystem implementations. | ||
| :param schema: Mapping of column names to DataFrame data types. | ||
| :param monitor_interval: Optional file discovery interval, for example ``"60s"``. |
There was a problem hiding this comment.
Could we document the exact semantics here? Setting monitor_interval changes the source from a bounded scan to continuous file discovery, while path_regex_pattern performs a regex full match against the file path. This distinction is important for users configuring streaming reads.
There was a problem hiding this comment.
Updated both reader docstrings and the I/O docs to spell this out: monitor_interval enables unbounded file discovery, while the regex matches the entire path, excluding the scheme and authority. The docs also explain that files aren't tailed and new partition directories aren't discovered. Added tests for full-path matching and rejection of unbounded reads in batch mode.
Add partition columns and connector/format options to JSON and Parquet readers and writers. Merge explicit settings before applying defaults and support computed columns and watermarks on both readers. Cover partitioned writes, continuous discovery and non-UTC partition commits, and document the filesystem and format semantics. Generated-by: OpenAI Codex (GPT-6)
3 participants
What is the purpose of the change
Add Parquet and JSON filesystem I/O to the PyFlink DataFrame API, as described in FLINK-40426 and FLIP-591.
Brief change log
read_parquet,read_json,write_parquet, andwrite_json, using the existing filesystem connector.write_genericbehavior.Verifying this change
test_filesystem_io.pyandtest_io.pypassed, including real JSON/Parquet reads and writes, write modes, format options, and argument validation.flink-pythonwith its dependencies passed (Java tests skipped in that build).clean verifywas attempted, but four OpenTelemetry tests failed because no Docker environment was available locally.Does this pull request potentially affect one of the following parts:
Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: OpenAI Codex (GPT-6)