pvary commented on code in PR #17859:
URL: https://github.com/apache/iceberg/pull/17859#discussion_r3986887006
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -683,6 +740,44 @@ static CatalogTable toCatalogTable(Table table) {
return toCatalogTableWithProps(table, table.properties());
}
+ private CatalogView toCatalogView(ObjectPath tablePath, View view) {
+ SQLViewRepresentation sqlRepresentation = view.sqlFor(FLINK_DIALECT);
+ Preconditions.checkState(sqlRepresentation != null, "Cannot load SQL for
view %s", view.name());
+
+ // Flink resolves against the view's location; mismatched stored defaults
risk wrong results
+ ViewVersion currentVersion = view.currentVersion();
+ String defaultCatalog = currentVersion.defaultCatalog();
+ if (defaultCatalog != null && !defaultCatalog.equals(getName())) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Cannot read view %s: its default-catalog '%s' does not match
the catalog '%s' it is read through, "
+ + "and Flink resolves references against the view's own
catalog",
+ tablePath, defaultCatalog, getName()));
+ }
+
+ Namespace defaultNamespace = currentVersion.defaultNamespace();
+ Namespace viewNamespace = appendLevel(baseNamespace,
tablePath.getDatabaseName());
+ if (!defaultNamespace.isEmpty() &&
!defaultNamespace.equals(viewNamespace)) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Cannot read view %s: its default-namespace '%s' does not match
the view's namespace '%s', "
+ + "and Flink resolves references against the view's own
namespace",
+ tablePath, defaultNamespace, viewNamespace));
+ }
+
+ ResolvedSchema resolvedSchema =
FlinkSchemaUtil.toResolvedSchema(view.schema());
+ org.apache.flink.table.api.Schema schema =
+
org.apache.flink.table.api.Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();
+
+ Map<String, String> options = Maps.newHashMap(view.properties());
Review Comment:
Question: Do we want to filter properties, like it is done in Spark?
See:
https://github.com/apache/iceberg/blob/89987a6174ba8cb8f1ab539e693321cea1604189/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkView.java#L196-L203
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
