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 bb9e0387a6 [#11003] feat(catalog-lakehouse-paimon): add paimon view
CRUD support and interoperability tests (#11013)
bb9e0387a6 is described below
commit bb9e0387a6fbe7b87d25b14225e2356c360f3365
Author: mchades <[email protected]>
AuthorDate: Thu May 21 22:13:24 2026 +0800
[#11003] feat(catalog-lakehouse-paimon): add paimon view CRUD support and
interoperability tests (#11013)
### What changes were proposed in this pull request?
- Added Paimon view CRUD support in Gravitino
(`list/load/create/alter/rename/drop`).
- Added Paimon view metadata conversion and extracted view logic into a
dedicated composed ops class.
- Exposed Paimon catalog as `ViewCatalog`.
- Added tests:
- unit tests for view ops/wiring
- integration tests for:
- Spark create view -> Gravitino load view
- Gravitino create view -> Spark query view
### Why are the changes needed?
Paimon catalog did not fully support view CRUD in Gravitino. This PR
completes Paimon view capability and adds interoperability coverage with
Spark.
Fix: #11003
### Does this PR introduce _any_ user-facing change?
Yes.
- Paimon catalog now supports `ViewCatalog` operations in Gravitino.
- Added internal reserved hidden properties for view default
catalog/schema handling.
### How was this patch tested?
- `./gradlew :catalogs:catalog-lakehouse-paimon:spotlessApply`
- `./gradlew :catalogs:catalog-lakehouse-paimon:test -PskipITs`
- `SKIP_DOCKER_TESTS=false ./gradlew
:catalogs:catalog-lakehouse-paimon:test -PskipDockerTests=false --tests
org.apache.gravitino.catalog.lakehouse.paimon.integration.test.CatalogPaimonHiveIT.testSparkCreateViewAndLoadByGravitino
--tests
org.apache.gravitino.catalog.lakehouse.paimon.integration.test.CatalogPaimonHiveIT.testGravitinoCreateViewAndReadBySpark`
---
catalogs/catalog-lakehouse-paimon/build.gradle.kts | 34 +-
.../catalog/lakehouse/paimon/PaimonCatalog.java | 6 +
.../lakehouse/paimon/PaimonCatalogOperations.java | 91 +++-
.../paimon/PaimonTablePropertiesMetadata.java | 6 +-
.../catalog/lakehouse/paimon/PaimonView.java | 221 +++++++++
.../lakehouse/paimon/PaimonViewCatalogOps.java | 310 ++++++++++++
.../lakehouse/paimon/ops/PaimonCatalogOps.java | 41 ++
.../lakehouse/paimon/TestPaimonCatalog.java | 29 ++
.../lakehouse/paimon/TestPaimonViewCatalogOps.java | 526 +++++++++++++++++++++
.../integration/test/CatalogPaimonBaseIT.java | 75 +++
.../integration/test/CatalogPaimonHiveIT.java | 91 ++++
.../lakehouse/paimon/ops/TestPaimonCatalogOps.java | 103 +++-
.../paimon/utils/PaimonViewTestCatalogHelper.java | 178 +++++++
13 files changed, 1706 insertions(+), 5 deletions(-)
diff --git a/catalogs/catalog-lakehouse-paimon/build.gradle.kts
b/catalogs/catalog-lakehouse-paimon/build.gradle.kts
index ab9eb42986..4ef2ad0584 100644
--- a/catalogs/catalog-lakehouse-paimon/build.gradle.kts
+++ b/catalogs/catalog-lakehouse-paimon/build.gradle.kts
@@ -125,6 +125,39 @@ dependencies {
exclude("*")
}
+ // Required by Paimon HiveCatalog#createView, which calls hive ql metadata
Table APIs.
+ runtimeOnly(libs.hive2.exec) {
+ // Use the lightweight core artifact instead of the shaded hive-exec fat
jar.
+ artifact {
+ classifier = "core"
+ }
+ // Keep hive-exec footprint minimal for catalog runtime packaging.
+ exclude("com.google.code.findbugs", "jsr305")
+ exclude("com.google.protobuf")
+ exclude("org.apache.ant")
+ exclude("org.apache.avro")
+ exclude("org.apache.calcite")
+ exclude("org.apache.calcite.avatica")
+ exclude("org.apache.curator")
+ exclude("org.apache.hadoop")
+ exclude("org.apache.hive", "hive-llap-tez")
+ exclude("org.apache.hive", "hive-vector-code-gen")
+ exclude("org.apache.ivy")
+ exclude("org.apache.logging.log4j")
+ exclude("org.apache.zookeeper")
+ exclude("org.codehaus.groovy")
+ exclude("org.datanucleus")
+ exclude("org.eclipse.jetty.aggregate", "jetty-all")
+ exclude("org.eclipse.jetty.orbit", "javax.servlet")
+ exclude("org.openjdk.jol")
+ // Avoid resolving non-Central transitive artifacts (e.g.
pentaho-aggdesigner-algorithm).
+ exclude("org.pentaho")
+ exclude("org.codehaus.janino")
+ exclude("net.hydromatic", "eigenbase-properties")
+ exclude("org.slf4j")
+ exclude("stax", "stax-api")
+ }
+
annotationProcessor(libs.lombok)
testImplementation(project(":api"))
@@ -153,7 +186,6 @@ dependencies {
exclude("org.rocksdb")
}
testImplementation(libs.awaitility)
- testImplementation(libs.awaitility)
testImplementation(libs.bundles.log4j)
testImplementation(libs.h2db)
testImplementation(libs.junit.jupiter.api)
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalog.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalog.java
index ba69cbfc0b..34c327bc41 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalog.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalog.java
@@ -31,6 +31,7 @@ import org.apache.gravitino.connector.PropertiesMetadata;
import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.JdbcCredential;
+import org.apache.gravitino.rel.ViewCatalog;
/**
* Implementation of {@link Catalog} that represents an Apache Paimon catalog
in Apache Gravitino.
@@ -65,6 +66,11 @@ public class PaimonCatalog extends
BaseCatalog<PaimonCatalog> {
return new PaimonCatalogOperations();
}
+ @Override
+ public ViewCatalog asViewCatalog() {
+ return (ViewCatalog) ops();
+ }
+
@Override
public Capability newCapability() {
return new PaimonCatalogCapability();
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
index aa2a298a62..8e9f31e79b 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogOperations.java
@@ -49,14 +49,20 @@ import
org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchColumnException;
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.meta.AuditInfo;
import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Representation;
import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.TableChange.RenameTable;
+import org.apache.gravitino.rel.View;
+import org.apache.gravitino.rel.ViewCatalog;
+import org.apache.gravitino.rel.ViewChange;
import org.apache.gravitino.rel.expressions.NamedReference;
import org.apache.gravitino.rel.expressions.distributions.Distribution;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -77,11 +83,13 @@ import org.slf4j.LoggerFactory;
* Implementation of {@link CatalogOperations} that represents operations for
interacting with the
* Apache Paimon catalog in Apache Gravitino.
*/
-public class PaimonCatalogOperations implements CatalogOperations,
SupportsSchemas, TableCatalog {
+public class PaimonCatalogOperations
+ implements CatalogOperations, SupportsSchemas, TableCatalog, ViewCatalog {
public static final Logger LOG =
LoggerFactory.getLogger(PaimonCatalogOperations.class);
@VisibleForTesting public PaimonCatalogOps paimonCatalogOps;
+ @VisibleForTesting PaimonViewCatalogOps paimonViewCatalogOps;
private static final String NO_SUCH_SCHEMA_EXCEPTION =
"Paimon schema (database) %s does not exist.";
@@ -117,6 +125,9 @@ public class PaimonCatalogOperations implements
CatalogOperations, SupportsSchem
resultConf.putAll(gravitinoConfig);
this.paimonCatalogOps = new PaimonCatalogOps(new PaimonConfig(resultConf));
+ this.paimonViewCatalogOps =
+ new PaimonViewCatalogOps(
+ paimonCatalogOps, this::buildPaimonNameIdentifier,
this::schemaExists);
}
/**
@@ -465,6 +476,84 @@ public class PaimonCatalogOperations implements
CatalogOperations, SupportsSchem
return true;
}
+ /**
+ * Lists all views under the specified namespace.
+ *
+ * @param namespace The namespace to list views for.
+ * @return An array of {@link NameIdentifier} representing the views in the
namespace.
+ * @throws NoSuchSchemaException If the schema with the provided namespace
does not exist.
+ */
+ @Override
+ public NameIdentifier[] listViews(Namespace namespace) throws
NoSuchSchemaException {
+ return paimonViewCatalogOps.listViews(namespace);
+ }
+
+ /**
+ * Loads the view with the provided identifier.
+ *
+ * @param identifier The identifier of the view to load.
+ * @return The loaded {@link View} instance representing the view metadata.
+ * @throws NoSuchViewException If the view with the provided identifier does
not exist.
+ */
+ @Override
+ public View loadView(NameIdentifier identifier) throws NoSuchViewException {
+ return paimonViewCatalogOps.loadView(identifier);
+ }
+
+ /**
+ * Creates a new view with the provided metadata.
+ *
+ * @param identifier The identifier of the view to create.
+ * @param comment The view comment.
+ * @param columns The output columns of the view.
+ * @param representations The view SQL representations.
+ * @param defaultCatalog The default catalog for unqualified identifiers.
+ * @param defaultSchema The default schema for unqualified identifiers.
+ * @param properties The view properties.
+ * @return The created view metadata.
+ * @throws NoSuchSchemaException If the target schema does not exist.
+ * @throws ViewAlreadyExistsException If the view already exists.
+ */
+ @Override
+ public View createView(
+ NameIdentifier identifier,
+ String comment,
+ Column[] columns,
+ Representation[] representations,
+ String defaultCatalog,
+ String defaultSchema,
+ Map<String, String> properties)
+ throws NoSuchSchemaException, ViewAlreadyExistsException {
+ return paimonViewCatalogOps.createView(
+ identifier, comment, columns, representations, defaultCatalog,
defaultSchema, properties);
+ }
+
+ /**
+ * Alters an existing view according to the provided changes.
+ *
+ * @param identifier The identifier of the view to alter.
+ * @param changes The changes to apply.
+ * @return The updated view metadata.
+ * @throws NoSuchViewException If the view does not exist.
+ * @throws IllegalArgumentException If any change type is unsupported by
Paimon.
+ */
+ @Override
+ public View alterView(NameIdentifier identifier, ViewChange... changes)
+ throws NoSuchViewException, IllegalArgumentException {
+ return paimonViewCatalogOps.alterView(identifier, changes);
+ }
+
+ /**
+ * Drops the view with the provided identifier.
+ *
+ * @param identifier The identifier of the view to drop.
+ * @return true if the view is successfully dropped, false if the view does
not exist.
+ */
+ @Override
+ public boolean dropView(NameIdentifier identifier) {
+ return paimonViewCatalogOps.dropView(identifier);
+ }
+
@Override
public void close() {
if (paimonCatalogOps != null) {
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonTablePropertiesMetadata.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonTablePropertiesMetadata.java
index e56e6ea6f3..b9f0139519 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonTablePropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonTablePropertiesMetadata.java
@@ -61,7 +61,11 @@ public class PaimonTablePropertiesMetadata extends
BasePropertiesMetadata {
stringImmutablePropertyEntry(
ROWKIND_FIELD, "The table rowkind field", false, null, false,
false),
stringReservedPropertyEntry(PRIMARY_KEY, "The table primary key",
false),
- stringReservedPropertyEntry(PARTITION, "The table partition",
false));
+ stringReservedPropertyEntry(PARTITION, "The table partition",
false),
+ stringReservedPropertyEntry(
+ PaimonView.DEFAULT_CATALOG_PROPERTY, "The view default
catalog", true),
+ stringReservedPropertyEntry(
+ PaimonView.DEFAULT_SCHEMA_PROPERTY, "The view default schema",
true));
PROPERTIES_METADATA = Maps.uniqueIndex(propertyEntries,
PropertyEntry::getName);
}
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
new file mode 100644
index 0000000000..0d0f22d656
--- /dev/null
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonView.java
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.lakehouse.paimon;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.SQLRepresentation;
+import org.apache.gravitino.rel.View;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.view.ViewImpl;
+
+/** Represents a Gravitino view object converted from Apache Paimon view
metadata. */
+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 final String name;
+ @Nullable private final String comment;
+ private final Column[] columns;
+ private final Representation[] representations;
+ @Nullable private final String defaultCatalog;
+ @Nullable private final String defaultSchema;
+ private final Map<String, String> properties;
+ private final AuditInfo auditInfo;
+
+ private PaimonView(
+ String name,
+ @Nullable String comment,
+ Column[] columns,
+ Representation[] representations,
+ @Nullable String defaultCatalog,
+ @Nullable String defaultSchema,
+ Map<String, String> properties,
+ AuditInfo auditInfo) {
+ this.name = name;
+ this.comment = comment;
+ this.columns = Arrays.copyOf(columns, columns.length);
+ this.representations = Arrays.copyOf(representations,
representations.length);
+ this.defaultCatalog = defaultCatalog;
+ this.defaultSchema = defaultSchema;
+ this.properties = new HashMap<>(properties);
+ this.auditInfo = auditInfo;
+ }
+
+ static PaimonView fromPaimonView(org.apache.paimon.view.View view) {
+ Map<String, String> options =
+ view.options() == null ? new HashMap<>() : new
HashMap<>(view.options());
+
+ String defaultCatalog = options.remove(DEFAULT_CATALOG_PROPERTY);
+ String defaultSchema = options.remove(DEFAULT_SCHEMA_PROPERTY);
+
+ return new PaimonView(
+ view.name(),
+ view.comment().orElse(null),
+ GravitinoPaimonColumn.fromPaimonRowType(view.rowType()).toArray(new
Column[0]),
+ toRepresentations(view.query(), view.dialects()),
+ defaultCatalog,
+ defaultSchema,
+ options,
+ AuditInfo.EMPTY);
+ }
+
+ static org.apache.paimon.view.View toPaimonView(
+ NameIdentifier ident,
+ @Nullable String comment,
+ @Nullable Column[] columns,
+ Representation[] representations,
+ @Nullable String defaultCatalog,
+ @Nullable String defaultSchema,
+ Map<String, String> properties) {
+ Preconditions.checkArgument(ident != null, "View identifier must not be
null");
+ Preconditions.checkArgument(
+ representations != null && representations.length > 0,
+ "representations must not be null or empty");
+
+ Column[] safeColumns = columns == null ? new Column[0] : columns;
+ List<DataField> fields = new ArrayList<>(safeColumns.length);
+ for (int index = 0; index < safeColumns.length; index++) {
+ fields.add(GravitinoPaimonColumn.toPaimonColumn(index,
safeColumns[index]));
+ }
+
+ Map<String, String> dialectQueries = new HashMap<>();
+ String query = null;
+ for (Representation representation : representations) {
+ Preconditions.checkArgument(
+ representation instanceof SQLRepresentation, "Paimon only supports
SQL representations");
+ SQLRepresentation sqlRepresentation = (SQLRepresentation) representation;
+ if (PAIMON_VIEW_QUERY.equalsIgnoreCase(sqlRepresentation.dialect())) {
+ Preconditions.checkArgument(
+ query == null,
+ "Only one representation with dialect '%s' is allowed",
+ PAIMON_VIEW_QUERY);
+ query = sqlRepresentation.sql();
+ } else {
+ String normalizedDialect =
sqlRepresentation.dialect().toLowerCase(Locale.ROOT);
+ Preconditions.checkArgument(
+ !dialectQueries.containsKey(normalizedDialect),
+ "Only one representation per dialect is allowed
(case-insensitive). Found duplicate: %s",
+ sqlRepresentation.dialect());
+ dialectQueries.put(normalizedDialect, sqlRepresentation.sql());
+ }
+ }
+
+ Preconditions.checkArgument(
+ query != null && !query.isEmpty(),
+ "View representation with dialect '%s' must not be null or empty",
+ PAIMON_VIEW_QUERY);
+
+ Map<String, String> options = properties == null ? new HashMap<>() : new
HashMap<>(properties);
+ if (defaultCatalog != null) {
+ options.put(DEFAULT_CATALOG_PROPERTY, defaultCatalog);
+ } else {
+ options.remove(DEFAULT_CATALOG_PROPERTY);
+ }
+
+ if (defaultSchema != null) {
+ options.put(DEFAULT_SCHEMA_PROPERTY, defaultSchema);
+ } else {
+ options.remove(DEFAULT_SCHEMA_PROPERTY);
+ }
+
+ String[] namespaceLevels = ident.namespace().levels();
+ Preconditions.checkArgument(namespaceLevels.length > 0, "View namespace
must not be empty");
+ Identifier paimonIdentifier =
+ Identifier.create(namespaceLevels[namespaceLevels.length - 1],
ident.name());
+
+ return new ViewImpl(paimonIdentifier, fields, query, dialectQueries,
comment, options);
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public String comment() {
+ return comment;
+ }
+
+ @Override
+ public Column[] columns() {
+ return Arrays.copyOf(columns, columns.length);
+ }
+
+ @Override
+ public Representation[] representations() {
+ return Arrays.copyOf(representations, representations.length);
+ }
+
+ @Override
+ public String defaultCatalog() {
+ return defaultCatalog;
+ }
+
+ @Override
+ public String defaultSchema() {
+ return defaultSchema;
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return Collections.unmodifiableMap(properties);
+ }
+
+ @Override
+ public AuditInfo auditInfo() {
+ return auditInfo;
+ }
+
+ private static Representation[] toRepresentations(String query, Map<String,
String> dialects) {
+ Preconditions.checkArgument(
+ query != null && !query.isEmpty(), "Paimon view query must not be null
or empty");
+
+ Map<String, String> safeDialects =
+ dialects == null ? Collections.emptyMap() : new HashMap<>(dialects);
+ List<Representation> sqlRepresentations = new
ArrayList<>(safeDialects.size() + 1);
+ sqlRepresentations.add(
+
SQLRepresentation.builder().withDialect(PAIMON_VIEW_QUERY).withSql(query).build());
+
+ safeDialects.forEach(
+ (dialect, sql) -> {
+ if (!PAIMON_VIEW_QUERY.equalsIgnoreCase(dialect)) {
+ sqlRepresentations.add(
+
SQLRepresentation.builder().withDialect(dialect).withSql(sql).build());
+ }
+ });
+
+ return sqlRepresentations.toArray(new Representation[0]);
+ }
+}
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonViewCatalogOps.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonViewCatalogOps.java
new file mode 100644
index 0000000000..179a9ff09a
--- /dev/null
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonViewCatalogOps.java
@@ -0,0 +1,310 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.lakehouse.paimon;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.catalog.lakehouse.paimon.ops.PaimonCatalogOps;
+import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NoSuchViewException;
+import org.apache.gravitino.exceptions.ViewAlreadyExistsException;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.View;
+import org.apache.gravitino.rel.ViewChange;
+import org.apache.paimon.catalog.Catalog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Encapsulates Paimon view operations so {@link PaimonCatalogOperations} can
compose view logic.
+ */
+final class PaimonViewCatalogOps {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PaimonViewCatalogOps.class);
+
+ private static final String NO_SUCH_SCHEMA_EXCEPTION =
+ "Paimon schema (database) %s does not exist.";
+ private static final String NO_SUCH_VIEW_EXCEPTION = "Paimon view %s does
not exist.";
+ private static final String VIEW_ALREADY_EXISTS_EXCEPTION = "Paimon view %s
already exists.";
+
+ private final PaimonCatalogOps paimonCatalogOps;
+ private final Function<NameIdentifier, NameIdentifier>
paimonIdentifierBuilder;
+ private final Predicate<NameIdentifier> schemaExistsChecker;
+
+ PaimonViewCatalogOps(
+ PaimonCatalogOps paimonCatalogOps,
+ Function<NameIdentifier, NameIdentifier> paimonIdentifierBuilder,
+ Predicate<NameIdentifier> schemaExistsChecker) {
+ Preconditions.checkArgument(paimonCatalogOps != null, "Paimon catalog ops
must not be null");
+ Preconditions.checkArgument(
+ paimonIdentifierBuilder != null, "Paimon identifier builder must not
be null");
+ Preconditions.checkArgument(
+ schemaExistsChecker != null, "Schema exists checker must not be null");
+
+ this.paimonCatalogOps = paimonCatalogOps;
+ this.paimonIdentifierBuilder = paimonIdentifierBuilder;
+ this.schemaExistsChecker = schemaExistsChecker;
+ }
+
+ NameIdentifier[] listViews(Namespace namespace) throws NoSuchSchemaException
{
+ String[] levels = namespace.levels();
+ NameIdentifier schemaIdentifier = NameIdentifier.of(levels[levels.length -
1]);
+
+ List<String> views;
+ try {
+ views = paimonCatalogOps.listViews(schemaIdentifier.name());
+ } catch (Catalog.DatabaseNotExistException e) {
+ throw new NoSuchSchemaException(e, NO_SUCH_SCHEMA_EXCEPTION,
namespace.toString());
+ }
+
+ return views.stream()
+ .map(
+ viewIdentifier ->
NameIdentifier.of(ArrayUtils.add(namespace.levels(), viewIdentifier)))
+ .toArray(NameIdentifier[]::new);
+ }
+
+ View loadView(NameIdentifier identifier) throws NoSuchViewException {
+ org.apache.paimon.view.View view;
+ try {
+ NameIdentifier viewIdentifier =
paimonIdentifierBuilder.apply(identifier);
+ view = paimonCatalogOps.loadView(viewIdentifier.toString());
+ } catch (Catalog.ViewNotExistException e) {
+ throw new NoSuchViewException(e, NO_SUCH_VIEW_EXCEPTION, identifier);
+ }
+ return PaimonView.fromPaimonView(view);
+ }
+
+ View createView(
+ NameIdentifier identifier,
+ String comment,
+ Column[] columns,
+ Representation[] representations,
+ String defaultCatalog,
+ String defaultSchema,
+ Map<String, String> properties)
+ throws NoSuchSchemaException, ViewAlreadyExistsException {
+ NameIdentifier nameIdentifier = paimonIdentifierBuilder.apply(identifier);
+ NameIdentifier schemaIdentifier =
NameIdentifier.of(nameIdentifier.namespace().levels());
+ if (!schemaExistsChecker.test(schemaIdentifier)) {
+ throw new NoSuchSchemaException(NO_SUCH_SCHEMA_EXCEPTION,
schemaIdentifier);
+ }
+
+ org.apache.paimon.view.View paimonView =
+ PaimonView.toPaimonView(
+ identifier,
+ comment,
+ columns,
+ representations,
+ defaultCatalog,
+ defaultSchema,
+ properties);
+
+ try {
+ paimonCatalogOps.createView(nameIdentifier.toString(), paimonView);
+ } catch (Catalog.DatabaseNotExistException e) {
+ throw new NoSuchSchemaException(e, NO_SUCH_SCHEMA_EXCEPTION,
schemaIdentifier);
+ } catch (Catalog.ViewAlreadyExistException e) {
+ throw new ViewAlreadyExistsException(e, VIEW_ALREADY_EXISTS_EXCEPTION,
identifier);
+ }
+
+ return PaimonView.fromPaimonView(paimonView);
+ }
+
+ View alterView(NameIdentifier identifier, ViewChange... changes)
+ throws NoSuchViewException, IllegalArgumentException {
+ if (changes == null || changes.length == 0) {
+ return loadView(identifier);
+ }
+
+ List<ViewChange.RenameView> renameViewChanges =
+ Arrays.stream(changes)
+ .filter(viewChange -> viewChange instanceof ViewChange.RenameView)
+ .map(viewChange -> (ViewChange.RenameView) viewChange)
+ .collect(Collectors.toList());
+ if (!renameViewChanges.isEmpty()) {
+ Preconditions.checkArgument(
+ renameViewChanges.size() == 1,
+ "Only one rename operation is allowed, but found: %s",
+ renameViewChanges.size());
+
+ List<String> otherChanges =
+ Arrays.stream(changes)
+ .filter(viewChange -> !(viewChange instanceof
ViewChange.RenameView))
+ .map(String::valueOf)
+ .collect(Collectors.toList());
+ Preconditions.checkArgument(
+ otherChanges.isEmpty(),
+ "The operation to change the view name cannot be performed together
with other operations. "
+ + "The list of operations that you cannot perform includes:
\n%s",
+ String.join("\n", otherChanges));
+ return renameView(identifier, renameViewChanges.get(0));
+ }
+
+ if (containsReplaceView(changes)) {
+ return replaceView(identifier, changes);
+ }
+
+ return alterViewWithoutReplace(identifier, changes);
+ }
+
+ boolean dropView(NameIdentifier identifier) {
+ try {
+ NameIdentifier viewIdentifier =
paimonIdentifierBuilder.apply(identifier);
+ paimonCatalogOps.dropView(viewIdentifier.toString());
+ } catch (Catalog.ViewNotExistException e) {
+ return false;
+ }
+
+ return true;
+ }
+
+ private View renameView(NameIdentifier identifier, ViewChange.RenameView
renameView)
+ throws NoSuchViewException, IllegalArgumentException {
+ NameIdentifier sourceIdentifier =
paimonIdentifierBuilder.apply(identifier);
+ NameIdentifier renamedIdentifier =
+ NameIdentifier.of(identifier.namespace(), renameView.getNewName());
+ NameIdentifier targetIdentifier =
paimonIdentifierBuilder.apply(renamedIdentifier);
+ try {
+ paimonCatalogOps.renameView(sourceIdentifier.toString(),
targetIdentifier.toString());
+ } catch (Catalog.ViewNotExistException e) {
+ throw new NoSuchViewException(e, NO_SUCH_VIEW_EXCEPTION, identifier);
+ } catch (Catalog.ViewAlreadyExistException e) {
+ throw new ViewAlreadyExistsException(e, VIEW_ALREADY_EXISTS_EXCEPTION,
renamedIdentifier);
+ }
+
+ try {
+ return loadView(renamedIdentifier);
+ } catch (NoSuchViewException e) {
+ throw new IllegalStateException(
+ String.format(
+ "Paimon view %s was renamed to %s, but loading the renamed view
failed.",
+ identifier, renamedIdentifier),
+ e);
+ }
+ }
+
+ private View replaceView(NameIdentifier identifier, ViewChange... changes)
+ throws NoSuchViewException, IllegalArgumentException {
+ View existingView = loadView(identifier);
+ Map<String, String> finalProperties = new
HashMap<>(existingView.properties());
+ ViewChange.ReplaceView replaceView = null;
+
+ for (ViewChange change : changes) {
+ if (change instanceof ViewChange.SetProperty) {
+ ViewChange.SetProperty setProperty = (ViewChange.SetProperty) change;
+ finalProperties.put(setProperty.getProperty(), setProperty.getValue());
+ } else if (change instanceof ViewChange.RemoveProperty) {
+ finalProperties.remove(((ViewChange.RemoveProperty)
change).getProperty());
+ } else if (change instanceof ViewChange.ReplaceView) {
+ replaceView = (ViewChange.ReplaceView) change;
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported view change type: " +
change.getClass().getSimpleName());
+ }
+ }
+
+ Preconditions.checkArgument(replaceView != null, "Replace view change is
required");
+
+ org.apache.paimon.view.View paimonView =
+ PaimonView.toPaimonView(
+ identifier,
+ replaceView.getComment(),
+ replaceView.getColumns(),
+ replaceView.getRepresentations(),
+ replaceView.getDefaultCatalog(),
+ replaceView.getDefaultSchema(),
+ finalProperties);
+
+ NameIdentifier sourceIdentifier =
paimonIdentifierBuilder.apply(identifier);
+
+ try {
+ // Paimon does not provide a native replace-view API. To align with
Paimon Spark's
+ // CREATE OR REPLACE VIEW behavior, replace is executed as
drop-then-create.
+ paimonCatalogOps.dropView(sourceIdentifier.toString());
+ } catch (Catalog.ViewNotExistException e) {
+ throw new NoSuchViewException(e, NO_SUCH_VIEW_EXCEPTION, identifier);
+ }
+
+ try {
+ paimonCatalogOps.createView(sourceIdentifier.toString(), paimonView);
+ } catch (Catalog.ViewAlreadyExistException e) {
+ LOG.error(
+ "Replacing Paimon view {} is non-atomic (drop-then-create). "
+ + "Create failed after drop and the original view may be lost.",
+ identifier,
+ e);
+ throw new IllegalArgumentException(
+ String.format(VIEW_ALREADY_EXISTS_EXCEPTION, identifier), e);
+ } catch (Catalog.DatabaseNotExistException e) {
+ LOG.error(
+ "Replacing Paimon view {} is non-atomic (drop-then-create). "
+ + "Create failed after drop and the original view may be lost.",
+ identifier,
+ e);
+ throw new IllegalArgumentException(
+ String.format(NO_SUCH_SCHEMA_EXCEPTION, identifier.namespace()), e);
+ }
+
+ return loadView(identifier);
+ }
+
+ private View alterViewWithoutReplace(NameIdentifier identifier,
ViewChange... changes)
+ throws NoSuchViewException, IllegalArgumentException {
+ List<org.apache.paimon.view.ViewChange> paimonViewChanges = new
ArrayList<>();
+
+ try {
+ for (ViewChange change : changes) {
+ if (change instanceof ViewChange.SetProperty) {
+ ViewChange.SetProperty setProperty = (ViewChange.SetProperty) change;
+ paimonViewChanges.add(
+ org.apache.paimon.view.ViewChange.setOption(
+ setProperty.getProperty(), setProperty.getValue()));
+ } else if (change instanceof ViewChange.RemoveProperty) {
+ ViewChange.RemoveProperty removeProperty =
(ViewChange.RemoveProperty) change;
+ paimonViewChanges.add(
+
org.apache.paimon.view.ViewChange.removeOption(removeProperty.getProperty()));
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported view change type: " +
change.getClass().getSimpleName());
+ }
+ }
+ paimonCatalogOps.alterView(
+ paimonIdentifierBuilder.apply(identifier).toString(),
paimonViewChanges);
+ } catch (Catalog.ViewNotExistException e) {
+ throw new NoSuchViewException(e, NO_SUCH_VIEW_EXCEPTION, identifier);
+ }
+
+ return loadView(identifier);
+ }
+
+ private boolean containsReplaceView(ViewChange... changes) {
+ return Arrays.stream(changes).anyMatch(change -> change instanceof
ViewChange.ReplaceView);
+ }
+}
diff --git
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/PaimonCatalogOps.java
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/PaimonCatalogOps.java
index 02996a2f5f..0851f2fa0b 100644
---
a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/PaimonCatalogOps.java
+++
b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/PaimonCatalogOps.java
@@ -32,11 +32,17 @@ import
org.apache.paimon.catalog.Catalog.ColumnNotExistException;
import org.apache.paimon.catalog.Catalog.DatabaseAlreadyExistException;
import org.apache.paimon.catalog.Catalog.DatabaseNotEmptyException;
import org.apache.paimon.catalog.Catalog.DatabaseNotExistException;
+import org.apache.paimon.catalog.Catalog.DialectAlreadyExistException;
+import org.apache.paimon.catalog.Catalog.DialectNotExistException;
import org.apache.paimon.catalog.Catalog.TableAlreadyExistException;
import org.apache.paimon.catalog.Catalog.TableNotExistException;
+import org.apache.paimon.catalog.Catalog.ViewAlreadyExistException;
+import org.apache.paimon.catalog.Catalog.ViewNotExistException;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.Table;
+import org.apache.paimon.view.View;
+import org.apache.paimon.view.ViewChange;
/** Table operation proxy that handles table operations of an underlying
Apache Paimon catalog. */
public class PaimonCatalogOps implements AutoCloseable {
@@ -104,7 +110,42 @@ public class PaimonCatalogOps implements AutoCloseable {
catalog.renameTable(tableIdentifier(fromTableName),
tableIdentifier(toTableName), false);
}
+ public List<String> listViews(String databaseName) throws
DatabaseNotExistException {
+ return catalog.listViews(databaseName);
+ }
+
+ public View loadView(String viewName) throws ViewNotExistException {
+ return catalog.getView(viewIdentifier(viewName));
+ }
+
+ public void createView(String viewName, View view)
+ throws ViewAlreadyExistException, DatabaseNotExistException {
+ catalog.createView(viewIdentifier(viewName), view, false);
+ }
+
+ public void alterView(String viewName, List<ViewChange> changes) throws
ViewNotExistException {
+ try {
+ catalog.alterView(viewIdentifier(viewName), changes, false);
+ } catch (DialectAlreadyExistException | DialectNotExistException e) {
+ throw new IllegalArgumentException(
+ String.format("Cannot alter view %s: %s", viewName, e.getMessage()),
e);
+ }
+ }
+
+ public void renameView(String fromViewName, String toViewName)
+ throws ViewNotExistException, ViewAlreadyExistException {
+ catalog.renameView(viewIdentifier(fromViewName),
viewIdentifier(toViewName), false);
+ }
+
+ public void dropView(String viewName) throws ViewNotExistException {
+ catalog.dropView(viewIdentifier(viewName), false);
+ }
+
private Identifier tableIdentifier(String tableName) {
return Identifier.fromString(tableName);
}
+
+ private Identifier viewIdentifier(String viewName) {
+ return Identifier.fromString(viewName);
+ }
}
diff --git
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestPaimonCatalog.java
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestPaimonCatalog.java
index e83f8c7d78..93510faf29 100644
---
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestPaimonCatalog.java
+++
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestPaimonCatalog.java
@@ -44,6 +44,7 @@ import org.apache.gravitino.credential.OSSSecretKeyCredential;
import org.apache.gravitino.credential.S3SecretKeyCredential;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.rel.ViewCatalog;
import org.apache.gravitino.storage.AzureProperties;
import org.apache.gravitino.storage.OSSProperties;
import org.apache.gravitino.storage.S3Properties;
@@ -145,6 +146,14 @@ public class TestPaimonCatalog {
ImmutableMap.of()));
}
+ @Test
+ void testAsViewCatalog() {
+ PaimonCatalog paimonCatalog = newPaimonCatalog("catalog_view");
+ ViewCatalog viewCatalog = paimonCatalog.asViewCatalog();
+ Assertions.assertNotNull(viewCatalog);
+ Assertions.assertSame(paimonCatalog.ops(), viewCatalog);
+ }
+
@Test
void testCatalogProperty() throws IOException {
AuditInfo auditInfo =
@@ -403,4 +412,24 @@ public class TestPaimonCatalog {
String credentialProviders =
properties.get(CredentialConstants.CREDENTIAL_PROVIDERS);
Assertions.assertEquals("custom-provider", credentialProviders);
}
+
+ private PaimonCatalog newPaimonCatalog(String catalogName) {
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+ CatalogEntity entity =
+ CatalogEntity.builder()
+ .withId(1L)
+ .withName(catalogName)
+ .withNamespace(Namespace.of("metalake"))
+ .withType(PaimonCatalog.Type.RELATIONAL)
+ .withProvider("lakehouse-paimon")
+ .withAuditInfo(auditInfo)
+ .build();
+
+ Map<String, String> conf = Maps.newHashMap();
+ conf.put(PaimonCatalogPropertiesMetadata.GRAVITINO_CATALOG_BACKEND,
"filesystem");
+ conf.put(PaimonCatalogPropertiesMetadata.WAREHOUSE, tempDir +
File.separator + catalogName);
+ return new PaimonCatalog().withCatalogConf(conf).withCatalogEntity(entity);
+ }
}
diff --git
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestPaimonViewCatalogOps.java
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestPaimonViewCatalogOps.java
new file mode 100644
index 0000000000..640b98a7f6
--- /dev/null
+++
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/TestPaimonViewCatalogOps.java
@@ -0,0 +1,526 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.lakehouse.paimon;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import java.io.File;
+import java.util.Arrays;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.catalog.lakehouse.paimon.ops.PaimonCatalogOps;
+import
org.apache.gravitino.catalog.lakehouse.paimon.utils.PaimonViewTestCatalogHelper;
+import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NoSuchViewException;
+import org.apache.gravitino.exceptions.ViewAlreadyExistsException;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.SQLRepresentation;
+import org.apache.gravitino.rel.View;
+import org.apache.gravitino.rel.ViewChange;
+import org.apache.gravitino.rel.types.Types;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Tests for {@link PaimonViewCatalogOps}. */
+public class TestPaimonViewCatalogOps {
+
+ private static final String DATABASE = "test_view_ops_database";
+ private static final String VIEW = "test_view_ops_view";
+ private static final NameIdentifier VIEW_IDENTIFIER =
+ NameIdentifier.of(Namespace.of(DATABASE), VIEW);
+
+ @TempDir private File warehouse;
+
+ private TestablePaimonCatalogOps paimonCatalogOps;
+ private PaimonViewCatalogOps paimonViewCatalogOps;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ paimonCatalogOps =
+ new TestablePaimonCatalogOps(
+ new PaimonConfig(
+ ImmutableMap.of(PaimonCatalogPropertiesMetadata.WAREHOUSE,
warehouse.getPath())));
+ paimonCatalogOps.setCatalog(
+
PaimonViewTestCatalogHelper.createViewSupportedCatalog(paimonCatalogOps.catalog()));
+ paimonCatalogOps.createDatabase(DATABASE, Maps.newHashMap());
+
+ paimonViewCatalogOps =
+ new PaimonViewCatalogOps(
+ paimonCatalogOps, this::buildPaimonNameIdentifier,
this::schemaExists);
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ if (paimonCatalogOps != null) {
+ paimonCatalogOps.close();
+ }
+ }
+
+ @Test
+ public void testRenameViewCannotBeCombinedWithOtherChanges() throws
Exception {
+ createView(VIEW_IDENTIFIER);
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ paimonViewCatalogOps.alterView(
+ VIEW_IDENTIFIER,
+ ViewChange.rename(VIEW + "_renamed"),
+ ViewChange.setProperty("k1", "v1")));
+
+ View unchangedView = paimonViewCatalogOps.loadView(VIEW_IDENTIFIER);
+ assertEquals(VIEW, unchangedView.name());
+ assertFalse(unchangedView.properties().containsKey("k1"));
+ }
+
+ @Test
+ public void testAlterViewPropertyChangesAndRenameOnly() throws Exception {
+ createView(VIEW_IDENTIFIER);
+
+ paimonViewCatalogOps.alterView(
+ VIEW_IDENTIFIER,
+ ViewChange.setProperty("k1", "v1"),
+ ViewChange.setProperty("k2", "v2"),
+ ViewChange.removeProperty("k1"));
+
+ View alteredView = paimonViewCatalogOps.loadView(VIEW_IDENTIFIER);
+ assertFalse(alteredView.properties().containsKey("k1"));
+ assertEquals("v2", alteredView.properties().get("k2"));
+
+ String renamedViewName = VIEW + "_renamed";
+ NameIdentifier renamedIdentifier =
+ NameIdentifier.of(VIEW_IDENTIFIER.namespace(), renamedViewName);
+ View renamedView =
+ paimonViewCatalogOps.alterView(VIEW_IDENTIFIER,
ViewChange.rename(renamedViewName));
+ assertEquals(renamedViewName, renamedView.name());
+ assertThrows(NoSuchViewException.class, () ->
paimonViewCatalogOps.loadView(VIEW_IDENTIFIER));
+ assertEquals("v2",
paimonViewCatalogOps.loadView(renamedIdentifier).properties().get("k2"));
+ }
+
+ @Test
+ public void testAlterViewReplaceUpdatesBodyAndPreservesProperties() throws
Exception {
+ NameIdentifier replaceIdentifier =
NameIdentifier.of(Namespace.of(DATABASE), VIEW + "_replace");
+ String originalQuery = "SELECT col_1 FROM source_table";
+
+ paimonViewCatalogOps.createView(
+ replaceIdentifier,
+ "original_view_comment",
+ new Column[] {Column.of("col_1", Types.IntegerType.get(), "col_1")},
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("query").withSql(originalQuery).build(),
+
SQLRepresentation.builder().withDialect("spark").withSql(originalQuery).build()
+ },
+ "paimon",
+ DATABASE,
+ Maps.newHashMap(ImmutableMap.of("keep_key", "keep_value",
"remove_key", "remove_value")));
+
+ String replacedQuery = "SELECT col_2, col_3 FROM source_table";
+ String replacedTrinoQuery = "SELECT CAST(col_2 AS BIGINT), col_3 FROM
source_table";
+ paimonViewCatalogOps.alterView(
+ replaceIdentifier,
+ ViewChange.setProperty("add_key", "add_value"),
+ ViewChange.removeProperty("remove_key"),
+ ViewChange.replaceView(
+ new Column[] {
+ Column.of("col_2", Types.LongType.get(), "col_2"),
+ Column.of("col_3", Types.StringType.get(), "col_3")
+ },
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("query").withSql(replacedQuery).build(),
+
SQLRepresentation.builder().withDialect("trino").withSql(replacedTrinoQuery).build()
+ },
+ "replaced_catalog",
+ "replaced_schema",
+ "replaced_view_comment"));
+
+ View replacedView = paimonViewCatalogOps.loadView(replaceIdentifier);
+ assertEquals("replaced_view_comment", replacedView.comment());
+ assertEquals(2, replacedView.columns().length);
+ assertEquals("col_2", replacedView.columns()[0].name());
+ assertEquals("col_3", replacedView.columns()[1].name());
+ assertEquals(2, replacedView.representations().length);
+ assertTrue(replacedView.sqlFor("query").isPresent());
+ assertEquals(replacedQuery, replacedView.sqlFor("query").get().sql());
+ assertTrue(replacedView.sqlFor("trino").isPresent());
+ assertEquals(replacedTrinoQuery, replacedView.sqlFor("trino").get().sql());
+ assertTrue(replacedView.sqlFor("spark").isEmpty());
+ assertEquals("replaced_catalog", replacedView.defaultCatalog());
+ assertEquals("replaced_schema", replacedView.defaultSchema());
+ assertEquals("keep_value", replacedView.properties().get("keep_key"));
+ assertFalse(replacedView.properties().containsKey("remove_key"));
+ assertEquals("add_value", replacedView.properties().get("add_key"));
+ }
+
+ @Test
+ public void testReplaceViewCreateFailureAfterDropLeavesViewDropped() throws
Exception {
+ NameIdentifier replaceIdentifier =
+ NameIdentifier.of(Namespace.of(DATABASE), VIEW +
"_replace_create_failure");
+ createView(replaceIdentifier);
+
+ NameIdentifier paimonReplaceIdentifier =
buildPaimonNameIdentifier(replaceIdentifier);
+ paimonCatalogOps.failCreateViewWithDatabaseNotExist(
+ paimonReplaceIdentifier.toString(), DATABASE);
+
+ try {
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ paimonViewCatalogOps.alterView(
+ replaceIdentifier,
+ ViewChange.replaceView(
+ new Column[] {Column.of("col_2",
Types.IntegerType.get(), "col_2")},
+ new Representation[] {
+ SQLRepresentation.builder()
+ .withDialect("query")
+ .withSql("SELECT col_2 FROM source_table")
+ .build()
+ },
+ "paimon",
+ DATABASE,
+ "replaced_view_comment")));
+
+ assertTrue(exception.getMessage().contains(DATABASE));
+ } finally {
+ paimonCatalogOps.clearFailCreateView();
+ }
+
+ assertThrows(NoSuchViewException.class, () ->
paimonViewCatalogOps.loadView(replaceIdentifier));
+ }
+
+ @Test
+ public void testListAndDropViewOperations() throws Exception {
+ NameIdentifier firstIdentifier = NameIdentifier.of(Namespace.of(DATABASE),
VIEW + "_list_1");
+ NameIdentifier secondIdentifier =
NameIdentifier.of(Namespace.of(DATABASE), VIEW + "_list_2");
+ createView(firstIdentifier);
+ createView(secondIdentifier);
+
+ NameIdentifier[] listedViews =
paimonViewCatalogOps.listViews(Namespace.of(DATABASE));
+ assertEquals(2, listedViews.length);
+ assertTrue(Arrays.asList(listedViews).contains(firstIdentifier));
+ assertTrue(Arrays.asList(listedViews).contains(secondIdentifier));
+
+ assertTrue(paimonViewCatalogOps.dropView(firstIdentifier));
+ assertFalse(paimonViewCatalogOps.dropView(firstIdentifier));
+ assertThrows(NoSuchViewException.class, () ->
paimonViewCatalogOps.loadView(firstIdentifier));
+
+ NameIdentifier[] remainingViews =
paimonViewCatalogOps.listViews(Namespace.of(DATABASE));
+ assertEquals(1, remainingViews.length);
+ assertEquals(secondIdentifier, remainingViews[0]);
+ }
+
+ @Test
+ public void testRenameAndReplaceCannotBeCombined() throws Exception {
+ createView(VIEW_IDENTIFIER);
+
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ paimonViewCatalogOps.alterView(
+ VIEW_IDENTIFIER,
+ ViewChange.rename(VIEW + "_renamed"),
+ ViewChange.replaceView(
+ new Column[] {Column.of("col_2",
Types.IntegerType.get(), "col_2")},
+ new Representation[] {
+ SQLRepresentation.builder()
+ .withDialect("query")
+ .withSql("SELECT col_2 FROM source_table")
+ .build()
+ },
+ "paimon",
+ DATABASE,
+ "replaced_view_comment")));
+
+ assertTrue(exception.getMessage().contains("cannot be performed
together"));
+ assertEquals(VIEW, paimonViewCatalogOps.loadView(VIEW_IDENTIFIER).name());
+ }
+
+ @Test
+ public void testMultipleRenameChangesAreRejected() throws Exception {
+ createView(VIEW_IDENTIFIER);
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ paimonViewCatalogOps.alterView(
+ VIEW_IDENTIFIER, ViewChange.rename("rename_1"),
ViewChange.rename("rename_2")));
+
+ assertEquals(VIEW, paimonViewCatalogOps.loadView(VIEW_IDENTIFIER).name());
+ }
+
+ @Test
+ public void testRenameViewFailsWithTypedExceptionWhenTargetExists() throws
Exception {
+ NameIdentifier sourceIdentifier =
NameIdentifier.of(Namespace.of(DATABASE), VIEW + "_source");
+ NameIdentifier targetIdentifier =
NameIdentifier.of(Namespace.of(DATABASE), VIEW + "_target");
+ createView(sourceIdentifier);
+ createView(targetIdentifier);
+
+ ViewAlreadyExistsException exception =
+ assertThrows(
+ ViewAlreadyExistsException.class,
+ () ->
+ paimonViewCatalogOps.alterView(
+ sourceIdentifier,
ViewChange.rename(targetIdentifier.name())));
+ assertTrue(exception.getMessage().contains(targetIdentifier.name()));
+ assertEquals(sourceIdentifier.name(),
paimonViewCatalogOps.loadView(sourceIdentifier).name());
+ assertEquals(targetIdentifier.name(),
paimonViewCatalogOps.loadView(targetIdentifier).name());
+ }
+
+ @Test
+ public void testRenameViewReportsLoadFailureAfterSuccessfulRename() throws
Exception {
+ NameIdentifier sourceIdentifier =
+ NameIdentifier.of(Namespace.of(DATABASE), VIEW +
"_rename_load_source");
+ createView(sourceIdentifier);
+ NameIdentifier renamedIdentifier =
+ NameIdentifier.of(sourceIdentifier.namespace(), VIEW +
"_rename_load_target");
+ NameIdentifier paimonRenamedIdentifier =
buildPaimonNameIdentifier(renamedIdentifier);
+ paimonCatalogOps.failLoadView(paimonRenamedIdentifier.toString());
+
+ try {
+ IllegalStateException exception =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ paimonViewCatalogOps.alterView(
+ sourceIdentifier,
ViewChange.rename(renamedIdentifier.name())));
+ assertTrue(exception.getMessage().contains(sourceIdentifier.toString()));
+
assertTrue(exception.getMessage().contains(renamedIdentifier.toString()));
+ } finally {
+ paimonCatalogOps.clearFailLoadView();
+ }
+
+ assertThrows(NoSuchViewException.class, () ->
paimonViewCatalogOps.loadView(sourceIdentifier));
+ assertEquals(renamedIdentifier.name(),
paimonViewCatalogOps.loadView(renamedIdentifier).name());
+ }
+
+ @Test
+ public void testCreateViewRequiresQueryRepresentation() {
+ String query = "SELECT col_1 FROM source_table";
+ Representation[] representations =
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("spark").withSql(query).build()
+ };
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ paimonViewCatalogOps.createView(
+ NameIdentifier.of(Namespace.of(DATABASE),
"missing_query_dialect"),
+ "test_view_comment",
+ new Column[] {Column.of("col_1", Types.IntegerType.get(),
"col_1")},
+ representations,
+ "paimon",
+ DATABASE,
+ Maps.newHashMap()));
+ }
+
+ @Test
+ public void testCreateViewAcceptsCaseInsensitiveQueryRepresentation() throws
Exception {
+ String query = "SELECT col_1 FROM source_table";
+ NameIdentifier identifier = NameIdentifier.of(Namespace.of(DATABASE),
"upper_case_query");
+ Representation[] representations =
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("QUERY").withSql(query).build(),
+
SQLRepresentation.builder().withDialect("Spark").withSql(query).build()
+ };
+
+ paimonViewCatalogOps.createView(
+ identifier,
+ "test_view_comment",
+ new Column[] {Column.of("col_1", Types.IntegerType.get(), "col_1")},
+ representations,
+ "paimon",
+ DATABASE,
+ Maps.newHashMap());
+
+ View loadedView = paimonViewCatalogOps.loadView(identifier);
+ boolean hasQueryRepresentation =
+ Arrays.stream(loadedView.representations())
+ .anyMatch(
+ representation ->
+ representation instanceof SQLRepresentation
+ && "query".equals(((SQLRepresentation)
representation).dialect())
+ && query.equals(((SQLRepresentation)
representation).sql()));
+ assertTrue(hasQueryRepresentation);
+
+ boolean hasNormalizedSparkRepresentation =
+ Arrays.stream(loadedView.representations())
+ .anyMatch(
+ representation ->
+ representation instanceof SQLRepresentation
+ && "spark".equals(((SQLRepresentation)
representation).dialect())
+ && query.equals(((SQLRepresentation)
representation).sql()));
+ assertTrue(hasNormalizedSparkRepresentation);
+ }
+
+ @Test
+ public void testCreateViewRejectsCaseInsensitiveDuplicateDialects() {
+ String query = "SELECT col_1 FROM source_table";
+ Representation[] representations =
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("query").withSql(query).build(),
+
SQLRepresentation.builder().withDialect("Spark").withSql(query).build(),
+ SQLRepresentation.builder()
+ .withDialect("spark")
+ .withSql(query + " WHERE col_1 > 1")
+ .build()
+ };
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ paimonViewCatalogOps.createView(
+ NameIdentifier.of(Namespace.of(DATABASE),
"duplicate_dialect_case"),
+ "test_view_comment",
+ new Column[] {Column.of("col_1", Types.IntegerType.get(),
"col_1")},
+ representations,
+ "paimon",
+ DATABASE,
+ Maps.newHashMap()));
+ }
+
+ @Test
+ public void
testCreateViewDatabaseNotExistUsesSchemaIdentifierInErrorMessage() {
+ String query = "SELECT col_1 FROM source_table";
+ NameIdentifier identifier =
NameIdentifier.of(Namespace.of("missing_schema"), "missing_view");
+ Representation[] representations =
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("query").withSql(query).build()
+ };
+
+ PaimonViewCatalogOps bypassedSchemaCheckOps =
+ new PaimonViewCatalogOps(
+ paimonCatalogOps, this::buildPaimonNameIdentifier,
schemaIdentifier -> true);
+
+ NoSuchSchemaException exception =
+ assertThrows(
+ NoSuchSchemaException.class,
+ () ->
+ bypassedSchemaCheckOps.createView(
+ identifier,
+ "test_view_comment",
+ new Column[] {Column.of("col_1", Types.IntegerType.get(),
"col_1")},
+ representations,
+ "paimon",
+ "missing_schema",
+ Maps.newHashMap()));
+
+ assertTrue(exception.getMessage().contains("missing_schema"));
+ assertFalse(exception.getMessage().contains("missing_view"));
+ }
+
+ private void createView(NameIdentifier identifier) throws Exception {
+ String query = "SELECT col_1 FROM source_table";
+ Representation[] representations =
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("query").withSql(query).build(),
+
SQLRepresentation.builder().withDialect("spark").withSql(query).build()
+ };
+
+ paimonViewCatalogOps.createView(
+ identifier,
+ "test_view_comment",
+ new Column[] {Column.of("col_1", Types.IntegerType.get(), "col_1")},
+ representations,
+ "paimon",
+ DATABASE,
+ Maps.newHashMap());
+ }
+
+ private boolean schemaExists(NameIdentifier identifier) {
+ try {
+ paimonCatalogOps.loadDatabase(identifier.name());
+ return true;
+ } catch (Catalog.DatabaseNotExistException e) {
+ return false;
+ }
+ }
+
+ private NameIdentifier buildPaimonNameIdentifier(NameIdentifier identifier) {
+ String[] levels = identifier.namespace().levels();
+ return NameIdentifier.of(levels[levels.length - 1], identifier.name());
+ }
+
+ private static class TestablePaimonCatalogOps extends PaimonCatalogOps {
+
+ private String failLoadViewName;
+ private String failCreateViewName;
+ private String failCreateViewDatabaseName;
+
+ TestablePaimonCatalogOps(PaimonConfig paimonConfig) {
+ super(paimonConfig);
+ }
+
+ Catalog catalog() {
+ return catalog;
+ }
+
+ void setCatalog(Catalog catalog) {
+ this.catalog = catalog;
+ }
+
+ void failLoadView(String viewName) {
+ this.failLoadViewName = viewName;
+ }
+
+ void clearFailLoadView() {
+ this.failLoadViewName = null;
+ }
+
+ void failCreateViewWithDatabaseNotExist(String viewName, String
databaseName) {
+ this.failCreateViewName = viewName;
+ this.failCreateViewDatabaseName = databaseName;
+ }
+
+ void clearFailCreateView() {
+ this.failCreateViewName = null;
+ this.failCreateViewDatabaseName = null;
+ }
+
+ @Override
+ public void createView(String viewName, org.apache.paimon.view.View view)
+ throws Catalog.ViewAlreadyExistException,
Catalog.DatabaseNotExistException {
+ if (failCreateViewName != null && failCreateViewName.equals(viewName)) {
+ throw new
Catalog.DatabaseNotExistException(failCreateViewDatabaseName);
+ }
+
+ super.createView(viewName, view);
+ }
+
+ @Override
+ public org.apache.paimon.view.View loadView(String viewName)
+ throws Catalog.ViewNotExistException {
+ if (failLoadViewName != null && failLoadViewName.equals(viewName)) {
+ throw new
Catalog.ViewNotExistException(Identifier.fromString(viewName));
+ }
+ return super.loadView(viewName);
+ }
+ }
+}
diff --git
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonBaseIT.java
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonBaseIT.java
index 677d8ca790..1e1a7a20d7 100644
---
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonBaseIT.java
+++
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonBaseIT.java
@@ -56,9 +56,12 @@ import org.apache.gravitino.integration.test.util.BaseIT;
import org.apache.gravitino.integration.test.util.GravitinoITUtils;
import org.apache.gravitino.integration.test.util.TestDatabaseName;
import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.SQLRepresentation;
import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.rel.ViewCatalog;
import org.apache.gravitino.rel.expressions.NamedReference;
import org.apache.gravitino.rel.expressions.distributions.Distribution;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -858,6 +861,55 @@ public abstract class CatalogPaimonBaseIT extends BaseIT {
}
}
+ @Test
+ void testSparkCreateViewAndLoadByGravitino() {
+ String viewName = GravitinoITUtils.genRandomName("spark_create_view");
+ String viewIdentifier = String.join(".", schemaName, viewName);
+
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () -> {
+ spark.sql(
+ String.format(
+ "CREATE VIEW paimon.%s AS SELECT 1 AS id, 'name_1' AS name",
viewIdentifier));
+ viewCatalog.loadView(NameIdentifier.of(schemaName, viewName));
+ });
+ }
+
+ @Test
+ void testGravitinoCreateViewAndReadBySpark() {
+ String viewName = GravitinoITUtils.genRandomName("gravitino_create_view");
+ NameIdentifier viewIdentifier = NameIdentifier.of(schemaName, viewName);
+
+ String query = "SELECT 1 AS id, 'name_1' AS name";
+ Representation[] representations =
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("query").withSql(query).build(),
+
SQLRepresentation.builder().withDialect("spark").withSql(query).build()
+ };
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () -> {
+ viewCatalog.createView(
+ viewIdentifier,
+ "view_for_spark_read",
+ new Column[] {
+ Column.of("id", Types.IntegerType.get(), "id column"),
+ Column.of("name", Types.StringType.get(), "name column")
+ },
+ representations,
+ "paimon",
+ schemaName,
+ Collections.emptyMap());
+ spark
+ .sql(String.format("SELECT * FROM paimon.%s.%s ORDER BY id",
schemaName, viewName))
+ .collectAsList();
+ });
+ }
+
@Test
void testTimeTypePrecision() throws
org.apache.paimon.catalog.Catalog.TableNotExistException {
String tableName = GravitinoITUtils.genRandomName("test_time_precision");
@@ -1040,6 +1092,29 @@ public abstract class CatalogPaimonBaseIT extends BaseIT
{
return values;
}
+ protected NameIdentifier createSimplePaimonTableForViewInterop(String
tablePrefix) {
+ Preconditions.checkNotNull(
+ spark, "Spark session is required for Spark view interoperability
tests, but it is null.");
+
+ String tableName = GravitinoITUtils.genRandomName(tablePrefix);
+ NameIdentifier tableIdentifier = NameIdentifier.of(schemaName, tableName);
+ catalog
+ .asTableCatalog()
+ .createTable(
+ tableIdentifier,
+ new Column[] {
+ Column.of("id", Types.IntegerType.get(), "id column"),
+ Column.of("name", Types.StringType.get(), "name column")
+ },
+ table_comment,
+ createProperties());
+
+ spark.sql(
+ String.format(
+ "INSERT INTO paimon.%s.%s VALUES (1, 'name_1'), (2, 'name_2')",
schemaName, tableName));
+ return tableIdentifier;
+ }
+
private void clearTableAndSchema() {
SupportsSchemas supportsSchema = catalog.asSchemas();
Arrays.stream(supportsSchema.listSchemas())
diff --git
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonHiveIT.java
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonHiveIT.java
index fdebf6839f..f943d971dc 100644
---
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonHiveIT.java
+++
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonHiveIT.java
@@ -19,6 +19,8 @@
package org.apache.gravitino.catalog.lakehouse.paimon.integration.test;
import com.google.common.collect.Maps;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Schema;
@@ -26,7 +28,16 @@ import org.apache.gravitino.SupportsSchemas;
import
org.apache.gravitino.catalog.lakehouse.paimon.PaimonCatalogPropertiesMetadata;
import org.apache.gravitino.integration.test.container.HiveContainer;
import org.apache.gravitino.integration.test.util.GravitinoITUtils;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Representation;
+import org.apache.gravitino.rel.SQLRepresentation;
+import org.apache.gravitino.rel.View;
+import org.apache.gravitino.rel.ViewCatalog;
+import org.apache.gravitino.rel.types.Types;
import org.apache.paimon.catalog.Catalog;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
@@ -61,6 +72,86 @@ public class CatalogPaimonHiveIT extends CatalogPaimonBaseIT
{
return catalogProperties;
}
+ @Override
+ protected void initSparkEnv() {
+ spark =
+ SparkSession.builder()
+ .master("local[1]")
+ .appName("Paimon Catalog integration test")
+ .config("spark.sql.catalog.paimon",
"org.apache.paimon.spark.SparkCatalog")
+ .config("spark.sql.catalog.paimon.metastore", "hive")
+ .config("spark.sql.catalog.paimon.uri", URI)
+ .config("spark.sql.catalog.paimon.warehouse", WAREHOUSE)
+ .config("spark.sql.catalog.paimon.cache-enabled", "false")
+ .config(
+ "spark.sql.extensions",
+
"org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
+ .enableHiveSupport()
+ .getOrCreate();
+ }
+
+ @Test
+ @Override
+ void testSparkCreateViewAndLoadByGravitino() {
+ NameIdentifier baseTableIdentifier =
+
createSimplePaimonTableForViewInterop("spark_create_view_source_table");
+ String viewName = GravitinoITUtils.genRandomName("spark_create_view");
+ String viewIdentifier = String.join(".", schemaName, viewName);
+ String tableIdentifier = String.join(".", schemaName,
baseTableIdentifier.name());
+
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+ spark.sql(
+ String.format(
+ "CREATE VIEW paimon.%s AS SELECT id, name FROM paimon.%s",
+ viewIdentifier, tableIdentifier));
+
+ View loadedView = viewCatalog.loadView(NameIdentifier.of(schemaName,
viewName));
+ Assertions.assertEquals(viewName, loadedView.name());
+ Assertions.assertEquals(2, loadedView.columns().length);
+ Assertions.assertEquals("id", loadedView.columns()[0].name());
+ Assertions.assertEquals("name", loadedView.columns()[1].name());
+ Assertions.assertTrue(loadedView.representations().length > 0);
+ }
+
+ @Test
+ @Override
+ void testGravitinoCreateViewAndReadBySpark() {
+ NameIdentifier baseTableIdentifier =
+
createSimplePaimonTableForViewInterop("gravitino_create_view_source_table");
+ String viewName = GravitinoITUtils.genRandomName("gravitino_create_view");
+ NameIdentifier viewIdentifier = NameIdentifier.of(schemaName, viewName);
+
+ String query =
+ String.format("SELECT id, name FROM paimon.%s.%s", schemaName,
baseTableIdentifier.name());
+ Representation[] representations =
+ new Representation[] {
+
SQLRepresentation.builder().withDialect("query").withSql(query).build(),
+
SQLRepresentation.builder().withDialect("spark").withSql(query).build()
+ };
+ ViewCatalog viewCatalog = catalog.asViewCatalog();
+
+ viewCatalog.createView(
+ viewIdentifier,
+ "view_for_spark_read",
+ new Column[] {
+ Column.of("id", Types.IntegerType.get(), "id column"),
+ Column.of("name", Types.StringType.get(), "name column")
+ },
+ representations,
+ "paimon",
+ schemaName,
+ Collections.emptyMap());
+
+ Dataset<Row> rows =
+ spark.sql(String.format("SELECT * FROM paimon.%s.%s ORDER BY id",
schemaName, viewName));
+ List<Row> results = rows.collectAsList();
+ Assertions.assertEquals(2, results.size());
+ Assertions.assertEquals(1, results.get(0).getInt(0));
+ Assertions.assertEquals("name_1", results.get(0).getString(1));
+ Assertions.assertEquals(2, results.get(1).getInt(0));
+ Assertions.assertEquals("name_2", results.get(1).getString(1));
+ }
+
@Test
void testPaimonSchemaProperties() throws Catalog.DatabaseNotExistException {
SupportsSchemas schemas = catalog.asSchemas();
diff --git
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/TestPaimonCatalogOps.java
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/TestPaimonCatalogOps.java
index ffdf2d5980..26095464c0 100644
---
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/TestPaimonCatalogOps.java
+++
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/ops/TestPaimonCatalogOps.java
@@ -38,6 +38,7 @@ import static
org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -52,6 +53,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import
org.apache.gravitino.catalog.lakehouse.paimon.PaimonCatalogPropertiesMetadata;
import org.apache.gravitino.catalog.lakehouse.paimon.PaimonConfig;
+import
org.apache.gravitino.catalog.lakehouse.paimon.utils.PaimonViewTestCatalogHelper;
import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.TableChange.ColumnPosition;
import org.apache.gravitino.rel.TableChange.UpdateColumnComment;
@@ -63,6 +65,7 @@ import
org.apache.paimon.catalog.Catalog.ColumnAlreadyExistException;
import org.apache.paimon.catalog.Catalog.ColumnNotExistException;
import org.apache.paimon.catalog.Catalog.DatabaseNotExistException;
import org.apache.paimon.catalog.Catalog.TableAlreadyExistException;
+import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaChange.AddColumn;
@@ -76,6 +79,8 @@ import org.apache.paimon.types.MapType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.TimestampType;
import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.view.ViewChange;
+import org.apache.paimon.view.ViewImpl;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -85,19 +90,22 @@ import org.junit.jupiter.api.io.TempDir;
/** Tests for {@link
org.apache.gravitino.catalog.lakehouse.paimon.ops.PaimonCatalogOps}. */
public class TestPaimonCatalogOps {
- private PaimonCatalogOps paimonCatalogOps;
+ private TestablePaimonCatalogOps paimonCatalogOps;
@TempDir private File warehouse;
private static final String DATABASE = "test_table_ops_database";
private static final String TABLE = "test_table_ops_table";
+ private static final String VIEW = "test_table_ops_view";
private static final String COMMENT = "table_ops_table_comment";
private static final NameIdentifier IDENTIFIER =
NameIdentifier.of(Namespace.of(DATABASE), TABLE);
+ private static final NameIdentifier VIEW_IDENTIFIER =
+ NameIdentifier.of(Namespace.of(DATABASE), VIEW);
private static final Map<String, String> OPTIONS =
ImmutableMap.of(BUCKET.key(), "10");
@BeforeEach
public void setUp() throws Exception {
paimonCatalogOps =
- new PaimonCatalogOps(
+ new TestablePaimonCatalogOps(
new PaimonConfig(
ImmutableMap.of(PaimonCatalogPropertiesMetadata.WAREHOUSE,
warehouse.getPath())));
createDatabase();
@@ -167,6 +175,82 @@ public class TestPaimonCatalogOps {
Assertions.assertNotNull(paimonCatalogOps.loadTable(IDENTIFIER.toString()));
}
+ @Test
+ void testCreateLoadAlterRenameAndDropViewOperations() throws Exception {
+ paimonCatalogOps.setCatalog(
+
PaimonViewTestCatalogHelper.createViewSupportedCatalog(paimonCatalogOps.catalog()));
+
+ org.apache.paimon.view.View createdView =
+ new ViewImpl(
+ Identifier.create(DATABASE, VIEW),
+ List.of(new DataField(0, "col_1", DataTypes.INT(), "col_1")),
+ "SELECT col_1 FROM " + TABLE,
+ ImmutableMap.of("spark", "SELECT col_1 FROM " + TABLE),
+ "test_view_comment",
+ Maps.newHashMap());
+ paimonCatalogOps.createView(VIEW_IDENTIFIER.toString(), createdView);
+
+ org.apache.paimon.view.View loadedView =
paimonCatalogOps.loadView(VIEW_IDENTIFIER.toString());
+ assertEquals("test_view_comment", loadedView.comment().orElse(null));
+ assertEquals("SELECT col_1 FROM " + TABLE, loadedView.query());
+ assertTrue(paimonCatalogOps.listViews(DATABASE).contains(VIEW));
+
+ paimonCatalogOps.alterView(
+ VIEW_IDENTIFIER.toString(),
+ List.of(
+ ViewChange.setOption("k1", "v1"),
+ ViewChange.updateComment("updated_view_comment"),
+ ViewChange.updateDialect("spark", "SELECT col_1 FROM " + TABLE + "
WHERE col_1 > 1")));
+
+ org.apache.paimon.view.View alteredView =
paimonCatalogOps.loadView(VIEW_IDENTIFIER.toString());
+ assertEquals("updated_view_comment", alteredView.comment().orElse(null));
+ assertEquals("v1", alteredView.options().get("k1"));
+ assertEquals("SELECT col_1 FROM " + TABLE, alteredView.query());
+ assertEquals(
+ "SELECT col_1 FROM " + TABLE + " WHERE col_1 > 1",
alteredView.dialects().get("spark"));
+
+ NameIdentifier renamedViewIdentifier =
+ NameIdentifier.of(VIEW_IDENTIFIER.namespace(), VIEW + "_renamed");
+ paimonCatalogOps.renameView(VIEW_IDENTIFIER.toString(),
renamedViewIdentifier.toString());
+
+ assertThrowsExactly(
+ Catalog.ViewNotExistException.class,
+ () -> paimonCatalogOps.loadView(VIEW_IDENTIFIER.toString()));
+ assertNotNull(paimonCatalogOps.loadView(renamedViewIdentifier.toString()));
+
+ paimonCatalogOps.dropView(renamedViewIdentifier.toString());
+ assertThrowsExactly(
+ Catalog.ViewNotExistException.class,
+ () -> paimonCatalogOps.loadView(renamedViewIdentifier.toString()));
+ }
+
+ @Test
+ void testAlterViewExceptionMessageContainsContext() throws Exception {
+ paimonCatalogOps.setCatalog(
+
PaimonViewTestCatalogHelper.createViewSupportedCatalog(paimonCatalogOps.catalog()));
+
+ org.apache.paimon.view.View createdView =
+ new ViewImpl(
+ Identifier.create(DATABASE, VIEW),
+ List.of(new DataField(0, "col_1", DataTypes.INT(), "col_1")),
+ "SELECT col_1 FROM " + TABLE,
+ ImmutableMap.of("spark", "SELECT col_1 FROM " + TABLE),
+ "test_view_comment",
+ Maps.newHashMap());
+ paimonCatalogOps.createView(VIEW_IDENTIFIER.toString(), createdView);
+
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ paimonCatalogOps.alterView(
+ VIEW_IDENTIFIER.toString(),
+ List.of(ViewChange.updateDialect("trino", "SELECT col_1
FROM " + TABLE))));
+
+ assertTrue(exception.getMessage().contains("Cannot alter view " +
VIEW_IDENTIFIER + ": "));
+ assertTrue(exception.getCause() instanceof
Catalog.DialectNotExistException);
+ }
+
@Test
void testAddColumn() throws Exception {
// Test AddColumn after column.
@@ -433,4 +517,19 @@ public class TestPaimonCatalogOps {
paimonCatalogOps.dropDatabase(DATABASE, true);
Assertions.assertTrue(paimonCatalogOps.listDatabases().isEmpty());
}
+
+ private static class TestablePaimonCatalogOps extends PaimonCatalogOps {
+
+ TestablePaimonCatalogOps(PaimonConfig paimonConfig) {
+ super(paimonConfig);
+ }
+
+ Catalog catalog() {
+ return catalog;
+ }
+
+ void setCatalog(Catalog catalog) {
+ this.catalog = catalog;
+ }
+ }
}
diff --git
a/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/PaimonViewTestCatalogHelper.java
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/PaimonViewTestCatalogHelper.java
new file mode 100644
index 0000000000..0dfce90d48
--- /dev/null
+++
b/catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/PaimonViewTestCatalogHelper.java
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.lakehouse.paimon.utils;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.catalog.DelegateCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.view.View;
+import org.apache.paimon.view.ViewChange;
+import org.apache.paimon.view.ViewImpl;
+
+/** Utilities for creating in-memory view-capable catalog wrappers in Paimon
unit tests. */
+public final class PaimonViewTestCatalogHelper {
+
+ private PaimonViewTestCatalogHelper() {}
+
+ public static Catalog createViewSupportedCatalog(Catalog wrappedCatalog) {
+ Map<Identifier, View> viewStore = new HashMap<>();
+ return new DelegateCatalog(wrappedCatalog) {
+ @Override
+ public CatalogLoader catalogLoader() {
+ return wrapped().catalogLoader();
+ }
+
+ @Override
+ public List<String> listViews(String databaseName) {
+ return viewStore.keySet().stream()
+ .filter(identifier ->
identifier.getDatabaseName().equals(databaseName))
+ .map(Identifier::getObjectName)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public View getView(Identifier identifier) throws
Catalog.ViewNotExistException {
+ View storedView = viewStore.get(identifier);
+ if (storedView == null) {
+ throw new Catalog.ViewNotExistException(identifier);
+ }
+ return storedView;
+ }
+
+ @Override
+ public void createView(Identifier identifier, View view, boolean
ignoreIfExists)
+ throws Catalog.ViewAlreadyExistException,
Catalog.DatabaseNotExistException {
+ if (viewStore.containsKey(identifier)) {
+ if (!ignoreIfExists) {
+ throw new Catalog.ViewAlreadyExistException(identifier);
+ }
+ return;
+ }
+
+ if (!listDatabases().contains(identifier.getDatabaseName())) {
+ throw new
Catalog.DatabaseNotExistException(identifier.getDatabaseName());
+ }
+
+ viewStore.put(identifier, copyView(identifier, view));
+ }
+
+ @Override
+ public void alterView(
+ Identifier identifier, List<ViewChange> changes, boolean
ignoreIfNotExists)
+ throws Catalog.ViewNotExistException,
Catalog.DialectAlreadyExistException,
+ Catalog.DialectNotExistException {
+ View storedView = viewStore.get(identifier);
+ if (storedView == null) {
+ if (!ignoreIfNotExists) {
+ throw new Catalog.ViewNotExistException(identifier);
+ }
+ return;
+ }
+
+ Map<String, String> updatedOptions = new
HashMap<>(storedView.options());
+ Map<String, String> updatedDialects = new
HashMap<>(storedView.dialects());
+ String updatedComment = storedView.comment().orElse(null);
+
+ for (ViewChange change : changes) {
+ if (change instanceof ViewChange.SetViewOption) {
+ ViewChange.SetViewOption setViewOption =
(ViewChange.SetViewOption) change;
+ updatedOptions.put(setViewOption.key(), setViewOption.value());
+ } else if (change instanceof ViewChange.RemoveViewOption) {
+ ViewChange.RemoveViewOption removeViewOption =
(ViewChange.RemoveViewOption) change;
+ updatedOptions.remove(removeViewOption.key());
+ } else if (change instanceof ViewChange.UpdateViewComment) {
+ ViewChange.UpdateViewComment updateViewComment =
(ViewChange.UpdateViewComment) change;
+ updatedComment = updateViewComment.comment();
+ } else if (change instanceof ViewChange.AddDialect) {
+ ViewChange.AddDialect addDialect = (ViewChange.AddDialect) change;
+ if (updatedDialects.containsKey(addDialect.dialect())) {
+ throw new Catalog.DialectAlreadyExistException(identifier,
addDialect.dialect());
+ }
+ updatedDialects.put(addDialect.dialect(), addDialect.query());
+ } else if (change instanceof ViewChange.UpdateDialect) {
+ ViewChange.UpdateDialect updateDialect =
(ViewChange.UpdateDialect) change;
+ if (!updatedDialects.containsKey(updateDialect.dialect())) {
+ throw new Catalog.DialectNotExistException(identifier,
updateDialect.dialect());
+ }
+ updatedDialects.put(updateDialect.dialect(),
updateDialect.query());
+ } else if (change instanceof ViewChange.DropDialect) {
+ ViewChange.DropDialect dropDialect = (ViewChange.DropDialect)
change;
+ if (!updatedDialects.containsKey(dropDialect.dialect())) {
+ throw new Catalog.DialectNotExistException(identifier,
dropDialect.dialect());
+ }
+ updatedDialects.remove(dropDialect.dialect());
+ }
+ }
+
+ viewStore.put(
+ identifier,
+ new ViewImpl(
+ identifier,
+ storedView.rowType().getFields(),
+ storedView.query(),
+ updatedDialects,
+ updatedComment,
+ updatedOptions));
+ }
+
+ @Override
+ public void renameView(
+ Identifier fromIdentifier, Identifier toIdentifier, boolean
ignoreIfNotExists)
+ throws Catalog.ViewNotExistException,
Catalog.ViewAlreadyExistException {
+ View storedView = viewStore.remove(fromIdentifier);
+ if (storedView == null) {
+ if (!ignoreIfNotExists) {
+ throw new Catalog.ViewNotExistException(fromIdentifier);
+ }
+ return;
+ }
+
+ if (viewStore.containsKey(toIdentifier)) {
+ viewStore.put(fromIdentifier, storedView);
+ throw new Catalog.ViewAlreadyExistException(toIdentifier);
+ }
+
+ viewStore.put(toIdentifier, copyView(toIdentifier, storedView));
+ }
+
+ @Override
+ public void dropView(Identifier identifier, boolean ignoreIfNotExists)
+ throws Catalog.ViewNotExistException {
+ if (viewStore.remove(identifier) == null && !ignoreIfNotExists) {
+ throw new Catalog.ViewNotExistException(identifier);
+ }
+ }
+ };
+ }
+
+ private static View copyView(Identifier identifier, View view) {
+ return new ViewImpl(
+ identifier,
+ view.rowType().getFields(),
+ view.query(),
+ new HashMap<>(view.dialects()),
+ view.comment().orElse(null),
+ new HashMap<>(view.options()));
+ }
+}
