godfreyhe commented on a change in pull request #12335:
URL: https://github.com/apache/flink/pull/12335#discussion_r435773307
##########
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
##########
@@ -490,8 +490,11 @@ private void insertIntoInternal(UnresolvedIdentifier
unresolvedIdentifier, Table
private Optional<CatalogQueryOperation>
scanInternal(UnresolvedIdentifier identifier) {
ObjectIdentifier tableIdentifier =
catalogManager.qualifyIdentifier(identifier);
- return catalogManager.getTable(tableIdentifier)
- .map(t -> new CatalogQueryOperation(tableIdentifier,
t.getTable().getSchema()));
+ return catalogManager.getTable(tableIdentifier).map(t -> {
+ CatalogTableSchemaResolver resolver = new
CatalogTableSchemaResolver(parser);
Review comment:
We can fix the `rowtime` type in `TableSchema` based on `WatermarkSpec`,
but we can‘t fix the `proctime` type, unless we just use string
compare(`proctime()`). As discussed in
https://github.com/apache/flink/pull/12260, we should use
`SqlExprToRexConverter` to get the correct type. For table api, there is the
entry point to create a flink `Table`, and for sql, `CatalogSchemaTable` is the
entry point to create a calcite `Table`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
