mchades commented on code in PR #10992: URL: https://github.com/apache/gravitino/pull/10992#discussion_r3217130056
########## catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergViewCatalogOperations.java: ########## @@ -0,0 +1,390 @@ +/* + * 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.iceberg; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Maps; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.catalog.lakehouse.iceberg.converter.ConvertUtil; +import org.apache.gravitino.catalog.lakehouse.iceberg.ops.IcebergCatalogWrapperHelper; +import org.apache.gravitino.exceptions.NoSuchSchemaException; +import org.apache.gravitino.exceptions.NoSuchViewException; +import org.apache.gravitino.exceptions.ViewAlreadyExistsException; +import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; +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.gravitino.rel.ViewChange; +import org.apache.gravitino.utils.PrincipalUtils; +import org.apache.iceberg.MetadataUpdate; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.rest.requests.ImmutableCreateViewRequest; +import org.apache.iceberg.rest.requests.RenameTableRequest; +import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.ListTablesResponse; +import org.apache.iceberg.rest.responses.LoadViewResponse; +import org.apache.iceberg.view.ImmutableSQLViewRepresentation; +import org.apache.iceberg.view.ImmutableViewVersion; +import org.apache.iceberg.view.SQLViewRepresentation; +import org.apache.iceberg.view.ViewMetadata; +import org.apache.iceberg.view.ViewRepresentation; +import org.apache.iceberg.view.ViewVersion; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** View catalog operations for Iceberg. */ +class IcebergViewCatalogOperations { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergViewCatalogOperations.class); + + private final IcebergCatalogWrapper icebergCatalogWrapper; + + IcebergViewCatalogOperations(IcebergCatalogWrapper icebergCatalogWrapper) { + this.icebergCatalogWrapper = + Preconditions.checkNotNull(icebergCatalogWrapper, "icebergCatalogWrapper must not be null"); + } + + public View loadView(NameIdentifier ident) throws NoSuchViewException { + try { + LoadViewResponse response = + icebergCatalogWrapper.loadView( + IcebergCatalogWrapperHelper.buildIcebergTableIdentifier(ident)); + return IcebergView.fromLoadViewResponse(response, ident.name()); + } catch (Exception e) { + throw new NoSuchViewException( + e, "Failed to load view %s from Iceberg catalog: %s", ident, e.getMessage()); + } Review Comment: Fixed in 586ae324c. loadView now only translates Iceberg NoSuchViewException to Gravitino NoSuchViewException, and other runtime exceptions are propagated. Added unit coverage in TestIcebergViewCatalogOperations for both paths. ########## catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java: ########## @@ -636,16 +640,79 @@ public void testConnection( */ @Override public View loadView(NameIdentifier ident) throws NoSuchViewException { - try { - LoadViewResponse response = - icebergCatalogWrapper.loadView( - IcebergCatalogWrapperHelper.buildIcebergTableIdentifier(ident)); + return viewCatalogOperations().loadView(ident); + } - return IcebergView.fromLoadViewResponse(response, ident.name()); - } catch (Exception e) { - throw new NoSuchViewException( - e, "Failed to load view %s from Iceberg catalog: %s", ident, e.getMessage()); - } + /** + * Lists all views in the given namespace from the Iceberg catalog. + * + * @param namespace A namespace. + * @return An array of view identifiers in the namespace. + * @throws NoSuchSchemaException If the schema does not exist. + */ + @Override + public NameIdentifier[] listViews(Namespace namespace) throws NoSuchSchemaException { + return viewCatalogOperations().listViews(namespace); + } Review Comment: Addressed in 586ae324c. Added viewExists(NameIdentifier) override in IcebergCatalogOperations and delegated to IcebergViewCatalogOperations.viewExists, which calls wrapper viewExists directly. This avoids default ViewCatalog.viewExists fallback to loadView. ########## catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergView.java: ########## @@ -90,6 +125,41 @@ public static Builder builder() { return new Builder(); } + private static Column[] extractColumns(ViewMetadata metadata) { + try { + Schema schema = metadata.schema(); + if (schema != null && schema.columns() != null) { + return schema.columns().stream().map(ConvertUtil::fromNestedField).toArray(Column[]::new); + } + } catch (Exception e) { + LOG.warn("Failed to extract columns from Iceberg view metadata: {}", e.getMessage()); + } + return new Column[0]; Review Comment: Fixed in 586ae324c. Updated warning log in extractColumns to pass throwable (LOG.warn(..., e)) so stack traces are preserved. ########## catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergView.java: ########## @@ -90,6 +125,41 @@ public static Builder builder() { return new Builder(); } + private static Column[] extractColumns(ViewMetadata metadata) { + try { + Schema schema = metadata.schema(); + if (schema != null && schema.columns() != null) { + return schema.columns().stream().map(ConvertUtil::fromNestedField).toArray(Column[]::new); + } + } catch (Exception e) { + LOG.warn("Failed to extract columns from Iceberg view metadata: {}", e.getMessage()); + } + return new Column[0]; + } + + private static Representation[] extractRepresentations(ViewMetadata metadata) { + try { + ViewVersion currentVersion = metadata.currentVersion(); + if (currentVersion != null && currentVersion.representations() != null) { + return currentVersion.representations().stream() + .filter(r -> r instanceof SQLViewRepresentation) + .map( + r -> { + SQLViewRepresentation sqlRep = (SQLViewRepresentation) r; + return (Representation) + SQLRepresentation.builder() + .withDialect(sqlRep.dialect()) + .withSql(sqlRep.sql()) + .build(); + }) + .toArray(Representation[]::new); + } + } catch (Exception e) { + LOG.warn("Failed to extract representations from Iceberg view metadata: {}", e.getMessage()); + } + return new Representation[0]; Review Comment: Fixed in 586ae324c. Updated warning log in extractRepresentations to pass throwable (LOG.warn(..., e)) so stack traces are preserved. ########## catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/CatalogIcebergBaseIT.java: ########## @@ -1541,6 +1556,431 @@ void testTimeTypePrecisionValidation() { } } + @Test + void testCreateAndLoadView() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + Column[] columns = { + Column.of("id", Types.LongType.get(), "id column"), + Column.of("name", Types.StringType.get(), "name column") + }; + SQLRepresentation sparkRep = + SQLRepresentation.builder() + .withDialect(SPARK_DIALECT) + .withSql("SELECT id, name FROM some_table") + .build(); + + String viewName = GravitinoITUtils.genRandomName("test_view"); + View created = + viewCatalog.createView( + NameIdentifier.of(schemaName, viewName), + VIEW_COMMENT, + columns, + new SQLRepresentation[] {sparkRep}, + null, + null, + Collections.singletonMap("created_by", "test")); + + Assertions.assertEquals(viewName, created.name()); + Assertions.assertEquals(VIEW_COMMENT, created.comment()); + Assertions.assertEquals(1, created.representations().length); + Assertions.assertEquals(Representation.TYPE_SQL, created.representations()[0].type()); + Assertions.assertEquals("test", created.properties().get("created_by")); + + View loaded = viewCatalog.loadView(NameIdentifier.of(schemaName, viewName)); + Assertions.assertEquals(viewName, loaded.name()); + Assertions.assertEquals(VIEW_COMMENT, loaded.comment()); + Assertions.assertEquals(1, loaded.representations().length); + Assertions.assertEquals(Representation.TYPE_SQL, loaded.representations()[0].type()); + Assertions.assertEquals("test", loaded.properties().get("created_by")); + } + + @Test + void testCreateViewWithMultipleRepresentations() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + Column[] columns = {Column.of("id", Types.IntegerType.get(), null)}; + SQLRepresentation sparkRep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + SQLRepresentation trinoRep = + SQLRepresentation.builder().withDialect(TRINO_DIALECT).withSql("SELECT id FROM t").build(); + + String viewName = GravitinoITUtils.genRandomName("multi_rep_view"); + View view = + viewCatalog.createView( + NameIdentifier.of(schemaName, viewName), + null, + columns, + new SQLRepresentation[] {sparkRep, trinoRep}, + null, + null, + Collections.emptyMap()); + + Assertions.assertEquals(2, view.representations().length); + Assertions.assertEquals(Representation.TYPE_SQL, view.representations()[0].type()); + Assertions.assertEquals(Representation.TYPE_SQL, view.representations()[1].type()); + Assertions.assertNull(view.comment()); + } + + @Test + void testCreateViewAndQueryWithSpark() { + TableCatalog tableCatalog = catalog.asTableCatalog(); + ViewCatalog viewCatalog = catalog.asViewCatalog(); + + String sourceTableName = GravitinoITUtils.genRandomName("view_source_tbl"); + NameIdentifier sourceTableIdent = NameIdentifier.of(schemaName, sourceTableName); + Column[] columns = { + Column.of("id", Types.IntegerType.get(), null), + Column.of("name", Types.StringType.get(), null) + }; + tableCatalog.createTable( + sourceTableIdent, + columns, + table_comment, + createProperties(), + Transforms.EMPTY_TRANSFORM, + Distributions.NONE, + new SortOrder[0]); + + TableIdentifier sourceTableIdentifier = TableIdentifier.of(schemaName, sourceTableName); + spark.sql( + String.format( + "INSERT INTO iceberg.%s VALUES (1, 'alice'), (2, 'bob')", sourceTableIdentifier)); + + String viewName = GravitinoITUtils.genRandomName("spark_query_view"); + TableIdentifier viewIdentifier = TableIdentifier.of(schemaName, viewName); + String viewSql = String.format("SELECT id, name FROM iceberg.%s", sourceTableIdentifier); + viewCatalog.createView( + NameIdentifier.of(schemaName, viewName), + VIEW_COMMENT, + columns, + new SQLRepresentation[] { + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql(viewSql).build() + }, + null, + null, + Collections.emptyMap()); + + List<Row> result = + spark + .sql(String.format("SELECT * FROM iceberg.%s ORDER BY id", viewIdentifier)) + .collectAsList(); + Assertions.assertEquals(2, result.size()); + Assertions.assertEquals(1, result.get(0).getInt(0)); + Assertions.assertEquals("alice", result.get(0).getString(1)); + Assertions.assertEquals(2, result.get(1).getInt(0)); + Assertions.assertEquals("bob", result.get(1).getString(1)); + } + + @Test + void testSparkCreateViewLoadWithGravitino() { + TableCatalog tableCatalog = catalog.asTableCatalog(); + ViewCatalog viewCatalog = catalog.asViewCatalog(); + + String sourceTableName = GravitinoITUtils.genRandomName("spark_create_view_src"); + NameIdentifier sourceTableIdent = NameIdentifier.of(schemaName, sourceTableName); + Column[] columns = { + Column.of("id", Types.IntegerType.get(), null), + Column.of("name", Types.StringType.get(), null) + }; + tableCatalog.createTable( + sourceTableIdent, + columns, + table_comment, + createProperties(), + Transforms.EMPTY_TRANSFORM, + Distributions.NONE, + new SortOrder[0]); + + TableIdentifier sourceTableIdentifier = TableIdentifier.of(schemaName, sourceTableName); + spark.sql( + String.format( + "INSERT INTO iceberg.%s VALUES (10, 'spark_created'), (20, 'spark_loaded')", + sourceTableIdentifier)); + + String viewName = GravitinoITUtils.genRandomName("spark_created_view"); + TableIdentifier viewIdentifier = TableIdentifier.of(schemaName, viewName); + spark.sql( + String.format( + "CREATE OR REPLACE VIEW iceberg.%s AS SELECT id, name FROM iceberg.%s", + viewIdentifier, sourceTableIdentifier)); + + View loaded = viewCatalog.loadView(NameIdentifier.of(schemaName, viewName)); + Assertions.assertEquals(viewName, loaded.name()); + Assertions.assertEquals(2, loaded.columns().length); + Assertions.assertEquals("id", loaded.columns()[0].name()); + Assertions.assertEquals("name", loaded.columns()[1].name()); + Assertions.assertTrue(loaded.representations().length > 0); + Assertions.assertEquals(Representation.TYPE_SQL, loaded.representations()[0].type()); + } + + @Test + void testCreateViewWithUnsupportedRepresentation() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + Representation unsupportedRepresentation = () -> "unsupported"; + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + + IllegalArgumentException exception = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + viewCatalog.createView( + NameIdentifier.of(schemaName, GravitinoITUtils.genRandomName("bad_rep_view")), + null, + columns, + new Representation[] {unsupportedRepresentation}, + null, + null, + Collections.emptyMap())); + Assertions.assertNotNull(exception.getMessage()); + } + + @Test + void testAlterViewReplaceWithUnsupportedRepresentation() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String viewName = GravitinoITUtils.genRandomName("replace_bad_rep_view"); + NameIdentifier viewIdent = NameIdentifier.of(schemaName, viewName); + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + SQLRepresentation sparkRep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + viewCatalog.createView( + viewIdent, + null, + columns, + new SQLRepresentation[] {sparkRep}, + null, + null, + Collections.emptyMap()); + + Representation unsupportedRepresentation = () -> "unsupported"; + IllegalArgumentException exception = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + viewCatalog.alterView( + viewIdent, + ViewChange.replaceView( + columns, + new Representation[] {unsupportedRepresentation}, + null, + schemaName, + null))); + Assertions.assertNotNull(exception.getMessage()); + } + + @Test + void testListViews() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String view1 = GravitinoITUtils.genRandomName("list_view1"); + String view2 = GravitinoITUtils.genRandomName("list_view2"); + Column[] columns = {Column.of("c1", Types.StringType.get(), null)}; + SQLRepresentation rep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT c1 FROM t").build(); + + viewCatalog.createView( + NameIdentifier.of(schemaName, view1), + null, + columns, + new SQLRepresentation[] {rep}, + null, + null, + Collections.emptyMap()); + viewCatalog.createView( + NameIdentifier.of(schemaName, view2), + null, + columns, + new SQLRepresentation[] {rep}, + null, + null, + Collections.emptyMap()); + + NameIdentifier[] views = viewCatalog.listViews(Namespace.of(schemaName)); + Assertions.assertTrue(views.length >= 2); + boolean foundView1 = false; + boolean foundView2 = false; + for (NameIdentifier v : views) { + if (v.name().equals(view1)) { + foundView1 = true; + } + if (v.name().equals(view2)) { + foundView2 = true; + } + } + Assertions.assertTrue(foundView1, "view1 not found in list"); + Assertions.assertTrue(foundView2, "view2 not found in list"); + } + + @Test + void testViewExists() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String viewName = GravitinoITUtils.genRandomName("exists_view"); + NameIdentifier ident = NameIdentifier.of(schemaName, viewName); + + Assertions.assertFalse(viewCatalog.viewExists(ident)); + + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + SQLRepresentation rep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + viewCatalog.createView( + ident, null, columns, new SQLRepresentation[] {rep}, null, null, Collections.emptyMap()); + + Assertions.assertTrue(viewCatalog.viewExists(ident)); + } + + @Test + void testDropView() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String viewName = GravitinoITUtils.genRandomName("drop_view"); + NameIdentifier ident = NameIdentifier.of(schemaName, viewName); + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + SQLRepresentation rep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + + viewCatalog.createView( + ident, null, columns, new SQLRepresentation[] {rep}, null, null, Collections.emptyMap()); + Assertions.assertTrue(viewCatalog.viewExists(ident)); + + Assertions.assertTrue(viewCatalog.dropView(ident)); + Assertions.assertFalse(viewCatalog.viewExists(ident)); + Assertions.assertFalse(viewCatalog.dropView(ident)); + } + + @Test + void testAlterViewRename() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String viewName = GravitinoITUtils.genRandomName("rename_view"); + String newName = GravitinoITUtils.genRandomName("renamed_view"); + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + SQLRepresentation rep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + + viewCatalog.createView( + NameIdentifier.of(schemaName, viewName), + null, + columns, + new SQLRepresentation[] {rep}, + null, + null, + Collections.emptyMap()); + + View renamed = + viewCatalog.alterView(NameIdentifier.of(schemaName, viewName), ViewChange.rename(newName)); + + Assertions.assertEquals(newName, renamed.name()); + Assertions.assertFalse(viewCatalog.viewExists(NameIdentifier.of(schemaName, viewName))); + Assertions.assertTrue(viewCatalog.viewExists(NameIdentifier.of(schemaName, newName))); + } + + @Test + void testAlterViewSetAndRemoveProperty() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String viewName = GravitinoITUtils.genRandomName("prop_view"); + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + SQLRepresentation rep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + + viewCatalog.createView( + NameIdentifier.of(schemaName, viewName), + null, + columns, + new SQLRepresentation[] {rep}, + null, + null, + Collections.singletonMap("initial_key", "initial_val")); + + View withProp = + viewCatalog.alterView( + NameIdentifier.of(schemaName, viewName), ViewChange.setProperty("new_key", "new_val")); + Assertions.assertEquals("new_val", withProp.properties().get("new_key")); + Assertions.assertEquals("initial_val", withProp.properties().get("initial_key")); + + View withoutProp = + viewCatalog.alterView( + NameIdentifier.of(schemaName, viewName), ViewChange.removeProperty("new_key")); + Assertions.assertNull(withoutProp.properties().get("new_key")); + } + + @Test + void testCreateViewAlreadyExists() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String viewName = GravitinoITUtils.genRandomName("dup_view"); + NameIdentifier ident = NameIdentifier.of(schemaName, viewName); + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + SQLRepresentation rep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + + viewCatalog.createView( + ident, null, columns, new SQLRepresentation[] {rep}, null, null, Collections.emptyMap()); + + Assertions.assertThrows( + ViewAlreadyExistsException.class, + () -> + viewCatalog.createView( + ident, + null, + columns, + new SQLRepresentation[] {rep}, + null, + null, + Collections.emptyMap())); + } + + @Test + void testLoadNonExistentView() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + NameIdentifier ident = NameIdentifier.of(schemaName, "non_existent_view_xyz"); + Assertions.assertThrows(NoSuchViewException.class, () -> viewCatalog.loadView(ident)); + } + + @Test + void testAlterViewReplace() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + String viewName = GravitinoITUtils.genRandomName("replace_view"); + Column[] columns = {Column.of("id", Types.LongType.get(), null)}; + SQLRepresentation sparkRep = + SQLRepresentation.builder().withDialect(SPARK_DIALECT).withSql("SELECT id FROM t").build(); + + viewCatalog.createView( + NameIdentifier.of(schemaName, viewName), + "original comment", + columns, + new SQLRepresentation[] {sparkRep}, + null, + null, + Collections.emptyMap()); + + SQLRepresentation trinoRep = + SQLRepresentation.builder() + .withDialect(TRINO_DIALECT) + .withSql("SELECT id, name FROM updated_table") + .build(); + Column[] replacedColumns = { + Column.of("id", Types.LongType.get(), null), Column.of("name", Types.StringType.get(), null) + }; + + View altered = + viewCatalog.alterView( + NameIdentifier.of(schemaName, viewName), + ViewChange.replaceView( + replacedColumns, + new SQLRepresentation[] {trinoRep}, + null, + schemaName, + "replaced comment")); + + Assertions.assertEquals("replaced comment", altered.comment()); + Assertions.assertEquals(1, altered.representations().length); + Assertions.assertTrue(altered.sqlFor(TRINO_DIALECT).isPresent()); + Assertions.assertFalse(altered.sqlFor(SPARK_DIALECT).isPresent()); + Assertions.assertEquals(2, altered.columns().length); + Assertions.assertEquals("name", altered.columns()[1].name()); + } + + @Test + void testListViewsInNonExistentSchema() { + ViewCatalog viewCatalog = catalog.asViewCatalog(); + Assertions.assertThrows( + Exception.class, () -> viewCatalog.listViews(Namespace.of("non_existent_schema_xyz"))); Review Comment: Fixed in 586ae324c. Updated integration test testListViewsInNonExistentSchema to assert NoSuchSchemaException instead of generic Exception. ########## catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergViewCatalogOperations.java: ########## @@ -0,0 +1,390 @@ +/* + * 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.iceberg; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Maps; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.catalog.lakehouse.iceberg.converter.ConvertUtil; +import org.apache.gravitino.catalog.lakehouse.iceberg.ops.IcebergCatalogWrapperHelper; +import org.apache.gravitino.exceptions.NoSuchSchemaException; +import org.apache.gravitino.exceptions.NoSuchViewException; +import org.apache.gravitino.exceptions.ViewAlreadyExistsException; +import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; +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.gravitino.rel.ViewChange; +import org.apache.gravitino.utils.PrincipalUtils; +import org.apache.iceberg.MetadataUpdate; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.rest.requests.ImmutableCreateViewRequest; +import org.apache.iceberg.rest.requests.RenameTableRequest; +import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.ListTablesResponse; +import org.apache.iceberg.rest.responses.LoadViewResponse; +import org.apache.iceberg.view.ImmutableSQLViewRepresentation; +import org.apache.iceberg.view.ImmutableViewVersion; +import org.apache.iceberg.view.SQLViewRepresentation; +import org.apache.iceberg.view.ViewMetadata; +import org.apache.iceberg.view.ViewRepresentation; +import org.apache.iceberg.view.ViewVersion; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** View catalog operations for Iceberg. */ +class IcebergViewCatalogOperations { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergViewCatalogOperations.class); + + private final IcebergCatalogWrapper icebergCatalogWrapper; + + IcebergViewCatalogOperations(IcebergCatalogWrapper icebergCatalogWrapper) { + this.icebergCatalogWrapper = + Preconditions.checkNotNull(icebergCatalogWrapper, "icebergCatalogWrapper must not be null"); + } + + public View loadView(NameIdentifier ident) throws NoSuchViewException { + try { + LoadViewResponse response = + icebergCatalogWrapper.loadView( + IcebergCatalogWrapperHelper.buildIcebergTableIdentifier(ident)); + return IcebergView.fromLoadViewResponse(response, ident.name()); + } catch (Exception e) { + throw new NoSuchViewException( + e, "Failed to load view %s from Iceberg catalog: %s", ident, e.getMessage()); + } + } + + public NameIdentifier[] listViews(Namespace namespace) throws NoSuchSchemaException { + org.apache.iceberg.catalog.Namespace icebergNamespace = + IcebergCatalogWrapperHelper.getIcebergNamespace(namespace); + if (!icebergCatalogWrapper.namespaceExists(icebergNamespace)) { + throw new NoSuchSchemaException("Schema %s does not exist", namespace); + } + try { + ListTablesResponse response = icebergCatalogWrapper.listView(icebergNamespace); + return response.identifiers().stream() + .map(id -> NameIdentifier.of(ArrayUtils.add(namespace.levels(), id.name()))) + .toArray(NameIdentifier[]::new); + } catch (NoSuchNamespaceException e) { + throw new NoSuchSchemaException("Schema does not exist %s in Iceberg", namespace); + } + } + + public View createView( + NameIdentifier ident, + String comment, + Column[] columns, + Representation[] representations, + String defaultCatalog, + String defaultSchema, + Map<String, String> properties) + throws NoSuchSchemaException, ViewAlreadyExistsException { + org.apache.iceberg.catalog.Namespace icebergNamespace = + IcebergCatalogWrapperHelper.getIcebergNamespace( + ident.namespace().level(ident.namespace().length() - 1)); + + Schema schema = ConvertUtil.toIcebergSchema(columns); + List<ViewRepresentation> viewRepresentations = new ArrayList<>(); + for (Representation representation : representations) { + viewRepresentations.add(toSqlViewRepresentation(representation)); + } Review Comment: Keeping current behavior intentionally. ViewOperations.createView already calls request.validate(), and ViewCreateRequest.validate() enforces representations not null and not empty, and validates each representation. Adding the same check again in IcebergViewCatalogOperations.createView would be redundant. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
