This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 3a8cc60bc0 [#11187] feat(flink-connector): Add view support (#11189)
3a8cc60bc0 is described below
commit 3a8cc60bc02f3f1c625454a99edcb593fafefb1f
Author: Yuhui <[email protected]>
AuthorDate: Wed May 27 10:59:32 2026 +0800
[#11187] feat(flink-connector): Add view support (#11189)
### What changes were proposed in this pull request?
Route Flink SQL view operations (CREATE/DROP/ALTER/LIST VIEW) through
Gravitino's `ViewCatalog` API in `BaseCatalog`. Any Gravitino catalog
that implements `ViewCatalog` gets view support automatically.
### Why are the changes needed?
Fix: #11187
### Does this PR introduce _any_ user-facing change?
Yes — users can now use `CREATE VIEW`, `DROP VIEW`, `ALTER VIEW`,
`SHOW VIEWS`, and `SELECT` from views via Flink SQL on Gravitino
catalogs.
### How was this patch tested?
- Unit tests: `TestBaseCatalog` — listViews, view change conversion
- Integration tests: 9 new cases in `FlinkHiveCatalogIT` covering
create, list, drop, rename, replace body, query, and idempotency
---
.../catalog/hive/HiveCatalogOperations.java | 4 +-
.../apache/gravitino/catalog/hive/HiveView.java | 9 +-
.../catalog/hive/HiveViewCatalogOperations.java | 52 +--
.../catalog/hive/TestHiveCatalogOperations.java | 8 +-
.../flink/connector/catalog/BaseCatalog.java | 354 ++++++++++++++++++---
.../flink/connector/hive/GravitinoHiveCatalog.java | 22 +-
.../flink/connector/catalog/TestBaseCatalog.java | 160 +++++++++-
.../integration/test/hive/FlinkHiveCatalogIT.java | 275 ++++++++++++++++
8 files changed, 804 insertions(+), 80 deletions(-)
diff --git
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
index 5ff4f27ddc..87825ca913 100644
---
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
+++
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java
@@ -476,7 +476,9 @@ public class HiveCatalogOperations
@Override
public Table loadTable(NameIdentifier tableIdent) throws
NoSuchTableException {
HiveTableHandle hiveTable = loadHiveTable(tableIdent);
-
+ if
(TableType.VIRTUAL_VIEW.name().equalsIgnoreCase(hiveTable.getTableType())) {
+ throw new NoSuchTableException("Table %s is a view, not a table",
tableIdent);
+ }
LOG.info("Loaded Hive table {} from Hive Metastore ", tableIdent.name());
return hiveTable;
}
diff --git
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
index 1e28045c55..bae828e7f6 100644
---
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
+++
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveView.java
@@ -41,8 +41,8 @@ import org.apache.gravitino.rel.View;
/**
* Represents a view stored in Hive Metastore (VIRTUAL_VIEW table type). The
SQL dialect is detected
* from table properties: Trino views start with "/* Presto View:", Spark
views carry {@code
- * spark.sql.create.version} in their parameters, and all other views are
treated as native Hive SQL
- * views.
+ * spark.sql.create.version}, Flink views carry properties prefixed with
{@code flink.}, and all
+ * other views are treated as native Hive SQL views.
*/
@Unstable
@NoArgsConstructor(access = AccessLevel.PRIVATE)
@@ -112,7 +112,7 @@ public class HiveView implements View {
*
* @param viewOriginalText The original view text from HMS.
* @param parameters The HMS table parameters map.
- * @return The detected dialect string: "trino", "spark", or "hive".
+ * @return The detected dialect string: "trino", "spark", "flink", or "hive".
*/
static String detectDialect(String viewOriginalText, Map<String, String>
parameters) {
if (parameters != null &&
"true".equalsIgnoreCase(parameters.get(TRINO_VIEW_MARKER_KEY))) {
@@ -124,6 +124,9 @@ public class HiveView implements View {
if (parameters != null && parameters.containsKey(SPARK_VERSION_KEY)) {
return Dialects.SPARK;
}
+ if (parameters != null && parameters.keySet().stream().anyMatch(k ->
k.startsWith("flink."))) {
+ return Dialects.FLINK;
+ }
return Dialects.HIVE;
}
diff --git
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
index e35b265949..1200391bbc 100644
---
a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
+++
b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java
@@ -363,18 +363,19 @@ class HiveViewCatalogOperations implements ViewCatalog {
Maps.newHashMap(properties != null ? properties : ImmutableMap.of());
String representationSql = viewOriginalText;
String detectedDialect = HiveView.detectDialect(representationSql, params);
- if (!Dialects.HIVE.equalsIgnoreCase(detectedDialect)) {
+ if (!Dialects.HIVE.equalsIgnoreCase(detectedDialect)
+ && !Dialects.FLINK.equalsIgnoreCase(detectedDialect)) {
// TODO(design-docs/gravitino-logical-view-management.md): support
loading trino/spark HMS
// views.
throw new UnsupportedOperationException(
String.format(
- "Hive catalog currently supports only '%s' view dialect, but
found '%s' for view %s",
- Dialects.HIVE, detectedDialect, ident));
+ "Hive catalog currently supports only '%s' and '%s' view
dialects, but found '%s' for view %s",
+ Dialects.HIVE, Dialects.FLINK, detectedDialect, ident));
}
SQLRepresentation rep =
SQLRepresentation.builder()
- .withDialect(Dialects.HIVE)
+ .withDialect(detectedDialect)
.withSql(StringUtils.defaultString(representationSql))
.build();
@@ -405,34 +406,37 @@ class HiveViewCatalogOperations implements ViewCatalog {
firstRepresentation == null ? "null" :
firstRepresentation.getClass().getSimpleName());
SQLRepresentation selected = (SQLRepresentation) firstRepresentation;
- boolean isHiveDialect = Dialects.HIVE.equalsIgnoreCase(selected.dialect());
- if (isHiveDialect) {
- Preconditions.checkArgument(
- defaultCatalog == null && defaultSchema == null,
- "Hive dialect '%s' does not support non-null
defaultCatalog/defaultSchema, but got "
- + "defaultCatalog=%s, defaultSchema=%s for view %s",
- Dialects.HIVE,
- defaultCatalog,
- defaultSchema,
- ident);
- return selected;
+ switch (selected.dialect().toLowerCase(java.util.Locale.ROOT)) {
+ case Dialects.HIVE:
+ case Dialects.FLINK:
+ Preconditions.checkArgument(
+ defaultCatalog == null && defaultSchema == null,
+ "Dialect '%s' does not support non-null
defaultCatalog/defaultSchema, but got "
+ + "defaultCatalog=%s, defaultSchema=%s for view %s",
+ selected.dialect(),
+ defaultCatalog,
+ defaultSchema,
+ ident);
+ return selected;
+ default:
+ // TODO(design-docs/gravitino-logical-view-management.md): support
creating trino/spark HMS
+ // views.
+ throw new UnsupportedOperationException(
+ String.format(
+ "Hive catalog currently supports only '%s' and '%s' view
dialects, but got '%s' for view %s",
+ Dialects.HIVE, Dialects.FLINK, selected.dialect(), ident));
}
- // TODO(design-docs/gravitino-logical-view-management.md): support
creating trino/spark HMS
- // views.
- throw new UnsupportedOperationException(
- String.format(
- "Hive catalog currently supports only '%s' view dialect, but got
'%s' for view %s",
- Dialects.HIVE, selected.dialect(), ident));
}
private String toHmsViewOriginalText(SQLRepresentation representation,
NameIdentifier ident) {
- if (!Dialects.HIVE.equalsIgnoreCase(representation.dialect())) {
+ if (!Dialects.HIVE.equalsIgnoreCase(representation.dialect())
+ && !Dialects.FLINK.equalsIgnoreCase(representation.dialect())) {
// TODO(design-docs/gravitino-logical-view-management.md): support
serializing trino/spark HMS
// view definitions.
throw new UnsupportedOperationException(
String.format(
- "Hive catalog currently supports only '%s' view dialect, but got
'%s' for view %s",
- Dialects.HIVE, representation.dialect(), ident));
+ "Hive catalog currently supports only '%s' and '%s' view
dialects, but got '%s' for view %s",
+ Dialects.HIVE, Dialects.FLINK, representation.dialect(), ident));
}
return representation.sql();
}
diff --git
a/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
b/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
index 56154cbf41..a8ab3142d9 100644
---
a/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
+++
b/catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java
@@ -245,7 +245,7 @@ class TestHiveCatalogOperations {
null,
null,
Maps.newHashMap()));
- Assertions.assertTrue(exception.getMessage().contains("supports only
'hive'"));
+ Assertions.assertTrue(exception.getMessage().contains("supports only"));
}
@Test
@@ -403,7 +403,7 @@ class TestHiveCatalogOperations {
null,
null,
Maps.newHashMap()));
- Assertions.assertTrue(exception.getMessage().contains("supports only
'hive'"));
+ Assertions.assertTrue(exception.getMessage().contains("supports only"));
}
@Test
@@ -510,7 +510,7 @@ class TestHiveCatalogOperations {
Assertions.assertThrows(
UnsupportedOperationException.class,
() -> op.loadView(NameIdentifier.of("db", "v_trino")));
- Assertions.assertTrue(exception.getMessage().contains("supports only
'hive'"));
+ Assertions.assertTrue(exception.getMessage().contains("supports only"));
}
@Test
@@ -628,7 +628,7 @@ class TestHiveCatalogOperations {
null,
null,
null)));
- Assertions.assertTrue(exception.getMessage().contains("supports only
'hive'"));
+ Assertions.assertTrue(exception.getMessage().contains("supports only"));
}
@Test
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
index 3fdbbb4eef..f3fa79c29d 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
@@ -41,8 +41,10 @@ import org.apache.flink.table.catalog.CatalogFunction;
import org.apache.flink.table.catalog.CatalogPartition;
import org.apache.flink.table.catalog.CatalogPartitionSpec;
import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
+import org.apache.flink.table.catalog.ResolvedCatalogView;
import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
@@ -69,9 +71,11 @@ import org.apache.gravitino.SchemaChange;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NoSuchTableException;
+import org.apache.gravitino.exceptions.NoSuchViewException;
import org.apache.gravitino.exceptions.NonEmptySchemaException;
import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
import org.apache.gravitino.exceptions.TableAlreadyExistsException;
+import org.apache.gravitino.exceptions.ViewAlreadyExistsException;
import org.apache.gravitino.flink.connector.PartitionConverter;
import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
import org.apache.gravitino.flink.connector.utils.CatalogCompat;
@@ -79,20 +83,31 @@ import
org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
import org.apache.gravitino.flink.connector.utils.TableUtils;
import org.apache.gravitino.flink.connector.utils.TypeUtils;
import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Dialects;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.SQLRepresentation;
import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.rel.View;
+import org.apache.gravitino.rel.ViewCatalog;
+import org.apache.gravitino.rel.ViewChange;
import org.apache.gravitino.rel.expressions.distributions.Distribution;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.rel.indexes.Indexes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* The BaseCatalog that provides a default implementation for all methods in
the {@link
* org.apache.flink.table.catalog.Catalog} interface.
*/
public abstract class BaseCatalog extends AbstractCatalog {
+ private static final Logger LOG = LoggerFactory.getLogger(BaseCatalog.class);
+ private static final String FLINK_SCHEMA_NUM_COLUMNS_KEY =
"flink.schema.num-columns";
+
private final SchemaAndTablePropertiesConverter
schemaAndTablePropertiesConverter;
private final PartitionConverter partitionConverter;
private final Map<String, String> catalogOptions;
@@ -210,32 +225,71 @@ public abstract class BaseCatalog extends AbstractCatalog
{
@Override
public List<String> listViews(String databaseName)
throws DatabaseNotExistException, CatalogException {
- // Gravitino does not support views yet; return empty to keep Flink
callers happy.
- return Collections.emptyList();
+ try {
+ ViewCatalog viewCatalog = catalog().asViewCatalog();
+ // TODO: Currently returns all VIRTUAL_VIEW entries from the underlying
catalog regardless of
+ // dialect. Views created by other engines (e.g. Trino, Spark) may
appear here but will fail
+ // when Flink attempts to load them. Consider filtering to only dialects
that Flink can handle
+ // (hive, flink), but this requires per-view property inspection which
is expensive.
+ return Arrays.stream(viewCatalog.listViews(Namespace.of(databaseName)))
+ .map(NameIdentifier::name)
+ .collect(Collectors.toList());
+ } catch (UnsupportedOperationException e) {
+ // Flink's listViews contract allows returning an empty list when views
are not supported.
+ LOG.debug("Catalog {} does not support views; returning empty view
list", catalogName(), e);
+ return Collections.emptyList();
+ } catch (NoSuchSchemaException e) {
+ throw new DatabaseNotExistException(catalogName(), databaseName, e);
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
}
@Override
public CatalogBaseTable getTable(ObjectPath tablePath)
throws TableNotExistException, CatalogException {
+ NameIdentifier ident =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
try {
- Table table =
- catalog()
- .asTableCatalog()
- .loadTable(NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()));
+ Table table = catalog().asTableCatalog().loadTable(ident);
return toFlinkTable(table, tablePath);
} catch (NoSuchTableException e) {
- throw new TableNotExistException(catalogName(), tablePath, e);
+ // Fall through to check views.
} catch (Exception e) {
+ LOG.warn("Failed to load table {} from catalog {}", ident,
catalogName(), e);
throw new CatalogException(e);
}
+
+ return loadViewOrThrow(tablePath);
}
@Override
public boolean tableExists(ObjectPath tablePath) throws CatalogException {
+ NameIdentifier ident =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
try {
- return catalog()
- .asTableCatalog()
- .tableExists(NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()));
+ if (catalog().asTableCatalog().tableExists(ident)) {
+ return true;
+ }
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
+ try {
+ return catalog().asViewCatalog().viewExists(ident);
+ } catch (UnsupportedOperationException e) {
+ LOG.debug(
+ "Catalog {} does not support views; viewExists returns false for {}",
+ catalogName(),
+ ident,
+ e);
+ return false;
+ } catch (NoSuchSchemaException e) {
+ LOG.debug(
+ "Schema not found when checking viewExists for {} in catalog {};
returning false",
+ ident,
+ catalogName(),
+ e);
+ return false;
} catch (Exception e) {
throw new CatalogException(e);
}
@@ -244,36 +298,64 @@ public abstract class BaseCatalog extends AbstractCatalog
{
@Override
public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
- boolean dropped =
- catalog()
- .asTableCatalog()
- .dropTable(NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()));
- if (!dropped && !ignoreIfNotExists) {
- throw new TableNotExistException(catalogName(), tablePath);
+ NameIdentifier ident =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
+
+ try {
+ boolean tableDropped = catalog().asTableCatalog().dropTable(ident);
+ boolean viewDropped = false;
+ try {
+ viewDropped = catalog().asViewCatalog().dropView(ident);
+ } catch (UnsupportedOperationException ignored) {
+ // catalog does not support views
+ }
+ if (!tableDropped && !viewDropped && !ignoreIfNotExists) {
+ throw new TableNotExistException(catalogName(), tablePath);
+ }
+ } catch (TableNotExistException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new CatalogException(e);
}
}
@Override
public void renameTable(ObjectPath tablePath, String newTableName, boolean
ignoreIfNotExists)
throws TableNotExistException, TableAlreadyExistException,
CatalogException {
- NameIdentifier identifier =
- NameIdentifier.of(Namespace.of(tablePath.getDatabaseName()),
newTableName);
+ NameIdentifier srcIdent =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
+ ObjectPath newPath = new ObjectPath(tablePath.getDatabaseName(),
newTableName);
+
+ try {
+ catalog().asTableCatalog().alterTable(srcIdent,
TableChange.rename(newTableName));
+ return;
+ } catch (NoSuchTableException ignored) {
+ // source is not a table, try as a view below
+ } catch (TableAlreadyExistsException e) {
+ throw new TableAlreadyExistException(catalogName(), newPath, e);
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
- if (catalog().asTableCatalog().tableExists(identifier)) {
- throw new TableAlreadyExistException(
- catalogName(), ObjectPath.fromString(tablePath.getDatabaseName() +
newTableName));
+ ViewCatalog viewCatalog;
+ try {
+ viewCatalog = catalog().asViewCatalog();
+ } catch (UnsupportedOperationException e) {
+ // catalog does not support views; source does not exist as table or view
+ if (!ignoreIfNotExists) {
+ throw new TableNotExistException(catalogName(), tablePath);
+ }
+ return;
}
try {
- catalog()
- .asTableCatalog()
- .alterTable(
- NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()),
- TableChange.rename(newTableName));
- } catch (NoSuchTableException e) {
+ viewCatalog.alterView(srcIdent, ViewChange.rename(newTableName));
+ } catch (NoSuchViewException e) {
if (!ignoreIfNotExists) {
throw new TableNotExistException(catalogName(), tablePath, e);
}
+ } catch (ViewAlreadyExistsException e) {
+ throw new TableAlreadyExistException(catalogName(), newPath, e);
} catch (Exception e) {
throw new CatalogException(e);
}
@@ -284,13 +366,19 @@ public abstract class BaseCatalog extends AbstractCatalog
{
throws TableAlreadyExistException, DatabaseNotExistException,
CatalogException {
Preconditions.checkArgument(
table instanceof ResolvedCatalogBaseTable, "table should be resolved");
+
+ if (table.getTableKind() == CatalogBaseTable.TableKind.VIEW) {
+ createView(tablePath, (ResolvedCatalogView) table, ignoreIfExists);
+ return;
+ }
+
NameIdentifier identifier =
NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
ResolvedCatalogBaseTable<?> resolvedTable = (ResolvedCatalogBaseTable<?>)
table;
Column[] columns =
resolvedTable.getResolvedSchema().getColumns().stream()
- .map(this::toGravitinoColumn)
+ .map(BaseCatalog::toGravitinoColumn)
.toArray(Column[]::new);
String comment = table.getComment();
Map<String, String> flinkOptions = table.getOptions();
@@ -323,6 +411,30 @@ public abstract class BaseCatalog extends AbstractCatalog {
}
}
+ private void createView(ObjectPath tablePath, ResolvedCatalogView view,
boolean ignoreIfExists)
+ throws TableAlreadyExistException, DatabaseNotExistException,
CatalogException {
+ NameIdentifier identifier =
+ NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
+ Column[] columns = toGravitinoColumns(view);
+ Representation[] representations =
+ buildSqlRepresentation(Dialects.FLINK, view.getExpandedQuery());
+ Map<String, String> options = new HashMap<>(view.getOptions());
+ options.put(FLINK_SCHEMA_NUM_COLUMNS_KEY, String.valueOf(columns.length));
+ try {
+ catalog()
+ .asViewCatalog()
+ .createView(identifier, view.getComment(), columns, representations,
null, null, options);
+ } catch (NoSuchSchemaException e) {
+ throw new DatabaseNotExistException(catalogName(),
tablePath.getDatabaseName(), e);
+ } catch (ViewAlreadyExistsException e) {
+ if (!ignoreIfExists) {
+ throw new TableAlreadyExistException(catalogName(), tablePath, e);
+ }
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
+ }
+
private static Index[] getGrivatinoIndices(ResolvedCatalogBaseTable<?>
resolvedTable) {
Optional<UniqueConstraint> primaryKey =
resolvedTable.getResolvedSchema().getPrimaryKey();
List<String> primaryColumns =
primaryKey.map(UniqueConstraint::getColumns).orElse(null);
@@ -371,9 +483,26 @@ public abstract class BaseCatalog extends AbstractCatalog {
NameIdentifier identifier =
NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
- catalog()
- .asTableCatalog()
- .alterTable(identifier, getGravitinoTableChanges(existingTable,
newTable));
+
+ if (newTable.getTableKind() == CatalogBaseTable.TableKind.VIEW) {
+ try {
+ catalog()
+ .asViewCatalog()
+ .alterView(
+ identifier,
+ toReplaceViewChange(existingTable, (ResolvedCatalogView)
newTable, Dialects.FLINK));
+ } catch (NoSuchViewException e) {
+ if (!ignoreIfNotExists) {
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ }
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
+ } else {
+ catalog()
+ .asTableCatalog()
+ .alterTable(identifier, getGravitinoTableChanges(existingTable,
newTable));
+ }
}
@Override
@@ -402,7 +531,24 @@ public abstract class BaseCatalog extends AbstractCatalog {
NameIdentifier identifier =
NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
- catalog().asTableCatalog().alterTable(identifier,
getGravitinoTableChanges(tableChanges));
+
+ if (newTable.getTableKind() == CatalogBaseTable.TableKind.VIEW) {
+ try {
+ catalog()
+ .asViewCatalog()
+ .alterView(
+ identifier,
+ toReplaceViewChange(tableChanges, (ResolvedCatalogView)
newTable, Dialects.FLINK));
+ } catch (NoSuchViewException e) {
+ if (!ignoreIfNotExists) {
+ throw new TableNotExistException(catalogName(), tablePath, e);
+ }
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
+ } else {
+ catalog().asTableCatalog().alterTable(identifier,
getGravitinoTableChanges(tableChanges));
+ }
}
@Override
@@ -561,14 +707,7 @@ public abstract class BaseCatalog extends AbstractCatalog {
}
protected CatalogBaseTable toFlinkTable(Table table, ObjectPath tablePath) {
- org.apache.flink.table.api.Schema.Builder builder =
- org.apache.flink.table.api.Schema.newBuilder();
- for (Column column : table.columns()) {
- DataType flinkType = TypeUtils.toFlinkType(column.dataType());
- builder
- .column(column.name(), column.nullable() ? flinkType.nullable() :
flinkType.notNull())
- .withComment(column.comment());
- }
+ org.apache.flink.table.api.Schema.Builder builder =
buildSchemaFromColumns(table.columns());
Optional<List<String>> flinkPrimaryKey = getFlinkPrimaryKey(table);
flinkPrimaryKey.ifPresent(builder::primaryKey);
Map<String, String> flinkTableProperties =
@@ -618,7 +757,7 @@ public abstract class BaseCatalog extends AbstractCatalog {
return Optional.of(primaryKeyFieldList);
}
- private Column toGravitinoColumn(org.apache.flink.table.catalog.Column
column) {
+ private static Column
toGravitinoColumn(org.apache.flink.table.catalog.Column column) {
return Column.of(
column.getName(),
TypeUtils.toGravitinoType(column.getDataType().getLogicalType()),
@@ -748,6 +887,139 @@ public abstract class BaseCatalog extends AbstractCatalog
{
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 sql =
+ view.sqlFor(Dialects.FLINK)
+ .map(SQLRepresentation::sql)
+ .orElseGet(
+ () ->
+ view.sqlFor(Dialects.HIVE)
+ .map(SQLRepresentation::sql)
+ .orElseThrow(
+ () ->
+ new CatalogException(
+ String.format(
+ "View '%s' in catalog '%s' has no SQL
representation for dialect '%s' or '%s'",
+ view.name(),
+ catalogName(),
+ Dialects.FLINK,
+ Dialects.HIVE))));
+
+ Map<String, String> properties =
+ view.properties() != null
+ ? Collections.unmodifiableMap(view.properties())
+ : Collections.emptyMap();
+ return CatalogView.of(builder.build(), view.comment(), sql, sql,
properties);
+ }
+
+ @VisibleForTesting
+ static ViewChange[] toReplaceViewChange(
+ CatalogBaseTable existingView, ResolvedCatalogView newView, String
dialect) {
+ return new ViewChange[] {
+ ViewChange.replaceView(
+ toGravitinoColumns(newView),
+ buildSqlRepresentation(dialect, newView.getExpandedQuery()),
+ null,
+ null,
+ newView.getComment())
+ };
+ }
+
+ @VisibleForTesting
+ static ViewChange[] toReplaceViewChange(
+ List<org.apache.flink.table.catalog.TableChange> tableChanges,
+ ResolvedCatalogView newView,
+ String dialect) {
+ List<ViewChange> changes = Lists.newArrayList();
+ boolean needsBodyReplace = false;
+
+ for (org.apache.flink.table.catalog.TableChange change : tableChanges) {
+ if (change instanceof
org.apache.flink.table.catalog.TableChange.SetOption) {
+ org.apache.flink.table.catalog.TableChange.SetOption setOption =
+ (org.apache.flink.table.catalog.TableChange.SetOption) change;
+ changes.add(ViewChange.setProperty(setOption.getKey(),
setOption.getValue()));
+ } else if (change instanceof
org.apache.flink.table.catalog.TableChange.ResetOption) {
+ org.apache.flink.table.catalog.TableChange.ResetOption resetOption =
+ (org.apache.flink.table.catalog.TableChange.ResetOption) change;
+ changes.add(ViewChange.removeProperty(resetOption.getKey()));
+ } else {
+ LOG.warn(
+ "Unrecognized TableChange type {} for view alter; triggering full
ReplaceView",
+ change.getClass().getSimpleName());
+ needsBodyReplace = true;
+ }
+ }
+
+ if (needsBodyReplace) {
+ changes.add(
+ ViewChange.replaceView(
+ toGravitinoColumns(newView),
+ buildSqlRepresentation(dialect, newView.getExpandedQuery()),
+ null,
+ null,
+ newView.getComment()));
+ }
+
+ return changes.toArray(new ViewChange[0]);
+ }
+
+ private static Column[] toGravitinoColumns(ResolvedCatalogView view) {
+ return view.getResolvedSchema().getColumns().stream()
+ .map(BaseCatalog::toGravitinoColumn)
+ .toArray(Column[]::new);
+ }
+
+ private static org.apache.flink.table.api.Schema.Builder
buildSchemaFromColumns(
+ Column[] columns) {
+ org.apache.flink.table.api.Schema.Builder builder =
+ org.apache.flink.table.api.Schema.newBuilder();
+ for (Column column : columns) {
+ DataType flinkType = TypeUtils.toFlinkType(column.dataType());
+ builder
+ .column(column.name(), column.nullable() ? flinkType.nullable() :
flinkType.notNull())
+ .withComment(column.comment());
+ }
+ return builder;
+ }
+
+ private static Representation[] buildSqlRepresentation(String dialect,
String sql) {
+ return new Representation[] {
+ SQLRepresentation.builder().withDialect(dialect).withSql(sql).build()
+ };
+ }
+
protected Catalog catalog() {
return GravitinoCatalogManager.get().getGravitinoCatalogInfo(getName());
}
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
index 0a32ce4373..a027b36dd4 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
@@ -30,6 +30,7 @@ import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogPropertiesUtil;
import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
@@ -54,12 +55,15 @@ import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* The GravitinoHiveCatalog class is an implementation of the BaseCatalog
class that is used to
* proxy the HiveCatalog class.
*/
public class GravitinoHiveCatalog extends BaseCatalog {
+ private static final Logger LOG =
LoggerFactory.getLogger(GravitinoHiveCatalog.class);
private HiveCatalog hiveCatalog;
@@ -100,6 +104,11 @@ public class GravitinoHiveCatalog extends BaseCatalog {
Preconditions.checkArgument(
table instanceof ResolvedCatalogBaseTable, "table should be resolved");
+ if (table instanceof CatalogView) {
+ super.createTable(tablePath, table, ignoreIfExists);
+ return;
+ }
+
if (!FlinkGenericTableUtil.isGenericTableWhenCreate(table.getOptions())) {
super.createTable(tablePath, table, ignoreIfExists);
return;
@@ -150,15 +159,22 @@ public class GravitinoHiveCatalog extends BaseCatalog {
}
return super.toFlinkTable(table, tablePath);
} catch (NoSuchTableException e) {
- throw new TableNotExistException(catalogName(), tablePath, e);
+ // Fall through to check views.
} catch (Exception e) {
+ LOG.warn("Failed to load table {} from catalog {}", tablePath,
catalogName(), e);
throw new CatalogException(e);
}
+
+ return loadViewOrThrow(tablePath);
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable,
boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
+ if (newTable instanceof CatalogView) {
+ super.alterTable(tablePath, newTable, ignoreIfNotExists);
+ return;
+ }
Table table = loadGravitinoTable(tablePath, ignoreIfNotExists);
if (table == null) {
return;
@@ -183,6 +199,10 @@ public class GravitinoHiveCatalog extends BaseCatalog {
java.util.List<org.apache.flink.table.catalog.TableChange> tableChanges,
boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
+ if (newTable instanceof CatalogView) {
+ super.alterTable(tablePath, newTable, tableChanges, ignoreIfNotExists);
+ return;
+ }
Table table = loadGravitinoTable(tablePath, ignoreIfNotExists);
if (table == null) {
return;
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
index ec934881df..34b7882fea 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
@@ -20,21 +20,34 @@ package org.apache.gravitino.flink.connector.catalog;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogDatabase;
import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedCatalogView;
+import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
import org.apache.gravitino.SchemaChange;
import org.apache.gravitino.flink.connector.PartitionConverter;
import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
+import org.apache.gravitino.rel.Dialects;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.SQLRepresentation;
+import org.apache.gravitino.rel.ViewCatalog;
+import org.apache.gravitino.rel.ViewChange;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
import org.apache.gravitino.rel.types.Types;
import org.junit.jupiter.api.Assertions;
@@ -125,7 +138,8 @@ public class TestBaseCatalog {
@Test
public void testToGravitinoDistributionDefaultsToNone() {
- TestableBaseCatalog catalog = new
TestableBaseCatalog(Mockito.mock(AbstractCatalog.class));
+ TestableBaseCatalog catalog =
+ new TestableBaseCatalog(Mockito.mock(AbstractCatalog.class),
mockUnsupportedViewCatalog());
Assertions.assertEquals(
Distributions.NONE,
@@ -136,21 +150,149 @@ public class TestBaseCatalog {
}
@Test
- public void testListViewsReturnsEmptyWithoutDelegation() throws Exception {
- AbstractCatalog delegate = Mockito.mock(AbstractCatalog.class);
- BaseCatalog catalog = new TestableBaseCatalog(delegate);
+ public void testListViewsReturnsEmptyWhenViewCatalogUnsupported() throws
Exception {
+ Catalog gravitinoCatalog = mockUnsupportedViewCatalog();
+ BaseCatalog catalog =
+ new TestableBaseCatalog(Mockito.mock(AbstractCatalog.class),
gravitinoCatalog);
List<String> views = catalog.listViews("db");
Assertions.assertTrue(views.isEmpty());
- Mockito.verifyNoInteractions(delegate);
+ }
+
+ @Test
+ public void testListViewsDelegatesToViewCatalog() throws Exception {
+ ViewCatalog viewCatalog = Mockito.mock(ViewCatalog.class);
+ Mockito.when(viewCatalog.listViews(Namespace.of("db")))
+ .thenReturn(
+ new NameIdentifier[] {NameIdentifier.of("db", "v1"),
NameIdentifier.of("db", "v2")});
+ Catalog gravitinoCatalog = Mockito.mock(Catalog.class);
+ Mockito.when(gravitinoCatalog.asViewCatalog()).thenReturn(viewCatalog);
+
+ BaseCatalog catalog =
+ new TestableBaseCatalog(Mockito.mock(AbstractCatalog.class),
gravitinoCatalog);
+
+ List<String> views = catalog.listViews("db");
+
+ Assertions.assertEquals(ImmutableList.of("v1", "v2"), views);
+ }
+
+ @Test
+ public void testGetGravitinoViewChangesSetAndRemoveProperty() {
+ List<TableChange> tableChanges =
+ ImmutableList.of(TableChange.set("k1", "v1"), TableChange.reset("k2"));
+
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+
+ ViewChange[] changes =
+ BaseCatalog.toReplaceViewChange(
+ tableChanges, resolveView(schema, "SELECT 1", "comment"),
Dialects.FLINK);
+
+ Assertions.assertEquals(2, changes.length);
+ Assertions.assertInstanceOf(ViewChange.SetProperty.class, changes[0]);
+ Assertions.assertEquals("k1", ((ViewChange.SetProperty)
changes[0]).getProperty());
+ Assertions.assertEquals("v1", ((ViewChange.SetProperty)
changes[0]).getValue());
+
+ Assertions.assertInstanceOf(ViewChange.RemoveProperty.class, changes[1]);
+ Assertions.assertEquals("k2", ((ViewChange.RemoveProperty)
changes[1]).getProperty());
+ }
+
+ @Test
+ public void testGetGravitinoViewChangesBodyReplaceOnStructuralChange() {
+ List<TableChange> tableChanges =
+ ImmutableList.of(
+ TableChange.add(Column.physical("id", DataTypes.INT())),
TableChange.set("k1", "v1"));
+
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+
+ ViewChange[] changes =
+ BaseCatalog.toReplaceViewChange(
+ tableChanges, resolveView(schema, "SELECT id FROM t", "new
comment"), Dialects.FLINK);
+
+ // Should have exactly one SetProperty and one ReplaceView (order may vary)
+ Assertions.assertEquals(2, changes.length);
+ ViewChange.SetProperty setProp =
+ (ViewChange.SetProperty)
+ Arrays.stream(changes)
+ .filter(c -> c instanceof ViewChange.SetProperty)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("expected SetProperty"));
+ Assertions.assertEquals("k1", setProp.getProperty());
+ Assertions.assertEquals("v1", setProp.getValue());
+
+ ViewChange.ReplaceView replaceView =
+ (ViewChange.ReplaceView)
+ Arrays.stream(changes)
+ .filter(c -> c instanceof ViewChange.ReplaceView)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("expected ReplaceView"));
+ Assertions.assertEquals("new comment", replaceView.getComment());
+ Assertions.assertEquals(1, replaceView.getRepresentations().length);
+ Assertions.assertInstanceOf(SQLRepresentation.class,
replaceView.getRepresentations()[0]);
+ SQLRepresentation sqlRep = (SQLRepresentation)
replaceView.getRepresentations()[0];
+ Assertions.assertEquals(Dialects.FLINK, sqlRep.dialect());
+ Assertions.assertEquals("SELECT id FROM t", sqlRep.sql());
+ }
+
+ @Test
+ public void testGetGravitinoViewChangesFullReplace() {
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+ CatalogView existing =
+ CatalogView.of(schema, "old comment", "SELECT 1", "SELECT 1",
Collections.emptyMap());
+
+ ViewChange[] changes =
+ BaseCatalog.toReplaceViewChange(
+ existing, resolveView(schema, "SELECT 2", "new comment"),
Dialects.FLINK);
+
+ Assertions.assertEquals(1, changes.length);
+ Assertions.assertInstanceOf(ViewChange.ReplaceView.class, changes[0]);
+ ViewChange.ReplaceView replaceView = (ViewChange.ReplaceView) changes[0];
+ Assertions.assertEquals("new comment", replaceView.getComment());
+ Representation[] reps = replaceView.getRepresentations();
+ Assertions.assertEquals(1, reps.length);
+ SQLRepresentation sqlRep = (SQLRepresentation) reps[0];
+ Assertions.assertEquals("SELECT 2", sqlRep.sql());
+ Assertions.assertEquals(Dialects.FLINK, sqlRep.dialect());
+ }
+
+ /**
+ * Helper to build a minimal {@link
org.apache.flink.table.catalog.ResolvedCatalogView} for
+ * testing view-change conversion.
+ */
+ private static ResolvedCatalogView resolveView(Schema schema, String query,
String comment) {
+ CatalogView catalogView = CatalogView.of(schema, comment, query, query,
Collections.emptyMap());
+ ResolvedSchema resolvedSchema =
+ ResolvedSchema.of(
+ schema.getColumns().stream()
+ .map(
+ c -> {
+ if (c instanceof Schema.UnresolvedPhysicalColumn) {
+ Schema.UnresolvedPhysicalColumn pc =
(Schema.UnresolvedPhysicalColumn) c;
+ return Column.physical(pc.getName(), DataTypes.INT());
+ }
+ return Column.physical(c.getName(), DataTypes.INT());
+ })
+ .collect(Collectors.toList()));
+ return new ResolvedCatalogView(catalogView, resolvedSchema);
+ }
+
+ /**
+ * Returns a mock {@link Catalog} whose {@code asViewCatalog()} throws
+ * UnsupportedOperationException.
+ */
+ private static Catalog mockUnsupportedViewCatalog() {
+ Catalog gravitinoCatalog = Mockito.mock(Catalog.class);
+ Mockito.when(gravitinoCatalog.asViewCatalog())
+ .thenThrow(new UnsupportedOperationException("views not supported"));
+ return gravitinoCatalog;
}
private static class TestableBaseCatalog extends BaseCatalog {
private final AbstractCatalog delegate;
+ private final Catalog gravitinoCatalog;
- TestableBaseCatalog(AbstractCatalog delegate) {
+ TestableBaseCatalog(AbstractCatalog delegate, Catalog gravitinoCatalog) {
super(
"test",
Collections.emptyMap(),
@@ -158,11 +300,17 @@ public class TestBaseCatalog {
Mockito.mock(SchemaAndTablePropertiesConverter.class),
Mockito.mock(PartitionConverter.class));
this.delegate = delegate;
+ this.gravitinoCatalog = gravitinoCatalog;
}
@Override
protected AbstractCatalog realCatalog() {
return delegate;
}
+
+ @Override
+ protected Catalog catalog() {
+ return gravitinoCatalog;
+ }
}
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java
index 27f60b558c..eac2c9ae3a 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java
@@ -60,6 +60,7 @@ import
org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
import org.apache.flink.table.catalog.hive.util.Constants;
import org.apache.flink.types.Row;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
import org.apache.gravitino.catalog.hive.HiveConstants;
import org.apache.gravitino.catalog.hive.HiveStorageConstants;
import org.apache.gravitino.flink.connector.CatalogPropertiesConverter;
@@ -71,7 +72,10 @@ import
org.apache.gravitino.flink.connector.store.GravitinoCatalogStoreFactoryOp
import org.apache.gravitino.integration.test.container.ContainerSuite;
import org.apache.gravitino.integration.test.util.TestDatabaseName;
import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.SQLRepresentation;
import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.View;
+import org.apache.gravitino.rel.ViewCatalog;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.expressions.transforms.Transforms;
import org.apache.gravitino.rel.types.Types;
@@ -745,6 +749,277 @@ public abstract class FlinkHiveCatalogIT extends
FlinkCommonIT {
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 =
+ flinkCatalog.get().getTable(new ObjectPath(schemaName,
viewName));
+ Assertions.assertEquals(CatalogBaseTable.TableKind.VIEW,
flinkTable.getTableKind());
+ } catch (TableNotExistException e) {
+ Assertions.fail("view should exist in Flink catalog: " +
e.getMessage());
+ }
+ },
+ true);
+ }
+
+ @Test
+ public void testListViews() {
+ String schemaName = "test_hive_list_views_db";
+ String tableName = "test_list_view_base";
+ String view1 = "test_list_view_1";
+ String view2 = "test_list_view_2";
+ doWithSchema(
+ currentCatalog(),
+ schemaName,
+ catalog -> {
+ TestUtils.assertTableResult(
+ sql("CREATE TABLE %s (id INT) WITH ('connector'='hive')",
tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", view1, tableName),
ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", view2, tableName),
ResultKind.SUCCESS);
+
+ List<String> views = Arrays.asList(tableEnv.listViews());
+ Assertions.assertTrue(views.contains(view1), "view1 not found in
SHOW VIEWS");
+ Assertions.assertTrue(views.contains(view2), "view2 not found in
SHOW VIEWS");
+
+ // Tables should not appear in listViews
+ Assertions.assertFalse(views.contains(tableName), "table should not
appear in listViews");
+
+ // Verify via Gravitino ViewCatalog
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ NameIdentifier[] gravitinoViews =
viewCatalog.listViews(Namespace.of(schemaName));
+ List<String> gravitinoViewNames =
+
Arrays.stream(gravitinoViews).map(NameIdentifier::name).collect(Collectors.toList());
+ Assertions.assertTrue(gravitinoViewNames.contains(view1));
+ Assertions.assertTrue(gravitinoViewNames.contains(view2));
+ },
+ true);
+ }
+
+ @Test
+ public void testDropView() {
+ String schemaName = "test_hive_drop_view_db";
+ String tableName = "test_drop_view_base";
+ String viewName = "test_view_drop";
+ doWithSchema(
+ currentCatalog(),
+ schemaName,
+ catalog -> {
+ TestUtils.assertTableResult(
+ sql("CREATE TABLE %s (id INT) WITH ('connector'='hive')",
tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ Assertions.assertTrue(
+ viewCatalog.viewExists(NameIdentifier.of(schemaName, viewName)),
+ "view should exist before drop");
+
+ TestUtils.assertTableResult(sql("DROP VIEW %s", viewName),
ResultKind.SUCCESS);
+
+ Assertions.assertFalse(
+ viewCatalog.viewExists(NameIdentifier.of(schemaName, viewName)),
+ "view should not exist after drop");
+ },
+ true);
+ }
+
+ @Test
+ public void testAlterViewRename() {
+ String schemaName = "test_hive_rename_view_db";
+ String tableName = "test_rename_view_base";
+ String viewName = "test_view_rename_src";
+ String newViewName = "test_view_rename_dst";
+ doWithSchema(
+ currentCatalog(),
+ schemaName,
+ catalog -> {
+ TestUtils.assertTableResult(
+ sql("CREATE TABLE %s (id INT) WITH ('connector'='hive')",
tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ TestUtils.assertTableResult(
+ sql("ALTER VIEW %s RENAME TO %s", viewName, newViewName),
ResultKind.SUCCESS);
+
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ Assertions.assertFalse(
+ viewCatalog.viewExists(NameIdentifier.of(schemaName, viewName)),
+ "old view name should not exist");
+ Assertions.assertTrue(
+ viewCatalog.viewExists(NameIdentifier.of(schemaName,
newViewName)),
+ "new view name should exist");
+ },
+ true);
+ }
+
+ @Test
+ public void testAlterViewReplaceBody() {
+ String schemaName = "test_hive_replace_view_db";
+ String tableName = "test_replace_view_base";
+ String viewName = "test_view_replace";
+ 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 AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ // Replace the view body to select a different column
+ TestUtils.assertTableResult(
+ sql("ALTER VIEW %s AS SELECT id, name FROM %s", viewName,
tableName),
+ ResultKind.SUCCESS);
+
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ View view = viewCatalog.loadView(NameIdentifier.of(schemaName,
viewName));
+ Assertions.assertEquals(1, view.representations().length);
+ SQLRepresentation rep = (SQLRepresentation)
view.representations()[0];
+ Assertions.assertTrue(
+ rep.sql().contains("name") && rep.sql().contains("id"),
+ "updated view SQL should select both id and name columns");
+ },
+ true);
+ }
+
+ @Test
+ public void testQueryView() {
+ String schemaName = "test_hive_query_view_db";
+ String tableName = "test_query_view_base";
+ String viewName = "test_view_query";
+ 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 AS SELECT id, name FROM %s WHERE id > 1",
viewName, tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("INSERT INTO %s VALUES (1, 'alice'), (2, 'bob'), (3,
'carol')", tableName),
+ ResultKind.SUCCESS_WITH_CONTENT,
+ Row.of(-1L));
+
+ TestUtils.assertTableResult(
+ sql("SELECT * FROM %s ORDER BY id", viewName),
+ ResultKind.SUCCESS_WITH_CONTENT,
+ Row.of(2, "bob"),
+ Row.of(3, "carol"));
+ },
+ true);
+ }
+
+ @Test
+ public void testCreateViewIfNotExists() {
+ String schemaName = "test_hive_create_view_ifte_db";
+ String tableName = "test_create_view_ifte_base";
+ String viewName = "test_view_ifte";
+ doWithSchema(
+ currentCatalog(),
+ schemaName,
+ catalog -> {
+ TestUtils.assertTableResult(
+ sql("CREATE TABLE %s (id INT) WITH ('connector'='hive')",
tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ // Second CREATE VIEW IF NOT EXISTS should succeed without error
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW IF NOT EXISTS %s AS SELECT id FROM %s",
viewName, tableName),
+ ResultKind.SUCCESS);
+
+ // View should still exist and be unchanged
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+
Assertions.assertTrue(viewCatalog.viewExists(NameIdentifier.of(schemaName,
viewName)));
+ },
+ true);
+ }
+
+ @Test
+ public void testDropViewIfExists() {
+ String schemaName = "test_hive_drop_view_ifte_db";
+ String tableName = "test_drop_view_ifte_base";
+ String viewName = "test_view_drop_ifte";
+ doWithSchema(
+ currentCatalog(),
+ schemaName,
+ catalog -> {
+ TestUtils.assertTableResult(
+ sql("CREATE TABLE %s (id INT) WITH ('connector'='hive')",
tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ TestUtils.assertTableResult(sql("DROP VIEW %s", viewName),
ResultKind.SUCCESS);
+
+ // DROP VIEW IF EXISTS on a non-existent view should not throw
+ TestUtils.assertTableResult(sql("DROP VIEW IF EXISTS %s", viewName),
ResultKind.SUCCESS);
+ },
+ true);
+ }
+
+ @Test
+ public void testListTablesDoesNotIncludeViews() {
+ String schemaName = "test_hive_list_tables_no_views_db";
+ String tableName = "test_list_no_view_base";
+ String viewName = "test_list_no_view_view";
+ doWithSchema(
+ currentCatalog(),
+ schemaName,
+ catalog -> {
+ TestUtils.assertTableResult(
+ sql("CREATE TABLE %s (id INT) WITH ('connector'='hive')",
tableName),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ List<String> tables = Arrays.asList(tableEnv.listTables());
+ Assertions.assertTrue(tables.contains(tableName), "table should
appear in listTables");
+ Assertions.assertFalse(tables.contains(viewName), "view should not
appear in listTables");
+
+ List<String> views = Arrays.asList(tableEnv.listViews());
+ Assertions.assertTrue(views.contains(viewName), "view should appear
in listViews");
+ Assertions.assertFalse(views.contains(tableName), "table should not
appear in listViews");
+ },
+ true);
+ }
+
private void assertHiveCatalogRead(String databaseName, String tableName,
Row expectedRow) {
EnvironmentSettings settings =
EnvironmentSettings.newInstance().inBatchMode().build();
TableEnvironment hiveEnv = TableEnvironment.create(settings);
