HIVE-29535: Iceberg: [V3] Add support for Table Level Encryption

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HMSTablePropertyHelper.java

@@ -19,6 +19,11 @@
 
 package org.apache.iceberg.hive;
 
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.Base64;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
@@ -27,6 +32,7 @@
 import org.apache.hadoop.hive.common.StatsSetupConst;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hive.iceberg.com.fasterxml.jackson.core.JsonGenerator;
 import org.apache.hive.iceberg.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.iceberg.BaseMetastoreTableOperations;
 import org.apache.iceberg.PartitionSpec;
@@ -38,12 +44,14 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.SortOrderParser;
 import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.collect.BiMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableBiMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.HashWriter;
 import org.apache.iceberg.util.JsonUtil;
 import org.apache.iceberg.view.ViewMetadata;
 import org.apache.parquet.Strings;
@@ -118,6 +126,7 @@ public static void updateHmsTableForIcebergTable(
         metadata.schema(),
         maxHiveTablePropertySize);
     setStorageHandler(parameters, hiveEngineEnabled);
+    setMetadataHash(metadata, parameters);
 
     // Set the basic statistics
     if (summary.get(SnapshotSummary.TOTAL_DATA_FILES_PROP) != null) {
@@ -295,4 +304,39 @@ private static void setField(
   private static boolean exposeInHmsProperties(long maxHiveTablePropertySize) {
     return maxHiveTablePropertySize > 0;
   }
+
+  @VisibleForTesting
+  static void setMetadataHash(TableMetadata metadata, Map<String, String> parameters) {
+    if (parameters.containsKey(TableProperties.ENCRYPTION_TABLE_KEY)) {
+      byte[] currentHashBytes = hashOf(metadata);
+      parameters.put(
+          BaseMetastoreTableOperations.METADATA_HASH_PROP,
+          Base64.getEncoder().encodeToString(currentHashBytes));
+    }
+  }
+
+  @VisibleForTesting
+  static void verifyMetadataHash(TableMetadata metadata, String metadataHashFromHMS) {
+    byte[] currentHashBytes = hashOf(metadata);
+    byte[] expectedHashBytes = Base64.getDecoder().decode(metadataHashFromHMS);
+
+    if (!Arrays.equals(expectedHashBytes, currentHashBytes)) {
+      throw new RuntimeException(
+          String.format(
+              "The current metadata file %s might have been modified. Hash of metadata loaded from storage differs " +
+                  "from HMS-stored metadata hash.",
+              metadata.metadataFileLocation()));
+    }
+  }
+
+  private static byte[] hashOf(TableMetadata tableMetadata) {
+    try (HashWriter hashWriter = new HashWriter("SHA-256", StandardCharsets.UTF_8)) {
+      JsonGenerator generator = JsonUtil.factory().createGenerator(hashWriter);
+      TableMetadataParser.toJson(tableMetadata, generator);
+      generator.flush();
+      return hashWriter.getHash();
+    } catch (NoSuchAlgorithmException | IOException e) {
+      throw new RuntimeException("Unable to produce hash of table metadata", e);
+    }
+  }
 }

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveCatalog.java

@@ -19,6 +19,7 @@
 
 package org.apache.iceberg.hive;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -47,6 +48,8 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.EncryptionUtil;
+import org.apache.iceberg.encryption.KeyManagementClient;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchIcebergViewException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
@@ -97,6 +100,7 @@ public class HiveCatalog extends BaseMetastoreViewCatalog
   private String name;
   private Configuration conf;
   private FileIO fileIO;
+  private KeyManagementClient keyManagementClient;
   private ClientPool<IMetaStoreClient, TException> clients;
   private boolean listAllTables = false;
   private Map<String, String> catalogProperties;
@@ -136,6 +140,11 @@ public void initialize(String inputName, Map<String, String> properties) {
     } else {
       this.fileIO = CatalogUtil.loadFileIO(fileIOImpl, properties, conf);
     }
+    if (catalogProperties.containsKey(CatalogProperties.ENCRYPTION_KMS_TYPE) ||
+        catalogProperties.containsKey(CatalogProperties.ENCRYPTION_KMS_IMPL)) {
+      this.keyManagementClient = EncryptionUtil.createKmsClient(properties);
+    }
+
     this.clients = new CachedClientPool(conf, properties);
   }
 
@@ -677,7 +686,7 @@ private boolean isValidateNamespace(Namespace namespace) {
   public TableOperations newTableOps(TableIdentifier tableIdentifier) {
     String dbName = tableIdentifier.namespace().level(0);
     String tableName = tableIdentifier.name();
-    return new HiveTableOperations(conf, clients, fileIO, name, dbName, tableName);
+    return new HiveTableOperations(conf, clients, fileIO, keyManagementClient, name, dbName, tableName);
   }
 
   @Override
@@ -810,6 +819,15 @@ public Map<String, String> properties() {
     return catalogProperties == null ? ImmutableMap.of() : catalogProperties;
   }
 
+  @Override
+  public void close() throws IOException {
+    super.close();
+
+    if (keyManagementClient != null) {
+      keyManagementClient.close();
+    }
+  }
+
   @VisibleForTesting
   void setListAllTables(boolean listAllTables) {
     this.listAllTables = listAllTables;