Skip to content

Commit c4b234b

Browse files
hantangwangdhantangwangd
committed
[Iceberg]Prefix table (loaded from Hive Catalog) with catalog name
1 parent 675e65e commit c4b234b

11 files changed

+73
-26
lines changed

‎presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergHiveMetadata.java

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,7 @@ public class IcebergHiveMetadata
154154
{
155155
public static final int MAXIMUM_PER_QUERY_TABLE_CACHE_SIZE = 1000;
156156

157+
private final IcebergCatalogName catalogName;
157158
private final ExtendedHiveMetastore metastore;
158159
private final HdfsEnvironment hdfsEnvironment;
159160
private final DateTimeZone timeZone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(ZoneId.of(TimeZone.getDefault().getID())));
@@ -162,6 +163,7 @@ public class IcebergHiveMetadata
162163
private final ManifestFileCache manifestFileCache;
163164

164165
public IcebergHiveMetadata(
166+
IcebergCatalogName catalogName,
165167
ExtendedHiveMetastore metastore,
166168
HdfsEnvironment hdfsEnvironment,
167169
TypeManager typeManager,
@@ -176,6 +178,7 @@ public IcebergHiveMetadata(
176178
IcebergTableProperties tableProperties)
177179
{
178180
super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion, filterStatsCalculatorService, statisticsFileCache, tableProperties);
181+
this.catalogName = requireNonNull(catalogName, "catalogName is null");
179182
this.metastore = requireNonNull(metastore, "metastore is null");
180183
this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");
181184
this.hiveTableOeprationsConfig = requireNonNull(hiveTableOeprationsConfig, "hiveTableOperationsConfig is null");
@@ -204,7 +207,7 @@ public boolean schemaExists(ConnectorSession session, String schemaName)
204207
@Override
205208
protected org.apache.iceberg.Table getRawIcebergTable(ConnectorSession session, SchemaTableName schemaTableName)
206209
{
207-
return getHiveIcebergTable(metastore, hdfsEnvironment, hiveTableOeprationsConfig, manifestFileCache, session, schemaTableName);
210+
return getHiveIcebergTable(metastore, hdfsEnvironment, hiveTableOeprationsConfig, manifestFileCache, session, catalogName, schemaTableName);
208211
}
209212

210213
@Override

‎presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergHiveMetadataFactory.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
public class IcebergHiveMetadataFactory
3232
implements IcebergMetadataFactory
3333
{
34+
final IcebergCatalogName catalogName;
3435
final ExtendedHiveMetastore metastore;
3536
final HdfsEnvironment hdfsEnvironment;
3637
final TypeManager typeManager;
@@ -46,6 +47,7 @@ public class IcebergHiveMetadataFactory
4647

4748
@Inject
4849
public IcebergHiveMetadataFactory(
50+
IcebergCatalogName catalogName,
4951
ExtendedHiveMetastore metastore,
5052
HdfsEnvironment hdfsEnvironment,
5153
TypeManager typeManager,
@@ -59,6 +61,7 @@ public IcebergHiveMetadataFactory(
5961
ManifestFileCache manifestFileCache,
6062
IcebergTableProperties tableProperties)
6163
{
64+
this.catalogName = requireNonNull(catalogName, "catalogName is null");
6265
this.metastore = requireNonNull(metastore, "metastore is null");
6366
this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");
6467
this.typeManager = requireNonNull(typeManager, "typeManager is null");
@@ -76,6 +79,7 @@ public IcebergHiveMetadataFactory(
7679
public ConnectorMetadata create()
7780
{
7881
return new IcebergHiveMetadata(
82+
catalogName,
7983
metastore,
8084
hdfsEnvironment,
8185
typeManager,

‎presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java

Lines changed: 29 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@
7171
import org.apache.iceberg.TableOperations;
7272
import org.apache.iceberg.TableScan;
7373
import org.apache.iceberg.catalog.Catalog;
74+
import org.apache.iceberg.catalog.TableIdentifier;
7475
import org.apache.iceberg.catalog.ViewCatalog;
7576
import org.apache.iceberg.expressions.Expression;
7677
import org.apache.iceberg.hive.HiveSchemaUtil;
@@ -101,7 +102,6 @@
101102
import java.util.Objects;
102103
import java.util.Optional;
103104
import java.util.Set;
104-
import java.util.regex.Pattern;
105105
import java.util.stream.Collectors;
106106
import java.util.stream.Stream;
107107

@@ -211,7 +211,6 @@
211211

212212
public final class IcebergUtil
213213
{
214-
private static final Pattern SIMPLE_NAME = Pattern.compile("[a-z][a-z0-9]*");
215214
private static final Logger log = Logger.get(IcebergUtil.class);
216215
public static final int MIN_FORMAT_VERSION_FOR_DELETE = 2;
217216

@@ -246,7 +245,7 @@ public static Table getShallowWrappedIcebergTable(Schema schema, PartitionSpec s
246245
return new PrestoIcebergTableForMetricsConfig(schema, spec, properties, sortOrder);
247246
}
248247

249-
public static Table getHiveIcebergTable(ExtendedHiveMetastore metastore, HdfsEnvironment hdfsEnvironment, IcebergHiveTableOperationsConfig config, ManifestFileCache manifestFileCache, ConnectorSession session, SchemaTableName table)
248+
public static Table getHiveIcebergTable(ExtendedHiveMetastore metastore, HdfsEnvironment hdfsEnvironment, IcebergHiveTableOperationsConfig config, ManifestFileCache manifestFileCache, ConnectorSession session, IcebergCatalogName catalogName, SchemaTableName table)
250249
{
251250
HdfsContext hdfsContext = new HdfsContext(session, table.getSchemaName(), table.getTableName());
252251
TableOperations operations = new HiveTableOperations(
@@ -258,7 +257,7 @@ public static Table getHiveIcebergTable(ExtendedHiveMetastore metastore, HdfsEnv
258257
manifestFileCache,
259258
table.getSchemaName(),
260259
table.getTableName());
261-
return new BaseTable(operations, quotedTableName(table));
260+
return new BaseTable(operations, fullTableName(catalogName.getCatalogName(), TableIdentifier.of(table.getSchemaName(), table.getTableName())));
262261
}
263262

264263
public static Table getNativeIcebergTable(IcebergNativeCatalogFactory catalogFactory, ConnectorSession session, SchemaTableName table)
@@ -411,19 +410,6 @@ public static Optional<String> getViewComment(View view)
411410
return Optional.ofNullable(view.properties().get(TABLE_COMMENT));
412411
}
413412

414-
private static String quotedTableName(SchemaTableName name)
415-
{
416-
return quotedName(name.getSchemaName()) + "." + quotedName(name.getTableName());
417-
}
418-
419-
private static String quotedName(String name)
420-
{
421-
if (SIMPLE_NAME.matcher(name).matches()) {
422-
return name;
423-
}
424-
return '"' + name.replace("\"", "\"\"") + '"';
425-
}
426-
427413
public static TableScan getTableScan(TupleDomain<IcebergColumnHandle> predicates, Optional<Long> snapshotId, Table icebergTable)
428414
{
429415
Expression expression = ExpressionConverter.toIcebergExpression(predicates);
@@ -1329,4 +1315,30 @@ public static DataSize getTargetSplitSize(ConnectorSession session, Scan<?, ?, ?
13291315
{
13301316
return getTargetSplitSize(IcebergSessionProperties.getTargetSplitSize(session), scan.targetSplitSize());
13311317
}
1318+
1319+
// This code is copied from Iceberg
1320+
private static String fullTableName(String catalogName, TableIdentifier identifier)
1321+
{
1322+
StringBuilder sb = new StringBuilder();
1323+
1324+
if (catalogName.contains("/") || catalogName.contains(":")) {
1325+
// use / for URI-like names: thrift://host:port/db.table
1326+
sb.append(catalogName);
1327+
if (!catalogName.endsWith("/")) {
1328+
sb.append("/");
1329+
}
1330+
}
1331+
else {
1332+
// use . for non-URI named catalogs: prod.db.table
1333+
sb.append(catalogName).append(".");
1334+
}
1335+
1336+
for (String level : identifier.namespace().levels()) {
1337+
sb.append(level).append(".");
1338+
}
1339+
1340+
sb.append(identifier.name());
1341+
1342+
return sb.toString();
1343+
}
13321344
}

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/IcebergDistributedTestBase.java

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -635,6 +635,21 @@ public void testCreateTableWithCustomLocation()
635635
}
636636
}
637637

638+
@Test
639+
protected void testCreateTableAndValidateIcebergTableName()
640+
{
641+
String tableName = "test_create_table_for_validate_name";
642+
Session session = getSession();
643+
assertUpdate(session, format("CREATE TABLE %s (col1 INTEGER, aDate DATE)", tableName));
644+
Table icebergTable = loadTable(tableName);
645+
646+
String catalog = session.getCatalog().get();
647+
String schemaName = session.getSchema().get();
648+
assertEquals(icebergTable.name(), catalog + "." + schemaName + "." + tableName);
649+
650+
assertUpdate("DROP TABLE IF EXISTS " + tableName);
651+
}
652+
638653
@Test
639654
public void testPartitionedByTimeType()
640655
{
@@ -2526,7 +2541,7 @@ private Table updateTable(String tableName)
25262541

25272542
protected Table loadTable(String tableName)
25282543
{
2529-
Catalog catalog = CatalogUtil.loadCatalog(catalogType.getCatalogImpl(), "test-hive", getProperties(), new Configuration());
2544+
Catalog catalog = CatalogUtil.loadCatalog(catalogType.getCatalogImpl(), ICEBERG_CATALOG, getProperties(), new Configuration());
25302545
return catalog.loadTable(TableIdentifier.of("tpch", tableName));
25312546
}
25322547

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/TestIcebergTableVersion.java

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -349,15 +349,15 @@ public void testTableVersionErrors()
349349
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF id", ".* cannot be resolved");
350350
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF (SELECT CURRENT_TIMESTAMP)", ".* Constant expression cannot contain a subquery");
351351
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF NULL", "Table version AS OF/BEFORE expression cannot be NULL for .*");
352-
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF TIMESTAMP " + "'" + tab2Timestamp1 + "' - INTERVAL '1' MONTH", "No history found based on timestamp for table \"test_tt_schema\".\"test_table_version_tab2\"");
353-
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF CAST ('2023-01-01' AS TIMESTAMP WITH TIME ZONE)", "No history found based on timestamp for table \"test_tt_schema\".\"test_table_version_tab2\"");
354-
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF CAST ('2023-01-01' AS TIMESTAMP)", "No history found based on timestamp for table \"test_tt_schema\".\"test_table_version_tab2\"");
352+
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF TIMESTAMP " + "'" + tab2Timestamp1 + "' - INTERVAL '1' MONTH", "No history found based on timestamp for table iceberg.test_tt_schema.test_table_version_tab2");
353+
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF CAST ('2023-01-01' AS TIMESTAMP WITH TIME ZONE)", "No history found based on timestamp for table iceberg.test_tt_schema.test_table_version_tab2");
354+
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF CAST ('2023-01-01' AS TIMESTAMP)", "No history found based on timestamp for table iceberg.test_tt_schema.test_table_version_tab2");
355355
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF CAST ('2023-01-01' AS DATE)", ".* Type date is invalid. Supported table version AS OF/BEFORE expression type is Timestamp or Timestamp with Time Zone.");
356356
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF CURRENT_DATE", ".* Type date is invalid. Supported table version AS OF/BEFORE expression type is Timestamp or Timestamp with Time Zone.");
357-
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF TIMESTAMP '2023-01-01 00:00:00.000'", "No history found based on timestamp for table \"test_tt_schema\".\"test_table_version_tab2\"");
357+
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP AS OF TIMESTAMP '2023-01-01 00:00:00.000'", "No history found based on timestamp for table iceberg.test_tt_schema.test_table_version_tab2");
358358

359-
assertQueryFails("SELECT desc FROM " + tableName1 + " FOR VERSION BEFORE " + tab1VersionId1 + " ORDER BY 1", "No history found based on timestamp for table \"test_tt_schema\".\"test_table_version_tab1\"");
360-
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP BEFORE TIMESTAMP " + "'" + tab2Timestamp1 + "' - INTERVAL '1' MONTH", "No history found based on timestamp for table \"test_tt_schema\".\"test_table_version_tab2\"");
359+
assertQueryFails("SELECT desc FROM " + tableName1 + " FOR VERSION BEFORE " + tab1VersionId1 + " ORDER BY 1", "No history found based on timestamp for table iceberg.test_tt_schema.test_table_version_tab1");
360+
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP BEFORE TIMESTAMP " + "'" + tab2Timestamp1 + "' - INTERVAL '1' MONTH", "No history found based on timestamp for table iceberg.test_tt_schema.test_table_version_tab2");
361361
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR VERSION BEFORE 100", ".* Type integer is invalid. Supported table version AS OF/BEFORE expression type is BIGINT or VARCHAR");
362362
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR VERSION BEFORE " + tab2VersionId1 + " - " + tab2VersionId1, "Iceberg snapshot ID does not exists: 0");
363363
assertQueryFails("SELECT desc FROM " + tableName2 + " FOR TIMESTAMP BEFORE 'bad'", ".* Type varchar\\(3\\) is invalid. Supported table version AS OF/BEFORE expression type is Timestamp or Timestamp with Time Zone.");

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/hadoop/TestIcebergDistributedOnS3Hadoop.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import java.net.URI;
3838

3939
import static com.facebook.presto.iceberg.CatalogType.HADOOP;
40+
import static com.facebook.presto.iceberg.IcebergQueryRunner.ICEBERG_CATALOG;
4041
import static com.facebook.presto.iceberg.container.IcebergMinIODataLake.ACCESS_KEY;
4142
import static com.facebook.presto.iceberg.container.IcebergMinIODataLake.SECRET_KEY;
4243
import static com.facebook.presto.testing.TestingConnectorSession.SESSION;
@@ -130,7 +131,7 @@ protected HdfsEnvironment getHdfsEnvironment()
130131
protected Table loadTable(String tableName)
131132
{
132133
Configuration configuration = getHdfsEnvironment().getConfiguration(new HdfsContext(SESSION), getCatalogDataDirectory());
133-
Catalog catalog = CatalogUtil.loadCatalog(HADOOP.getCatalogImpl(), "test-hive", getProperties(), configuration);
134+
Catalog catalog = CatalogUtil.loadCatalog(HADOOP.getCatalogImpl(), ICEBERG_CATALOG, getProperties(), configuration);
134135
return catalog.loadTable(TableIdentifier.of("tpch", tableName));
135136
}
136137
}

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/hive/TestIcebergDistributedHive.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import com.facebook.presto.Session;
1717
import com.facebook.presto.common.QualifiedObjectName;
1818
import com.facebook.presto.hive.metastore.ExtendedHiveMetastore;
19+
import com.facebook.presto.iceberg.IcebergCatalogName;
1920
import com.facebook.presto.iceberg.IcebergDistributedTestBase;
2021
import com.facebook.presto.iceberg.IcebergHiveMetadata;
2122
import com.facebook.presto.iceberg.IcebergHiveTableOperationsConfig;
@@ -191,6 +192,7 @@ protected Table loadTable(String tableName)
191192
new IcebergHiveTableOperationsConfig(),
192193
new ManifestFileCache(CacheBuilder.newBuilder().build(), false, 0, 1024 * 1024),
193194
getQueryRunner().getDefaultSession().toConnectorSession(connectorId),
195+
new IcebergCatalogName(ICEBERG_CATALOG),
194196
SchemaTableName.valueOf("tpch." + tableName));
195197
}
196198

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/hive/TestIcebergHiveStatistics.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import com.facebook.presto.hive.authentication.NoHdfsAuthentication;
3232
import com.facebook.presto.hive.metastore.ExtendedHiveMetastore;
3333
import com.facebook.presto.iceberg.CatalogType;
34+
import com.facebook.presto.iceberg.IcebergCatalogName;
3435
import com.facebook.presto.iceberg.IcebergColumnHandle;
3536
import com.facebook.presto.iceberg.IcebergHiveTableOperationsConfig;
3637
import com.facebook.presto.iceberg.IcebergMetadataColumn;
@@ -594,6 +595,7 @@ private Table loadTable(String tableName)
594595
new IcebergHiveTableOperationsConfig(),
595596
new ManifestFileCache(CacheBuilder.newBuilder().build(), false, 0, 1024),
596597
getQueryRunner().getDefaultSession().toConnectorSession(connectorId),
598+
new IcebergCatalogName(ICEBERG_CATALOG),
597599
SchemaTableName.valueOf("tpch." + tableName));
598600
}
599601

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/hive/TestIcebergSmokeHive.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,9 @@
1313
*/
1414
package com.facebook.presto.iceberg.hive;
1515

16+
import com.facebook.presto.FullConnectorSession;
1617
import com.facebook.presto.hive.metastore.ExtendedHiveMetastore;
18+
import com.facebook.presto.iceberg.IcebergCatalogName;
1719
import com.facebook.presto.iceberg.IcebergConfig;
1820
import com.facebook.presto.iceberg.IcebergDistributedSmokeTestBase;
1921
import com.facebook.presto.iceberg.IcebergHiveTableOperationsConfig;
@@ -60,11 +62,13 @@ protected ExtendedHiveMetastore getFileHiveMetastore()
6062
@Override
6163
protected Table getIcebergTable(ConnectorSession session, String schema, String tableName)
6264
{
65+
String defaultCatalog = ((FullConnectorSession) session).getSession().getCatalog().get();
6366
return IcebergUtil.getHiveIcebergTable(getFileHiveMetastore(),
6467
getHdfsEnvironment(),
6568
new IcebergHiveTableOperationsConfig(),
6669
new ManifestFileCache(CacheBuilder.newBuilder().build(), false, 0, 1024),
6770
session,
71+
new IcebergCatalogName(defaultCatalog),
6872
SchemaTableName.valueOf(schema + "." + tableName));
6973
}
7074
}

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/hive/TestRenameTableOnFragileFileSystem.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import com.facebook.presto.hive.metastore.file.FileHiveMetastoreConfig;
3737
import com.facebook.presto.hive.metastore.file.TableMetadata;
3838
import com.facebook.presto.iceberg.CommitTaskData;
39+
import com.facebook.presto.iceberg.IcebergCatalogName;
3940
import com.facebook.presto.iceberg.IcebergConfig;
4041
import com.facebook.presto.iceberg.IcebergHiveMetadata;
4142
import com.facebook.presto.iceberg.IcebergHiveMetadataFactory;
@@ -405,6 +406,7 @@ private ConnectorMetadata getIcebergHiveMetadata(ExtendedHiveMetastore metastore
405406
{
406407
HdfsEnvironment hdfsEnvironment = new TestingHdfsEnvironment();
407408
IcebergHiveMetadataFactory icebergHiveMetadataFactory = new IcebergHiveMetadataFactory(
409+
new IcebergCatalogName("unimportant"),
408410
metastore,
409411
hdfsEnvironment,
410412
FUNCTION_AND_TYPE_MANAGER,

‎presto-iceberg/src/test/java/com/facebook/presto/iceberg/procedure/TestRemoveOrphanFilesProcedureHive.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import com.facebook.presto.hive.metastore.ExtendedHiveMetastore;
1919
import com.facebook.presto.hive.metastore.MetastoreContext;
2020
import com.facebook.presto.iceberg.HiveTableOperations;
21+
import com.facebook.presto.iceberg.IcebergCatalogName;
2122
import com.facebook.presto.iceberg.IcebergHiveTableOperationsConfig;
2223
import com.facebook.presto.iceberg.IcebergUtil;
2324
import com.facebook.presto.iceberg.ManifestFileCache;
@@ -97,6 +98,7 @@ Table loadTable(String tableName)
9798
new IcebergHiveTableOperationsConfig(),
9899
new ManifestFileCache(CacheBuilder.newBuilder().build(), false, 0, 1024),
99100
getQueryRunner().getDefaultSession().toConnectorSession(connectorId),
101+
new IcebergCatalogName(ICEBERG_CATALOG),
100102
SchemaTableName.valueOf("tpch." + tableName));
101103
}
102104

0 commit comments

Comments
 (0)