This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new cff10cd880 [Cherry-pick to branch-1.3] [#11348] feat(flink-connector):
Flink connector view support to Iceberg and Paimon catalogs (#11349) (#11471)
cff10cd880 is described below
commit cff10cd88023ebfc4db7bbb172c9e9a64300eabe
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Jun 8 12:55:33 2026 +0800
[Cherry-pick to branch-1.3] [#11348] feat(flink-connector): Flink connector
view support to Iceberg and Paimon catalogs (#11349) (#11471)
**Cherry-pick Information:**
- Original commit: baebcb2bc6badc8e9c909d2741e7c6f7db81780e
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: Yuhui <[email protected]>
---
.../catalog/lakehouse/paimon/PaimonConstants.java | 4 +
.../catalog/lakehouse/paimon/PaimonView.java | 2 +-
.../flink/connector/catalog/BaseCatalog.java | 125 ++++++---
.../connector/paimon/GravitinoPaimonCatalog.java | 42 +--
.../flink/connector/catalog/TestBaseCatalog.java | 85 ++++++
.../connector/integration/test/FlinkCommonIT.java | 299 +++++++++++++++++++++
.../connector/integration/test/FlinkEnvIT.java | 16 +-
.../integration/test/hive/FlinkHiveCatalogIT.java | 277 +------------------
.../test/iceberg/FlinkIcebergCatalogIT.java | 5 +
.../test/paimon/FlinkPaimonHiveBackendIT.java | 5 +
.../test/paimon/FlinkPaimonJdbcBackendIT.java | 6 +
.../paimon/TestGravitinoPaimonCatalog.java | 1 +
12 files changed, 540 insertions(+), 327 deletions(-)
diff --git
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonConstants.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonConstants.java
index 9c2f472f39..922167e325 100644
---
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonConstants.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonConstants.java
@@ -68,6 +68,10 @@ public class PaimonConstants {
public static final String OWNER = "owner";
public static final String BUCKET_KEY = "bucket-key";
public static final String BUCKET_NUM = "bucket";
+
+ /** The dialect key used to store the canonical SQL query in a Paimon view.
*/
+ public static final String VIEW_QUERY_DIALECT = "query";
+
public static final String MERGE_ENGINE = "merge-engine";
public static final String SEQUENCE_FIELD = "sequence.field";
public static final String ROWKIND_FIELD = "rowkind.field";
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonView.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonView.java
index 0d0f22d656..9990925a0a 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonView.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonView.java
@@ -43,7 +43,7 @@ final class PaimonView implements View {
static final String DEFAULT_CATALOG_PROPERTY =
"gravitino.view.default-catalog";
static final String DEFAULT_SCHEMA_PROPERTY =
"gravitino.view.default-schema";
- private static final String PAIMON_VIEW_QUERY = "query";
+ private static final String PAIMON_VIEW_QUERY =
PaimonConstants.VIEW_QUERY_DIALECT;
private final String name;
@Nullable private final String comment;
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 9d9d407468..59874aff4a 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
@@ -312,12 +312,13 @@ public abstract class BaseCatalog extends AbstractCatalog
{
NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
try {
- boolean tableDropped = catalog().asTableCatalog().dropTable(ident);
+ boolean tableDropped = dropTableEntry(ident);
boolean viewDropped = false;
try {
viewDropped = catalog().asViewCatalog().dropView(ident);
- } catch (UnsupportedOperationException ignored) {
- // catalog does not support views
+ } catch (UnsupportedOperationException e) {
+ LOG.debug(
+ "Catalog {} does not support views; skipping dropView for {}",
catalogName(), ident, e);
}
if (!tableDropped && !viewDropped && !ignoreIfNotExists) {
throw new TableNotExistException(catalogName(), tablePath);
@@ -332,6 +333,16 @@ public abstract class BaseCatalog extends AbstractCatalog {
}
}
+ /**
+ * Drops the table entry. Subclasses may override to use a different drop
strategy (e.g., purge).
+ *
+ * @param ident the Gravitino name identifier of the table to drop
+ * @return {@code true} if the table existed and was dropped.
+ */
+ protected boolean dropTableEntry(NameIdentifier ident) {
+ return catalog().asTableCatalog().dropTable(ident);
+ }
+
@Override
public void renameTable(ObjectPath tablePath, String newTableName, boolean
ignoreIfNotExists)
throws TableNotExistException, TableAlreadyExistException,
CatalogException {
@@ -431,8 +442,7 @@ public abstract class BaseCatalog extends AbstractCatalog {
NameIdentifier identifier =
NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName());
Column[] columns = toGravitinoColumns(view);
- Representation[] representations =
- buildSqlRepresentation(Dialects.FLINK, view.getExpandedQuery());
+ Representation[] representations = buildViewRepresentations(view);
Map<String, String> options = new HashMap<>(view.getOptions());
options.put(FLINK_SCHEMA_NUM_COLUMNS_KEY, String.valueOf(columns.length));
try {
@@ -505,12 +515,16 @@ public abstract class BaseCatalog extends AbstractCatalog
{
.asViewCatalog()
.alterView(
identifier,
- toReplaceViewChange(existingTable, (ResolvedCatalogView)
newTable, Dialects.FLINK));
+ toReplaceViewChange(
+ existingTable,
+ (ResolvedCatalogView) newTable,
+ buildViewRepresentations((ResolvedCatalogView) newTable)));
} catch (NoSuchViewException e) {
if (!ignoreIfNotExists) {
throw new TableNotExistException(catalogName(), tablePath, e);
}
} catch (Exception e) {
+ LOG.warn("Failed to alter view {} in catalog {}", identifier,
catalogName(), e);
throw new CatalogException(e);
}
} else {
@@ -555,12 +569,16 @@ public abstract class BaseCatalog extends AbstractCatalog
{
.asViewCatalog()
.alterView(
identifier,
- toReplaceViewChange(tableChanges, (ResolvedCatalogView)
newTable, Dialects.FLINK));
+ toReplaceViewChange(
+ tableChanges,
+ (ResolvedCatalogView) newTable,
+ buildViewRepresentations((ResolvedCatalogView) newTable)));
} catch (NoSuchViewException e) {
if (!ignoreIfNotExists) {
throw new TableNotExistException(catalogName(), tablePath, e);
}
} catch (Exception e) {
+ LOG.warn("Failed to alter view {} in catalog {}", identifier,
catalogName(), e);
throw new CatalogException(e);
}
} else {
@@ -989,6 +1007,15 @@ public abstract class BaseCatalog extends AbstractCatalog
{
}
}
+ /**
+ * Returns the ordered list of SQL dialects to try when loading a view. The
first dialect with an
+ * available representation wins. Subclasses may override to change the
fallback order. Must be
+ * consistent with the dialects stored by {@link #buildViewRepresentations}.
+ */
+ protected List<String> viewDialectFallbackOrder() {
+ return Arrays.asList(Dialects.FLINK, Dialects.HIVE);
+ }
+
/**
* Converts a Gravitino {@link View} to a Flink {@link CatalogView}.
*
@@ -997,22 +1024,19 @@ public abstract class BaseCatalog extends
AbstractCatalog {
*/
protected CatalogView toFlinkView(View view) {
org.apache.flink.table.api.Schema.Builder builder =
buildSchemaFromColumns(view.columns());
+ List<String> dialects = viewDialectFallbackOrder();
String sql =
- view.sqlFor(Dialects.FLINK)
- .map(SQLRepresentation::sql)
- .orElseGet(
+ dialects.stream()
+ .map(d -> view.sqlFor(d).map(SQLRepresentation::sql))
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .findFirst()
+ .orElseThrow(
() ->
- 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))));
+ new CatalogException(
+ String.format(
+ "View '%s' in catalog '%s' has no SQL
representation for dialects %s",
+ view.name(), catalogName(), dialects)));
Map<String, String> properties =
view.properties() != null
@@ -1024,13 +1048,18 @@ public abstract class BaseCatalog extends
AbstractCatalog {
@VisibleForTesting
static ViewChange[] toReplaceViewChange(
CatalogBaseTable existingView, ResolvedCatalogView newView, String
dialect) {
+ return toReplaceViewChange(
+ existingView, newView, buildSqlRepresentation(dialect,
newView.getExpandedQuery()));
+ }
+
+ @VisibleForTesting
+ static ViewChange[] toReplaceViewChange(
+ CatalogBaseTable existingView,
+ ResolvedCatalogView newView,
+ Representation[] representations) {
return new ViewChange[] {
ViewChange.replaceView(
- toGravitinoColumns(newView),
- buildSqlRepresentation(dialect, newView.getExpandedQuery()),
- null,
- null,
- newView.getComment())
+ toGravitinoColumns(newView), representations, null, null,
newView.getComment())
};
}
@@ -1039,6 +1068,15 @@ public abstract class BaseCatalog extends
AbstractCatalog {
List<org.apache.flink.table.catalog.TableChange> tableChanges,
ResolvedCatalogView newView,
String dialect) {
+ return toReplaceViewChange(
+ tableChanges, newView, buildSqlRepresentation(dialect,
newView.getExpandedQuery()));
+ }
+
+ @VisibleForTesting
+ static ViewChange[] toReplaceViewChange(
+ List<org.apache.flink.table.catalog.TableChange> tableChanges,
+ ResolvedCatalogView newView,
+ Representation[] representations) {
List<ViewChange> changes = Lists.newArrayList();
boolean needsBodyReplace = false;
@@ -1062,11 +1100,7 @@ public abstract class BaseCatalog extends
AbstractCatalog {
if (needsBodyReplace) {
changes.add(
ViewChange.replaceView(
- toGravitinoColumns(newView),
- buildSqlRepresentation(dialect, newView.getExpandedQuery()),
- null,
- null,
- newView.getComment()));
+ toGravitinoColumns(newView), representations, null, null,
newView.getComment()));
}
return changes.toArray(new ViewChange[0]);
@@ -1078,7 +1112,13 @@ public abstract class BaseCatalog extends
AbstractCatalog {
.toArray(Column[]::new);
}
- private static org.apache.flink.table.api.Schema.Builder
buildSchemaFromColumns(
+ /**
+ * Builds a Flink {@link org.apache.flink.table.api.Schema.Builder} from
Gravitino columns.
+ *
+ * @param columns the Gravitino column definitions
+ * @return a Flink schema builder populated with the given columns
+ */
+ protected 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();
@@ -1091,7 +1131,26 @@ public abstract class BaseCatalog extends
AbstractCatalog {
return builder;
}
- private static Representation[] buildSqlRepresentation(String dialect,
String sql) {
+ /**
+ * Builds the SQL representations to store when creating or replacing a
view. Subclasses may
+ * override to include additional dialect representations. The returned
array must contain at
+ * least one representation whose dialect appears in {@link
#viewDialectFallbackOrder()}.
+ *
+ * @param view the resolved Flink view being created or replaced
+ * @return representations to pass to the Gravitino view catalog API
+ */
+ protected Representation[] buildViewRepresentations(ResolvedCatalogView
view) {
+ return buildSqlRepresentation(Dialects.FLINK, view.getExpandedQuery());
+ }
+
+ /**
+ * Builds a single-element {@link Representation} array for the given
dialect and SQL text.
+ *
+ * @param dialect the SQL dialect identifier (see {@link Dialects})
+ * @param sql the SQL text of the view
+ * @return a single-element array containing the built {@link
SQLRepresentation}
+ */
+ protected static Representation[] buildSqlRepresentation(String dialect,
String sql) {
return new Representation[] {
SQLRepresentation.builder().withDialect(dialect).withSql(sql).build()
};
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
index 3776b02da3..aa9edabd84 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
@@ -23,6 +23,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -31,6 +32,7 @@ import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogView;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.factories.CatalogFactory;
@@ -40,6 +42,9 @@ import
org.apache.gravitino.catalog.lakehouse.paimon.PaimonConstants;
import org.apache.gravitino.flink.connector.PartitionConverter;
import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+import org.apache.gravitino.rel.Dialects;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.SQLRepresentation;
import org.apache.gravitino.rel.expressions.Expression;
import org.apache.gravitino.rel.expressions.NamedReference;
import org.apache.gravitino.rel.expressions.distributions.Distribution;
@@ -103,6 +108,28 @@ public class GravitinoPaimonCatalog extends BaseCatalog {
// DDL — route through Gravitino (single source of truth)
//
---------------------------------------------------------------------------
+ @Override
+ protected List<String> viewDialectFallbackOrder() {
+ return Arrays.asList(Dialects.FLINK, Dialects.HIVE,
PaimonConstants.VIEW_QUERY_DIALECT);
+ }
+
+ @Override
+ protected Representation[] buildViewRepresentations(ResolvedCatalogView
view) {
+ String sql = view.getExpandedQuery();
+ return new Representation[] {
+
SQLRepresentation.builder().withDialect(Dialects.FLINK).withSql(sql).build(),
+ SQLRepresentation.builder()
+ .withDialect(PaimonConstants.VIEW_QUERY_DIALECT)
+ .withSql(sql)
+ .build()
+ };
+ }
+
+ @Override
+ protected boolean dropTableEntry(NameIdentifier ident) {
+ return catalog().asTableCatalog().purgeTable(ident);
+ }
+
@VisibleForTesting
static Map<String, String> distributionToProperties(Distribution
distribution) {
if (distribution == null || distribution.strategy() == Strategy.NONE) {
@@ -188,21 +215,6 @@ public class GravitinoPaimonCatalog extends BaseCatalog {
return paimonCatalog.getFactory();
}
- @Override
- public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
- throws TableNotExistException, CatalogException {
- boolean dropped =
- catalog()
- .asTableCatalog()
- .purgeTable(NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()));
- if (!dropped && !ignoreIfNotExists) {
- throw new TableNotExistException(catalogName(), tablePath);
- }
- if (dropped) {
- invalidateNativeTableCache(tablePath);
- }
- }
-
@Override
protected Distribution toGravitinoDistribution(Map<String, String>
properties) {
return getDistribution(properties);
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 34b7882fea..e24dc13c6c 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
@@ -40,6 +40,7 @@ import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.catalog.lakehouse.paimon.PaimonConstants;
import org.apache.gravitino.flink.connector.PartitionConverter;
import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
@@ -276,6 +277,53 @@ public class TestBaseCatalog {
return new ResolvedCatalogView(catalogView, resolvedSchema);
}
+ @Test
+ public void testBaseCatalogViewDialectFallbackOrderIsFLINKThenHIVE() {
+ BaseCatalog catalog = new TestableBaseCatalog(null, null);
+ List<String> order = catalog.viewDialectFallbackOrder();
+ Assertions.assertEquals(2, order.size());
+ Assertions.assertEquals(Dialects.FLINK, order.get(0));
+ Assertions.assertEquals(Dialects.HIVE, order.get(1));
+ }
+
+ @Test
+ public void
testBaseCatalogBuildViewRepresentationsProducesSingleFlinkEntry() {
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+ ResolvedCatalogView view = resolveView(schema, "SELECT id FROM t",
"comment");
+ BaseCatalog catalog = new TestableBaseCatalog(null, null);
+ Representation[] reps = catalog.buildViewRepresentations(view);
+ Assertions.assertEquals(1, reps.length);
+ Assertions.assertInstanceOf(SQLRepresentation.class, reps[0]);
+ SQLRepresentation sqlRep = (SQLRepresentation) reps[0];
+ Assertions.assertEquals(Dialects.FLINK, sqlRep.dialect());
+ Assertions.assertEquals("SELECT id FROM t", sqlRep.sql());
+ }
+
+ @Test
+ public void testPaimonLikeViewDialectFallbackOrderIsFLINKThenHIVEThenQUERY()
{
+ BaseCatalog catalog = new PaimonLikeBaseCatalog();
+ List<String> order = catalog.viewDialectFallbackOrder();
+ Assertions.assertEquals(3, order.size());
+ Assertions.assertEquals(Dialects.FLINK, order.get(0));
+ Assertions.assertEquals(Dialects.HIVE, order.get(1));
+ Assertions.assertEquals(PaimonConstants.VIEW_QUERY_DIALECT, order.get(2));
+ }
+
+ @Test
+ public void
testPaimonLikeBuildViewRepresentationsProducesBothFlinkAndQueryDialects() {
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+ ResolvedCatalogView view = resolveView(schema, "SELECT id FROM t",
"comment");
+ BaseCatalog catalog = new PaimonLikeBaseCatalog();
+ Representation[] reps = catalog.buildViewRepresentations(view);
+ Assertions.assertEquals(2, reps.length);
+ SQLRepresentation first = (SQLRepresentation) reps[0];
+ Assertions.assertEquals(Dialects.FLINK, first.dialect());
+ Assertions.assertEquals("SELECT id FROM t", first.sql());
+ SQLRepresentation second = (SQLRepresentation) reps[1];
+ Assertions.assertEquals(PaimonConstants.VIEW_QUERY_DIALECT,
second.dialect());
+ Assertions.assertEquals("SELECT id FROM t", second.sql());
+ }
+
/**
* Returns a mock {@link Catalog} whose {@code asViewCatalog()} throws
* UnsupportedOperationException.
@@ -287,6 +335,43 @@ public class TestBaseCatalog {
return gravitinoCatalog;
}
+ /** Simulates a Paimon-like catalog that overrides dialect hooks. */
+ private static class PaimonLikeBaseCatalog extends BaseCatalog {
+
+ PaimonLikeBaseCatalog() {
+ super(
+ "paimon-test",
+ Collections.emptyMap(),
+ "default",
+ Mockito.mock(SchemaAndTablePropertiesConverter.class),
+ Mockito.mock(PartitionConverter.class));
+ }
+
+ @Override
+ protected AbstractCatalog realCatalog() {
+ return null;
+ }
+
+ @Override
+ protected Catalog catalog() {
+ return null;
+ }
+
+ @Override
+ protected List<String> viewDialectFallbackOrder() {
+ return Arrays.asList(Dialects.FLINK, Dialects.HIVE,
PaimonConstants.VIEW_QUERY_DIALECT);
+ }
+
+ @Override
+ protected Representation[] buildViewRepresentations(ResolvedCatalogView
view) {
+ String sql = view.getExpandedQuery();
+ return new Representation[] {
+ buildSqlRepresentation(Dialects.FLINK, sql)[0],
+ buildSqlRepresentation(PaimonConstants.VIEW_QUERY_DIALECT, sql)[0]
+ };
+ }
+ }
+
private static class TestableBaseCatalog extends BaseCatalog {
private final AbstractCatalog delegate;
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
index 993aa794ce..159fe00775 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
@@ -48,12 +48,16 @@ import
org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.types.Row;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.catalog.hive.HiveConstants;
import org.apache.gravitino.flink.connector.integration.test.utils.TestUtils;
import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
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.literals.Literals;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.rel.types.Types;
@@ -95,6 +99,19 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
return false;
}
+ protected boolean supportViewOperation() {
+ return false;
+ }
+
+ /**
+ * Returns the WITH clause to append to a base-table DDL inside view tests.
Hive requires {@code
+ * WITH ('connector'='hive')}; Iceberg/Paimon use their catalog-native
tables so return {@code
+ * ""}.
+ */
+ protected String baseTableConnectorClause() {
+ return "";
+ }
+
protected String defaultDatabaseName() {
return "default";
}
@@ -812,4 +829,286 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
Assertions.assertEquals(expected[i].nullable(), actual[i].nullable());
}
}
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testCreateView() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql(
+ "CREATE VIEW %s COMMENT 'view comment' AS SELECT id, name
FROM %s",
+ viewName, tableName),
+ ResultKind.SUCCESS);
+
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ View view = viewCatalog.loadView(NameIdentifier.of(schemaName,
viewName));
+ Assertions.assertEquals(viewName, view.name());
+ Assertions.assertEquals("view comment", view.comment());
+ Assertions.assertTrue(view.representations().length >= 1);
+ Assertions.assertInstanceOf(SQLRepresentation.class,
view.representations()[0]);
+
+
Assertions.assertTrue(tableEnv.getCatalog(catalog.name()).isPresent());
+ try {
+ CatalogBaseTable flinkTable =
+ tableEnv
+ .getCatalog(catalog.name())
+ .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,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testListViews() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ 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");
+ Assertions.assertFalse(views.contains(tableName), "table should not
appear in listViews");
+
+ 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,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testDropView() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ 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,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testAlterViewRename() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ 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,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testAlterViewReplaceBody() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ 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.assertTrue(view.representations().length >= 1);
+ 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,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testQueryView() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ 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,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testCreateViewIfNotExists() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ ResultKind.SUCCESS);
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW %s AS SELECT id FROM %s", viewName, tableName),
ResultKind.SUCCESS);
+
+ TestUtils.assertTableResult(
+ sql("CREATE VIEW IF NOT EXISTS %s AS SELECT id FROM %s",
viewName, tableName),
+ ResultKind.SUCCESS);
+
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+
Assertions.assertTrue(viewCatalog.viewExists(NameIdentifier.of(schemaName,
viewName)));
+ },
+ true,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testDropViewIfExists() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ 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);
+
+ TestUtils.assertTableResult(sql("DROP VIEW IF EXISTS %s", viewName),
ResultKind.SUCCESS);
+ },
+ true,
+ supportDropCascade());
+ }
+
+ @Test
+ @EnabledIf("supportViewOperation")
+ public void testListTablesDoesNotIncludeViews() {
+ String schemaName = "test_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)%s", tableName,
baseTableConnectorClause()),
+ 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,
+ supportDropCascade());
+ }
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkEnvIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkEnvIT.java
index a0b2c2a10e..e0b0e7f2ea 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkEnvIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkEnvIT.java
@@ -25,7 +25,6 @@ import com.google.errorprone.annotations.FormatString;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import org.apache.flink.configuration.Configuration;
@@ -258,14 +257,17 @@ public abstract class FlinkEnvIT extends BaseIT {
action.accept(catalog);
}
- /** Iceberg requires deleting the table first, then deleting the schema. */
+ /** Iceberg requires deleting tables and views first, then deleting the
schema. */
protected static void clearTableInSchema() {
TableResult result = sql("SHOW TABLES");
- List<Row> rows = Lists.newArrayList(result.collect());
- for (Row row : rows) {
- String tableName = row.getField(0).toString();
- TableResult deleteResult = sql("DROP TABLE IF EXISTS %s", tableName);
- TestUtils.assertTableResult(deleteResult, ResultKind.SUCCESS);
+ for (Row row : Lists.newArrayList(result.collect())) {
+ TestUtils.assertTableResult(
+ sql("DROP TABLE IF EXISTS %s", row.getField(0).toString()),
ResultKind.SUCCESS);
+ }
+ TableResult viewResult = sql("SHOW VIEWS");
+ for (Row row : Lists.newArrayList(viewResult.collect())) {
+ TestUtils.assertTableResult(
+ sql("DROP VIEW IF EXISTS %s", row.getField(0).toString()),
ResultKind.SUCCESS);
}
}
}
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 eac2c9ae3a..a48738a8a6 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,7 +60,6 @@ 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;
@@ -72,10 +71,7 @@ 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;
@@ -749,275 +745,14 @@ 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);
+ @Override
+ protected boolean supportViewOperation() {
+ return 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);
+ @Override
+ protected String baseTableConnectorClause() {
+ return " WITH ('connector'='hive')";
}
private void assertHiveCatalogRead(String databaseName, String tableName,
Row expectedRow) {
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
index e21c911ccb..8bbbe32e56 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/iceberg/FlinkIcebergCatalogIT.java
@@ -67,6 +67,11 @@ public abstract class FlinkIcebergCatalogIT extends
FlinkCommonIT {
return false;
}
+ @Override
+ protected boolean supportViewOperation() {
+ return true;
+ }
+
@BeforeAll
public void before() {
Preconditions.checkArgument(metalake != null, "metalake should not be
null");
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT.java
index 72160ecf8c..7ae9396c33 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonHiveBackendIT.java
@@ -56,4 +56,9 @@ public abstract class FlinkPaimonHiveBackendIT extends
FlinkPaimonCatalogIT {
protected String getWarehouse() {
return warehouse;
}
+
+ @Override
+ protected boolean supportViewOperation() {
+ return true;
+ }
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT.java
index 2f8e117411..7b93f3ff49 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonJdbcBackendIT.java
@@ -94,4 +94,10 @@ public abstract class FlinkPaimonJdbcBackendIT extends
FlinkPaimonCatalogIT {
protected String getWarehouse() {
return warehouseDir.toString();
}
+
+ @Override
+ protected boolean supportViewOperation() {
+ // Paimon JDBC metastore backend does not support view operations.
+ return false;
+ }
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
index 1df9150b4e..3e29e30e2e 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
@@ -399,6 +399,7 @@ public class TestGravitinoPaimonCatalog {
TableCatalog mockTableCatalog = mock(TableCatalog.class);
when(mockCatalog.asTableCatalog()).thenReturn(mockTableCatalog);
when(mockTableCatalog.purgeTable(any())).thenReturn(true);
+ when(mockCatalog.asViewCatalog()).thenThrow(new
UnsupportedOperationException("no views"));
TestablePaimonCatalog cat = new TestablePaimonCatalog(mockFlinkCatalog,
mockCatalog);
ObjectPath path = new ObjectPath("mydb", "mytable");
