You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bo...@apache.org on 2022/03/10 16:49:23 UTC
[impala] branch master updated: Revert "IMPALA-10737: Optimize the number of Iceberg API Metadata requests"
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new aef30d0 Revert "IMPALA-10737: Optimize the number of Iceberg API Metadata requests"
aef30d0 is described below
commit aef30d0442f8272e5d8639ebce61ccd28db2a5e9
Author: Tamas Mate <tm...@apache.org>
AuthorDate: Thu Mar 10 09:46:49 2022 +0100
Revert "IMPALA-10737: Optimize the number of Iceberg API Metadata requests"
This reverts commit cd10acdbb15c74eae21fa630bcb174104283560a.
This commit has been reverted, because it blocks upgrading the Iceberg
version to 0.13.
In the newer Iceberg version the BaseTable serialization has been
changed, it serializes the BaseTable to a SerializableTable sibiling
class. This is a lightweigth Table class which does not have the
necessary metadata that could be cached and reused by the ImpalaDs.
SerializableTable utilization has to be further considered.
Change-Id: I21e65cb3ab38d9e683223fb100d7ced90caa6edd
Reviewed-on: http://gerrit.cloudera.org:8080/18305
Reviewed-by: Zoltan Borok-Nagy <bo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
common/thrift/CatalogObjects.thrift | 15 ++--
common/thrift/CatalogService.thrift | 14 +++-
.../impala/catalog/CatalogServiceCatalog.java | 1 +
.../org/apache/impala/catalog/FeIcebergTable.java | 90 +++++-----------------
.../org/apache/impala/catalog/IcebergTable.java | 37 +++++----
.../catalog/IcebergTableLoadingException.java | 32 --------
.../impala/catalog/iceberg/IcebergCtasTarget.java | 12 ---
.../impala/catalog/local/CatalogdMetaProvider.java | 66 +++++++---------
.../impala/catalog/local/DirectMetaProvider.java | 13 +++-
.../impala/catalog/local/LocalIcebergTable.java | 64 +++++++--------
.../apache/impala/catalog/local/LocalTable.java | 2 +-
.../apache/impala/catalog/local/MetaProvider.java | 8 +-
.../org/apache/impala/planner/IcebergScanNode.java | 11 ++-
.../apache/impala/service/CatalogOpExecutor.java | 10 +--
.../java/org/apache/impala/service/Frontend.java | 2 +-
.../java/org/apache/impala/util/IcebergUtil.java | 30 +++++++-
.../impala/catalog/local/LocalCatalogTest.java | 8 +-
.../queries/QueryTest/iceberg-insert.test | 1 -
18 files changed, 182 insertions(+), 234 deletions(-)
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 77568bb..e6bc987 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -578,19 +578,18 @@ struct TIcebergTable {
2: required list<TIcebergPartitionSpec> partition_spec
3: required i32 default_partition_spec_id
// Map from 128-bit Murmur3 hash of data file path to its file descriptor
- 4: optional map<string, THdfsFileDesc> path_hash_to_file_descriptor
+ 4: optional map<string,THdfsFileDesc> path_hash_to_file_descriptor
+ // Iceberg snapshot id of the table
+ 5: optional i64 snapshot_id
// Iceberg 'write.parquet.compression-codec' and 'write.parquet.compression-level' table
// properties
- 5: optional TCompressionCodec parquet_compression_codec
+ 6: optional TCompressionCodec parquet_compression_codec
// Iceberg 'write.parquet.row-group-size-bytes' table property
- 6: optional i64 parquet_row_group_size
+ 7: optional i64 parquet_row_group_size
// Iceberg 'write.parquet.page-size-bytes' and 'write.parquet.dict-size-bytes' table
// properties
- 7: optional i64 parquet_plain_page_size;
- 8: optional i64 parquet_dict_page_size;
- // Serialized org.apache.iceberg.BaseTable object, it is used to execute operations
- // on the encapsulated Iceberg org.apache.iceberg.TableMetadata.
- 9: optional binary iceberg_base_table;
+ 8: optional i64 parquet_plain_page_size;
+ 9: optional i64 parquet_dict_page_size;
}
// Represents a table or view.
diff --git a/common/thrift/CatalogService.thrift b/common/thrift/CatalogService.thrift
index ba98df9..886c41b 100644
--- a/common/thrift/CatalogService.thrift
+++ b/common/thrift/CatalogService.thrift
@@ -390,8 +390,9 @@ struct TTableInfoSelector {
// it in cases the clients do need HMS partition structs.
12: bool want_hms_partition
- // The response should contain information about the Iceberg table.
- 13: bool want_iceberg_table
+ // The response should contain information about the Iceberg snapshot, i.e. the snapshot
+ // id and the file descriptors.
+ 13: bool want_iceberg_snapshot
}
// Returned information about a particular partition.
@@ -441,6 +442,11 @@ struct TPartialPartitionInfo {
13: optional CatalogObjects.THdfsPartitionLocation location
}
+struct TIcebergSnapshot {
+ 1: required i64 snapshot_id
+ 2: optional map<string, CatalogObjects.THdfsFileDesc> iceberg_file_desc_map
+}
+
// Returned information about a Table, as selected by TTableInfoSelector.
struct TPartialTableInfo {
1: optional hive_metastore.Table hms_table
@@ -481,8 +487,8 @@ struct TPartialTableInfo {
// the description of how a prefix is computed.
11: optional list<string> partition_prefixes
- // Iceberg table information
- 12: optional CatalogObjects.TIcebergTable iceberg_table
+ // Iceberg snapshot information
+ 12: optional TIcebergSnapshot iceberg_snapshot
}
struct TBriefTableMeta {
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index c3dc4e7..c85af60 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -59,6 +59,7 @@ import org.apache.impala.authorization.AuthorizationPolicy;
import org.apache.impala.catalog.FeFsTable.Utils;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
+import org.apache.impala.catalog.events.EventFactory;
import org.apache.impala.catalog.events.ExternalEventsProcessor;
import org.apache.impala.catalog.events.MetastoreEvents.EventFactoryForSyncToLatestEvent;
import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventFactory;
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
index 991bef5..7473346 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
@@ -17,12 +17,7 @@
package org.apache.impala.catalog;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -36,7 +31,6 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
-import org.apache.iceberg.BaseTable;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
@@ -60,6 +54,7 @@ import org.apache.impala.thrift.THdfsTable;
import org.apache.impala.thrift.THdfsPartition;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergFileFormat;
+import org.apache.impala.thrift.TIcebergSnapshot;
import org.apache.impala.thrift.TIcebergTable;
import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.thrift.TResultSet;
@@ -95,18 +90,6 @@ public interface FeIcebergTable extends FeFsTable {
TIcebergCatalog getIcebergCatalog();
/**
- * Returns the BaseTable that stores the metadata loaded by Iceberg.
- */
- BaseTable getIcebergBaseTable();
-
- /**
- * Returns the TableMetadata from the BaseTable.
- */
- default TableMetadata getIcebergTableMetadata() {
- return getIcebergBaseTable().operations().current();
- }
-
- /**
* Return Iceberg catalog location, we use this location to load metadata from Iceberg
* When using 'hadoop.tables', this value equals to table location
* When using 'hadoop.catalog', this value equals to 'iceberg.catalog_location'
@@ -286,11 +269,18 @@ public interface FeIcebergTable extends FeFsTable {
/**
* Current snapshot id of the table.
*/
- default long snapshotId() {
- if (getIcebergBaseTable().currentSnapshot() != null) {
- return getIcebergTableMetadata().currentSnapshot().snapshotId();
+ long snapshotId();
+
+ /**
+ * Utility class to hold information about Iceberg snapshots.
+ */
+ public static class Snapshot {
+ public Snapshot(long snapshotId, Map<String, FileDescriptor> pathHashToFileDescMap) {
+ this.snapshotId = snapshotId;
+ this.pathHashToFileDescMap = pathHashToFileDescMap;
}
- return -1;
+ public long snapshotId;
+ public Map<String, FileDescriptor> pathHashToFileDescMap;
}
/**
@@ -318,7 +308,7 @@ public interface FeIcebergTable extends FeFsTable {
resultSchema.addToColumns(new TColumn("Field Partition Transform",
Type.STRING.toThrift()));
- TableMetadata metadata = table.getIcebergTableMetadata();
+ TableMetadata metadata = IcebergUtil.getIcebergTableMetadata(table);
if (!metadata.specs().isEmpty()) {
// Just show the current PartitionSpec from Iceberg table metadata
PartitionSpec latestSpec = metadata.spec();
@@ -429,6 +419,7 @@ public interface FeIcebergTable extends FeFsTable {
tIcebergTable.setPath_hash_to_file_descriptor(
convertPathHashToFileDescMap(icebergTable));
+ tIcebergTable.setSnapshot_id(icebergTable.snapshotId());
tIcebergTable.setParquet_compression_codec(
icebergTable.getIcebergParquetCompressionCodec());
tIcebergTable.setParquet_row_group_size(
@@ -436,9 +427,7 @@ public interface FeIcebergTable extends FeFsTable {
tIcebergTable.setParquet_plain_page_size(
icebergTable.getIcebergParquetPlainPageSize());
tIcebergTable.setParquet_dict_page_size(
- icebergTable.getIcebergParquetDictPageSize());
- tIcebergTable.setIceberg_base_table(serializeIcebergBaseTable(
- icebergTable.getIcebergBaseTable()));
+ icebergTable.getIcebergParquetDictPageSize());
return tIcebergTable;
}
@@ -476,48 +465,11 @@ public interface FeIcebergTable extends FeFsTable {
return fileDescMap;
}
- public static byte[] serializeIcebergBaseTable(BaseTable baseTable) {
- ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
- ObjectOutputStream objectOutputStream = null;
- byte[] result = new byte[]{};
- try {
- objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);
- objectOutputStream.writeObject(baseTable);
- objectOutputStream.flush();
- result = byteArrayOutputStream.toByteArray();
- } catch (IOException e) {
- LOG.warn("Failed to serialize Iceberg BaseTable: " + e.getMessage());
- }
- finally {
- try {
- byteArrayOutputStream.close();
- } catch (IOException ex) {
- // ignore close exception
- }
- }
- return result;
- }
-
- public static BaseTable deserializeIcebergBaseTable(byte[] serializedBaseTable)
- throws TableLoadingException {
- Object baseTable = null;
- ByteArrayInputStream inputStream = new ByteArrayInputStream(serializedBaseTable);
- ObjectInput objectInput = null;
- try {
- objectInput = new ObjectInputStream(inputStream);
- baseTable = objectInput.readObject();
- } catch (Exception e) {
- throw new TableLoadingException("Failed to deserialize Iceberg BaseTable.", e);
- } finally {
- try {
- if (objectInput != null) {
- objectInput.close();
- }
- } catch (IOException ex) {
- // ignore close exception
- }
- }
- return (BaseTable)baseTable;
+ public static TIcebergSnapshot createTIcebergSnapshot(FeIcebergTable icebergTable) {
+ TIcebergSnapshot snapshot = new TIcebergSnapshot();
+ snapshot.setSnapshot_id(icebergTable.snapshotId());
+ snapshot.setIceberg_file_desc_map(convertPathHashToFileDescMap(icebergTable));
+ return snapshot;
}
/**
@@ -572,7 +524,7 @@ public interface FeIcebergTable extends FeFsTable {
Map<String, HdfsPartition.FileDescriptor> fileDescMap = new HashMap<>();
List<DataFile> dataFileList = IcebergUtil.getIcebergDataFiles(table,
new ArrayList<>(), /*timeTravelSpecl=*/null);
- Table iceTable = table.getIcebergBaseTable();
+ Table iceTable = IcebergUtil.loadTable(table);
for (DataFile dataFile : dataFileList) {
Path path = new Path(dataFile.path().toString());
if (hdfsFileDescMap.containsKey(path.toUri().getPath())) {
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
index 519bbb2..42a6ae9 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
@@ -28,8 +28,8 @@ import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.iceberg.Schema;
-import org.apache.iceberg.BaseTable;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.types.Types;
import org.apache.impala.analysis.IcebergPartitionField;
import org.apache.impala.analysis.IcebergPartitionSpec;
import org.apache.impala.analysis.IcebergPartitionTransform;
@@ -39,6 +39,7 @@ import org.apache.impala.thrift.TCompressionCodec;
import org.apache.impala.thrift.TGetPartialCatalogObjectRequest;
import org.apache.impala.thrift.TGetPartialCatalogObjectResponse;
import org.apache.impala.thrift.THdfsCompression;
+import org.apache.impala.thrift.THdfsFileDesc;
import org.apache.impala.thrift.THdfsTable;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergFileFormat;
@@ -164,8 +165,7 @@ public class IcebergTable extends Table implements FeIcebergTable {
// Treat iceberg table as a non-partitioned hdfs table in backend
private HdfsTable hdfsTable_;
- // Treat iceberg table as a non-partitioned hdfs table in backend
- private BaseTable icebergBaseTable_;
+ private long snapshotId_ = -1;
protected IcebergTable(org.apache.hadoop.hive.metastore.api.Table msTable,
Db db, String name, String owner) {
@@ -204,11 +204,6 @@ public class IcebergTable extends Table implements FeIcebergTable {
}
@Override
- public BaseTable getIcebergBaseTable() {
- return icebergBaseTable_;
- }
-
- @Override
public TCatalogObjectType getCatalogObjectType() {
return TCatalogObjectType.TABLE;
}
@@ -304,6 +299,11 @@ public class IcebergTable extends Table implements FeIcebergTable {
}
@Override
+ public long snapshotId() {
+ return snapshotId_;
+ }
+
+ @Override
public TTable toThrift() {
TTable table = super.toThrift();
table.setTable_type(TTableType.ICEBERG_TABLE);
@@ -333,8 +333,10 @@ public class IcebergTable extends Table implements FeIcebergTable {
final Timer.Context ctxStorageLdTime =
getMetrics().getTimer(Table.LOAD_DURATION_STORAGE_METADATA).time();
try {
- icebergBaseTable_ = (BaseTable)IcebergUtil.loadTable(this);
- TableMetadata metadata = getIcebergTableMetadata();
+ TableMetadata metadata = IcebergUtil.getIcebergTableMetadata(this);
+ if (metadata.currentSnapshot() != null) {
+ snapshotId_ = metadata.currentSnapshot().snapshotId();
+ }
loadSchemaFromIceberg(metadata);
// Loading hdfs table after loaded schema from Iceberg,
// in case we create external Iceberg table skipping column info in sql.
@@ -348,8 +350,8 @@ public class IcebergTable extends Table implements FeIcebergTable {
pathHashToFileDescMap_ = Utils.loadAllPartition(this);
loadAllColumnStats(msClient);
} catch (Exception e) {
- throw new IcebergTableLoadingException("Error loading metadata for Iceberg table "
- + icebergTableLocation_, e);
+ throw new TableLoadingException("Error loading metadata for Iceberg table " +
+ icebergTableLocation_, e);
} finally {
storageMetadataLoadTime_ = ctxStorageLdTime.stop();
}
@@ -420,10 +422,10 @@ public class IcebergTable extends Table implements FeIcebergTable {
// so there's no need for translation.
pathHashToFileDescMap_ = FeIcebergTable.Utils.loadFileDescMapFromThrift(
ticeberg.getPath_hash_to_file_descriptor(), null, null);
+ snapshotId_ = ticeberg.getSnapshot_id();
hdfsTable_.loadFromThrift(thriftTable);
- icebergBaseTable_ = FeIcebergTable.Utils.deserializeIcebergBaseTable(
- ticeberg.getIceberg_base_table());
- icebergSchema_ = icebergBaseTable_.operations().current().schema();
+ TableMetadata metadata = IcebergUtil.getIcebergTableMetadata(this);
+ icebergSchema_ = metadata.schema();
}
private List<IcebergPartitionSpec> loadPartitionBySpecsFromThrift(
@@ -480,8 +482,9 @@ public class IcebergTable extends Table implements FeIcebergTable {
Map<HdfsPartition, TPartialPartitionInfo> missingPartialInfos = new HashMap<>();
TGetPartialCatalogObjectResponse resp =
getHdfsTable().getPartialInfo(req, missingPartialInfos);
- if (req.table_info_selector.want_iceberg_table) {
- resp.table_info.setIceberg_table(Utils.getTIcebergTable(this));
+ if (req.table_info_selector.want_iceberg_snapshot) {
+ resp.table_info.setIceberg_snapshot(
+ FeIcebergTable.Utils.createTIcebergSnapshot(this));
if (!resp.table_info.isSetNetwork_addresses()) {
resp.table_info.setNetwork_addresses(getHostIndex().getList());
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergTableLoadingException.java b/fe/src/main/java/org/apache/impala/catalog/IcebergTableLoadingException.java
deleted file mode 100644
index 4b2d691..0000000
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergTableLoadingException.java
+++ /dev/null
@@ -1,32 +0,0 @@
-// 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.impala.catalog;
-
-/**
- * Thrown when the iceberg table metadata cannot be loaded due to an error.
- */
-public class IcebergTableLoadingException extends TableLoadingException {
-
- public IcebergTableLoadingException(String s, Throwable cause) {
- super(s, cause);
- }
-
- public IcebergTableLoadingException(String s) {
- super(s);
- }
-};
\ No newline at end of file
diff --git a/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java b/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java
index 3657801..e0c986f 100644
--- a/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java
+++ b/fe/src/main/java/org/apache/impala/catalog/iceberg/IcebergCtasTarget.java
@@ -26,11 +26,9 @@ import java.util.Set;
import com.google.common.base.Preconditions;
-import org.apache.iceberg.BaseTable;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
-import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.types.TypeUtil;
@@ -256,16 +254,6 @@ public class IcebergCtasTarget extends CtasTargetTable implements FeIcebergTable
return -1;
}
- @Override
- public BaseTable getIcebergBaseTable() {
- return null;
- }
-
- @Override
- public TableMetadata getIcebergTableMetadata() {
- return null;
- }
-
public void addColumn(IcebergColumn col) {
colsByPos_.add(col);
colsByName_.put(col.getName().toLowerCase(), col);
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
index d60bb64..7dff758 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
@@ -54,6 +54,7 @@ import org.apache.impala.catalog.Catalog;
import org.apache.impala.catalog.CatalogDeltaLog;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogObjectCache;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HdfsCachePool;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
@@ -83,6 +84,7 @@ import org.apache.impala.thrift.TFunctionName;
import org.apache.impala.thrift.TGetPartialCatalogObjectRequest;
import org.apache.impala.thrift.TGetPartialCatalogObjectResponse;
import org.apache.impala.thrift.THdfsFileDesc;
+import org.apache.impala.thrift.TIcebergSnapshot;
import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.thrift.TPartialPartitionInfo;
import org.apache.impala.thrift.TPartialTableInfo;
@@ -104,6 +106,7 @@ import org.ehcache.sizeof.SizeOf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
@@ -606,7 +609,7 @@ public class CatalogdMetaProvider implements MetaProvider {
req.catalog_info_selector.want_db_names = true;
TGetPartialCatalogObjectResponse resp = sendRequest(req);
checkResponse(resp.catalog_info != null && resp.catalog_info.db_names != null,
- req, "missing database names");
+ req, "missing table names");
return ImmutableList.copyOf(resp.catalog_info.db_names);
}
});
@@ -1021,35 +1024,34 @@ public class CatalogdMetaProvider implements MetaProvider {
return ret;
}
- @Override
- public TPartialTableInfo loadIcebergTable(final TableMetaRef table) throws TException {
- Preconditions.checkArgument(table instanceof TableMetaRefImpl);
- return loadTableInfoWithIcebergTable(table);
- }
-
/**
- * Utility function for to retrieve table info with Iceberg table metadata.
+ * Utility function for to retrieve table info with Iceberg snapshot. Exists for testing
+ * purposes, use loadIcebergSnapshot() which translates the file descriptors to the
+ * table's host index.
*/
- TPartialTableInfo loadTableInfoWithIcebergTable(final TableMetaRef table)
+ @VisibleForTesting
+ TPartialTableInfo loadTableInfoWithIcebergSnapshot(final TableMetaRef table)
throws TException {
Preconditions.checkArgument(table instanceof TableMetaRefImpl);
- TableMetaRefImpl tableRef = (TableMetaRefImpl)table;
- String itemStr = "iceberg metadata for " + tableRef.dbName_ + "."
- + tableRef.tableName_;
- IcebergMetaCacheKey cacheKey = new IcebergMetaCacheKey(tableRef);
- return loadWithCaching(itemStr, TABLE_METADATA_CACHE_CATEGORY, cacheKey,
- new Callable<TPartialTableInfo>() {
- @Override
- public TPartialTableInfo call() throws Exception {
- TGetPartialCatalogObjectRequest req = newReqForTable(table);
- req.table_info_selector.want_iceberg_table = true;
- TGetPartialCatalogObjectResponse resp = sendRequest(req);
- checkResponse(resp.table_info != null &&
- resp.table_info.iceberg_table != null, req,
- "missing Iceberg table metadata");
- return resp.getTable_info();
- }
- });
+ TGetPartialCatalogObjectRequest req = newReqForTable(table);
+ req.table_info_selector.want_iceberg_snapshot = true;
+ TGetPartialCatalogObjectResponse resp = sendRequest(req);
+ return resp.table_info;
+ }
+
+ @Override
+ public FeIcebergTable.Snapshot loadIcebergSnapshot(final TableMetaRef table,
+ ListMap<TNetworkAddress> hostIndex)
+ throws TException {
+ TPartialTableInfo tableInfo = loadTableInfoWithIcebergSnapshot(table);
+ Map<String, FileDescriptor> pathToFds =
+ FeIcebergTable.Utils.loadFileDescMapFromThrift(
+ tableInfo.getIceberg_snapshot().getIceberg_file_desc_map(),
+ tableInfo.getNetwork_addresses(),
+ hostIndex);
+ return new FeIcebergTable.Snapshot(
+ tableInfo.getIceberg_snapshot().getSnapshot_id(),
+ pathToFds);
}
private ImmutableList<FileDescriptor> convertThriftFdList(List<THdfsFileDesc> thriftFds,
@@ -1940,18 +1942,6 @@ public class CatalogdMetaProvider implements MetaProvider {
}
}
- /**
- * Cache key for an entry storing Iceberg table metadata.
- *
- * Values for these keys are 'TPartialTableInfo' objects.
- */
- private static class IcebergMetaCacheKey extends VersionedTableCacheKey {
-
- public IcebergMetaCacheKey(TableMetaRefImpl table) {
- super(table);
- }
- }
-
@VisibleForTesting
static class SizeOfWeigher implements Weigher<Object, Object> {
// Bypass flyweight objects like small boxed integers, Boolean.TRUE, enums, etc.
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
index 36c33ef..41c7d15 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
@@ -19,13 +19,18 @@ package org.apache.impala.catalog.local;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import com.google.common.base.Stopwatch;
+import com.google.common.collect.Iterables;
import org.apache.commons.lang.NotImplementedException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
@@ -39,6 +44,7 @@ import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.TableMeta;
import org.apache.impala.authorization.AuthorizationPolicy;
import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FileMetadataLoader;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HdfsCachePool;
@@ -54,8 +60,8 @@ import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TBackendGflags;
import org.apache.impala.thrift.TBriefTableMeta;
+import org.apache.impala.thrift.TIcebergSnapshot;
import org.apache.impala.thrift.TNetworkAddress;
-import org.apache.impala.thrift.TPartialTableInfo;
import org.apache.impala.thrift.TValidWriteIdList;
import org.apache.impala.util.AcidUtils;
import org.apache.impala.util.ListMap;
@@ -540,9 +546,10 @@ class DirectMetaProvider implements MetaProvider {
}
@Override
- public TPartialTableInfo loadIcebergTable(final TableMetaRef table)
+ public FeIcebergTable.Snapshot loadIcebergSnapshot(final TableMetaRef table,
+ ListMap<TNetworkAddress> hostIndex)
throws TException {
throw new NotImplementedException(
- "loadIcebergTable() is not implemented for DirectMetaProvider");
+ "loadIcebergSnapshot() is not implemented for DirectMetaProvider");
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
index 077e9d9..09bec8f 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalIcebergTable.java
@@ -17,6 +17,7 @@
package org.apache.impala.catalog.local;
+import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
@@ -25,8 +26,8 @@ import java.util.Set;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.iceberg.BaseTable;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
import org.apache.impala.analysis.IcebergPartitionSpec;
import org.apache.impala.catalog.CatalogObject;
import org.apache.impala.catalog.Column;
@@ -41,12 +42,13 @@ import org.apache.impala.thrift.THdfsPartition;
import org.apache.impala.thrift.THdfsTable;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergFileFormat;
-import org.apache.impala.thrift.TIcebergTable;
-import org.apache.impala.thrift.TPartialTableInfo;
+import org.apache.impala.thrift.TIcebergSnapshot;
+import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.thrift.TTableDescriptor;
import org.apache.impala.thrift.TTableType;
import org.apache.impala.util.IcebergSchemaConverter;
import org.apache.impala.util.IcebergUtil;
+import org.apache.thrift.TException;
import com.google.common.base.Preconditions;
import com.google.errorprone.annotations.Immutable;
@@ -68,32 +70,29 @@ public class LocalIcebergTable extends LocalTable implements FeIcebergTable {
private int defaultPartitionSpecId_;
private Map<String, FileDescriptor> pathHashToFileDescMap_;
private LocalFsTable localFsTable_;
+ private long snapshotId_ = -1;
private Schema icebergSchema_;
- // BaseTable contains the metadata that Iceberg keeps track of.
- private BaseTable icebergBaseTable_;
-
- /**
- * Loads the Iceberg metadata from the CatalogD then initializes a LocalIcebergTable.
- */
- static LocalTable loadIcebergTableViaMetaProvider(LocalDb db, Table msTable,
+ static LocalTable loadFromIceberg(LocalDb db, Table msTable,
MetaProvider.TableMetaRef ref) throws TableLoadingException {
Preconditions.checkNotNull(db);
Preconditions.checkNotNull(msTable);
try {
- TPartialTableInfo tableInfo =
- db.getCatalog().getMetaProvider().loadIcebergTable(ref);
- TIcebergTable table = tableInfo.getIceberg_table();
- BaseTable icebergBaseTable = FeIcebergTable.Utils.deserializeIcebergBaseTable(
- table.getIceberg_base_table());
- List<Column> iceColumns = IcebergSchemaConverter.convertToImpalaSchema(
- icebergBaseTable.operations().current().schema());
+ TableParams params = new TableParams(msTable);
+ TableMetadata metadata =
+ IcebergUtil.getIcebergTableMetadata(params.icebergCatalog_,
+ IcebergUtil.getIcebergTableIdentifier(msTable),
+ params.icebergCatalogLocation_,
+ msTable.getParameters());
+ List<Column> iceColumns =
+ IcebergSchemaConverter.convertToImpalaSchema(metadata.schema());
validateColumns(iceColumns, msTable.getSd().getCols());
ColumnMap colMap = new ColumnMap(iceColumns,
/*numClusteringCols=*/ 0,
db.getName() + "." + msTable.getTableName(),
/*isFullAcidSchema=*/false);
- return new LocalIcebergTable(db, msTable, ref, colMap, tableInfo, icebergBaseTable);
+
+ return new LocalIcebergTable(db, msTable, ref, colMap, metadata);
} catch (Exception e) {
String fullTableName = msTable.getDbName() + "." + msTable.getTableName();
throw new TableLoadingException(
@@ -102,20 +101,25 @@ public class LocalIcebergTable extends LocalTable implements FeIcebergTable {
}
private LocalIcebergTable(LocalDb db, Table msTable, MetaProvider.TableMetaRef ref,
- ColumnMap cmap, TPartialTableInfo tableInfo, BaseTable icebergBaseTable)
+ ColumnMap cmap, TableMetadata metadata)
throws TableLoadingException {
super(db, msTable, ref, cmap);
- Preconditions.checkNotNull(tableInfo);
localFsTable_ = LocalFsTable.load(db, msTable, ref);
tableParams_ = new TableParams(msTable);
- pathHashToFileDescMap_ = FeIcebergTable.Utils.loadFileDescMapFromThrift(
- tableInfo.getIceberg_table().getPath_hash_to_file_descriptor(),
- tableInfo.getNetwork_addresses(),
- getHostIndex());
- icebergBaseTable_ = icebergBaseTable;
- partitionSpecs_ = Utils.loadPartitionSpecByIceberg(getIcebergTableMetadata());
- defaultPartitionSpecId_ = getIcebergTableMetadata().defaultSpecId();
- icebergSchema_ = getIcebergTableMetadata().schema();
+ FeIcebergTable.Snapshot snapshot;
+ try {
+ snapshot = db_.getCatalog().getMetaProvider().loadIcebergSnapshot(
+ ref, getHostIndex());
+ Preconditions.checkNotNull(snapshot);
+ snapshotId_ = snapshot.snapshotId;
+ pathHashToFileDescMap_ = snapshot.pathHashToFileDescMap;
+ } catch (TException e) {
+ throw new TableLoadingException(String.format(
+ "Failed to load table: %s.%s", msTable.getDbName(), msTable.getTableName()), e);
+ }
+ partitionSpecs_ = Utils.loadPartitionSpecByIceberg(metadata);
+ defaultPartitionSpecId_ = metadata.defaultSpecId();
+ icebergSchema_ = metadata.schema();
icebergFileFormat_ = IcebergUtil.getIcebergFileFormat(msTable);
icebergParquetCompressionCodec_ = Utils.getIcebergParquetCompressionCodec(msTable);
icebergParquetRowGroupSize_ = Utils.getIcebergParquetRowGroupSize(msTable);
@@ -200,8 +204,8 @@ public class LocalIcebergTable extends LocalTable implements FeIcebergTable {
}
@Override
- public BaseTable getIcebergBaseTable() {
- return icebergBaseTable_;
+ public long snapshotId() {
+ return snapshotId_;
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
index f056c2b..f09b656 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
@@ -104,7 +104,7 @@ abstract class LocalTable implements FeTable {
} else if (KuduTable.isKuduTable(msTbl)) {
t = LocalKuduTable.loadFromKudu(db, msTbl, ref);
} else if (IcebergTable.isIcebergTable(msTbl)) {
- t = LocalIcebergTable.loadIcebergTableViaMetaProvider(db, msTbl, ref);
+ t = LocalIcebergTable.loadFromIceberg(db, msTbl, ref);
} else if (DataSourceTable.isDataSourceTable(msTbl)) {
// TODO(todd) support datasource table
} else if (HdfsFileFormat.isHdfsInputFormatClass(
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
index ca437be..524871d 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
import org.apache.impala.authorization.AuthorizationPolicy;
import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HdfsCachePool;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
@@ -36,8 +37,8 @@ import org.apache.impala.catalog.HdfsStorageDescriptor;
import org.apache.impala.catalog.SqlConstraints;
import org.apache.impala.common.Pair;
import org.apache.impala.thrift.TBriefTableMeta;
+import org.apache.impala.thrift.TIcebergSnapshot;
import org.apache.impala.thrift.TNetworkAddress;
-import org.apache.impala.thrift.TPartialTableInfo;
import org.apache.impala.thrift.TValidWriteIdList;
import org.apache.impala.util.ListMap;
import org.apache.thrift.TException;
@@ -125,9 +126,10 @@ public interface MetaProvider {
List<String> colNames) throws TException;
/**
- * Loads Iceberg related table metadata.
+ * Loads Iceberg snapshot information, i.e. snapshot id and file descriptors.
*/
- public TPartialTableInfo loadIcebergTable(final TableMetaRef table)
+ public FeIcebergTable.Snapshot loadIcebergSnapshot(final TableMetaRef table,
+ ListMap<TNetworkAddress> hostIndex)
throws TException;
/**
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
index 8f02d8f..7330149 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
@@ -138,9 +138,14 @@ public class IcebergScanNode extends HdfsScanNode {
"Cannot load file descriptor for: " + dataFile.path());
}
// Add file descriptor to the cache.
- org.apache.iceberg.Table iceTable = icebergTable_.getIcebergBaseTable();
- fileDesc = fileDesc.cloneWithFileMetadata(
- IcebergUtil.createIcebergMetadata(iceTable, dataFile));
+ try {
+ org.apache.iceberg.Table iceTable = IcebergUtil.loadTable(icebergTable_);
+ fileDesc = fileDesc.cloneWithFileMetadata(
+ IcebergUtil.createIcebergMetadata(iceTable, dataFile));
+ } catch (TableLoadingException ex) {
+ throw new ImpalaRuntimeException(String.format(
+ "Cannot load Iceberg table for: %s", icebergTable_.getFullName()), ex);
+ }
icebergTable_.getPathHashToFileDescMap().put(
IcebergUtil.getDataFilePathHash(dataFile), fileDesc);
}
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index e3b24db..4d6a59a 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -102,7 +102,6 @@ import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.FeCatalogUtils;
import org.apache.impala.catalog.FileMetadataLoadOpts;
import org.apache.impala.catalog.IcebergTable;
-import org.apache.impala.catalog.IcebergTableLoadingException;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeIcebergTable;
@@ -6062,13 +6061,8 @@ public class CatalogOpExecutor {
// ACID tables, there is a Jira to cover this: HIVE-22062.
// 2: If no need for a full table reload then fetch partition level
// writeIds and reload only the ones that changed.
- try {
- updatedThriftTable = catalog_.reloadTable(tbl, req, resultType, cmdString,
- -1);
- } catch (IcebergTableLoadingException e) {
- updatedThriftTable = catalog_.invalidateTable(
- req.getTable_name(), tblWasRemoved, dbWasAdded);
- }
+ updatedThriftTable = catalog_.reloadTable(tbl, req, resultType, cmdString,
+ -1);
}
} else {
// Table was loaded from scratch, so it's already "refreshed".
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index 865da57..a921bb9 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -1117,7 +1117,7 @@ public class Frontend {
FeTable feTable = getCatalog().getTable(dbName, tableName);
Preconditions.checkState(feTable instanceof FeIcebergTable);
FeIcebergTable feIcebergTable = (FeIcebergTable) feTable;
- TableMetadata metadata = feIcebergTable.getIcebergTableMetadata();
+ TableMetadata metadata = IcebergUtil.getIcebergTableMetadata(feIcebergTable);
Table table = IcebergUtil.loadTable(feIcebergTable);
Set<Long> ancestorIds = Sets.newHashSet(IcebergUtil.currentAncestorIds(table));
diff --git a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
index edab096..a7f184c 100644
--- a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
@@ -61,8 +61,10 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.mr.Catalogs;
import org.apache.iceberg.transforms.PartitionSpecVisitor;
+import org.apache.iceberg.transforms.Transform;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
@@ -84,12 +86,14 @@ import org.apache.impala.catalog.iceberg.IcebergCatalog;
import org.apache.impala.catalog.iceberg.IcebergCatalogs;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.thrift.TCompressionCodec;
+import org.apache.impala.thrift.TCreateTableParams;
import org.apache.impala.thrift.THdfsCompression;
import org.apache.impala.thrift.THdfsFileFormat;
import org.apache.impala.thrift.TIcebergCatalog;
import org.apache.impala.thrift.TIcebergFileFormat;
import org.apache.impala.thrift.TIcebergPartitionField;
import org.apache.impala.thrift.TIcebergPartitionSpec;
+import org.apache.impala.thrift.TIcebergPartitionTransform;
import org.apache.impala.thrift.TIcebergPartitionTransformType;
public class IcebergUtil {
@@ -146,6 +150,27 @@ public class IcebergUtil {
}
/**
+ * Get TableMetadata by FeIcebergTable
+ */
+ public static TableMetadata getIcebergTableMetadata(FeIcebergTable table)
+ throws TableLoadingException {
+ BaseTable iceTable = (BaseTable)IcebergUtil.loadTable(table);
+ return iceTable.operations().current();
+ }
+
+ /**
+ * Get TableMetadata by related info tableName is table full name, usually
+ * database.table
+ */
+ public static TableMetadata getIcebergTableMetadata(TIcebergCatalog catalog,
+ TableIdentifier tableId, String location, Map<String, String> tableProps)
+ throws TableLoadingException {
+ BaseTable baseTable = (BaseTable)IcebergUtil.loadTable(catalog,
+ tableId, location, tableProps);
+ return baseTable.operations().current();
+ }
+
+ /**
* Get Iceberg table identifier by table property
*/
public static TableIdentifier getIcebergTableIdentifier(FeIcebergTable table) {
@@ -176,7 +201,7 @@ public class IcebergUtil {
*/
public static Transaction getIcebergTransaction(FeIcebergTable feTable)
throws TableLoadingException, ImpalaRuntimeException {
- return feTable.getIcebergBaseTable().newTransaction();
+ return getIcebergCatalog(feTable).loadTable(feTable).newTransaction();
}
/**
@@ -539,7 +564,8 @@ public class IcebergUtil {
private static TableScan createScanAsOf(FeIcebergTable table,
TimeTravelSpec timeTravelSpec) throws TableLoadingException {
- TableScan scan = table.getIcebergBaseTable().newScan();
+ BaseTable baseTable = (BaseTable)IcebergUtil.loadTable(table);
+ TableScan scan = baseTable.newScan();
if (timeTravelSpec == null) {
scan = scan.useSnapshot(table.snapshotId());
} else {
diff --git a/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
index c72de6c..61721e4 100644
--- a/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
@@ -26,6 +26,7 @@ import java.util.Set;
import org.apache.hadoop.hive.metastore.api.SQLForeignKey;
import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
import org.apache.hive.service.rpc.thrift.TGetColumnsReq;
+import org.apache.hive.service.rpc.thrift.TGetSchemasReq;
import org.apache.hive.service.rpc.thrift.TGetTablesReq;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.ToSqlUtils;
@@ -45,12 +46,14 @@ import org.apache.impala.catalog.Type;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.service.FeSupport;
import org.apache.impala.service.Frontend;
+import org.apache.impala.service.MetadataOp;
import org.apache.impala.testutil.TestUtils;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TMetadataOpRequest;
import org.apache.impala.thrift.TMetadataOpcode;
import org.apache.impala.thrift.TNetworkAddress;
import org.apache.impala.thrift.TPartialTableInfo;
+import org.apache.impala.thrift.TResultRow;
import org.apache.impala.thrift.TResultSet;
import org.apache.impala.util.ListMap;
import org.apache.impala.util.MetaStoreUtil;
@@ -58,6 +61,7 @@ import org.apache.impala.util.PatternMatcher;
import org.hamcrest.CoreMatchers;
import org.junit.Assert;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
import com.google.common.base.Joiner;
@@ -260,12 +264,12 @@ public class LocalCatalogTest {
LocalIcebergTable t = (LocalIcebergTable)catalog_.getTable(
"functional_parquet", "iceberg_partitioned");
Map<String, FileDescriptor> localTblFdMap = t.getPathHashToFileDescMap();
- TPartialTableInfo tblInfo = provider_.loadTableInfoWithIcebergTable(t.ref_);
+ TPartialTableInfo tblInfo = provider_.loadTableInfoWithIcebergSnapshot(t.ref_);
ListMap<TNetworkAddress> catalogdHostIndexes = new ListMap<>();
catalogdHostIndexes.populate(tblInfo.getNetwork_addresses());
Map<String, FileDescriptor> catalogFdMap =
FeIcebergTable.Utils.loadFileDescMapFromThrift(
- tblInfo.getIceberg_table().getPath_hash_to_file_descriptor(),
+ tblInfo.getIceberg_snapshot().getIceberg_file_desc_map(),
null, null);
for (Map.Entry<String, FileDescriptor> entry : localTblFdMap.entrySet()) {
String path = entry.getKey();
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-insert.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-insert.test
index bffbd97..a05b440 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-insert.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-insert.test
@@ -239,7 +239,6 @@ drop table iceberg_hive_cat
====
---- QUERY
# The data has been purged, so querying the external table fails.
-refresh iceberg_hive_cat_ext_2;
select * from iceberg_hive_cat_ext_2
---- CATCH
Table does not exist