mchades commented on code in PR #11189:
URL: https://github.com/apache/gravitino/pull/11189#discussion_r3302167270
##########
flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -748,6 +889,146 @@ static SchemaChange[] getSchemaChange(CatalogDatabase
current, CatalogDatabase u
return schemaChanges.toArray(new SchemaChange[0]);
}
+ /**
+ * Loads the entity at {@code tablePath} as a Flink {@link CatalogView} if
it is a view in the
+ * underlying ViewCatalog. Throws {@link TableNotExistException} if no view
is found (or if the
+ * catalog does not support views).
+ */
+ protected CatalogBaseTable loadViewOrThrow(ObjectPath tablePath) throws
TableNotExistException {
+ NameIdentifier ident =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
+ try {
+ View view = catalog().asViewCatalog().loadView(ident);
+ return toFlinkView(view);
+ } catch (NoSuchViewException e) {
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ } catch (UnsupportedOperationException e) {
+ LOG.debug(
+ "Catalog {} does not support views; treating {} as not found",
+ catalogName(),
+ tablePath,
+ e);
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
+ }
+
+ /**
+ * Converts a Gravitino {@link View} to a Flink {@link CatalogView}.
+ *
+ * @param view The Gravitino view to convert.
+ * @return The corresponding Flink CatalogView.
+ */
+ protected CatalogView toFlinkView(View view) {
+ org.apache.flink.table.api.Schema.Builder builder =
buildSchemaFromColumns(view.columns());
+ String dialect = Dialects.FLINK;
+ String sql =
+ view.sqlFor(dialect)
+ .map(SQLRepresentation::sql)
+ .orElseGet(
+ () -> {
+ for (Representation rep : view.representations()) {
+ if (rep instanceof SQLRepresentation) {
+ SQLRepresentation fallback = (SQLRepresentation) rep;
+ LOG.warn(
+ "View {} has no SQL representation for dialect {};
falling back to dialect {}",
Review Comment:
Should we explicitly fallback to the Hive dialect here? Does Flink support
views with other dialects?
##########
flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -748,6 +889,146 @@ static SchemaChange[] getSchemaChange(CatalogDatabase
current, CatalogDatabase u
return schemaChanges.toArray(new SchemaChange[0]);
}
+ /**
+ * Loads the entity at {@code tablePath} as a Flink {@link CatalogView} if
it is a view in the
+ * underlying ViewCatalog. Throws {@link TableNotExistException} if no view
is found (or if the
+ * catalog does not support views).
+ */
+ protected CatalogBaseTable loadViewOrThrow(ObjectPath tablePath) throws
TableNotExistException {
+ NameIdentifier ident =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
+ try {
+ View view = catalog().asViewCatalog().loadView(ident);
+ return toFlinkView(view);
+ } catch (NoSuchViewException e) {
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ } catch (UnsupportedOperationException e) {
+ LOG.debug(
+ "Catalog {} does not support views; treating {} as not found",
+ catalogName(),
+ tablePath,
+ e);
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
+ }
+
+ /**
+ * Converts a Gravitino {@link View} to a Flink {@link CatalogView}.
+ *
+ * @param view The Gravitino view to convert.
+ * @return The corresponding Flink CatalogView.
+ */
+ protected CatalogView toFlinkView(View view) {
+ org.apache.flink.table.api.Schema.Builder builder =
buildSchemaFromColumns(view.columns());
+ String dialect = Dialects.FLINK;
+ String sql =
+ view.sqlFor(dialect)
+ .map(SQLRepresentation::sql)
+ .orElseGet(
+ () -> {
+ for (Representation rep : view.representations()) {
+ if (rep instanceof SQLRepresentation) {
+ SQLRepresentation fallback = (SQLRepresentation) rep;
+ LOG.warn(
+ "View {} has no SQL representation for dialect {};
falling back to dialect {}",
+ view.name(),
+ dialect,
+ fallback.dialect());
+ return fallback.sql();
+ }
+ }
+ throw new CatalogException(
+ String.format(
+ "View '%s' in catalog '%s' has no SQL representation
for dialect '%s'",
+ view.name(), catalogName(), dialect));
+ });
+
+ Map<String, String> properties =
+ view.properties() != null
+ ? Collections.unmodifiableMap(view.properties())
+ : Collections.emptyMap();
+ String comment = view.comment() != null ? view.comment() : "";
Review Comment:
The comment is nullable, so it seems unnecessary to do the conversion?
##########
flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java:
##########
@@ -145,20 +157,31 @@ public CatalogBaseTable getTable(ObjectPath tablePath)
catalog()
.asTableCatalog()
.loadTable(NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()));
+ if
(HIVE_VIRTUAL_VIEW_TYPE.equalsIgnoreCase(table.properties().get(HIVE_TABLE_TYPE_KEY)))
{
+ // Hive HMS stores VIRTUAL_VIEW entries as table entries also returned
by loadTable.
+ return loadViewOrThrow(tablePath);
Review Comment:
It appears that we should modify the server-side Hive catalog logic to
ensure that the `loadTable` method does not return views, rather than
performing special additional checks on the client side.
##########
flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -748,6 +889,146 @@ static SchemaChange[] getSchemaChange(CatalogDatabase
current, CatalogDatabase u
return schemaChanges.toArray(new SchemaChange[0]);
}
+ /**
+ * Loads the entity at {@code tablePath} as a Flink {@link CatalogView} if
it is a view in the
+ * underlying ViewCatalog. Throws {@link TableNotExistException} if no view
is found (or if the
+ * catalog does not support views).
+ */
+ protected CatalogBaseTable loadViewOrThrow(ObjectPath tablePath) throws
TableNotExistException {
+ NameIdentifier ident =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
+ try {
+ View view = catalog().asViewCatalog().loadView(ident);
+ return toFlinkView(view);
+ } catch (NoSuchViewException e) {
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ } catch (UnsupportedOperationException e) {
+ LOG.debug(
+ "Catalog {} does not support views; treating {} as not found",
+ catalogName(),
+ tablePath,
+ e);
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
+ }
+
+ /**
+ * Converts a Gravitino {@link View} to a Flink {@link CatalogView}.
+ *
+ * @param view The Gravitino view to convert.
+ * @return The corresponding Flink CatalogView.
+ */
+ protected CatalogView toFlinkView(View view) {
+ org.apache.flink.table.api.Schema.Builder builder =
buildSchemaFromColumns(view.columns());
+ String dialect = Dialects.FLINK;
+ String sql =
+ view.sqlFor(dialect)
+ .map(SQLRepresentation::sql)
+ .orElseGet(
+ () -> {
+ for (Representation rep : view.representations()) {
+ if (rep instanceof SQLRepresentation) {
+ SQLRepresentation fallback = (SQLRepresentation) rep;
+ LOG.warn(
+ "View {} has no SQL representation for dialect {};
falling back to dialect {}",
+ view.name(),
+ dialect,
+ fallback.dialect());
+ return fallback.sql();
+ }
+ }
+ throw new CatalogException(
+ String.format(
+ "View '%s' in catalog '%s' has no SQL representation
for dialect '%s'",
+ view.name(), catalogName(), dialect));
+ });
+
+ Map<String, String> properties =
+ view.properties() != null
+ ? Collections.unmodifiableMap(view.properties())
+ : Collections.emptyMap();
+ String comment = view.comment() != null ? view.comment() : "";
+ return CatalogView.of(builder.build(), comment, sql, sql, properties);
+ }
+
+ @VisibleForTesting
+ static ViewChange[] getGravitinoViewChanges(
Review Comment:
It is preferable to use an explicit name such as `toReplaceViewChange`.
--
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]
