Skip to content

Commit 96fa476

Browse files
Dmitriy FingermanDmitriy Fingerman
Dmitriy Fingerman
authored and
Dmitriy Fingerman
committed
HIVE-29578: Iceberg: support for Iceberg native views
1 parent 4d4fb6a commit 96fa476

19 files changed

Lines changed: 1052 additions & 18 deletions

File tree

‎common/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -445,6 +445,8 @@ public enum ErrorMsg {
445445
@Deprecated // kept for backwards reference
446446
REPLACE_VIEW_WITH_MATERIALIZED(10400, "Attempt to replace view {0} with materialized view", true),
447447
REPLACE_MATERIALIZED_WITH_VIEW(10401, "Attempt to replace materialized view {0} with view", true),
448+
VIEW_STORAGE_HANDLER_UNSUPPORTED(10448, "CREATE VIEW only supports STORED BY ICEBERG for native "
449+
+ "Iceberg views; unsupported storage clause: {0}", true),
448450
UPDATE_DELETE_VIEW(10402, "You cannot update or delete records in a view"),
449451
MATERIALIZED_VIEW_DEF_EMPTY(10403, "Query for the materialized view rebuild could not be retrieved"),
450452
MERGE_PREDIACTE_REQUIRED(10404, "MERGE statement with both UPDATE and DELETE clauses " +

‎iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -302,7 +302,7 @@ private Table newHMSView(ViewMetadata metadata) {
302302
tableType().name());
303303
}
304304

305-
private String sqlFor(ViewMetadata metadata) {
305+
public static String sqlFor(ViewMetadata metadata) {
306306
SQLViewRepresentation closest = null;
307307
for (ViewRepresentation representation : metadata.currentVersion().representations()) {
308308
if (representation instanceof SQLViewRepresentation) {

‎iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/MetastoreUtil.java‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,11 @@
4646
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
4747
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
4848
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
49+
import org.apache.iceberg.util.PropertyUtil;
50+
import org.apache.iceberg.view.BaseView;
51+
import org.apache.iceberg.view.SQLViewRepresentation;
52+
import org.apache.iceberg.view.View;
53+
import org.apache.iceberg.view.ViewMetadata;
4954
import org.apache.thrift.TException;
5055

5156
public class MetastoreUtil {
@@ -148,6 +153,65 @@ public static Table toHiveTable(org.apache.iceberg.Table table, Configuration co
148153
return result;
149154
}
150155

156+
/**
157+
* Builds a Hive metastore {@link Table} representation for an Iceberg {@link View}, for clients
158+
* (e.g. {@code HiveRESTCatalogClient}) that bridge Iceberg catalog metadata into the HMS API.
159+
*/
160+
public static Table toHiveView(View view, Configuration conf) {
161+
Table result = new Table();
162+
TableName tableName =
163+
TableName.fromString(
164+
view.name(), MetaStoreUtils.getDefaultCatalog(conf), Warehouse.DEFAULT_DATABASE_NAME);
165+
result.setCatName(tableName.getCat());
166+
result.setDbName(tableName.getDb());
167+
result.setTableName(tableName.getTable());
168+
result.setTableType(TableType.VIRTUAL_VIEW.toString());
169+
170+
ViewMetadata metadata = ((BaseView) view).operations().current();
171+
String sqlText = viewSqlText(view, metadata);
172+
result.setViewOriginalText(sqlText);
173+
result.setViewExpandedText(sqlText);
174+
175+
long nowMillis = System.currentTimeMillis();
176+
int nowSec = (int) (nowMillis / 1000);
177+
String owner =
178+
PropertyUtil.propertyAsString(
179+
metadata.properties(), HiveCatalog.HMS_TABLE_OWNER, System.getProperty("user.name"));
180+
result.setOwner(owner);
181+
result.setCreateTime(nowSec);
182+
result.setLastAccessTime(nowSec);
183+
result.setRetention(Integer.MAX_VALUE);
184+
185+
boolean hiveEngineEnabled = false;
186+
result.setSd(HiveOperationsBase.storageDescriptor(metadata.schema(), metadata.location(), hiveEngineEnabled));
187+
188+
long maxHiveTablePropertySize =
189+
conf.getLong(
190+
HiveOperationsBase.HIVE_TABLE_PROPERTY_MAX_SIZE,
191+
HiveOperationsBase.HIVE_TABLE_PROPERTY_MAX_SIZE_DEFAULT);
192+
HMSTablePropertyHelper.updateHmsTableForIcebergView(
193+
metadata.metadataFileLocation(),
194+
result,
195+
metadata,
196+
Collections.emptySet(),
197+
maxHiveTablePropertySize,
198+
null);
199+
200+
String catalogType = IcebergCatalogProperties.getCatalogType(conf);
201+
if (!StringUtils.isEmpty(catalogType) && !IcebergCatalogProperties.NO_CATALOG_TYPE.equals(catalogType)) {
202+
result.getParameters().put(CatalogUtil.ICEBERG_CATALOG_TYPE, IcebergCatalogProperties.getCatalogType(conf));
203+
}
204+
return result;
205+
}
206+
207+
private static String viewSqlText(View view, ViewMetadata metadata) {
208+
SQLViewRepresentation hiveRepr = view.sqlFor("hive");
209+
if (hiveRepr != null) {
210+
return hiveRepr.sql();
211+
}
212+
return HiveViewOperations.sqlFor(metadata);
213+
}
214+
151215
private static StorageDescriptor getHiveStorageDescriptor(org.apache.iceberg.Table table) {
152216
var result = new StorageDescriptor();
153217
result.setCols(HiveSchemaUtil.convert(table.schema()));

‎iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/NativeIcebergViewSupport.java‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iceberg.hive;
21+
22+
import java.io.Closeable;
23+
import java.util.Collections;
24+
import java.util.List;
25+
import java.util.Map;
26+
import org.apache.hadoop.conf.Configuration;
27+
import org.apache.hadoop.hive.metastore.api.FieldSchema;
28+
import org.apache.iceberg.CatalogUtil;
29+
import org.apache.iceberg.catalog.Catalog;
30+
import org.apache.iceberg.catalog.Namespace;
31+
import org.apache.iceberg.catalog.TableIdentifier;
32+
import org.apache.iceberg.catalog.ViewCatalog;
33+
import org.apache.iceberg.view.ViewBuilder;
34+
35+
/**
36+
* Commits a native Iceberg view through the configured default Iceberg catalog (HiveCatalog or REST
37+
* catalog, etc.) when {@code Catalog} also implements {@link ViewCatalog}.
38+
*/
39+
public final class NativeIcebergViewSupport {
40+
41+
/** HMS parameter aligned with Hive's {@code CreateViewDesc#ICEBERG_NATIVE_VIEW_PROPERTY}. */
42+
public static final String ICEBERG_NATIVE_VIEW_PROPERTY = "hive.iceberg.native.view";
43+
44+
private NativeIcebergViewSupport() {
45+
}
46+
47+
/**
48+
* Creates or replaces a view in the Iceberg catalog.
49+
*
50+
* @return {@code false} if skipped because {@code ifNotExists} is true and the view already exists
51+
*/
52+
public static boolean createOrReplaceNativeView(Configuration conf, String databaseName, String viewName,
53+
List<FieldSchema> fieldSchemas, String viewSql, Map<String, String> tblProperties, String comment,
54+
boolean replace, boolean ifNotExists) throws Exception {
55+
56+
TableIdentifier identifier = TableIdentifier.of(databaseName, viewName);
57+
String catalogName = IcebergCatalogProperties.getCatalogName(conf);
58+
Map<String, String> catalogProps = IcebergCatalogProperties.getCatalogProperties(conf, catalogName);
59+
Catalog catalog = CatalogUtil.buildIcebergCatalog(catalogName, catalogProps, conf);
60+
try {
61+
ViewCatalog viewCatalog = asViewCatalog(catalog, catalogName);
62+
if (!replace && ifNotExists && viewCatalog.viewExists(identifier)) {
63+
return false;
64+
}
65+
66+
ViewBuilder builder = startViewBuilder(viewCatalog, identifier, fieldSchemas, viewSql);
67+
builder = applyCommentAndTblProps(builder, tblProperties, comment);
68+
commitView(builder, replace);
69+
return true;
70+
} finally {
71+
if (catalog instanceof Closeable) {
72+
((Closeable) catalog).close();
73+
}
74+
}
75+
}
76+
77+
private static ViewCatalog asViewCatalog(Catalog catalog, String catalogName) {
78+
if (!(catalog instanceof ViewCatalog)) {
79+
throw new UnsupportedOperationException(
80+
String.format(
81+
"Iceberg catalog '%s' does not implement ViewCatalog.",
82+
catalogName) +
83+
" Native views require a catalog that implements ViewCatalog (e.g. HiveCatalog or REST).");
84+
}
85+
return (ViewCatalog) catalog;
86+
}
87+
88+
private static ViewBuilder startViewBuilder(
89+
ViewCatalog viewCatalog,
90+
TableIdentifier identifier,
91+
List<FieldSchema> fieldSchemas,
92+
String viewSql) {
93+
return viewCatalog
94+
.buildView(identifier)
95+
.withSchema(HiveSchemaUtil.convert(fieldSchemas, Collections.emptyMap(), true))
96+
.withDefaultNamespace(Namespace.of(identifier.namespace().level(0)))
97+
.withQuery("hive", viewSql)
98+
.withProperty(ICEBERG_NATIVE_VIEW_PROPERTY, "true");
99+
}
100+
101+
private static ViewBuilder applyCommentAndTblProps(
102+
ViewBuilder builder, Map<String, String> tblProperties, String comment) {
103+
ViewBuilder viewBuilder = builder;
104+
if (comment != null && !comment.isEmpty()) {
105+
viewBuilder = viewBuilder.withProperty("comment", comment);
106+
}
107+
if (tblProperties != null) {
108+
for (Map.Entry<String, String> e : tblProperties.entrySet()) {
109+
if (e.getKey() != null && e.getValue() != null) {
110+
viewBuilder = viewBuilder.withProperty(e.getKey(), e.getValue());
111+
}
112+
}
113+
}
114+
return viewBuilder;
115+
}
116+
117+
private static void commitView(ViewBuilder builder, boolean replace) {
118+
if (replace) {
119+
builder.createOrReplace();
120+
} else {
121+
builder.create();
122+
}
123+
}
124+
}

‎iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/HiveRESTCatalogClient.java‎

Lines changed: 40 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,12 @@
2121

2222
import java.io.IOException;
2323
import java.util.Collections;
24+
import java.util.LinkedHashSet;
2425
import java.util.List;
2526
import java.util.Map;
2627
import java.util.Optional;
2728
import java.util.Properties;
29+
import java.util.Set;
2830
import java.util.regex.Pattern;
2931
import org.apache.hadoop.conf.Configuration;
3032
import org.apache.hadoop.hive.metastore.api.CreateTableRequest;
@@ -43,7 +45,9 @@
4345
import org.apache.iceberg.SortOrder;
4446
import org.apache.iceberg.catalog.Namespace;
4547
import org.apache.iceberg.catalog.TableIdentifier;
48+
import org.apache.iceberg.catalog.ViewCatalog;
4649
import org.apache.iceberg.exceptions.NoSuchTableException;
50+
import org.apache.iceberg.exceptions.NoSuchViewException;
4751
import org.apache.iceberg.hive.HMSTablePropertyHelper;
4852
import org.apache.iceberg.hive.HiveSchemaUtil;
4953
import org.apache.iceberg.hive.IcebergCatalogProperties;
@@ -54,6 +58,7 @@
5458
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
5559
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
5660
import org.apache.iceberg.rest.RESTCatalog;
61+
import org.apache.iceberg.view.View;
5762
import org.apache.thrift.TException;
5863
import org.slf4j.Logger;
5964
import org.slf4j.LoggerFactory;
@@ -123,10 +128,19 @@ public List<String> getTables(String catName, String dbName, String tablePattern
123128
Pattern pattern = Pattern.compile(regex);
124129

125130
// List tables from the specific database (namespace) and filter them.
126-
return restCatalog.listTables(Namespace.of(dbName)).stream()
131+
Set<String> names = new LinkedHashSet<>();
132+
restCatalog.listTables(Namespace.of(dbName)).stream()
127133
.map(TableIdentifier::name)
128134
.filter(pattern.asPredicate())
129-
.toList();
135+
.forEach(names::add);
136+
if (restCatalog instanceof ViewCatalog) {
137+
((ViewCatalog) restCatalog)
138+
.listViews(Namespace.of(dbName)).stream()
139+
.map(TableIdentifier::name)
140+
.filter(pattern.asPredicate())
141+
.forEach(names::add);
142+
}
143+
return Lists.newArrayList(names);
130144
}
131145

132146
@Override
@@ -136,7 +150,12 @@ public List<String> getAllTables(String catName, String dbName) {
136150

137151
@Override
138152
public void dropTable(Table table, boolean deleteData, boolean ignoreUnknownTab, boolean ifPurge) throws TException {
139-
restCatalog.dropTable(TableIdentifier.of(table.getDbName(), table.getTableName()));
153+
TableIdentifier id = TableIdentifier.of(table.getDbName(), table.getTableName());
154+
if (restCatalog instanceof ViewCatalog && ((ViewCatalog) restCatalog).viewExists(id)) {
155+
((ViewCatalog) restCatalog).dropView(id);
156+
} else {
157+
restCatalog.dropTable(id);
158+
}
140159
}
141160

142161
private void validateCurrentCatalog(String catName) {
@@ -149,7 +168,11 @@ private void validateCurrentCatalog(String catName) {
149168
@Override
150169
public boolean tableExists(String catName, String dbName, String tableName) {
151170
validateCurrentCatalog(catName);
152-
return restCatalog.tableExists(TableIdentifier.of(dbName, tableName));
171+
TableIdentifier id = TableIdentifier.of(dbName, tableName);
172+
if (restCatalog.tableExists(id)) {
173+
return true;
174+
}
175+
return restCatalog instanceof ViewCatalog && ((ViewCatalog) restCatalog).viewExists(id);
153176
}
154177

155178
@Override
@@ -178,14 +201,22 @@ public Database getDatabase(String catName, String dbName) throws NoSuchObjectEx
178201
@Override
179202
public Table getTable(GetTableRequest tableRequest) throws TException {
180203
validateCurrentCatalog(tableRequest.getCatName());
181-
org.apache.iceberg.Table icebergTable;
204+
TableIdentifier id =
205+
TableIdentifier.of(tableRequest.getDbName(), tableRequest.getTblName());
182206
try {
183-
icebergTable = restCatalog.loadTable(TableIdentifier.of(tableRequest.getDbName(),
184-
tableRequest.getTblName()));
185-
} catch (NoSuchTableException exception) {
207+
org.apache.iceberg.Table icebergTable = restCatalog.loadTable(id);
208+
return MetastoreUtil.toHiveTable(icebergTable, conf);
209+
} catch (NoSuchTableException tableMissing) {
210+
if (restCatalog instanceof ViewCatalog) {
211+
try {
212+
View icebergView = ((ViewCatalog) restCatalog).loadView(id);
213+
return MetastoreUtil.toHiveView(icebergView, conf);
214+
} catch (NoSuchViewException viewMissing) {
215+
throw new NoSuchObjectException();
216+
}
217+
}
186218
throw new NoSuchObjectException();
187219
}
188-
return MetastoreUtil.toHiveTable(icebergTable, conf);
189220
}
190221

191222
@Override

0 commit comments

Comments
 (0)