fdolce commented on code in PR #29262:
URL: https://github.com/apache/flink/pull/29262#discussion_r4092525400
##########
flink-python/pyflink/dataframe/io.py:
##########
@@ -189,4 +185,155 @@ def read_generic(
source_schema = _build_source_schema(schema, computed_columns, watermark)
descriptor = _build_generic_descriptor(connector, options, source_schema)
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)
+
+
+def _extend_catalog_table(
+ table: Table,
+ computed_columns: Optional[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"):
Review Comment:
Ok, rebased
--
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]
