Copilot commented on code in PR #11189:
URL: https://github.com/apache/gravitino/pull/11189#discussion_r3296212160
##########
flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java:
##########
@@ -748,6 +904,155 @@ 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 = preferredDialect();
+ String sql =
+ view.sqlFor(dialect)
+ .map(SQLRepresentation::sql)
+ .orElseGet(
+ () -> {
+ Representation[] reps = view.representations();
+ if (reps.length > 0 && reps[0] instanceof SQLRepresentation)
{
+ SQLRepresentation fallback = (SQLRepresentation) reps[0];
+ 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",
+ view.name(), catalogName()));
Review Comment:
toFlinkView() falls back to only reps[0] when the preferred dialect isn’t
present. If representations are ordered with a non-SQL representation first (or
the SQL representation is later in the array), this will throw even though a
usable SQLRepresentation exists. Consider scanning representations() for the
first SQLRepresentation (or the first any-dialect SQLRepresentation) instead of
assuming index 0.
##########
flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java:
##########
@@ -94,12 +102,23 @@ protected AbstractCatalog realCatalog() {
return hiveCatalog;
}
+ /** {@inheritDoc} Returns {@link Dialects#HIVE} for Hive catalog views. */
+ @Override
+ protected String preferredDialect() {
+ return Dialects.HIVE;
Review Comment:
GravitinoHiveCatalog.preferredDialect() hard-codes Dialects.HIVE, but
BaseCatalog stores the view SQL using ResolvedCatalogView.getExpandedQuery()
generated by Flink’s current SQL dialect (often FLINK unless explicitly
switched). This can mislabel FLINK-dialect SQL as HIVE, which may break
consumers that look up representations by dialect. Consider keeping the default
Dialects.FLINK here, or deriving the dialect from Flink’s configured SqlDialect
when creating/altering views.
##########
flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java:
##########
@@ -745,6 +749,278 @@ public void testDefaultFormatAndSerdeApplied() {
true);
}
+ @Test
+ public void testCreateView() {
+ String schemaName = "test_hive_create_view_db";
+ String viewName = "test_view_create";
+ String tableName = "test_view_base_table";
+ doWithSchema(
+ currentCatalog(),
+ schemaName,
+ catalog -> {
+ TestUtils.assertTableResult(
+ sql("CREATE TABLE %s (id INT, name STRING) WITH
('connector'='hive')", tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql(
+ "CREATE VIEW %s COMMENT 'view comment' AS SELECT id, name
FROM %s",
+ viewName, tableName),
+ ResultKind.SUCCESS);
+
+ // Verify via Gravitino ViewCatalog
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ View view = viewCatalog.loadView(NameIdentifier.of(schemaName,
viewName));
+ Assertions.assertEquals(viewName, view.name());
+ Assertions.assertEquals("view comment", view.comment());
+ Assertions.assertEquals(1, view.representations().length);
+ Assertions.assertInstanceOf(SQLRepresentation.class,
view.representations()[0]);
+
+ // Verify via Flink catalog API
+ Optional<Catalog> flinkCatalog = tableEnv.getCatalog(catalog.name());
+ Assertions.assertTrue(flinkCatalog.isPresent());
+ try {
+ CatalogBaseTable flinkTable =
+ ((GravitinoHiveCatalog) flinkCatalog.get())
+ .getTable(new ObjectPath(schemaName, viewName));
Review Comment:
The test unnecessarily casts the catalog from tableEnv.getCatalog(...) to
GravitinoHiveCatalog just to call getTable(). The returned type is already
Flink's Catalog interface which exposes getTable(ObjectPath), so the cast can
be removed to keep the test resilient to wrapper/proxy catalog implementations.
--
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]
