talatuyarer commented on code in PR #17859:
URL: https://github.com/apache/iceberg/pull/17859#discussion_r3990639426
##########
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());
Review Comment:
The base-namespace parameterization was specific to `HadoopCatalog`, which
the view tests skip. I have rebased onto main and added REST parameterizations
to `TestFlinkCatalogView` to properly exercise namespace comparisons and
provide coverage for non-Hive `ViewCatalog` implementations.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -321,8 +339,30 @@ public void alterDatabase(String name, CatalogDatabase
newDatabase, boolean igno
@Override
public List<String> listTables(String databaseName)
throws DatabaseNotExistException, CatalogException {
+ List<String> results;
try {
- return icebergCatalog.listTables(appendLevel(baseNamespace,
databaseName)).stream()
+ results =
+ icebergCatalog.listTables(appendLevel(baseNamespace,
databaseName)).stream()
+ .map(TableIdentifier::name)
+ .collect(Collectors.toList());
Review Comment:
Done
--
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]
