This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7c0c43eeca9 [FLINK-40427][python] Add catalog management and catalog
table I/O to DataFrame API
7c0c43eeca9 is described below
commit 7c0c43eeca9ea1057b9c2722e1793b23c4c7724c
Author: Federico Dolce <[email protected]>
AuthorDate: Tue Sep 15 16:00:21 2026 +0200
[FLINK-40427][python] Add catalog management and catalog table I/O to
DataFrame API
This closes #29262.
---
.../pyflink.dataframe/{index.rst => catalog.rst} | 51 ++-
.../docs/reference/pyflink.dataframe/index.rst | 1 +
.../docs/reference/pyflink.dataframe/io.rst | 2 +
flink-python/pyflink/dataframe/__init__.py | 21 +-
flink-python/pyflink/dataframe/catalog.py | 254 ++++++++++++++
flink-python/pyflink/dataframe/dataframe.py | 55 ++-
flink-python/pyflink/dataframe/errors.py | 48 +++
flink-python/pyflink/dataframe/io.py | 175 +++++++++-
.../pyflink/dataframe/tests/test_catalog.py | 177 ++++++++++
.../pyflink/dataframe/tests/test_filesystem_io.py | 4 +-
flink-python/pyflink/dataframe/tests/test_io.py | 373 ++++++++++++++++++++-
11 files changed, 1118 insertions(+), 43 deletions(-)
diff --git a/flink-python/docs/reference/pyflink.dataframe/index.rst
b/flink-python/docs/reference/pyflink.dataframe/catalog.rst
similarity index 57%
copy from flink-python/docs/reference/pyflink.dataframe/index.rst
copy to flink-python/docs/reference/pyflink.dataframe/catalog.rst
index 547cb0f2202..cf22cc65c1e 100644
--- a/flink-python/docs/reference/pyflink.dataframe/index.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/catalog.rst
@@ -16,20 +16,37 @@
limitations under the License.
################################################################################
-==================
-PyFlink DataFrame
-==================
-
-This page gives an overview of all public PyFlink DataFrame APIs.
-
-.. toctree::
- :maxdepth: 1
-
- dataframe
- udf
- creation
- io
- sql
- datatype
- environment
- config
+========
+Catalogs
+========
+
+Functions for creating catalogs and navigating catalogs and databases. Setting
the current
+catalog and database lets :func:`~pyflink.dataframe.read_catalog_table` and
+:meth:`~pyflink.dataframe.DataFrame.write_catalog_table` reference tables by
short paths instead
+of the full ``catalog_name.db_name.table_name``.
+
+Catalog Management
+------------------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ create_catalog
+ get_catalog
+ use_catalog
+ get_current_catalog
+ list_catalogs
+
+Database Navigation
+-------------------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ use_database
+ get_current_database
+ list_databases
diff --git a/flink-python/docs/reference/pyflink.dataframe/index.rst
b/flink-python/docs/reference/pyflink.dataframe/index.rst
index 547cb0f2202..d92c806a027 100644
--- a/flink-python/docs/reference/pyflink.dataframe/index.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/index.rst
@@ -29,6 +29,7 @@ This page gives an overview of all public PyFlink DataFrame
APIs.
udf
creation
io
+ catalog
sql
datatype
environment
diff --git a/flink-python/docs/reference/pyflink.dataframe/io.rst
b/flink-python/docs/reference/pyflink.dataframe/io.rst
index a5c34a926ab..a92bd5a13fd 100644
--- a/flink-python/docs/reference/pyflink.dataframe/io.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/io.rst
@@ -32,6 +32,7 @@ Readers
.. autosummary::
:toctree: api/
+ read_catalog_table
read_generic
read_json
read_parquet
@@ -44,6 +45,7 @@ Writers
.. autosummary::
:toctree: api/
+ DataFrame.write_catalog_table
DataFrame.write_generic
DataFrame.write_json
DataFrame.write_parquet
diff --git a/flink-python/pyflink/dataframe/__init__.py
b/flink-python/pyflink/dataframe/__init__.py
index 94697f87029..2ac311e66b4 100644
--- a/flink-python/pyflink/dataframe/__init__.py
+++ b/flink-python/pyflink/dataframe/__init__.py
@@ -38,6 +38,16 @@ Example::
<Row(1, 'Alice', 31)>
"""
+from pyflink.dataframe.catalog import (
+ create_catalog,
+ get_catalog,
+ get_current_catalog,
+ get_current_database,
+ list_catalogs,
+ list_databases,
+ use_catalog,
+ use_database,
+)
from pyflink.dataframe.convert import (
from_arrow,
from_dict,
@@ -54,7 +64,7 @@ from pyflink.dataframe.context import (
from pyflink.dataframe.dataframe import DataFrame, GroupedDataFrame, col, lit
from pyflink.dataframe.dataframe_config import config
from pyflink.dataframe.datatype import DataType
-from pyflink.dataframe.io import read_generic, read_json, read_parquet
+from pyflink.dataframe.io import read_catalog_table, read_generic, read_json,
read_parquet
from pyflink.dataframe.sql import sql
from pyflink.dataframe.udf import udf
from pyflink.dataframe.udtf import udtf
@@ -73,9 +83,18 @@ __all__ = [
"from_records",
"from_table",
"range",
+ "read_catalog_table",
"read_generic",
"read_json",
"read_parquet",
+ "create_catalog",
+ "get_catalog",
+ "use_catalog",
+ "get_current_catalog",
+ "list_catalogs",
+ "use_database",
+ "get_current_database",
+ "list_databases",
"sql",
"config",
"set_table_environment",
diff --git a/flink-python/pyflink/dataframe/catalog.py
b/flink-python/pyflink/dataframe/catalog.py
new file mode 100644
index 00000000000..33dab3fc4fe
--- /dev/null
+++ b/flink-python/pyflink/dataframe/catalog.py
@@ -0,0 +1,254 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from typing import Dict, List, Optional
+
+from pyflink.common import Configuration
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.errors import _raise_as_value_error
+from pyflink.table.catalog import Catalog, CatalogDescriptor
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = [
+ "create_catalog",
+ "get_catalog",
+ "use_catalog",
+ "get_current_catalog",
+ "list_catalogs",
+ "use_database",
+ "get_current_database",
+ "list_databases",
+]
+
+
+def _validate_name(name: str, what: str) -> None:
+ if not isinstance(name, str):
+ raise TypeError(f"{what} must be a string")
+ if not name:
+ raise ValueError(f"{what} must not be empty")
+
+
+def _validate_catalog_options(options: Dict[str, str]) -> None:
+ if not isinstance(options, dict):
+ raise TypeError("options must be a dict of string keys and values")
+ for key, value in options.items():
+ if not isinstance(key, str):
+ raise TypeError("option keys must be strings")
+ if not key:
+ raise ValueError("option keys must not be empty")
+ if not isinstance(value, str):
+ raise TypeError(f"option {key!r} must have a string value")
+
+
+@PublicEvolving()
+def create_catalog(name: str, options: Dict[str, str]) -> None:
+ """
+ Create a catalog in the environment used by DataFrame operations.
+
+ The ``type`` option picks which kind of catalog to create, for example
``generic_in_memory``.
+ Flink has to know about that type, either built in or on the classpath.
Every other option is
+ passed through to the catalog. This does the same as ``CREATE CATALOG`` in
SQL.
+
+ :param name: Name under which the catalog is created.
+ :param options: Catalog options, including the ``type`` option.
+ :raises TypeError: If ``name`` is not a string or ``options`` is not a
dict of strings.
+ :raises ValueError: If ``name`` or an option key is empty, or if Flink
cannot create a catalog
+ from ``options``.
+ :raises ~pyflink.util.exceptions.CatalogException: If a catalog named
``name`` already exists,
+ or if the catalog reports an error while being created.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ >>> pf.list_catalogs()
+ ['default_catalog', 'my_catalog']
+
+ .. versionadded:: 2.4.0
+ """
+ _validate_name(name, "name")
+ _validate_catalog_options(options)
+ configuration = Configuration()
+ for key, value in options.items():
+ configuration.set_string(key, value)
+ descriptor = CatalogDescriptor.of(name, configuration)
+ try:
+ get_or_create_table_environment().create_catalog(name, descriptor)
+ except Exception as error:
+ _raise_as_value_error(error)
+
+
+@PublicEvolving()
+def get_catalog(name: str) -> Optional[Catalog]:
+ """
+ Return a registered catalog by name.
+
+ :param name: Name of the catalog.
+ :return: The :class:`~pyflink.table.catalog.Catalog`, or ``None`` if no
catalog is registered
+ under ``name``.
+ :raises TypeError: If ``name`` is not a string.
+ :raises ValueError: If ``name`` is empty.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ >>> pf.get_catalog("my_catalog").list_databases()
+ ['default']
+ >>> pf.get_catalog("missing") is None
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ _validate_name(name, "name")
+ return get_or_create_table_environment().get_catalog(name)
+
+
+@PublicEvolving()
+def use_catalog(name: str) -> None:
+ """
+ Set the current catalog.
+
+ Table paths without a catalog part, such as ``my_table`` or
``my_database.my_table``, are
+ resolved against the current catalog. Switching catalogs also moves you to
that catalog's
+ default database.
+
+ :param name: Name of a registered catalog.
+ :raises TypeError: If ``name`` is not a string.
+ :raises ValueError: If ``name`` is empty.
+ :raises ~pyflink.util.exceptions.CatalogException: If no catalog named
``name`` exists.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ >>> pf.use_catalog("my_catalog")
+ >>> pf.get_current_catalog()
+ 'my_catalog'
+
+ .. versionadded:: 2.4.0
+ """
+ _validate_name(name, "name")
+ get_or_create_table_environment().use_catalog(name)
+
+
+@PublicEvolving()
+def get_current_catalog() -> Optional[str]:
+ """
+ Return the name of the current catalog.
+
+ :return: The current catalog name, or ``None`` if no current catalog is
set.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.get_current_catalog()
+ 'default_catalog'
+
+ .. versionadded:: 2.4.0
+ """
+ return get_or_create_table_environment().get_current_catalog()
+
+
+@PublicEvolving()
+def list_catalogs() -> List[str]:
+ """
+ Return the names of all registered catalogs.
+
+ :return: Catalog names.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.list_catalogs()
+ ['default_catalog']
+
+ .. versionadded:: 2.4.0
+ """
+ return get_or_create_table_environment().list_catalogs()
+
+
+@PublicEvolving()
+def use_database(name: str) -> None:
+ """
+ Set the current database within the current catalog.
+
+ Table paths without a database part, such as ``my_table``, are resolved
against the current
+ database.
+
+ :param name: Name of a database in the current catalog.
+ :raises TypeError: If ``name`` is not a string.
+ :raises ValueError: If ``name`` is empty.
+ :raises ~pyflink.util.exceptions.CatalogException: If the current catalog
has no database
+ named ``name``.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.use_database("default_database")
+ >>> pf.get_current_database()
+ 'default_database'
+
+ .. versionadded:: 2.4.0
+ """
+ _validate_name(name, "name")
+ get_or_create_table_environment().use_database(name)
+
+
+@PublicEvolving()
+def get_current_database() -> Optional[str]:
+ """
+ Return the name of the current database.
+
+ :return: The current database name, or ``None`` if no current database is
set.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.get_current_database()
+ 'default_database'
+
+ .. versionadded:: 2.4.0
+ """
+ return get_or_create_table_environment().get_current_database()
+
+
+@PublicEvolving()
+def list_databases() -> List[str]:
+ """
+ Return the names of all databases in the current catalog.
+
+ :return: Database names.
+ :raises ValueError: If no current catalog is set.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.list_databases()
+ ['default_database']
+
+ .. versionadded:: 2.4.0
+ """
+ table_environment = get_or_create_table_environment()
+ # TableEnvironment.list_databases() fails with a bare
NoSuchElementException in this case.
+ if table_environment.get_current_catalog() is None:
+ raise ValueError(
+ "A current catalog has not been set. Set one with
pf.use_catalog(), or list the "
+ "databases of a specific catalog with
pf.get_catalog(name).list_databases()."
+ )
+ return table_environment.list_databases()
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
index a3635f876b4..c48f8af4397 100644
--- a/flink-python/pyflink/dataframe/dataframe.py
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -51,6 +51,7 @@ from pyflink.table.expressions import (
)
from pyflink.table.table import Table
from pyflink.table.types import ArrayType, MapType, MultisetType, RowType
+from pyflink.table.table_descriptor import TableDescriptor
from pyflink.util.api_stability_decorators import PublicEvolving
__all__ = ["DataFrame", "GroupedDataFrame", "col", "lit"]
@@ -1963,8 +1964,8 @@ class DataFrame:
:param connector: Factory identifier used as the ``connector`` Table
option.
:param options: Connector options, excluding the reserved
``connector`` option.
:raises TypeError: If an argument has an invalid type.
- :raises ValueError: If the connector or an option key is empty, or if
``options`` contains
- the reserved ``connector`` key.
+ :raises ValueError: If the connector or an option key is empty, if
``options`` contains
+ the reserved ``connector`` key, or if Flink rejects the connector
or its options.
Example::
@@ -1990,6 +1991,7 @@ class DataFrame:
mode: str = "append",
partition_by: Optional[Union[str, List[str]]] = None,
) -> None:
+ from pyflink.dataframe.errors import _raise_as_value_error
from pyflink.dataframe.io import _build_generic_descriptor
if not isinstance(mode, str):
@@ -1997,7 +1999,54 @@ class DataFrame:
if mode not in ("append", "overwrite"):
raise ValueError("mode must be 'append' or 'overwrite'")
descriptor = _build_generic_descriptor(connector, options,
partition_by=partition_by)
- result = self._table.execute_insert(descriptor, overwrite=mode ==
"overwrite")
+ try:
+ self._execute_insert(descriptor, mode == "overwrite")
+ except Exception as error:
+ _raise_as_value_error(error)
+
+ @PublicEvolving()
+ def write_catalog_table(self, path: str, *, overwrite: bool = False) ->
None:
+ """
+ Write this DataFrame to a table registered in a catalog.
+
+ ``path`` is ``table_name``, ``db_name.table_name``, or
``catalog_name.db_name.table_name``.
+ Missing parts are resolved against the current catalog and database,
see
+ :func:`~pyflink.dataframe.use_catalog` and
:func:`~pyflink.dataframe.use_database`. The
+ write runs right away. On a local or MiniCluster setup the call blocks
until the write is
+ done.
+
+ :param path: Path of the catalog table.
+ :param overwrite: Whether existing data should be replaced, like
``INSERT OVERWRITE``.
+ Not every connector supports overwriting.
+ :raises TypeError: If ``path`` is not a string or ``overwrite`` is not
a bool.
+ :raises ValueError: If ``path`` is empty, malformed, or does not name
a table, or if the
+ DataFrame's columns do not match the table.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> events = pf.from_records([(1, "login")], schema=["id",
"event"])
+ >>> events.write_catalog_table("my_catalog.my_database.events")
+ >>> pf.use_catalog("my_catalog")
+ >>> events.write_catalog_table("my_database.events",
overwrite=True)
+
+ .. versionadded:: 2.4.0
+ """
+ from pyflink.dataframe.catalog import _validate_name
+ from pyflink.dataframe.errors import _raise_as_value_error
+
+ _validate_name(path, "path")
+ if not isinstance(overwrite, bool):
+ raise TypeError("overwrite must be a bool")
+ try:
+ self._execute_insert(path, overwrite)
+ except Exception as error:
+ _raise_as_value_error(error)
+
+ def _execute_insert(
+ self, target: Union[str, TableDescriptor], overwrite: bool = False
+ ) -> None:
+ result = self._table.execute_insert(target, overwrite=overwrite)
execution_target = self._table._t_env.get_config().get(
"execution.target", None
)
diff --git a/flink-python/pyflink/dataframe/errors.py
b/flink-python/pyflink/dataframe/errors.py
new file mode 100644
index 00000000000..79b2ba73d48
--- /dev/null
+++ b/flink-python/pyflink/dataframe/errors.py
@@ -0,0 +1,48 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from typing import NoReturn
+
+from py4j.protocol import Py4JJavaError
+
+from pyflink.util.java_utils import is_instance_of
+
+_USER_ERROR_CLASSES = (
+ "org.apache.flink.table.api.ValidationException",
+ "org.apache.flink.table.api.SqlParserException",
+)
+
+
+def _is_user_error(j_exception) -> bool:
+ return any(is_instance_of(j_exception, j_class) for j_class in
_USER_ERROR_CLASSES)
+
+
+def _raise_as_value_error(error: Exception) -> NoReturn:
+ """
+ Re-raise ``error`` from a Flink call, turning user mistakes into
:class:`ValueError`.
+
+ When the user gets something wrong, an unknown table, a bad path, invalid
options, Flink
+ throws ``ValidationException`` or ``SqlParserException``. PyFlink hands
those to us as a bare
+ ``Py4JJavaError``. This turns them into a :class:`ValueError` with Flink's
message, chained to
+ the original so the Java trace is still there. Anything else is re-raised
as is, including
+ exceptions PyFlink already maps to Python classes such as
+ :class:`~pyflink.util.exceptions.CatalogException`.
+ """
+ if isinstance(error, Py4JJavaError) and
_is_user_error(error.java_exception):
+ raise ValueError(error.java_exception.getMessage()) from error
+ raise error
diff --git a/flink-python/pyflink/dataframe/io.py
b/flink-python/pyflink/dataframe/io.py
index 0f629674690..14b2ba206a7 100644
--- a/flink-python/pyflink/dataframe/io.py
+++ b/flink-python/pyflink/dataframe/io.py
@@ -18,13 +18,16 @@
from typing import Dict, List, Optional, Tuple, Union
+from pyflink.dataframe.catalog import _validate_name
from pyflink.dataframe.context import get_or_create_table_environment
from pyflink.dataframe.dataframe import DataFrame, _normalize_subset
from pyflink.dataframe.datatype import DataType
-from pyflink.table import Schema, TableDescriptor
+from pyflink.dataframe.errors import _raise_as_value_error
+from pyflink.java_gateway import get_gateway
+from pyflink.table import Schema, Table, TableDescriptor
from pyflink.util.api_stability_decorators import PublicEvolving
-__all__ = ["read_generic", "read_json", "read_parquet"]
+__all__ = ["read_catalog_table", "read_generic", "read_json", "read_parquet"]
def _build_filesystem_options(
@@ -296,9 +299,7 @@ def _validate_options(options: Dict[str, str]) -> None:
raise TypeError(f"option {key!r} must have a string value")
-def _validate_computed_columns(
- computed_columns: Optional[Dict[str, str]], physical_columns: Dict[str,
DataType]
-) -> Dict[str, str]:
+def _validate_computed_columns(computed_columns: Optional[Dict[str, str]]) ->
Dict[str, str]:
if computed_columns is None:
return {}
if not isinstance(computed_columns, dict):
@@ -310,10 +311,6 @@ def _validate_computed_columns(
raise TypeError("computed column names must be strings")
if not name:
raise ValueError("computed column names must not be empty")
- if name in physical_columns:
- raise ValueError(
- f"computed column {name!r} conflicts with a physical column"
- )
if not isinstance(expression, str):
raise TypeError(f"computed column {name!r} must use a string
expression")
if not expression:
@@ -358,9 +355,7 @@ def _build_source_schema(
raise TypeError(f"schema column {name!r} must use a DataType
value")
schema_builder.column(name, data_type._to_table_data_type())
- for name, expression in _validate_computed_columns(
- computed_columns, schema
- ).items():
+ for name, expression in
_validate_computed_columns(computed_columns).items():
schema_builder.column_by_expression(name, expression)
validated_watermark = _validate_watermark(watermark)
@@ -408,7 +403,11 @@ def _read(
descriptor = _build_generic_descriptor(
connector, options, schema=source_schema, partition_by=partition_by)
table_environment = get_or_create_table_environment()
- return DataFrame(table_environment.from_descriptor(descriptor))
+ try:
+ table = table_environment.from_descriptor(descriptor)
+ except Exception as error:
+ _raise_as_value_error(error)
+ return DataFrame(table)
@PublicEvolving()
@@ -435,7 +434,8 @@ def read_generic(
:return: A DataFrame backed by the configured source.
:raises TypeError: If an argument has an invalid type.
:raises ValueError: If a connector, schema, option key, computed column,
or watermark value is
- empty, or if a computed column conflicts with a physical column.
+ empty, or if Flink rejects the schema, such as a computed column that
duplicates a
+ physical column or an invalid computed column or watermark expression.
Example::
@@ -461,3 +461,150 @@ def read_generic(
connector, schema=schema, options=options,
computed_columns=computed_columns, watermark=watermark,
)
+
+
+def _extend_catalog_table(
+ table: Table,
+ computed_columns: Dict[str, str],
+ watermark: Optional[Tuple[str, str]],
+) -> Table:
+ """
+ Read the catalog table behind ``table`` with extra computed columns and,
optionally, a
+ different watermark.
+
+ Flink has no way to add columns to an existing table at read time, so this
builds a nameless
+ copy of the table definition with the new schema and reads from that
instead. Everything else
+ is copied over unchanged: options, partition keys, distribution, snapshot,
connection and
+ comment. The connector sees the same table it always did, only with more
columns. The real
+ catalog table is not touched.
+
+ This is the same thing ``TableEnvironment.from(TableDescriptor)`` does in
Java. We build the
+ ``CatalogTable`` ourselves because ``TableDescriptor`` cannot take a
distribution or snapshot
+ object directly.
+
+ Tables whose catalog supplies the connector, such as Paimon or Hive,
cannot be extended this
+ way: the copy has no catalog to ask, so it would not know which connector
to use. That is why
+ the ``connector`` option is required.
+ """
+ jvm = get_gateway().jvm
+ catalog = jvm.org.apache.flink.table.catalog
+
+ source = table._j_table.getQueryOperation().getContextResolvedTable()
+ identifier = source.getIdentifier().asSummaryString()
+ source_table = source.getTable()
+ table_kind = source_table.getTableKind()
+ if table_kind != catalog.CatalogBaseTable.TableKind.TABLE:
+ raise ValueError(
+ f"{identifier} is a {table_kind.name().lower()}; computed columns
and watermarks can "
+ f"only be added to tables"
+ )
+ if not source_table.getOptions().get("connector"):
+ raise ValueError(
+ f"{identifier} has no 'connector' option, so its connector comes
from the catalog and "
+ f"cannot be reused for an extended copy. Read it without
computed_columns and "
+ f"watermark, or use read_generic."
+ )
+
+ source_schema = source_table.getUnresolvedSchema()
+
+ schema_builder = jvm.org.apache.flink.table.api.Schema.newBuilder()
+ if watermark is None:
+ schema_builder.fromSchema(source_schema)
+ else:
+ # A schema can only have one watermark. Copy the columns and primary
key by hand so the
+ # table's own watermark is left out, then add the new one below.
+ schema_builder.fromColumns(source_schema.getColumns())
+ primary_key = source_schema.getPrimaryKey()
+ if primary_key.isPresent():
+ schema_builder.primaryKeyNamed(
+ primary_key.get().getConstraintName(),
primary_key.get().getColumnNames()
+ )
+ for name, expression in computed_columns.items():
+ schema_builder.columnByExpression(name, expression)
+ if watermark is not None:
+ schema_builder.watermark(*watermark)
+
+ catalog_table = (
+ catalog.CatalogTable.newBuilder()
+ .schema(schema_builder.build())
+ .options(source_table.getOptions())
+ .partitionKeys(source_table.getPartitionKeys())
+ .distribution(source_table.getDistribution().orElse(None))
+ .snapshot(source_table.getSnapshot().orElse(None))
+ .connection(source_table.getConnection().orElse(None))
+ .comment(source_table.getComment())
+ .build()
+ )
+
+ j_table_environment = table._t_env._j_tenv
+ try:
+ resolved_table =
j_table_environment.getCatalogManager().resolveCatalogTable(catalog_table)
+ except Exception as error:
+ _raise_as_value_error(error)
+ query_operation =
jvm.org.apache.flink.table.operations.SourceQueryOperation(
+ catalog.ContextResolvedTable.anonymous(resolved_table)
+ )
+ return Table(j_table_environment.createTable(query_operation),
table._t_env)
+
+
+@PublicEvolving()
+def read_catalog_table(
+ path: str,
+ *,
+ computed_columns: Optional[Dict[str, str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Read a table registered in a catalog.
+
+ ``path`` is ``table_name``, ``db_name.table_name``, or
``catalog_name.db_name.table_name``.
+ Missing parts are resolved against the current catalog and database, see
+ :func:`~pyflink.dataframe.use_catalog` and
:func:`~pyflink.dataframe.use_database`. Names that
+ are reserved keywords or contain dots must be escaped with backticks.
+
+ Use ``computed_columns`` and ``watermark`` to add columns or change the
watermark for this read
+ only. The new columns go after the table's own columns, in the order
given. The watermark
+ replaces whatever the table declares and may use the new columns. Nothing
is written back to
+ the catalog.
+
+ This only works for tables that set the ``connector`` option. Tables from
catalogs that supply
+ their own connector, such as Paimon or Hive, can still be read, just
without these two
+ arguments.
+
+ :param path: Path of the catalog table.
+ :param computed_columns: Optional SQL expressions keyed by computed column
name.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame backed by the catalog table.
+ :raises TypeError: If ``path`` is not a string or another argument has an
invalid type.
+ :raises ValueError: If ``path`` is empty, malformed, or does not name a
table. Also if
+ ``computed_columns`` or ``watermark`` contain empty strings, are used
on a view or on a
+ table without a ``connector`` option, or if Flink rejects the
resulting schema, for example
+ a duplicate column name or an expression that does not compile.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ >>> orders = pf.read_catalog_table("my_catalog.default.orders")
+ >>> pf.use_catalog("my_catalog")
+ >>> orders = pf.read_catalog_table("default.orders")
+ >>> pf.use_database("default")
+ >>> orders = pf.read_catalog_table(
+ ... "orders",
+ ... computed_columns={"order_time": "TO_TIMESTAMP_LTZ(ts_millis,
3)"},
+ ... watermark=("order_time", "order_time - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ _validate_name(path, "path")
+ table_environment = get_or_create_table_environment()
+ try:
+ table = table_environment.from_path(path)
+ except Exception as error:
+ _raise_as_value_error(error)
+ validated_columns = _validate_computed_columns(computed_columns)
+ validated_watermark = _validate_watermark(watermark)
+ if validated_columns or validated_watermark is not None:
+ table = _extend_catalog_table(table, validated_columns,
validated_watermark)
+ return DataFrame(table)
diff --git a/flink-python/pyflink/dataframe/tests/test_catalog.py
b/flink-python/pyflink/dataframe/tests/test_catalog.py
new file mode 100644
index 00000000000..3360a7a7dd7
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_catalog.py
@@ -0,0 +1,177 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import unittest
+from unittest.mock import patch
+
+from py4j.protocol import Py4JJavaError
+
+import pyflink.dataframe as pf
+from pyflink.java_gateway import get_gateway
+from pyflink.table.catalog import Catalog
+from pyflink.testing.test_case_utils import PyFlinkDataFrameUTTestCase
+from pyflink.util.exceptions import CatalogException
+
+
+class CatalogTests(PyFlinkDataFrameUTTestCase):
+ def test_create_catalog_builds_descriptor(self):
+ with patch.object(
+ self.t_env, "create_catalog", wraps=self.t_env.create_catalog
+ ) as create_catalog:
+ pf.create_catalog(
+ "my_catalog",
+ {"type": "generic_in_memory", "default-database":
"my_database"},
+ )
+
+ name, descriptor = create_catalog.call_args.args
+ self.assertEqual(name, "my_catalog")
+ options = descriptor._j_catalog_descriptor.getConfiguration().toMap()
+ self.assertEqual(
+ dict(options),
+ {"type": "generic_in_memory", "default-database": "my_database"},
+ )
+ self.assertIn("my_catalog", pf.list_catalogs())
+ self.assertEqual(
+ pf.get_catalog("my_catalog").get_default_database(), "my_database"
+ )
+
+ def test_create_catalog_rejects_invalid_arguments(self):
+ cases = [
+ (None, {"type": "generic_in_memory"}, TypeError, "name must be a
string"),
+ ("", {"type": "generic_in_memory"}, ValueError, "name must not be
empty"),
+ ("my_catalog", ["type"], TypeError, "options must be a dict"),
+ ("my_catalog", {1: "x"}, TypeError, "option keys must be strings"),
+ ("my_catalog", {"": "x"}, ValueError, "option keys must not be
empty"),
+ (
+ "my_catalog",
+ {"type": "generic_in_memory", "flag": True},
+ TypeError,
+ "option 'flag' must have a string value",
+ ),
+ ]
+
+ for name, options, error_type, message in cases:
+ with self.subTest(name=name, options=options):
+ with self.assertRaisesRegex(error_type, message):
+ pf.create_catalog(name, options)
+
+ self.assertNotIn("my_catalog", pf.list_catalogs())
+
+ def test_create_catalog_rejects_existing_catalog(self):
+ with self.assertRaisesRegex(CatalogException, "Catalog default_catalog
already exists"):
+ pf.create_catalog("default_catalog", {"type": "generic_in_memory"})
+
+ def test_create_catalog_translates_flink_errors(self):
+ with self.assertRaises(ValueError) as context:
+ pf.create_catalog("my_catalog", {"type": "no_such_catalog_type"})
+ self.assertIn("Unable to create catalog 'my_catalog'",
str(context.exception))
+ self.assertIn("'type'='no_such_catalog_type'", str(context.exception))
+ self.assertIsInstance(context.exception.__cause__, Py4JJavaError)
+ self.assertNotIn("my_catalog", pf.list_catalogs())
+
+ def test_get_catalog(self):
+ self.assertIsNone(pf.get_catalog("missing"))
+
+ pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ catalog = pf.get_catalog("my_catalog")
+ self.assertIsInstance(catalog, Catalog)
+ self.assertEqual(catalog.list_databases(), ["default"])
+
+ def test_catalog_navigation(self):
+ self.assertEqual(pf.get_current_catalog(), "default_catalog")
+ self.assertEqual(pf.get_current_database(), "default_database")
+ self.assertEqual(pf.list_catalogs(), ["default_catalog"])
+ self.assertEqual(pf.list_databases(), ["default_database"])
+
+ pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ self.assertEqual(pf.list_catalogs(), ["default_catalog", "my_catalog"])
+
+ pf.use_catalog("my_catalog")
+ self.assertEqual(pf.get_current_catalog(), "my_catalog")
+ self.assertEqual(pf.get_current_database(), "default")
+ self.assertEqual(pf.list_databases(), ["default"])
+
+ self.t_env.execute_sql("CREATE DATABASE my_database")
+ self.assertEqual(sorted(pf.list_databases()), ["default",
"my_database"])
+
+ pf.use_database("my_database")
+ self.assertEqual(pf.get_current_database(), "my_database")
+
+ pf.use_catalog("default_catalog")
+ self.assertEqual(pf.get_current_catalog(), "default_catalog")
+ self.assertEqual(pf.get_current_database(), "default_database")
+
+ def test_navigation_rejects_unknown_names(self):
+ with self.assertRaisesRegex(
+ CatalogException, r"A catalog with name \[missing\] does not exist"
+ ):
+ pf.use_catalog("missing")
+
+ with self.assertRaisesRegex(
+ CatalogException,
+ r"A database with name \[missing\] does not exist in the catalog:
\[default_catalog\]",
+ ):
+ pf.use_database("missing")
+
+ self.assertEqual(pf.get_current_catalog(), "default_catalog")
+ self.assertEqual(pf.get_current_database(), "default_database")
+
+ def test_navigation_without_current_catalog(self):
+ self.t_env.use_catalog(None)
+ self.assertIsNone(pf.get_current_catalog())
+ self.assertIsNone(pf.get_current_database())
+
+ with self.assertRaisesRegex(ValueError, "A current catalog has not
been set"):
+ pf.list_databases()
+ self.assertEqual(pf.get_catalog("default_catalog").list_databases(),
["default_database"])
+
+ pf.use_catalog("default_catalog")
+ self.assertEqual(pf.list_databases(), ["default_database"])
+
+ def test_unexpected_errors_are_not_translated(self):
+ gateway = get_gateway()
+ j_error = gateway.jvm.org.apache.flink.table.api.TableException("boom")
+ java_error = Py4JJavaError("An error occurred while calling
createCatalog.", j_error)
+ catalog_error = CatalogException("boom")
+ python_error = RuntimeError("boom")
+
+ for error in [java_error, catalog_error, python_error]:
+ with self.subTest(error=type(error).__name__):
+ with patch.object(self.t_env, "create_catalog",
side_effect=error):
+ with self.assertRaises(type(error)) as context:
+ pf.create_catalog("my_catalog", {"type":
"generic_in_memory"})
+ self.assertIs(context.exception, error)
+
+ def test_name_validation(self):
+ for function in [pf.get_catalog, pf.use_catalog, pf.use_database]:
+ with self.subTest(function=function.__name__):
+ with self.assertRaisesRegex(TypeError, "name must be a
string"):
+ function(None)
+ with self.assertRaisesRegex(ValueError, "name must not be
empty"):
+ function("")
+
+ def test_uses_dataframe_table_environment(self):
+ pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ self.assertIn("my_catalog", self.t_env.list_catalogs())
+
+ pf.use_catalog("my_catalog")
+ self.assertEqual(self.t_env.get_current_catalog(), "my_catalog")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/flink-python/pyflink/dataframe/tests/test_filesystem_io.py
b/flink-python/pyflink/dataframe/tests/test_filesystem_io.py
index f1012ae9cd0..ef265be8015 100644
--- a/flink-python/pyflink/dataframe/tests/test_filesystem_io.py
+++ b/flink-python/pyflink/dataframe/tests/test_filesystem_io.py
@@ -217,12 +217,12 @@ class FilesystemIOTests(PyFlinkDataFrameUTTestCase):
def test_invalid_partition_columns_are_rejected_by_the_table_api(self):
for reader in (pf.read_json, pf.read_parquet):
with self.subTest(reader=reader.__name__):
- with self.assertRaisesRegex(Py4JJavaError, "missing"):
+ with self.assertRaisesRegex(ValueError, "missing"):
reader("/input", schema=self._SCHEMA,
partition_by="missing")
dataframe = pf.from_records([(1,)], schema=["id"])
for writer in (dataframe.write_json, dataframe.write_parquet):
with self.subTest(writer=writer.__name__):
- with self.assertRaisesRegex(Py4JJavaError, "missing"):
+ with self.assertRaisesRegex(ValueError, "missing"):
writer("/output", partition_by="missing")
def
test_writers_accept_raw_options_without_overriding_them_with_defaults(self):
diff --git a/flink-python/pyflink/dataframe/tests/test_io.py
b/flink-python/pyflink/dataframe/tests/test_io.py
index ef57cfde616..64c2e84a8de 100644
--- a/flink-python/pyflink/dataframe/tests/test_io.py
+++ b/flink-python/pyflink/dataframe/tests/test_io.py
@@ -22,10 +22,12 @@ from unittest.mock import MagicMock, patch
import pyflink.dataframe as pf
from pyflink.dataframe import DataType
+from pyflink.table import DataTypes
from pyflink.testing.test_case_utils import (
PyFlinkDataFrameUTTestCase,
PyFlinkStreamDataFrameTestCase,
)
+from py4j.protocol import Py4JJavaError
class GenericIOTests(PyFlinkDataFrameUTTestCase):
@@ -146,12 +148,6 @@ class GenericIOTests(PyFlinkDataFrameUTTestCase):
TypeError,
"computed_columns must be a dict",
),
- (
- "duplicate computed column",
- {"computed_columns": {"id": "id + 1"}},
- ValueError,
- "conflicts with a physical column",
- ),
(
"computed column name type",
{"computed_columns": {1: "id + 1"}},
@@ -276,8 +272,373 @@ class GenericIOTests(PyFlinkDataFrameUTTestCase):
with self.assertRaisesRegex(error_type, message):
dataframe.write_generic(connector, options=options)
+ def test_read_generic_translates_flink_errors(self):
+ with self.assertRaises(ValueError) as context:
+ pf.read_generic(
+ "datagen",
+ schema=self._SCHEMA,
+ options={},
+ computed_columns={"bad": "NO_SUCH_FUNCTION(id)"},
+ )
+ self.assertEqual(
+ str(context.exception), "Invalid expression for computed column
'bad'."
+ )
+ self.assertIsInstance(context.exception.__cause__, Py4JJavaError)
+
+ with self.assertRaises(ValueError) as context:
+ pf.read_generic(
+ "datagen",
+ schema=self._SCHEMA,
+ options={},
+ computed_columns={"id": "id + 1"},
+ )
+ self.assertEqual(
+ str(context.exception),
+ "Schema must not contain duplicate column names. Found duplicates:
[id]",
+ )
+ self.assertIsInstance(context.exception.__cause__, Py4JJavaError)
+
+ def test_write_generic_translates_flink_errors(self):
+ dataframe = pf.from_records([(1,)], schema=["id"])
+
+ with self.assertRaises(ValueError) as context:
+ dataframe.write_generic("no_such_connector", options={})
+ self.assertIn("no_such_connector", str(context.exception))
+ self.assertIsInstance(context.exception.__cause__, Py4JJavaError)
+
+
+class CatalogTableIOTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ self.t_env.execute_sql("CREATE DATABASE my_catalog.my_database")
+ self.t_env.execute_sql(
+ "CREATE TABLE my_catalog.my_database.events ("
+ " id BIGINT, name STRING"
+ ") WITH ('connector' = 'datagen', 'number-of-rows' = '1')"
+ )
+
+ def test_read_catalog_table_resolves_paths(self):
+ cases = [
+ ("default_catalog", "default_database",
"my_catalog.my_database.events"),
+ ("my_catalog", "default", "my_database.events"),
+ ("my_catalog", "my_database", "events"),
+ ]
+
+ for catalog, database, path in cases:
+ with self.subTest(path=path):
+ pf.use_catalog(catalog)
+ pf.use_database(database)
+ with patch.object(
+ self.t_env, "from_path", wraps=self.t_env.from_path
+ ) as from_path:
+ dataframe = pf.read_catalog_table(path)
+
+ from_path.assert_called_once_with(path)
+ self.assertIsInstance(dataframe, pf.DataFrame)
+ self.assert_dataframe_schema(
+ dataframe,
+ ["id", "name"],
+ [DataTypes.BIGINT(), DataTypes.STRING()],
+ )
+
+ def test_read_catalog_table_rejects_invalid_path(self):
+ with self.assertRaisesRegex(TypeError, "path must be a string"):
+ pf.read_catalog_table(None)
+ with self.assertRaisesRegex(ValueError, "path must not be empty"):
+ pf.read_catalog_table("")
+
+ def test_read_catalog_table_translates_flink_errors(self):
+ with self.assertRaises(ValueError) as context:
+ pf.read_catalog_table("my_catalog.my_database.missing")
+ self.assertEqual(
+ str(context.exception),
+ "Table `my_catalog`.`my_database`.`missing` was not found.",
+ )
+ self.assertIsInstance(context.exception.__cause__, Py4JJavaError)
+
+ with self.assertRaises(ValueError) as context:
+ pf.read_catalog_table("too.many.path.parts")
+ self.assertEqual(
+ str(context.exception), "Invalid SQL identifier
too.many.path.parts."
+ )
+ self.assertIsInstance(context.exception.__cause__, Py4JJavaError)
+
+ def _create_timed_table(self):
+ self.t_env.execute_sql(
+ "CREATE TABLE my_catalog.my_database.timed ("
+ " id BIGINT,"
+ " ts_millis BIGINT,"
+ " doubled AS id * 2,"
+ " old_ts AS TO_TIMESTAMP_LTZ(ts_millis, 3),"
+ " WATERMARK FOR old_ts AS old_ts,"
+ " PRIMARY KEY (id) NOT ENFORCED"
+ ") COMMENT 'timed events' PARTITIONED BY (id) WITH ("
+ " 'connector' = 'datagen', 'number-of-rows' = '1'"
+ ")"
+ )
+
+ def test_read_catalog_table_extends_schema(self):
+ self._create_timed_table()
+ extension = dict(
+ computed_columns={"event_time": "TO_TIMESTAMP_LTZ(ts_millis, 3)"},
+ watermark=("event_time", "event_time - INTERVAL '5' SECOND"),
+ )
+
+ dataframe = pf.read_catalog_table("my_catalog.my_database.timed",
**extension)
+
+ self.assert_dataframe_schema(
+ dataframe, ["id", "ts_millis", "doubled", "old_ts", "event_time"]
+ )
+ resolved_schema = dataframe._table.get_resolved_schema()
+ (watermark_spec,) = resolved_schema.get_watermark_specs()
+ self.assertEqual(watermark_spec.get_rowtime_attribute(), "event_time")
+
self.assertEqual(list(resolved_schema.get_primary_key().get_columns()), ["id"])
+
+ # The copy keeps everything the connector needs from the original
table.
+ j_source =
dataframe._table._j_table.getQueryOperation().getContextResolvedTable()
+ self.assertTrue(j_source.isAnonymous())
+ j_catalog_table = j_source.getTable()
+ self.assertEqual(
+ dict(j_catalog_table.getOptions()),
+ {"connector": "datagen", "number-of-rows": "1"},
+ )
+ self.assertEqual(list(j_catalog_table.getPartitionKeys()), ["id"])
+ self.assertEqual(j_catalog_table.getComment(), "timed events")
+
+ # The catalog table itself is left untouched.
+ catalog_watermarks = self.t_env.from_path(
+ "my_catalog.my_database.timed"
+ ).get_resolved_schema().get_watermark_specs()
+ self.assertEqual(
+ [spec.get_rowtime_attribute() for spec in catalog_watermarks],
["old_ts"]
+ )
+
+ def test_read_catalog_table_keeps_source_watermark_without_override(self):
+ self._create_timed_table()
+ dataframe = pf.read_catalog_table(
+ "my_catalog.my_database.timed", computed_columns={"tripled": "id *
3"}
+ )
+ self.assert_dataframe_schema(
+ dataframe, ["id", "ts_millis", "doubled", "old_ts", "tripled"]
+ )
+ resolved_schema = dataframe._table.get_resolved_schema()
+ self.assertEqual(
+ [spec.get_rowtime_attribute() for spec in
resolved_schema.get_watermark_specs()],
+ ["old_ts"],
+ )
+
+ def test_read_catalog_table_without_extension_keeps_catalog_identity(self):
+ cases = [{}, {"computed_columns": {}}]
+
+ for arguments in cases:
+ with self.subTest(arguments=arguments):
+ dataframe = pf.read_catalog_table(
+ "my_catalog.my_database.events", **arguments
+ )
+ j_source = (
+
dataframe._table._j_table.getQueryOperation().getContextResolvedTable()
+ )
+ self.assertFalse(j_source.isAnonymous())
+ self.assertEqual(
+ j_source.getIdentifier().asSummaryString(),
+ "my_catalog.my_database.events",
+ )
+
+ def test_read_catalog_table_rejects_invalid_extensions(self):
+ self.t_env.execute_sql("CREATE TABLE
my_catalog.my_database.no_connector (id BIGINT)")
+ self.t_env.execute_sql(
+ "CREATE VIEW my_catalog.my_database.events_view AS "
+ "SELECT id FROM my_catalog.my_database.events"
+ )
+ cases = [
+ (
+ "events",
+ {"computed_columns": ["id + 1"]},
+ TypeError,
+ "computed_columns must be a dict",
+ ),
+ (
+ "events",
+ {"computed_columns": []},
+ TypeError,
+ "computed_columns must be a dict",
+ ),
+ (
+ "events",
+ {"watermark": "id"},
+ TypeError,
+ "watermark must be a tuple",
+ ),
+ (
+ "events_view",
+ {"computed_columns": {"tripled": "id * 3"}},
+ ValueError,
+ "my_catalog.my_database.events_view is a view",
+ ),
+ (
+ "no_connector",
+ {"computed_columns": {"tripled": "id * 3"}},
+ ValueError,
+ "my_catalog.my_database.no_connector has no 'connector'
option",
+ ),
+ ]
+ for table, extension, error_type, message in cases:
+ with self.subTest(table=table, extension=extension):
+ with self.assertRaisesRegex(error_type, message):
+ pf.read_catalog_table(f"my_catalog.my_database.{table}",
**extension)
+
+ def test_read_catalog_table_translates_extension_errors(self):
+ cases = [
+ (
+ {"computed_columns": {"broken": "no_such_column + 1"}},
+ "Invalid expression for computed column 'broken'",
+ ),
+ (
+ {"computed_columns": {"id": "id + 1"}},
+ "Schema must not contain duplicate column names. Found
duplicates: [id]",
+ ),
+ (
+ {"watermark": ("no_such_column", "no_such_column")},
+ "Invalid column name 'no_such_column' for rowtime attribute",
+ ),
+ ]
+ for extension, message in cases:
+ with self.subTest(extension=extension):
+ with self.assertRaises(ValueError) as context:
+ pf.read_catalog_table("my_catalog.my_database.events",
**extension)
+ self.assertIn(message, str(context.exception))
+ self.assertIsInstance(context.exception.__cause__,
Py4JJavaError)
+
+ def test_write_catalog_table_translates_flink_errors(self):
+ dataframe = pf.from_records([(1, "a")], schema=["id", "name"])
+
+ with self.assertRaises(ValueError) as context:
+ dataframe.write_catalog_table("my_catalog.my_database.missing")
+ self.assertIn(
+ "Cannot find table '`my_catalog`.`my_database`.`missing`'",
+ str(context.exception),
+ )
+
+ self.t_env.execute_sql(
+ "CREATE TABLE my_catalog.my_database.sink (id BIGINT) "
+ "WITH ('connector' = 'blackhole')"
+ )
+ with self.assertRaises(ValueError) as context:
+ dataframe.write_catalog_table("my_catalog.my_database.sink")
+ self.assertIn("Column types of query result and sink",
str(context.exception))
+ self.assertIsInstance(context.exception.__cause__, Py4JJavaError)
+
+ def test_write_catalog_table_passes_path_and_overwrite(self):
+ dataframe = pf.from_records([(1, "a")], schema=["id", "name"])
+
+ for overwrite in [False, True]:
+ with self.subTest(overwrite=overwrite):
+ with patch.object(
+ dataframe._table, "execute_insert",
return_value=MagicMock()
+ ) as execute_insert:
+ result = dataframe.write_catalog_table(
+ "my_catalog.my_database.events", overwrite=overwrite
+ )
+
+ self.assertIsNone(result)
+ execute_insert.assert_called_once_with(
+ "my_catalog.my_database.events", overwrite=overwrite
+ )
+
+ def
test_write_catalog_table_waits_for_local_and_minicluster_execution(self):
+ dataframe = pf.from_records([(1,)], schema=["id"])
+
+ for execution_target, waits in [
+ ("local", True),
+ ("minicluster", True),
+ ("remote", False),
+ ]:
+ with self.subTest(execution_target=execution_target):
+ table_result = MagicMock()
+ table_config = MagicMock()
+ table_config.get.return_value = execution_target
+ with patch.object(
+ dataframe._table,
+ "execute_insert",
+ return_value=table_result,
+ ), patch.object(
+ dataframe._table._t_env,
+ "get_config",
+ return_value=table_config,
+ ):
+ dataframe.write_catalog_table("events")
+
+ if waits:
+ table_result.wait.assert_called_once_with()
+ else:
+ table_result.wait.assert_not_called()
+
+ def test_write_catalog_table_rejects_invalid_arguments(self):
+ dataframe = pf.from_records([(1,)], schema=["id"])
+ cases = [
+ (None, False, TypeError, "path must be a string"),
+ ("", False, ValueError, "path must not be empty"),
+ ("events", "yes", TypeError, "overwrite must be a bool"),
+ ]
+
+ for path, overwrite, error_type, message in cases:
+ with self.subTest(path=path, overwrite=overwrite):
+ with self.assertRaisesRegex(error_type, message):
+ dataframe.write_catalog_table(path, overwrite=overwrite)
+
class GenericIOITTests(PyFlinkStreamDataFrameTestCase):
+ def test_catalog_table_round_trip(self):
+ output_path = os.path.join(self.tempdir, "catalog_output")
+ pf.create_catalog("my_catalog", {"type": "generic_in_memory"})
+ self.addCleanup(pf.use_catalog, pf.get_current_catalog())
+ self.t_env.execute_sql("CREATE DATABASE my_catalog.my_database")
+ self.t_env.execute_sql(
+ "CREATE TABLE my_catalog.my_database.events ("
+ " id BIGINT, name STRING"
+ ") WITH ("
+ " 'connector' = 'filesystem',"
+ f" 'path' = '{output_path}',"
+ " 'format' = 'csv'"
+ ")"
+ )
+
+ pf.use_catalog("my_catalog")
+ pf.use_database("my_database")
+ source = pf.from_records([(1, "a"), (2, "b"), (3, "c")], schema=["id",
"name"])
+ source.write_catalog_table("events")
+
+ result = pf.read_catalog_table("events").collect()
+ self.assertEqual(
+ sorted(tuple(row) for row in result), [(1, "a"), (2, "b"), (3,
"c")]
+ )
+
+ def test_catalog_table_with_computed_columns_and_watermark(self):
+ input_path = os.path.join(self.tempdir, "timed.csv")
+ with open(input_path, "w", encoding="utf-8") as input_file:
+ input_file.write("1,1000\n2,2000\n3,3000\n")
+ pf.create_catalog("timed_catalog", {"type": "generic_in_memory"})
+ self.t_env.execute_sql(
+ "CREATE TABLE timed_catalog.`default`.timed ("
+ " id BIGINT, ts_millis BIGINT"
+ ") WITH ("
+ " 'connector' = 'filesystem',"
+ f" 'path' = '{input_path}',"
+ " 'format' = 'csv'"
+ ")"
+ )
+
+ result = pf.read_catalog_table(
+ "timed_catalog.`default`.timed",
+ computed_columns={
+ "ts_seconds": "ts_millis / 1000",
+ "event_time": "TO_TIMESTAMP_LTZ(ts_millis, 3)",
+ },
+ watermark=("event_time", "event_time - INTERVAL '1' SECOND"),
+ ).select("id", "ts_seconds").collect()
+ self.assertEqual(sorted(tuple(row) for row in result), [(1, 1), (2,
2), (3, 3)])
+
def test_filesystem_csv_round_trip(self):
input_path = os.path.join(self.tempdir, "input.csv")
with open(input_path, "w", encoding="utf-8") as input_file:
