Address review comments.

Co-authored-by: smaheshwar-pltr <[email protected] >

Author: ggershinsky

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java

@@ -194,14 +194,18 @@ the table key parameter (along with existing snapshots) in the file, making the
     refreshFromMetadataLocation(metadataLocation, metadataRefreshMaxRetries);
 
     if (tableKeyIdFromHMS != null) {
+      checkEncryptionProperties(tableKeyIdFromHMS, dekLengthFromHMS);
+
       tableKeyId = tableKeyIdFromHMS;
       encryptionDekLength =
           (dekLengthFromHMS != null)
               ? Integer.parseInt(dekLengthFromHMS)
               : TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT;
 
-      checkEncryptionProperties(tableKeyIdFromHMS, dekLengthFromHMS);
       encryptedKeysFromMetadata = current().encryptionKeys();
+      // Force re-creation of encryption manager with updated keys
+      encryptingFileIO = null;
+      encryptionManager = null;
     }
   }
 
@@ -212,11 +216,12 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
     encryptionPropsFromMetadata(metadata.properties());
 
     String newMetadataLocation;
-    if (encryption() instanceof StandardEncryptionManager) {
+    EncryptionManager encryptionManager = encryption();
+    if (encryptionManager instanceof StandardEncryptionManager) {
       // Add new encryption keys to the metadata
       TableMetadata.Builder builder = TableMetadata.buildFrom(metadata);
       for (Map.Entry<String, EncryptedKey> entry :
-          EncryptionUtil.encryptionKeys(encryption()).entrySet()) {
+          EncryptionUtil.encryptionKeys(encryptionManager).entrySet()) {
         builder.addEncryptionKey(entry.getValue());
       }
 

spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestTableEncryption.java

@@ -30,9 +30,12 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.MetadataTableType;
 import org.apache.iceberg.Parameters;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.encryption.Ciphers;
 import org.apache.iceberg.encryption.UnitestKMS;
 import org.apache.iceberg.io.InputFile;
@@ -92,6 +95,25 @@ public void testSelect() {
     assertEquals("Should return all expected rows", expected, sql("SELECT * FROM %s", tableName));
   }
 
+  private static List<DataFile> currentDataFiles(Table table) {
+    return Streams.stream(table.newScan().planFiles())
+        .map(FileScanTask::file)
+        .collect(Collectors.toList());
+  }
+
+  @TestTemplate
+  public void testRefresh() {
+    catalog.initialize(catalogName, catalogConfig);
+    Table table = catalog.loadTable(tableIdent);
+
+    assertThat(currentDataFiles(table)).isNotEmpty();
+
+    sql("INSERT INTO %s VALUES (4, 'd', 4.0), (5, 'e', 5.0), (6, 'f', float('NaN'))", tableName);
+
+    table.refresh();
+    assertThat(currentDataFiles(table)).isNotEmpty();
+  }
+
   @TestTemplate
   public void testInsertAndDelete() {
     sql("INSERT INTO %s VALUES (4, 'd', 4.0), (5, 'e', 5.0), (6, 'f', float('NaN'))", tableName);