fdolce commented on code in PR #29262:
URL: https://github.com/apache/flink/pull/29262#discussion_r4091925474
##########
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:
Sorry for the long message, but I want to add some context 😄
The way this is implemented, Flink needs to find a connector for the table
we read, because the connector is what actually reads the data. Instead of
reading the original table directly, we build a new table definition with the
extra columns or watermark, and Flink has to work out which connector serves
it. Most tables make that easy with a connector option, like 'connector' =
'kafka', which Flink uses to find the matching connector on the classpath.
Catalog-native tables like Paimon or Fluss don't have that option, because the
catalog itself is the connector. Flink asks the catalog for its connector
through Catalog.getFactory(). The catalog also knows things the table's own
options don't include, such as where the warehouse lives or how to reach the
cluster.
The catch is that Flink only asks the catalog when the table being read is
stored in that catalog. Our new table definition isn't stored anywhere, so
Flink falls back to the connector option, finds nothing, and the read fails.
Setting the option by hand wouldn't fix it in general either, since the
connector often still needs those catalog-level settings.
Fixing this properly means changing the Java side of the planner. The new
table definition would need to remember which catalog it came from, so the
planner can still ask that catalog for its connector. That touches how the
planner picks a connector for every table it reads. It also touches how
compiled plans are saved and restored, because a restored plan would lose that
catalog link unless we store it. I think those are outside the scope of this
PR, so a separate ticket for it makes sense.
That said, computed_columns on its own could work for these tables today.
We'd read the table as it is and add the columns as a projection on top, which
doesn't need a new table definition at all. The watermark is the only part that
has to be declared in the schema. The downside is that the two arguments would
then behave differently on catalog-native tables: computed_columns would work
and watermark would still be rejected. I think it makes sense to do both in the
follow-up ticket.
--
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]
