csurong commented on code in PR #29245:
URL: https://github.com/apache/flink/pull/29245#discussion_r4090276170
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -16,15 +16,269 @@
# limitations under the License.
################################################################################
-from typing import Dict, Optional, Tuple
+from typing import Dict, List, Optional, Tuple, Union
from pyflink.dataframe.context import get_or_create_table_environment
-from pyflink.dataframe.dataframe import DataFrame
+from pyflink.dataframe.dataframe import DataFrame, _normalize_subset
from pyflink.dataframe.datatype import DataType
from pyflink.table import Schema, TableDescriptor
from pyflink.util.api_stability_decorators import PublicEvolving
-__all__ = ["read_generic"]
+__all__ = ["read_generic", "read_json", "read_parquet"]
+
+
+def _build_filesystem_options(
+ path: str,
+ file_format: str,
+ options: Dict[str, Optional[str]],
+ format_options: Optional[Dict[str, str]] = None,
+ *,
+ connector_options: Optional[Dict[str, str]] = None,
+ format_parameters: Optional[Dict[str, Optional[str]]] = None,
+) -> Dict[str, str]:
+ if not isinstance(path, str):
+ raise TypeError("path must be a string")
+ if not path:
+ raise ValueError("path must not be empty")
+
+ result = {"path": path, "format": file_format}
+ if connector_options is not None:
+ _validate_options(connector_options)
+ for key in connector_options:
+ if key in ("path", "format"):
+ raise ValueError(f"{key!r} must not be specified in
connector_options")
+ if key.startswith(file_format + "."):
+ raise ValueError(f"format option {key!r} must be specified in
format_options")
+ _merge_options(result, connector_options)
+ _merge_options(result, {key: value for key, value in options.items() if
value is not None})
+
+ normalized_format_options: Dict[str, str] = {}
+ if format_options is not None:
+ _validate_options(format_options)
+ for key, value in format_options.items():
+ option = key if key.startswith(file_format + ".") else file_format
+ "." + key
+ if option in normalized_format_options:
+ raise ValueError(f"duplicate format option: {option!r}")
+ normalized_format_options[option] = value
+ if format_parameters is not None:
+ _merge_options(normalized_format_options, {
+ file_format + "." + key: value
+ for key, value in format_parameters.items() if value is not None
+ })
+ _merge_options(result, normalized_format_options)
+ return result
+
+
+def _merge_options(target: Dict[str, str], options: Dict[str, str]) -> None:
+ _validate_options(options)
+ for key, value in options.items():
+ if key in target and target[key] != value:
+ raise ValueError(f"conflicting values for option {key!r}")
+ target[key] = value
+
+
+def _boolean_option(value: Optional[bool], name: str) -> Optional[str]:
+ if value is None:
+ return None
+ if not isinstance(value, bool):
+ raise TypeError(f"{name} must be a bool or None")
+ return str(value).lower()
+
+
+def _parallelism_option(value: Optional[int]) -> Optional[str]:
+ if value is None:
+ return None
+ if isinstance(value, bool) or not isinstance(value, int):
+ raise TypeError("sink_parallelism must be an int or None")
+ return str(value)
+
+
+def _build_filesystem_sink_options(
+ path: str,
+ file_format: str,
+ options: Dict[str, Optional[str]],
+ format_parameters: Dict[str, Optional[str]],
+ connector_options: Optional[Dict[str, str]] = None,
+ format_options: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+ result = _build_filesystem_options(
+ path, file_format, options, format_options,
+ connector_options=connector_options,
format_parameters=format_parameters,
+ )
+ # Apply defaults after merging explicit settings from either API entry
point.
+ defaults = {
Review Comment:
Removed the Python-side defaults. Options left unspecified are now resolved
by the connector or format factory, and the docstrings describe the current
defaults.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
