This is an automated email from the ASF dual-hosted git repository.
ayushtkn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 56928052cc4 HIVE-29535: Iceberg: [V3] Add support for Table Level
Encryption (#6396)
56928052cc4 is described below
commit 56928052cc415a009f98ac6ad50a79622d558fcd
Author: Ayush Saxena <[email protected]>
AuthorDate: Mon Jul 27 20:20:44 2026 +0530
HIVE-29535: Iceberg: [V3] Add support for Table Level Encryption (#6396)
---
.../iceberg/hive/HMSTablePropertyHelper.java | 44 +++
.../java/org/apache/iceberg/hive/HiveCatalog.java | 20 +-
.../apache/iceberg/hive/HiveTableOperations.java | 243 ++++++++++++-
.../iceberg/hive/StagingTableOperations.java | 4 +-
.../apache/iceberg/hive/TestHiveCommitLocks.java | 2 +
.../mr/hive/vector/HiveVectorizedReader.java | 29 +-
.../iceberg/mr/mapreduce/IcebergRecordReader.java | 4 +-
.../apache/iceberg/parquet/HiveParquetUtil.java | 77 +++++
.../parquet/ParquetFooterInputFromCache.java | 71 ++--
.../iceberg/mr/hive/TestHiveCatalogEncryption.java | 380 +++++++++++++++++++++
.../apache/iceberg/mr/hive/test/TestTables.java | 4 +-
.../mr/hive/vector/TestHiveVectorizedReader.java | 6 +-
.../parquet/TestParquetFooterInputFromCache.java | 216 ++++++++++++
.../src/main/java/org/apache/iceberg/DVUtil.java | 214 ++++++++++++
.../apache/iceberg/deletes/BaseDVFileWriter.java | 240 +++++++++++++
.../java/org/apache/iceberg/deletes/Deletes.java | 242 +++++++++++++
16 files changed, 1745 insertions(+), 51 deletions(-)
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HMSTablePropertyHelper.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HMSTablePropertyHelper.java
index 52d9ac84821..0639e50debe 100644
---
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HMSTablePropertyHelper.java
+++
b/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) {
@@ -300,4 +309,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);
+ }
+ }
}
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
index 24b962738c5..b7bd371dfe1 100644
---
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
+++
b/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;
@@ -46,6 +47,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.AlreadyExistsException;
import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
import org.apache.iceberg.exceptions.NoSuchIcebergViewException;
@@ -100,6 +103,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 boolean uniqueTableLocation;
@@ -147,6 +151,11 @@ public void initialize(String inputName, Map<String,
String> properties) {
CatalogProperties.UNIQUE_TABLE_LOCATION,
CatalogProperties.UNIQUE_TABLE_LOCATION_DEFAULT);
+ if (catalogProperties.containsKey(CatalogProperties.ENCRYPTION_KMS_TYPE) ||
+ catalogProperties.containsKey(CatalogProperties.ENCRYPTION_KMS_IMPL)) {
+ this.keyManagementClient = EncryptionUtil.createKmsClient(properties);
+ }
+
this.clients = new CachedClientPool(conf, properties);
}
@@ -684,7 +693,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
@@ -819,6 +828,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;
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
index 64e1a075f46..49ddf140bf9 100644
---
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
+++
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
@@ -23,8 +23,10 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
@@ -35,9 +37,19 @@
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.iceberg.BaseMetastoreOperations;
import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.LocationProviders;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.encryption.EncryptedKey;
+import org.apache.iceberg.encryption.EncryptingFileIO;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.encryption.EncryptionUtil;
+import org.apache.iceberg.encryption.KeyManagementClient;
+import org.apache.iceberg.encryption.PlaintextEncryptionManager;
+import org.apache.iceberg.encryption.StandardEncryptionManager;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.CommitStateUnknownException;
@@ -45,6 +57,10 @@
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.hadoop.ConfigProperties;
import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.thrift.TException;
import org.slf4j.Logger;
@@ -70,18 +86,28 @@ public class HiveTableOperations extends
BaseMetastoreTableOperations
private final long maxHiveTablePropertySize;
private final int metadataRefreshMaxRetries;
private final FileIO fileIO;
+ private final KeyManagementClient keyManagementClient;
private final ClientPool<IMetaStoreClient, TException> metaClients;
+ private EncryptionManager encryptionManager;
+ private EncryptingFileIO encryptingFileIO;
+ private String tableKeyId;
+ private int encryptionDekLength;
+
+ private List<EncryptedKey> encryptedKeys = List.of();
+
protected HiveTableOperations(
Configuration conf,
ClientPool<IMetaStoreClient, TException> metaClients,
FileIO fileIO,
+ KeyManagementClient keyManagementClient,
String catalogName,
String database,
String table) {
this.conf = conf;
this.metaClients = metaClients;
this.fileIO = fileIO;
+ this.keyManagementClient = keyManagementClient;
this.fullName = catalogName + "." + database + "." + table;
this.catalogName = catalogName;
this.database = database;
@@ -99,7 +125,7 @@ protected HiveTableOperations(
* Used by HiveTransaction to defer HMS updates to coordinator for atomic
batch commits.
*/
public StagingTableOperations toStagingOps() {
- return new StagingTableOperations(conf, metaClients, fileIO, catalogName,
database, tableName);
+ return new StagingTableOperations(conf, metaClients, fileIO,
keyManagementClient, catalogName, database, tableName);
}
@Override
@@ -109,12 +135,49 @@ protected String tableName() {
@Override
public FileIO io() {
- return fileIO;
+ if (tableKeyId == null) {
+ return fileIO;
+ }
+
+ if (encryptingFileIO == null) {
+ encryptingFileIO = EncryptingFileIO.combine(fileIO, encryption());
+ }
+
+ return encryptingFileIO;
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ if (encryptionManager != null) {
+ return encryptionManager;
+ }
+
+ if (tableKeyId != null) {
+ Preconditions.checkArgument(
+ keyManagementClient != null,
+ "Cannot create encryption manager without a key management client. "
+
+ "Consider setting the '%s' catalog property",
+ CatalogProperties.ENCRYPTION_KMS_IMPL);
+
+ Map<String, String> encryptionProperties =
+ ImmutableMap.of(
+ TableProperties.ENCRYPTION_TABLE_KEY, tableKeyId,
+ TableProperties.ENCRYPTION_DEK_LENGTH,
String.valueOf(encryptionDekLength));
+ encryptionManager =
+ EncryptionUtil.createEncryptionManager(encryptedKeys,
encryptionProperties, keyManagementClient);
+ } else {
+ return PlaintextEncryptionManager.instance();
+ }
+
+ return encryptionManager;
}
@Override
protected void doRefresh() {
String metadataLocation = null;
+ String tableKeyIdFromHMS = null;
+ String dekLengthFromHMS = null;
+ String metadataHashFromHMS = null;
try {
Table table = metaClients.run(client -> client.getTable(database,
tableName));
@@ -124,7 +187,9 @@ protected void doRefresh() {
HiveOperationsBase.validateTableIsIceberg(table, fullName);
metadataLocation = table.getParameters().get(METADATA_LOCATION_PROP);
-
+ tableKeyIdFromHMS =
table.getParameters().get(TableProperties.ENCRYPTION_TABLE_KEY);
+ dekLengthFromHMS =
table.getParameters().get(TableProperties.ENCRYPTION_DEK_LENGTH);
+ metadataHashFromHMS = table.getParameters().get(METADATA_HASH_PROP);
} catch (NoSuchObjectException e) {
if (currentMetadataLocation() != null) {
throw new NoSuchTableException("No such table: %s.%s", database,
tableName);
@@ -141,21 +206,67 @@ protected void doRefresh() {
}
refreshFromMetadataLocation(metadataLocation, metadataRefreshMaxRetries);
+ if (tableKeyIdFromHMS != null) {
+ tableKeyId = tableKeyIdFromHMS;
+ encryptionDekLength = (dekLengthFromHMS != null) ?
+ Integer.parseInt(dekLengthFromHMS) :
+ TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT;
+
+ if (current() != null) {
+ checkIntegrityForEncryption(tableKeyIdFromHMS, dekLengthFromHMS,
metadataHashFromHMS);
+
+ encryptedKeys =
+ Optional.ofNullable(current().encryptionKeys())
+ .map(Lists::newLinkedList)
+ .orElseGet(Lists::newLinkedList);
+
+ if (encryptionManager != null) {
+ Set<String> keyIdsFromMetadata =
+
encryptedKeys.stream().map(EncryptedKey::keyId).collect(Collectors.toSet());
+
+ for (EncryptedKey keyFromEM :
+ EncryptionUtil.encryptionKeys(encryptionManager).values()) {
+ if (!keyIdsFromMetadata.contains(keyFromEM.keyId())) {
+ encryptedKeys.add(keyFromEM);
+ }
+ }
+ }
+ }
+
+ // Force re-creation of encryption manager with updated keys
+ encryptingFileIO = null;
+ encryptionManager = null;
+ }
}
@SuppressWarnings({"checkstyle:CyclomaticComplexity", "MethodLength"})
@Override
protected void doCommit(TableMetadata base, TableMetadata metadata) {
boolean newTable = base == null;
- String newMetadataLocation = writeNewMetadataIfRequired(newTable,
metadata);
- boolean hiveEngineEnabled = hiveEngineEnabled(metadata, conf);
+ encryptionPropsFromMetadata(metadata.properties());
boolean keepHiveStats = conf.getBoolean(ConfigProperties.KEEP_HIVE_STATS,
false);
+ final TableMetadata tableMetadata;
+ EncryptionManager encrManager = encryption();
+ if (encrManager instanceof StandardEncryptionManager) {
+ // Add new encryption keys to the metadata
+ TableMetadata.Builder builder = TableMetadata.buildFrom(metadata);
+ for (Map.Entry<String, EncryptedKey> entry :
EncryptionUtil.encryptionKeys(encrManager).entrySet()) {
+ builder.addEncryptionKey(entry.getValue());
+ }
+
+ tableMetadata = builder.build();
+ } else {
+ tableMetadata = metadata;
+ }
+
+ boolean hiveEngineEnabled = hiveEngineEnabled(tableMetadata, conf);
+ String newMetadataLocation = writeNewMetadataIfRequired(newTable,
tableMetadata);
BaseMetastoreOperations.CommitStatus commitStatus =
BaseMetastoreOperations.CommitStatus.FAILURE;
boolean updateHiveTable = false;
- HiveLock lock = lockObject(base != null ? base : metadata);
+ HiveLock lock = lockObject(base != null ? base : tableMetadata);
try {
lock.lock();
@@ -175,7 +286,7 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
LOG.debug("Committing existing table: {}", fullName);
} else {
tbl = newHmsTable(
- metadata.property(HiveCatalog.HMS_TABLE_OWNER,
HiveHadoopUtil.currentUser())
+ tableMetadata.property(HiveCatalog.HMS_TABLE_OWNER,
HiveHadoopUtil.currentUser())
);
LOG.debug("Committing new table: {}", fullName);
}
@@ -192,8 +303,8 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
tbl.setSd(
HiveOperationsBase.storageDescriptor(
- metadata.schema(),
- metadata.location(),
+ tableMetadata.schema(),
+ tableMetadata.location(),
hiveEngineEnabled, // set to pickup any schema changes
bucketCols,
numBuckets));
@@ -212,14 +323,24 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
if (base != null) {
removedProps =
base.properties().keySet().stream()
- .filter(key -> !metadata.properties().containsKey(key))
+ .filter(key -> !tableMetadata.properties().containsKey(key))
.collect(Collectors.toSet());
+
+ Preconditions.checkArgument(
+ !removedProps.contains(TableProperties.ENCRYPTION_TABLE_KEY),
+ "Cannot remove key ID from an encrypted table");
+
+ Preconditions.checkArgument(
+ Objects.equals(
+ base.properties().get(TableProperties.ENCRYPTION_TABLE_KEY),
+
metadata.properties().get(TableProperties.ENCRYPTION_TABLE_KEY)),
+ "Cannot modify key ID of an encrypted table");
}
HMSTablePropertyHelper.updateHmsTableForIcebergTable(
newMetadataLocation,
tbl,
- metadata,
+ tableMetadata,
removedProps,
hiveEngineEnabled,
maxHiveTablePropertySize,
@@ -273,7 +394,7 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
// issue for example, and triggers this exception. So we need
double-check to make sure
// this is really a concurrent modification. Hitting this exception
means no pending
// requests, if any, can succeed later, so it's safe to check status
in strict mode
- commitStatus = checkCommitStatusStrict(newMetadataLocation,
metadata);
+ commitStatus = checkCommitStatusStrict(newMetadataLocation,
tableMetadata);
if (commitStatus == BaseMetastoreOperations.CommitStatus.FAILURE) {
throw new CommitFailedException(
e, "The table %s.%s has been modified concurrently", database,
tableName);
@@ -284,7 +405,7 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
database,
tableName,
e);
- commitStatus = checkCommitStatus(newMetadataLocation, metadata);
+ commitStatus = checkCommitStatus(newMetadataLocation, tableMetadata);
}
switch (commitStatus) {
@@ -340,6 +461,52 @@ public ClientPool<IMetaStoreClient, TException>
metaClients() {
return metaClients;
}
+ @Override
+ public TableOperations temp(TableMetadata uncommittedMetadata) {
+ return new TableOperations() {
+ @Override
+ public TableMetadata current() {
+ return uncommittedMetadata;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ throw new UnsupportedOperationException("Cannot call refresh on
temporary table operations");
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ throw new UnsupportedOperationException("Cannot call commit on
temporary table operations");
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ return
HiveTableOperations.this.metadataFileLocation(uncommittedMetadata, fileName);
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ return LocationProviders.locationsFor(uncommittedMetadata.location(),
uncommittedMetadata.properties());
+ }
+
+ @Override
+ public FileIO io() {
+
HiveTableOperations.this.encryptionPropsFromMetadata(uncommittedMetadata.properties());
+ return HiveTableOperations.this.io();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return HiveTableOperations.this.encryption();
+ }
+
+ @Override
+ public long newSnapshotId() {
+ return HiveTableOperations.this.newSnapshotId();
+ }
+ };
+ }
+
/**
* Returns if the hive engine related values should be enabled on the table,
or not.
*
@@ -398,6 +565,56 @@ public static boolean hiveLockEnabled(Map<String, String>
properties, Configurat
ConfigProperties.LOCK_HIVE_ENABLED,
TableProperties.HIVE_LOCK_ENABLED_DEFAULT);
}
+ private void encryptionPropsFromMetadata(Map<String, String>
tableProperties) {
+ if (tableKeyId == null) {
+ tableKeyId = tableProperties.get(TableProperties.ENCRYPTION_TABLE_KEY);
+ }
+
+ if (tableKeyId != null && encryptionDekLength <= 0) {
+ encryptionDekLength =
+ PropertyUtil.propertyAsInt(
+ tableProperties,
+ TableProperties.ENCRYPTION_DEK_LENGTH,
+ TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT);
+ }
+ }
+
+ private void checkIntegrityForEncryption(
+ String encryptionKeyIdFromHMS, String dekLengthFromHMS, String
metadataHashFromHMS) {
+ TableMetadata metadata = current();
+ if (StringUtils.isNotEmpty(metadataHashFromHMS)) {
+ HMSTablePropertyHelper.verifyMetadataHash(metadata, metadataHashFromHMS);
+ return;
+ }
+
+ LOG.warn(
+ "Full metadata integrity check skipped because no metadata hash was
recorded in HMS for table {}." +
+ " Falling back to encryption property based check.",
+ tableName);
+
+ Map<String, String> propertiesFromMetadata = metadata.properties();
+
+ String encryptionKeyIdFromMetadata =
+ propertiesFromMetadata.get(TableProperties.ENCRYPTION_TABLE_KEY);
+ if (!Objects.equals(encryptionKeyIdFromHMS, encryptionKeyIdFromMetadata)) {
+ String errMsg =
+ String.format(
+ "Metadata file might have been modified. Encryption key id %s
differs from HMS value %s",
+ encryptionKeyIdFromMetadata, encryptionKeyIdFromHMS);
+ throw new RuntimeException(errMsg);
+ }
+
+ String dekLengthFromMetadata =
+ propertiesFromMetadata.get(TableProperties.ENCRYPTION_DEK_LENGTH);
+ if (!Objects.equals(dekLengthFromHMS, dekLengthFromMetadata)) {
+ String errMsg =
+ String.format(
+ "Metadata file might have been modified. DEK length %s differs
from HMS value %s",
+ dekLengthFromMetadata, dekLengthFromHMS);
+ throw new RuntimeException(errMsg);
+ }
+ }
+
HiveLock lockObject(TableMetadata metadata) {
if (hiveLockEnabled(metadata, conf)) {
return new MetastoreLock(conf, metaClients, catalogName, database,
tableName);
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/StagingTableOperations.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/StagingTableOperations.java
index c61bde44d65..7715b122677 100644
---
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/StagingTableOperations.java
+++
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/StagingTableOperations.java
@@ -23,6 +23,7 @@
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.iceberg.ClientPool;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.encryption.KeyManagementClient;
import org.apache.iceberg.io.FileIO;
import org.apache.thrift.TException;
@@ -38,10 +39,11 @@ public StagingTableOperations(
Configuration conf,
ClientPool<IMetaStoreClient, TException> metaClients,
FileIO fileIO,
+ KeyManagementClient keyManagementClient,
String catalogName,
String database,
String table) {
- super(conf, metaClients, fileIO, catalogName, database, table);
+ super(conf, metaClients, fileIO, keyManagementClient, catalogName,
database, table);
}
@Override
diff --git
a/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java
b/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java
index d36f41b191e..fb08127683b 100644
---
a/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java
+++
b/iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java
@@ -187,6 +187,7 @@ public void before() throws Exception {
overriddenHiveConf,
spyCachedClientPool,
ops.io(),
+ null,
catalog.name(),
dbName,
tableName));
@@ -608,6 +609,7 @@ public void testNoLockCallsWithNoLock() throws TException {
confWithLock,
spyCachedClientPool,
ops.io(),
+ null,
catalog.name(),
TABLE_IDENTIFIER.namespace().level(0),
TABLE_IDENTIFIER.name()));
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
index 4189b3aea20..bfc7cc1fc5b 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java
@@ -51,15 +51,19 @@
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.data.CachingDeleteLoader;
+import org.apache.iceberg.data.DeleteLoader;
+import org.apache.iceberg.encryption.EncryptingFileIO;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.hive.HiveSchemaUtil;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.mr.InputFormatConfig;
import org.apache.iceberg.mr.hive.HiveIcebergInputFormat;
import org.apache.iceberg.mr.mapred.MapredIcebergInputFormat;
import org.apache.iceberg.orc.VectorizedReadUtils;
-import org.apache.iceberg.parquet.ParquetFooterInputFromCache;
+import org.apache.iceberg.parquet.HiveParquetUtil;
import org.apache.iceberg.parquet.ParquetSchemaUtil;
import org.apache.iceberg.parquet.TypeWithSchemaVisitor;
import org.apache.iceberg.parquet.VariantParquetFilters;
@@ -67,8 +71,6 @@
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
import org.apache.orc.impl.OrcTail;
-import org.apache.parquet.format.converter.ParquetMetadataConverter;
-import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.schema.MessageType;
@@ -91,8 +93,16 @@ public static CloseableIterable<HiveBatchContext>
reader(Table table, Path path,
Schema requiredSchema = readSchema;
if (!task.deletes().isEmpty()) {
- deleteFilter = new HiveDeleteFilter(table.io(), task, table.schema(),
prepareSchemaForDeleteFilter(readSchema),
- context.getConfiguration());
+ deleteFilter =
+ new HiveDeleteFilter(EncryptingFileIO.combine(table.io(),
table.encryption()), task, table.schema(),
+ prepareSchemaForDeleteFilter(readSchema),
context.getConfiguration()) {
+ @Override
+ protected DeleteLoader newDeleteLoader() {
+ return new CachingDeleteLoader(
+ deleteFile -> EncryptingFileIO.combine(table.io(),
table.encryption()).newInputFile(deleteFile),
+ context.getConfiguration());
+ }
+ };
requiredSchema = deleteFilter.requiredSchema();
// TODO: take requiredSchema and adjust readColumnIds below accordingly
for equality delete cases
// and remove below limitation
@@ -166,7 +176,8 @@ public static CloseableIterable<HiveBatchContext>
reader(Table table, Path path,
case PARQUET:
recordReader = parquetRecordReader(job, reporter, task, path, start,
length, fileId,
getInitialColumnDefaults(table.schema().columns()),
- HiveIcebergInputFormat.residualForReaderPruning(task, job));
+ HiveIcebergInputFormat.residualForReaderPruning(task, job),
+ table.io());
break;
default:
throw new UnsupportedOperationException("Vectorized Hive reading
unimplemented for format: " + format);
@@ -242,7 +253,7 @@ private static RecordReader<NullWritable,
VectorizedRowBatch> orcRecordReader(Jo
private static RecordReader<NullWritable, VectorizedRowBatch>
parquetRecordReader(JobConf job, Reporter reporter,
FileScanTask task, Path path, long start, long length, SyntheticFileId
fileId,
- Map<String, Object> initialColumnDefaults, Expression residual) throws
IOException {
+ Map<String, Object> initialColumnDefaults, Expression residual, FileIO
io) throws IOException {
InputSplit split = new FileSplit(path, start, length, job);
VectorizedParquetInputFormat inputFormat = new
VectorizedParquetInputFormat();
@@ -253,9 +264,7 @@ private static RecordReader<NullWritable,
VectorizedRowBatch> parquetRecordReade
footerData = LlapProxy.getIo().getParquetFooterBuffersFromCache(path,
job, fileId);
}
- ParquetMetadata parquetMetadata = footerData != null ?
- ParquetFileReader.readFooter(new
ParquetFooterInputFromCache(footerData), ParquetMetadataConverter.NO_FILTER) :
- ParquetFileReader.readFooter(job, path);
+ ParquetMetadata parquetMetadata = HiveParquetUtil.readFooter(task.file(),
io, job, footerData);
MessageType fileSchema = parquetMetadata.getFileMetaData().getSchema();
ParquetMetadata prunedMetadata =
VariantParquetFilters.pruneVariantRowGroups(parquetMetadata,
fileSchema, residual);
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java
index 96f8705a3f7..635a311a6c0 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java
@@ -52,6 +52,7 @@
import org.apache.iceberg.data.orc.GenericOrcReader;
import org.apache.iceberg.data.parquet.GenericParquetReaders;
import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptingFileIO;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.hive.HiveSchemaUtil;
import org.apache.iceberg.io.CloseableIterable;
@@ -141,7 +142,8 @@ private CloseableIterable<T> open(FileScanTask currentTask,
Schema readSchema) {
DeleteFilter<Record> deletes = new GenericDeleteFilter(table.io(),
currentTask, table.schema(), readSchema) {
@Override
protected DeleteLoader newDeleteLoader() {
- return new CachingDeleteLoader(this::loadInputFile, conf);
+ return new CachingDeleteLoader(
+ deleteFile -> EncryptingFileIO.combine(table.io(),
table.encryption()).newInputFile(deleteFile), conf);
}
};
Schema requiredSchema = deletes.requiredSchema();
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/HiveParquetUtil.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/HiveParquetUtil.java
new file mode 100644
index 00000000000..0b97927e9e4
--- /dev/null
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/HiveParquetUtil.java
@@ -0,0 +1,77 @@
+/*
+ * 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.iceberg.parquet;
+
+import java.io.IOException;
+import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.encryption.NativeEncryptionInputFile;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.util.ByteBuffers;
+import org.apache.parquet.HadoopReadOptions;
+import org.apache.parquet.ParquetReadOptions;
+import org.apache.parquet.crypto.FileDecryptionProperties;
+import org.apache.parquet.format.converter.ParquetMetadataConverter;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetFileWriter;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+
+public class HiveParquetUtil {
+
+ private HiveParquetUtil() {
+ }
+
+ public static ParquetMetadata readFooter(DataFile dataFile, FileIO io,
JobConf job, MemoryBufferOrBuffers footerData)
+ throws IOException {
+
+ InputFile icebergInputFile = io.newInputFile(dataFile);
+ InputFile rawFileToRead = icebergInputFile;
+
+ ParquetReadOptions.Builder optionsBuilder =
+
HadoopReadOptions.builder(job).withMetadataFilter(ParquetMetadataConverter.NO_FILTER);
+
+ if (icebergInputFile instanceof NativeEncryptionInputFile nativeEncFile) {
+ rawFileToRead = nativeEncFile.encryptedInputFile();
+
+ byte[] footerKey =
ByteBuffers.toByteArray(nativeEncFile.keyMetadata().encryptionKey());
+ byte[] aadPrefix =
ByteBuffers.toByteArray(nativeEncFile.keyMetadata().aadPrefix());
+
+ FileDecryptionProperties decryptProps =
+
FileDecryptionProperties.builder().withFooterKey(footerKey).withAADPrefix(aadPrefix).build();
+
+ optionsBuilder.withDecryption(decryptProps);
+ }
+
+ org.apache.parquet.io.InputFile parquetInputFile;
+ if (footerData != null) {
+ byte[] magic = (icebergInputFile instanceof NativeEncryptionInputFile) ?
+ ParquetFileWriter.EFMAGIC : ParquetFileWriter.MAGIC;
+ parquetInputFile = new ParquetFooterInputFromCache(footerData, magic);
+ } else {
+ parquetInputFile = ParquetIO.file(rawFileToRead);
+ }
+
+ try (ParquetFileReader reader = ParquetFileReader.open(parquetInputFile,
optionsBuilder.build())) {
+ return reader.getFooter();
+ }
+ }
+}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/ParquetFooterInputFromCache.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/ParquetFooterInputFromCache.java
index 64eae4ccfaf..adea4cfc06d 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/ParquetFooterInputFromCache.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/ParquetFooterInputFromCache.java
@@ -40,8 +40,6 @@
public final class ParquetFooterInputFromCache
extends SeekableInputStream implements InputFile {
public static final int FOOTER_LENGTH_SIZE = 4; // For the file size check.
- private static final int TAIL_LENGTH = ParquetFileWriter.MAGIC.length +
FOOTER_LENGTH_SIZE;
- private static final int FAKE_PREFIX_LENGTH = ParquetFileWriter.MAGIC.length;
private final int length;
private final int footerLength;
private int position = 0;
@@ -51,27 +49,37 @@ public final class ParquetFooterInputFromCache
private final int[] positions;
public ParquetFooterInputFromCache(MemoryBufferOrBuffers footerData) {
+ this(footerData, ParquetFileWriter.MAGIC);
+ }
+
+ public ParquetFooterInputFromCache(MemoryBufferOrBuffers footerData, byte[]
magic) {
MemoryBuffer oneBuffer = footerData.getSingleBuffer();
+ MemoryBuffer[] footerBuffers;
if (oneBuffer != null) {
- cacheData = new MemoryBuffer[2];
- cacheData[0] = oneBuffer;
+ footerBuffers = new MemoryBuffer[] { oneBuffer };
} else {
- MemoryBuffer[] bufs = footerData.getMultipleBuffers();
- cacheData = new MemoryBuffer[bufs.length + 1];
- System.arraycopy(bufs, 0, cacheData, 0, bufs.length);
+ footerBuffers = footerData.getMultipleBuffers();
}
+
+ // Layout: [prefix magic] [footer data buffers...] [tail: footer length +
magic]
+ cacheData = new MemoryBuffer[1 + footerBuffers.length + 1];
+ cacheData[0] = new ByteArrayBuffer(magic);
+ System.arraycopy(footerBuffers, 0, cacheData, 1, footerBuffers.length);
+
int footerLen = 0;
positions = new int[cacheData.length];
- for (int i = 0; i < cacheData.length - 1; ++i) {
- positions[i] = footerLen;
+ positions[0] = 0;
+ positions[1] = magic.length;
+ for (int i = 1; i <= footerBuffers.length; ++i) {
int dataLen = cacheData[i].getByteBufferRaw().remaining();
assert dataLen > 0;
footerLen += dataLen;
+ positions[i + 1] = magic.length + footerLen;
}
- positions[cacheData.length - 1] = footerLen;
- cacheData[cacheData.length - 1] = new FooterEndBuffer(footerLen);
+
+ cacheData[cacheData.length - 1] = new FooterEndBuffer(footerLen, magic);
this.footerLength = footerLen;
- this.length = footerLen + FAKE_PREFIX_LENGTH + TAIL_LENGTH;
+ this.length = magic.length + footerLen + FOOTER_LENGTH_SIZE + magic.length;
}
@Override
@@ -94,17 +102,16 @@ public long getPos() throws IOException {
@Override
public void seek(long pos) throws IOException {
this.position = (int) pos;
- long targetPos = pos - FAKE_PREFIX_LENGTH;
// Not efficient, but we don't expect this to be called frequently.
for (int i = 1; i <= positions.length; ++i) {
- int endPos = (i == positions.length) ? (length - FAKE_PREFIX_LENGTH) :
positions[i];
- if (endPos > targetPos) {
+ int endPos = (i == positions.length) ? length : positions[i];
+ if (endPos > pos) {
bufferIx = i - 1;
- bufferPos = (int) (targetPos - positions[i - 1]);
+ bufferPos = (int) (pos - positions[i - 1]);
return;
}
}
- throw new IOException("Incorrect seek " + targetPos + "; footer length " +
footerLength +
+ throw new IOException("Incorrect seek " + pos + "; footer length " +
footerLength +
Arrays.toString(positions));
}
@@ -138,6 +145,7 @@ public int readInternal(byte[] bytes, int offset, int len) {
}
argPos += toConsume;
}
+ position += len;
return len;
}
@@ -184,6 +192,27 @@ public void readFully(ByteBuffer arg0) throws IOException {
read(arg0);
}
+ /**
+ * Simple MemoryBuffer backed by a byte array.
+ */
+ private static final class ByteArrayBuffer implements MemoryBuffer {
+ private final ByteBuffer bb;
+
+ ByteArrayBuffer(byte[] data) {
+ bb = ByteBuffer.wrap(data);
+ }
+
+ @Override
+ public ByteBuffer getByteBufferRaw() {
+ return bb;
+ }
+
+ @Override
+ public ByteBuffer getByteBufferDup() {
+ return bb.duplicate();
+ }
+ }
+
/**
* The fake buffer that emulates end of file, with footer length and magic.
Given that these
* can be generated based on the footer buffer itself, we don't cache them.
@@ -191,14 +220,14 @@ public void readFully(ByteBuffer arg0) throws IOException
{
private static final class FooterEndBuffer implements MemoryBuffer {
private final ByteBuffer bb;
- FooterEndBuffer(int footerLength) {
- byte[] bytes = new byte[8];
+ FooterEndBuffer(int footerLength, byte[] magic) {
+ byte[] bytes = new byte[FOOTER_LENGTH_SIZE + magic.length];
bytes[0] = (byte) ((footerLength >>> 0) & 0xFF);
bytes[1] = (byte) ((footerLength >>> 8) & 0xFF);
bytes[2] = (byte) ((footerLength >>> 16) & 0xFF);
bytes[3] = (byte) ((footerLength >>> 24) & 0xFF);
- for (int i = 0; i < ParquetFileWriter.MAGIC.length; ++i) {
- bytes[4 + i] = ParquetFileWriter.MAGIC[i];
+ for (int i = 0; i < magic.length; ++i) {
+ bytes[FOOTER_LENGTH_SIZE + i] = magic[i];
}
bb = ByteBuffer.wrap(bytes);
}
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveCatalogEncryption.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveCatalogEncryption.java
new file mode 100644
index 00000000000..b45df4f53af
--- /dev/null
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveCatalogEncryption.java
@@ -0,0 +1,380 @@
+/*
+ * 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.iceberg.mr.hive;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.StreamSupport;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.ChecksumFileSystem;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.Ciphers;
+import org.apache.iceberg.encryption.UnitestKMS;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.mr.hive.test.TestTables.TestTableType;
+import
org.apache.iceberg.mr.hive.test.utils.HiveIcebergStorageHandlerTestUtils;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.crypto.ParquetCryptoRuntimeException;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runners.Parameterized.Parameters;
+
+import static org.apache.iceberg.Files.localInput;
+
+public class TestHiveCatalogEncryption extends
HiveIcebergStorageHandlerWithEngineBase {
+
+ private String encryptionProps() {
+ return String.format(
+ "TBLPROPERTIES ('%s'='%s', '%s'='3', '%s'='%s')",
+ TableProperties.ENCRYPTION_TABLE_KEY, UnitestKMS.MASTER_KEY_NAME1,
+ TableProperties.FORMAT_VERSION,
+ TableProperties.DEFAULT_FILE_FORMAT, fileFormat.toString());
+ }
+
+ @Parameters(name = "fileFormat={0}, catalog={1}, isVectorized={2},
formatVersion={3}")
+ public static Collection<Object[]> parameters() {
+ return HiveIcebergStorageHandlerWithEngineBase.getParameters(p ->
+ p.testTableType() == TestTableType.HIVE_CATALOG && p.formatVersion()
== 3 &&
+ // Iceberg has no native ORC encryption
(org.apache.iceberg.orc.ORC#writeData throws), so ORC is excluded.
+ p.fileFormat() != FileFormat.ORC);
+ }
+
+ @BeforeClass
+ public static void beforeClass() {
+ shell = HiveIcebergStorageHandlerTestUtils.shell(
+ ImmutableMap.of("iceberg.catalog.default_iceberg.encryption.kms-impl",
UnitestKMS.class.getCanonicalName()));
+ }
+
+ @Before
+ public void configureEncryption() {
+
shell.getHiveConf().set("iceberg.catalog.default_iceberg.encryption.kms-impl",
UnitestKMS.class.getCanonicalName());
+ }
+
+ @Test
+ public void testWriteAndReadEncryptedTable() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_v3_table");
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint, data string) STORED BY iceberg
%s",
+ identifier.name(), encryptionProps()));
+
+ // Insert initial set of rows
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd')",
identifier.name()));
+
+ // Verify all rows were inserted correctly
+ List<Object[]> rows = shell.executeStatement("SELECT * FROM " +
identifier.name() + " ORDER BY id");
+ Assert.assertEquals("Should have 4 rows after insert", 4, rows.size());
+ Assert.assertEquals(1L, rows.get(0)[0]);
+ Assert.assertEquals("a", rows.get(0)[1]);
+ Assert.assertEquals(4L, rows.get(3)[0]);
+ Assert.assertEquals("d", rows.get(3)[1]);
+
+ // Perform the DELETE operation
+ shell.executeStatement(String.format(
+ "DELETE FROM %s WHERE id < 3", identifier.name()));
+
+ // Verify that the rows were actually deleted
+ List<Object[]> rowsAfterDelete = shell.executeStatement("SELECT * FROM " +
identifier.name() + " ORDER BY id");
+ Assert.assertEquals("Should have 2 rows remaining after delete", 2,
rowsAfterDelete.size());
+ Assert.assertEquals(3L, rowsAfterDelete.get(0)[0]);
+ Assert.assertEquals("c", rowsAfterDelete.get(0)[1]);
+ Assert.assertEquals(4L, rowsAfterDelete.get(1)[0]);
+ Assert.assertEquals("d", rowsAfterDelete.get(1)[1]);
+ }
+
+ @Test
+ public void testManifestEncryption() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"manifest_check");
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint) STORED BY iceberg %s",
+ identifier.name(), encryptionProps()));
+ shell.executeStatement("INSERT INTO " + identifier.name() + " VALUES (1)");
+
+ Table table = testTables.loadTable(identifier);
+ Assert.assertEquals(3, ((BaseTable)
table).operations().current().formatVersion());
+
+ table.currentSnapshot().allManifests(table.io()).forEach(manifest -> {
+ try (SeekableInputStream stream =
localInput(manifest.path()).newStream()) {
+ byte[] magic = new byte[4];
+ stream.read(magic);
+ Assert.assertEquals("Should have GCM magic",
+ Ciphers.GCM_STREAM_MAGIC_STRING, new String(magic,
StandardCharsets.UTF_8));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ @Test
+ public void testAppendTransaction() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_append_transaction");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint, data string) STORED BY iceberg
%s",
+ identifier.name(), encryptionProps()));
+
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (1, 'a')", identifier.name()));
+
+ Table table = testTables.loadTable(identifier);
+ org.apache.iceberg.DataFile dataFile =
table.newScan().planFiles().iterator().next().file();
+
+ long initialFiles =
StreamSupport.stream(table.newScan().planFiles().spliterator(), false).count();
+
+ // Start a transaction and append the data file
+ org.apache.iceberg.Transaction transaction = table.newTransaction();
+ org.apache.iceberg.AppendFiles append = transaction.newAppend();
+ append.appendFile(dataFile);
+ append.commit();
+ transaction.commitTransaction();
+
+ table.refresh();
+ long finalFiles =
StreamSupport.stream(table.newScan().planFiles().spliterator(), false).count();
+
+ Assert.assertEquals("Should have added 1 data file via transaction",
+ initialFiles + 1, finalFiles);
+ }
+
+ @Test
+ public void testConcurrentAppendTransactions() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_concurrent_append");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint, data string) STORED BY iceberg
%s",
+ identifier.name(), encryptionProps()));
+
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (1, 'a')", identifier.name()));
+
+ Table table = testTables.loadTable(identifier);
+ org.apache.iceberg.DataFile dataFile =
table.newScan().planFiles().iterator().next().file();
+
+ long initialFiles = java.util.stream.StreamSupport
+ .stream(table.newScan().planFiles().spliterator(), false).count();
+
+ // Start the first transaction
+ Transaction transaction = table.newTransaction();
+ AppendFiles append = transaction.newAppend();
+ append.appendFile(dataFile);
+
+ // Concurrently append to the table before committing the first transaction
+ // using a separate table instance load
+
testTables.loadTable(identifier).newFastAppend().appendFile(dataFile).commit();
+
+ // Now commit the first transaction
+ append.commit();
+ transaction.commitTransaction();
+
+ table.refresh();
+ long finalFiles =
StreamSupport.stream(table.newScan().planFiles().spliterator(), false).count();
+
+ Assert.assertEquals("Should have added 2 data files total via concurrent
transactions",
+ initialFiles + 2, finalFiles);
+ }
+
+ @Test
+ public void testConcurrentReplaceTransactions() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_concurrent_replace");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint, data string) STORED BY iceberg
%s",
+ identifier.name(), encryptionProps()));
+
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (1, 'a')", identifier.name()));
+
+ Table table = testTables.loadTable(identifier);
+ DataFile dataFile = table.newScan().planFiles().iterator().next().file();
+
+ Transaction firstReplace = table.newTransaction();
+ firstReplace.newOverwrite()
+ .overwriteByRowFilter(Expressions.alwaysTrue())
+ .addFile(dataFile)
+ .commit();
+
+ org.apache.iceberg.Transaction secondReplace = table.newTransaction();
+ secondReplace.newOverwrite()
+ .overwriteByRowFilter(Expressions.alwaysTrue())
+ .addFile(dataFile)
+ .commit();
+
+ // Commit the first replacement
+ firstReplace.commitTransaction();
+
+ // Commit the second replacement
+ secondReplace.commitTransaction();
+
+ table.refresh();
+ long finalFiles =
StreamSupport.stream(table.newScan().planFiles().spliterator(), false).count();
+
+ // Since both were full table replace/overwrites with a single file, the
final count should be exactly 1
+ Assert.assertEquals("Should only have 1 data file after concurrent replace
transactions",
+ 1, finalFiles);
+ }
+
+ @Test
+ public void testRefresh() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_refresh_table");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint, data string) STORED BY iceberg
%s",
+ identifier.name(), encryptionProps()));
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (1, 'a')", identifier.name()));
+
+ // Load the table and check current snapshot
+ Table table = testTables.loadTable(identifier);
+ long snapshotIdBefore = table.currentSnapshot().snapshotId();
+
+ // Insert more data via Hive
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (2, 'b')", identifier.name()));
+
+ // Refresh the table object and verify the snapshot has updated
+ table.refresh();
+ long snapshotIdAfter = table.currentSnapshot().snapshotId();
+
+ Assert.assertNotEquals("Snapshot ID should change after insert and
refresh",
+ snapshotIdBefore, snapshotIdAfter);
+ }
+
+ @Test
+ public void testMetadataTamperproofing() throws IOException {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_tamperproof_table");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint, data string) STORED BY iceberg
%s",
+ identifier.name(), encryptionProps()));
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (1, 'a')", identifier.name()));
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (2, 'b')", identifier.name()));
+
+ Table table = testTables.loadTable(identifier);
+ TableMetadata metadata = ((BaseTable) table).operations().current();
+
+ Path currentMetadataPath = new Path(metadata.metadataFileLocation());
+ Path previousMetadataPath = new
Path(metadata.previousFiles().get(0).file());
+
+ ChecksumFileSystem fs = (ChecksumFileSystem) FileSystem.newInstance(new
Configuration());
+
+ // Tamper: replace current metadata file with a previous version
+ Path checksumFile = fs.getChecksumFile(currentMetadataPath);
+ fs.delete(checksumFile, false);
+ fs.delete(currentMetadataPath, false);
+ fs.rename(previousMetadataPath, currentMetadataPath);
+
+ AssertHelpers.assertThrows("Should detect metadata tampering",
+ RuntimeException.class,
+ "might have been modified. Hash of metadata loaded from storage
differs from HMS-stored metadata hash",
+ () -> testTables.loadTable(identifier));
+ }
+
+ @Test
+ public void testKeyDelete() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_key_delete_table");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint) STORED BY iceberg %s",
+ identifier.name(), encryptionProps()));
+
+ AssertHelpers.assertThrows("Should not allow removing encryption key",
+ IllegalArgumentException.class,
+ "Cannot remove key ID from an encrypted table",
+ () -> shell.executeStatement(String.format(
+ "ALTER TABLE %s UNSET TBLPROPERTIES ('%s')",
+ identifier.name(), TableProperties.ENCRYPTION_TABLE_KEY)));
+ }
+
+ @Test
+ public void testKeyAlter() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_key_alter_table");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id bigint) STORED BY iceberg %s",
+ identifier.name(), encryptionProps()));
+
+ AssertHelpers.assertThrows("Should not allow modifying encryption key",
+ IllegalArgumentException.class,
+ "Cannot modify key ID of an encrypted table",
+ () -> shell.executeStatement(String.format(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s'='abcd')",
+ identifier.name(), TableProperties.ENCRYPTION_TABLE_KEY)));
+ }
+
+ @Test
+ public void testDirectDataFileRead() {
+ // Reads the raw data file and asserts a Parquet encrypted-footer
exception, so it is Parquet-specific.
+ Assume.assumeTrue(fileFormat == FileFormat.PARQUET);
+ TableIdentifier identifier = TableIdentifier.of("default",
"encrypted_direct_read_table");
+
+ shell.executeStatement(String.format(
+ "CREATE EXTERNAL TABLE %s (id int, data string) STORED BY iceberg %s",
+ identifier.name(), encryptionProps()));
+ shell.executeStatement(String.format(
+ "INSERT INTO %s VALUES (1, 'a')", identifier.name()));
+
+ // Load table and extract data files via the Iceberg Java API
+ Table table = testTables.loadTable(identifier);
+ List<DataFile> dataFiles = Lists.newArrayList();
+ table.newScan().planFiles().forEach(task -> dataFiles.add(task.file()));
+
+ Assert.assertFalse("Data files should not be empty", dataFiles.isEmpty());
+
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.IntegerType.get()));
+
+ // Attempt to read the encrypted Parquet files directly without keys
+ for (DataFile dataFile : dataFiles) {
+ try {
+ Parquet.read(localInput(dataFile.path().toString()))
+ .project(schema)
+ .callInit()
+ .build()
+ .iterator()
+ .next();
+ Assert.fail("Should have thrown an exception when reading encrypted
file directly");
+ } catch (Exception e) {
+ Assert.assertTrue("Exception should be related to encrypted footer",
+ e instanceof ParquetCryptoRuntimeException ||
+ String.valueOf(e).contains("Trying to read file with encrypted
footer. No keys available"));
+ }
+ }
+ }
+}
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/TestTables.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/TestTables.java
index 9d073aa81f1..23771129e86 100644
---
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/TestTables.java
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/TestTables.java
@@ -46,6 +46,7 @@
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.data.Record;
+import org.apache.iceberg.encryption.UnitestKMS;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.hadoop.HadoopTables;
@@ -626,7 +627,8 @@ static class HiveTestTables extends TestTables {
HiveTestTables(Configuration conf, TemporaryFolder temp, String
catalogName) {
super(CatalogUtil.loadCatalog(HiveCatalog.class.getName(),
CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE,
- ImmutableMap.of(), conf), temp, catalogName);
+ ImmutableMap.of(CatalogProperties.ENCRYPTION_KMS_IMPL,
UnitestKMS.class.getCanonicalName()), conf), temp,
+ catalogName);
}
@Override
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveVectorizedReader.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveVectorizedReader.java
index c73cc0f332d..91a77404eed 100644
---
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveVectorizedReader.java
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveVectorizedReader.java
@@ -40,7 +40,7 @@
import
org.apache.iceberg.mr.mapred.MapredIcebergInputFormat.CompatibilityTaskAttemptContextImpl;
import org.apache.iceberg.mr.mapreduce.IcebergInputFormat;
import org.apache.iceberg.types.Types;
-import org.apache.parquet.format.converter.ParquetMetadataConverter;
+import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.io.InputFile;
import org.junit.Assert;
@@ -103,8 +103,8 @@ public void testRecordReaderShouldReuseFooter() throws
IOException, InterruptedE
r.nextKeyValue();
}
}
- mockedParquetFileReader.verify(() ->
ParquetFileReader.readFooter(any(InputFile.class),
- any(ParquetMetadataConverter.MetadataFilter.class)), times(1)
+ mockedParquetFileReader.verify(() ->
ParquetFileReader.open(any(InputFile.class),
+ any(ParquetReadOptions.class)), times(1)
);
}
}
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/parquet/TestParquetFooterInputFromCache.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/parquet/TestParquetFooterInputFromCache.java
new file mode 100644
index 00000000000..ca79fd948ed
--- /dev/null
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/parquet/TestParquetFooterInputFromCache.java
@@ -0,0 +1,216 @@
+/*
+ * 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.iceberg.parquet;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer;
+import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers;
+import org.apache.parquet.hadoop.ParquetFileWriter;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestParquetFooterInputFromCache {
+
+ @Test
+ public void testUnencryptedMagicAtPrefixAndTail() throws IOException {
+ byte[] footerBytes = new byte[] {10, 20, 30, 40, 50};
+ MemoryBufferOrBuffers footerData = singleBuffer(footerBytes);
+
+ ParquetFooterInputFromCache input = new
ParquetFooterInputFromCache(footerData);
+
+ // Total length: prefix(4) + footer(5) + tail(4 + 4) = 17
+ Assert.assertEquals(17, input.getLength());
+
+ // Read prefix: should be PAR1
+ input.seek(0);
+ byte[] prefix = new byte[4];
+ input.readFully(prefix);
+ Assert.assertArrayEquals(ParquetFileWriter.MAGIC, prefix);
+
+ // Read tail magic: last 4 bytes should be PAR1
+ input.seek(input.getLength() - 4);
+ byte[] tailMagic = new byte[4];
+ input.readFully(tailMagic);
+ Assert.assertArrayEquals(ParquetFileWriter.MAGIC, tailMagic);
+
+ // Read footer length from tail: 4 bytes before the tail magic
+ input.seek(input.getLength() - 8);
+ byte[] lenBytes = new byte[4];
+ input.readFully(lenBytes);
+ int footerLength = (lenBytes[0] & 0xFF) | ((lenBytes[1] & 0xFF) << 8) |
+ ((lenBytes[2] & 0xFF) << 16) | ((lenBytes[3] & 0xFF) << 24);
+ Assert.assertEquals(5, footerLength);
+ }
+
+ @Test
+ public void testEncryptedMagicAtPrefixAndTail() throws IOException {
+ byte[] footerBytes = new byte[] {1, 2, 3, 4, 5, 6, 7};
+ MemoryBufferOrBuffers footerData = singleBuffer(footerBytes);
+
+ ParquetFooterInputFromCache input =
+ new ParquetFooterInputFromCache(footerData, ParquetFileWriter.EFMAGIC);
+
+ // Total length: prefix(4) + footer(7) + tail(4 + 4) = 19
+ Assert.assertEquals(19, input.getLength());
+
+ // Read prefix: should be PARE
+ input.seek(0);
+ byte[] prefix = new byte[4];
+ input.readFully(prefix);
+ Assert.assertArrayEquals(ParquetFileWriter.EFMAGIC, prefix);
+
+ // Read tail magic: last 4 bytes should be PARE
+ input.seek(input.getLength() - 4);
+ byte[] tailMagic = new byte[4];
+ input.readFully(tailMagic);
+ Assert.assertArrayEquals(ParquetFileWriter.EFMAGIC, tailMagic);
+
+ // Read footer length from tail
+ input.seek(input.getLength() - 8);
+ byte[] lenBytes = new byte[4];
+ input.readFully(lenBytes);
+ int footerLength = (lenBytes[0] & 0xFF) | ((lenBytes[1] & 0xFF) << 8) |
+ ((lenBytes[2] & 0xFF) << 16) | ((lenBytes[3] & 0xFF) << 24);
+ Assert.assertEquals(7, footerLength);
+ }
+
+ @Test
+ public void testFooterDataReadCorrectly() throws IOException {
+ byte[] footerBytes = new byte[] {0x11, 0x22, 0x33, 0x44, 0x55};
+ MemoryBufferOrBuffers footerData = singleBuffer(footerBytes);
+
+ ParquetFooterInputFromCache input =
+ new ParquetFooterInputFromCache(footerData, ParquetFileWriter.EFMAGIC);
+
+ // Footer data starts after prefix (4 bytes)
+ input.seek(4);
+ byte[] readBack = new byte[5];
+ input.readFully(readBack);
+ Assert.assertArrayEquals(footerBytes, readBack);
+ }
+
+ @Test
+ public void testSequentialReadAcrossAllRegions() throws IOException {
+ byte[] footerBytes = new byte[] {(byte) 0xAA, (byte) 0xBB};
+ MemoryBufferOrBuffers footerData = singleBuffer(footerBytes);
+
+ ParquetFooterInputFromCache input =
+ new ParquetFooterInputFromCache(footerData, ParquetFileWriter.EFMAGIC);
+
+ // Length: prefix(4) + footer(2) + tail(8) = 14
+ Assert.assertEquals(14, input.getLength());
+
+ // Read entire content sequentially from position 0
+ input.seek(0);
+ byte[] all = new byte[14];
+ input.readFully(all);
+
+ // Verify prefix: PARE
+ Assert.assertEquals('P', all[0]);
+ Assert.assertEquals('A', all[1]);
+ Assert.assertEquals('R', all[2]);
+ Assert.assertEquals('E', all[3]);
+
+ // Verify footer data
+ Assert.assertEquals((byte) 0xAA, all[4]);
+ Assert.assertEquals((byte) 0xBB, all[5]);
+
+ // Verify tail: footer length (2, little-endian) + PARE
+ Assert.assertEquals(2, all[6]);
+ Assert.assertEquals(0, all[7]);
+ Assert.assertEquals(0, all[8]);
+ Assert.assertEquals(0, all[9]);
+ Assert.assertEquals('P', all[10]);
+ Assert.assertEquals('A', all[11]);
+ Assert.assertEquals('R', all[12]);
+ Assert.assertEquals('E', all[13]);
+ }
+
+ @Test
+ public void testMultipleBuffers() throws IOException {
+ byte[] part1 = new byte[] {1, 2, 3};
+ byte[] part2 = new byte[] {4, 5};
+ MemoryBufferOrBuffers footerData = multipleBuffers(part1, part2);
+
+ ParquetFooterInputFromCache input =
+ new ParquetFooterInputFromCache(footerData, ParquetFileWriter.MAGIC);
+
+ // Length: prefix(4) + footer(5) + tail(8) = 17
+ Assert.assertEquals(17, input.getLength());
+
+ // Read footer region which spans two buffers
+ input.seek(4);
+ byte[] footer = new byte[5];
+ input.readFully(footer);
+ Assert.assertArrayEquals(new byte[] {1, 2, 3, 4, 5}, footer);
+ }
+
+ private static MemoryBufferOrBuffers singleBuffer(byte[] data) {
+ MemoryBuffer buf = new TestMemoryBuffer(data);
+ return new MemoryBufferOrBuffers() {
+ @Override
+ public MemoryBuffer getSingleBuffer() {
+ return buf;
+ }
+
+ @Override
+ public MemoryBuffer[] getMultipleBuffers() {
+ return null;
+ }
+ };
+ }
+
+ private static MemoryBufferOrBuffers multipleBuffers(byte[]... parts) {
+ MemoryBuffer[] bufs = new MemoryBuffer[parts.length];
+ for (int i = 0; i < parts.length; i++) {
+ bufs[i] = new TestMemoryBuffer(parts[i]);
+ }
+ return new MemoryBufferOrBuffers() {
+ @Override
+ public MemoryBuffer getSingleBuffer() {
+ return null;
+ }
+
+ @Override
+ public MemoryBuffer[] getMultipleBuffers() {
+ return bufs;
+ }
+ };
+ }
+
+ private static class TestMemoryBuffer implements MemoryBuffer {
+ private final ByteBuffer bb;
+
+ TestMemoryBuffer(byte[] data) {
+ bb = ByteBuffer.wrap(data);
+ }
+
+ @Override
+ public ByteBuffer getByteBufferRaw() {
+ return bb;
+ }
+
+ @Override
+ public ByteBuffer getByteBufferDup() {
+ return bb.duplicate();
+ }
+ }
+}
diff --git
a/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/DVUtil.java
b/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/DVUtil.java
new file mode 100644
index 00000000000..5f67d57e540
--- /dev/null
+++ b/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/DVUtil.java
@@ -0,0 +1,214 @@
+/*
+ * 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.iceberg;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.Deletes;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptingFileIO;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.Tasks;
+
+class DVUtil {
+ private DVUtil() {
+ }
+
+ static PositionDeleteIndex readDV(DeleteFile deleteFile, FileIO fileIO) {
+ Preconditions.checkArgument(
+ ContentFileUtil.isDV(deleteFile),
+ "Cannot read, not a deletion vector: %s",
+ deleteFile.location());
+ InputFile inputFile = fileIO.newInputFile(deleteFile);
+ long offset = deleteFile.contentOffset();
+ int length = deleteFile.contentSizeInBytes().intValue();
+ byte[] bytes = new byte[length];
+ try {
+ IOUtil.readFully(inputFile, offset, bytes, 0, length);
+ return PositionDeleteIndex.deserialize(bytes, deleteFile);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /**
+ * Merges duplicate DVs for the same data file and writes the merged DV
Puffin files. If there is
+ * exactly 1 DV for a given data file then it is return as is
+ *
+ * @param dvsByReferencedFile map of data file location to DVs
+ * @param mergedOutputLocation output location of the merged DVs
+ * @param fileIO fileIO to use when reading and writing
+ * @param specs partition specs
+ * @param pool executor service for reading DVs
+ * @return a list containing both any newly merged DVs and any DVs that are
already valid
+ */
+ static List<DeleteFile> mergeAndWriteDVsIfRequired(
+ Map<String, List<DeleteFile>> dvsByReferencedFile,
+ String mergedOutputLocation,
+ FileIO fileIO,
+ Map<Integer, PartitionSpec> specs,
+ ExecutorService pool) {
+ List<DeleteFile> finalDVs = Lists.newArrayList();
+ Multimap<String, DeleteFile> duplicates =
+ Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList);
+ Map<String, Pair<PartitionSpec, StructLike>> partitions =
Maps.newHashMap();
+
+ for (Map.Entry<String, List<DeleteFile>> entry :
dvsByReferencedFile.entrySet()) {
+ if (entry.getValue().size() > 1) {
+ duplicates.putAll(entry.getKey(), entry.getValue());
+ DeleteFile first = entry.getValue().get(0);
+ partitions.put(entry.getKey(), Pair.of(specs.get(first.specId()),
first.partition()));
+ } else {
+ finalDVs.addAll(entry.getValue());
+ }
+ }
+
+ if (duplicates.isEmpty()) {
+ return finalDVs;
+ }
+
+ validateCanMerge(duplicates, partitions);
+
+ Map<String, PositionDeleteIndex> deletes =
+ readAndMergeDVs(duplicates.values().toArray(DeleteFile[]::new),
fileIO, pool);
+
+ finalDVs.addAll(writeDVs(deletes, fileIO, mergedOutputLocation,
partitions));
+ return finalDVs;
+ }
+
+ private static void validateCanMerge(
+ Multimap<String, DeleteFile> duplicates,
+ Map<String, Pair<PartitionSpec, StructLike>> partitions) {
+ Map<Integer, Comparator<StructLike>> comparatorsBySpecId =
Maps.newHashMap();
+ for (Map.Entry<String, Collection<DeleteFile>> entry :
duplicates.asMap().entrySet()) {
+ String referencedFile = entry.getKey();
+
+ // validate that each file matches the expected partition
+ Pair<PartitionSpec, StructLike> partition =
partitions.get(referencedFile);
+ Long sequenceNumber = Iterables.getFirst(entry.getValue(),
null).dataSequenceNumber();
+ PartitionSpec spec = partition.first();
+ StructLike tuple = partition.second();
+ Comparator<StructLike> comparator =
+ comparatorsBySpecId.computeIfAbsent(
+ spec.specId(), id -> Comparators.forType(spec.partitionType()));
+
+ for (DeleteFile dv : entry.getValue()) {
+ Preconditions.checkArgument(
+ Objects.equals(sequenceNumber, dv.dataSequenceNumber()),
+ "Cannot merge DVs, mismatched sequence numbers (%s, %s) for %s",
+ sequenceNumber,
+ dv.dataSequenceNumber(),
+ referencedFile);
+
+ Preconditions.checkArgument(
+ spec.specId() == dv.specId(),
+ "Cannot merge DVs, mismatched partition specs (%s, %s) for %s",
+ spec.specId(),
+ dv.specId(),
+ referencedFile);
+
+ Preconditions.checkArgument(
+ comparator.compare(tuple, dv.partition()) == 0,
+ "Cannot merge DVs, mismatched partition tuples (%s, %s) for %s",
+ tuple,
+ dv.partition(),
+ referencedFile);
+ }
+ }
+ }
+
+ /**
+ * Reads all DVs, and merge the position indices per referenced data file
+ *
+ * @param duplicateDVs list of dvs to read and merge
+ * @param io the FileIO to use for reading DV files
+ * @param pool executor service for reading DVs
+ * @return map of referenced data file location to the merged position
delete index
+ */
+ private static Map<String, PositionDeleteIndex> readAndMergeDVs(
+ DeleteFile[] duplicateDVs, FileIO io, ExecutorService pool) {
+ // Read all duplicate DVs in parallel
+ PositionDeleteIndex[] duplicatedDVPositions = new
PositionDeleteIndex[duplicateDVs.length];
+ Tasks.range(duplicatedDVPositions.length)
+ .executeWith(pool)
+ .stopOnFailure()
+ .throwFailureWhenFinished()
+ .run(i -> duplicatedDVPositions[i] = readDV(duplicateDVs[i], io));
+
+ Map<String, PositionDeleteIndex> mergedDVs = Maps.newHashMap();
+ for (int i = 0; i < duplicatedDVPositions.length; i++) {
+ DeleteFile dv = duplicateDVs[i];
+ PositionDeleteIndex previousDV =
mergedDVs.get(duplicateDVs[i].referencedDataFile());
+ if (previousDV != null) {
+ previousDV.merge(duplicatedDVPositions[i]);
+ } else {
+ mergedDVs.put(dv.referencedDataFile(), duplicatedDVPositions[i]);
+ }
+ }
+
+ return mergedDVs;
+ }
+
+ // Produces a single Puffin file containing the merged DVs
+ private static List<DeleteFile> writeDVs(
+ Map<String, PositionDeleteIndex> mergedIndexByFile,
+ FileIO fileIO,
+ String dvOutputLocation,
+ Map<String, Pair<PartitionSpec, StructLike>> partitions) {
+ EncryptedOutputFile dvOutputFile =
+ fileIO instanceof EncryptingFileIO encryptingFileIO ?
+ encryptingFileIO.newEncryptingOutputFile(dvOutputLocation) :
+
EncryptedFiles.plainAsEncryptedOutput(fileIO.newOutputFile(dvOutputLocation));
+
+ try (DVFileWriter dvFileWriter = Deletes.writeDVs(dvOutputFile, path ->
null)) {
+ for (Map.Entry<String, PositionDeleteIndex> entry :
mergedIndexByFile.entrySet()) {
+ String referencedLocation = entry.getKey();
+ PositionDeleteIndex mergedPositions = entry.getValue();
+ Pair<PartitionSpec, StructLike> partition =
partitions.get(referencedLocation);
+ dvFileWriter.delete(
+ referencedLocation, mergedPositions, partition.first(),
partition.second());
+ }
+ dvFileWriter.close();
+ return dvFileWriter.result().deleteFiles();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+}
diff --git
a/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java
b/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java
new file mode 100644
index 00000000000..ab8c85b50eb
--- /dev/null
+++
b/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java
@@ -0,0 +1,240 @@
+/*
+ * 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.iceberg.deletes;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.IcebergBuild;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.encryption.NativeEncryptionKeyMetadata;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.BlobMetadata;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.CharSequenceSet;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.apache.iceberg.util.StructLikeUtil;
+
+public class BaseDVFileWriter implements DVFileWriter {
+
+ private static final String REFERENCED_DATA_FILE_KEY =
"referenced-data-file";
+ private static final String CARDINALITY_KEY = "cardinality";
+
+ private final Supplier<EncryptedOutputFile> dvOutputFile;
+ private final Function<String, PositionDeleteIndex> loadPreviousDeletes;
+ private final Map<String, Deletes> deletesByPath = Maps.newHashMap();
+ private final Map<String, BlobMetadata> blobsByPath = Maps.newHashMap();
+ private DeleteWriteResult result = null;
+
+ public BaseDVFileWriter(
+ OutputFileFactory fileFactory, Function<String, PositionDeleteIndex>
loadPreviousDeletes) {
+ this(loadPreviousDeletes, fileFactory::newOutputFile);
+ }
+
+ public BaseDVFileWriter(
+ Supplier<OutputFile> dvOutputFile,
+ Function<String, PositionDeleteIndex> loadPreviousDeletes) {
+ this(loadPreviousDeletes, () ->
EncryptedFiles.plainAsEncryptedOutput(dvOutputFile.get()));
+ }
+
+ BaseDVFileWriter(
+ Function<String, PositionDeleteIndex> loadPreviousDeletes,
+ Supplier<EncryptedOutputFile> dvOutputFile) {
+ this.dvOutputFile = dvOutputFile;
+ this.loadPreviousDeletes = loadPreviousDeletes;
+ }
+
+ @Override
+ public void delete(String path, long pos, PartitionSpec spec, StructLike
partition) {
+ Deletes deletes =
+ deletesByPath.computeIfAbsent(path, key -> new Deletes(path, spec,
partition));
+ PositionDeleteIndex positions = deletes.positions();
+ positions.delete(pos);
+ }
+
+ @Override
+ public void delete(
+ String path,
+ PositionDeleteIndex positionDeleteIndex,
+ PartitionSpec spec,
+ StructLike partition) {
+ Deletes deletes =
+ deletesByPath.computeIfAbsent(path, key -> new Deletes(path, spec,
partition));
+ deletes.positions().merge(positionDeleteIndex);
+ }
+
+ @Override
+ public DeleteWriteResult result() {
+ Preconditions.checkState(result != null, "Cannot get result from unclosed
writer");
+ return result;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (result == null) {
+ List<DeleteFile> dvs = Lists.newArrayList();
+ CharSequenceSet referencedDataFiles = CharSequenceSet.empty();
+ List<DeleteFile> rewrittenDeleteFiles = Lists.newArrayList();
+
+ // Only create PuffinWriter if there are deletes to write
+ if (deletesByPath.isEmpty()) {
+ this.result = new DeleteWriteResult(dvs, referencedDataFiles,
rewrittenDeleteFiles);
+ return;
+ }
+
+ EncryptedOutputFile outputFile = dvOutputFile.get();
+ EncryptionKeyMetadata keyMetadata = outputFile.keyMetadata();
+ PuffinWriter writer =
+ Puffin.write(outputFile.encryptingOutputFile())
+ .createdBy(IcebergBuild.fullVersion())
+ .build();
+
+ try (PuffinWriter closeableWriter = writer) {
+ for (Deletes deletes : deletesByPath.values()) {
+ String path = deletes.path();
+ PositionDeleteIndex positions = deletes.positions();
+ PositionDeleteIndex previousPositions =
loadPreviousDeletes.apply(path);
+ if (previousPositions != null) {
+ positions.merge(previousPositions);
+ for (DeleteFile previousDeleteFile :
previousPositions.deleteFiles()) {
+ // only DVs and file-scoped deletes can be discarded from the
table state
+ if (ContentFileUtil.isFileScoped(previousDeleteFile)) {
+ rewrittenDeleteFiles.add(previousDeleteFile);
+ }
+ }
+ }
+ write(closeableWriter, deletes);
+ referencedDataFiles.add(path);
+ }
+ }
+
+ // DVs share the Puffin path and file size but have different offsets
+ String puffinPath = writer.location();
+ long puffinFileSize = writer.fileSize();
+
+ for (String path : deletesByPath.keySet()) {
+ DeleteFile dv = createDV(puffinPath, puffinFileSize, path,
keyMetadata);
+ dvs.add(dv);
+ }
+
+ this.result = new DeleteWriteResult(dvs, referencedDataFiles,
rewrittenDeleteFiles);
+ }
+ }
+
+ private DeleteFile createDV(
+ String path, long size, String referencedDataFile, EncryptionKeyMetadata
keyMetadata) {
+ Deletes deletes = deletesByPath.get(referencedDataFile);
+ BlobMetadata blobMetadata = blobsByPath.get(referencedDataFile);
+ return FileMetadata.deleteFileBuilder(deletes.spec())
+ .ofPositionDeletes()
+ .withFormat(FileFormat.PUFFIN)
+ .withPath(path)
+ .withPartition(deletes.partition())
+ .withFileSizeInBytes(size)
+ .withEncryptionKeyMetadata(encryptionKeyMetadata(size, keyMetadata))
+ .withReferencedDataFile(referencedDataFile)
+ .withContentOffset(blobMetadata.offset())
+ .withContentSizeInBytes(blobMetadata.length())
+ .withRecordCount(deletes.positions().cardinality())
+ .build();
+ }
+
+ private void write(PuffinWriter writer, Deletes deletes) {
+ String path = deletes.path();
+ PositionDeleteIndex positions = deletes.positions();
+ BlobMetadata blobMetadata = writer.write(toBlob(positions, path));
+ blobsByPath.put(path, blobMetadata);
+ }
+
+ private ByteBuffer encryptionKeyMetadata(
+ long fileSizeInBytes, EncryptionKeyMetadata keyMetadata) {
+ if (keyMetadata instanceof NativeEncryptionKeyMetadata nativeKeyMetadata) {
+ return nativeKeyMetadata.copyWithLength(fileSizeInBytes).buffer();
+ }
+
+ return keyMetadata.buffer();
+ }
+
+ private Blob toBlob(PositionDeleteIndex positions, String path) {
+ return new Blob(
+ StandardBlobTypes.DV_V1,
+ ImmutableList.of(MetadataColumns.ROW_POSITION.fieldId()),
+ -1 /* snapshot ID is inherited */,
+ -1 /* sequence number is inherited */,
+ positions.serialize(),
+ null /* uncompressed */,
+ ImmutableMap.of(
+ REFERENCED_DATA_FILE_KEY,
+ path,
+ CARDINALITY_KEY,
+ String.valueOf(positions.cardinality())));
+ }
+
+ private static class Deletes {
+ private final String path;
+ private final PositionDeleteIndex positions;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private Deletes(String path, PartitionSpec spec, StructLike partition) {
+ this.path = path;
+ this.positions = new BitmapPositionDeleteIndex();
+ this.spec = spec;
+ this.partition = StructLikeUtil.copy(partition);
+ }
+
+ public String path() {
+ return path;
+ }
+
+ public PositionDeleteIndex positions() {
+ return positions;
+ }
+
+ public PartitionSpec spec() {
+ return spec;
+ }
+
+ public StructLike partition() {
+ return partition;
+ }
+ }
+}
diff --git
a/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/Deletes.java
b/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/Deletes.java
new file mode 100644
index 00000000000..127c8046f9e
--- /dev/null
+++
b/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/Deletes.java
@@ -0,0 +1,242 @@
+/*
+ * 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.iceberg.deletes;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import org.apache.iceberg.Accessor;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.CharSequenceMap;
+import org.apache.iceberg.util.Filter;
+import org.apache.iceberg.util.SortedMerge;
+import org.apache.iceberg.util.StructLikeSet;
+
+public class Deletes {
+
+ private static final Schema POSITION_DELETE_SCHEMA =
+ new Schema(MetadataColumns.DELETE_FILE_PATH,
MetadataColumns.DELETE_FILE_POS);
+
+ private static final Accessor<StructLike> FILENAME_ACCESSOR =
+
POSITION_DELETE_SCHEMA.accessorForField(MetadataColumns.DELETE_FILE_PATH.fieldId());
+ private static final Accessor<StructLike> POSITION_ACCESSOR =
+
POSITION_DELETE_SCHEMA.accessorForField(MetadataColumns.DELETE_FILE_POS.fieldId());
+
+ private Deletes() {
+ }
+
+ public static DVFileWriter writeDVs(
+ EncryptedOutputFile outputFile, Function<String, PositionDeleteIndex>
loadPreviousDeletes) {
+ return new BaseDVFileWriter(loadPreviousDeletes, () -> outputFile);
+ }
+
+ public static <T> CloseableIterable<T> filter(
+ CloseableIterable<T> rows, Function<T, StructLike> rowToDeleteKey,
StructLikeSet deleteSet) {
+ if (deleteSet.isEmpty()) {
+ return rows;
+ }
+
+ EqualitySetDeleteFilter<T> equalityFilter =
+ new EqualitySetDeleteFilter<>(rowToDeleteKey, deleteSet);
+ return equalityFilter.filter(rows);
+ }
+
+ /**
+ * Returns the same rows that are input, while marking the deleted ones.
+ *
+ * @param rows the rows to process
+ * @param isDeleted a predicate that determines if a row is deleted
+ * @param deleteMarker a function that marks a row as deleted
+ * @return the processed rows
+ */
+ public static <T> CloseableIterable<T> markDeleted(
+ CloseableIterable<T> rows, Predicate<T> isDeleted, Consumer<T>
deleteMarker) {
+ return CloseableIterable.transform(
+ rows,
+ row -> {
+ if (isDeleted.test(row)) {
+ deleteMarker.accept(row);
+ }
+
+ return row;
+ });
+ }
+
+ /**
+ * Returns the remaining rows (the ones that are not deleted), while
counting the deleted ones.
+ *
+ * @param rows the rows to process
+ * @param isDeleted a predicate that determines if a row is deleted
+ * @param counter a counter that counts deleted rows
+ * @return the processed rows
+ */
+ public static <T> CloseableIterable<T> filterDeleted(
+ CloseableIterable<T> rows, Predicate<T> isDeleted, DeleteCounter
counter) {
+ Filter<T> remainingRowsFilter =
+ new Filter<T>() {
+ @Override
+ protected boolean shouldKeep(T item) {
+ boolean deleted = isDeleted.test(item);
+ if (deleted) {
+ counter.increment();
+ }
+
+ return !deleted;
+ }
+ };
+
+ return remainingRowsFilter.filter(rows);
+ }
+
+ public static StructLikeSet toEqualitySet(
+ CloseableIterable<StructLike> eqDeletes, Types.StructType eqType) {
+ try (CloseableIterable<StructLike> deletes = eqDeletes) {
+ StructLikeSet deleteSet = StructLikeSet.create(eqType);
+ Iterables.addAll(deleteSet, deletes);
+ return deleteSet;
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to close equality delete source",
e);
+ }
+ }
+
+ public static <T extends StructLike> CharSequenceMap<PositionDeleteIndex>
toPositionIndexes(
+ CloseableIterable<T> posDeletes) {
+ return toPositionIndexes(posDeletes, null /* unknown delete file */);
+ }
+
+ /**
+ * Builds a map of position delete indexes by path.
+ *
+ * <p>This method builds a position delete index for each referenced data
file and does not filter
+ * deletes. This can be useful when the entire delete file content is needed
(e.g. caching).
+ *
+ * @param posDeletes position deletes
+ * @param file the source delete file for the deletes
+ * @return the map of position delete indexes by path
+ */
+ public static <T extends StructLike> CharSequenceMap<PositionDeleteIndex>
toPositionIndexes(
+ CloseableIterable<T> posDeletes, DeleteFile file) {
+ CharSequenceMap<PositionDeleteIndex> indexes = CharSequenceMap.create();
+
+ try (CloseableIterable<T> deletes = posDeletes) {
+ for (T delete : deletes) {
+ CharSequence filePath = (CharSequence) FILENAME_ACCESSOR.get(delete);
+ long position = (long) POSITION_ACCESSOR.get(delete);
+ PositionDeleteIndex index =
+ indexes.computeIfAbsent(filePath, key -> new
BitmapPositionDeleteIndex(file));
+ index.delete(position);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to close position delete source",
e);
+ }
+
+ return indexes;
+ }
+
+ public static <T extends StructLike> PositionDeleteIndex toPositionIndex(
+ CharSequence dataLocation, CloseableIterable<T> posDeletes, DeleteFile
file) {
+ CloseableIterable<Long> positions = extractPositions(dataLocation,
posDeletes);
+ List<DeleteFile> files = ImmutableList.of(file);
+ return toPositionIndex(positions, files);
+ }
+
+ private static <T extends StructLike> CloseableIterable<Long>
extractPositions(
+ CharSequence dataLocation, CloseableIterable<T> rows) {
+ DataFileFilter<T> filter = new DataFileFilter<>(dataLocation);
+ CloseableIterable<T> filteredRows = filter.filter(rows);
+ return CloseableIterable.transform(filteredRows, row -> (Long)
POSITION_ACCESSOR.get(row));
+ }
+
+ public static PositionDeleteIndex toPositionIndex(CloseableIterable<Long>
posDeletes) {
+ return toPositionIndex(posDeletes, ImmutableList.of());
+ }
+
+ private static PositionDeleteIndex toPositionIndex(
+ CloseableIterable<Long> posDeletes, List<DeleteFile> files) {
+ try (CloseableIterable<Long> deletes = posDeletes) {
+ PositionDeleteIndex positionDeleteIndex = new
BitmapPositionDeleteIndex(files);
+ deletes.forEach(positionDeleteIndex::delete);
+ return positionDeleteIndex;
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to close position delete source",
e);
+ }
+ }
+
+ public static CloseableIterable<Long> deletePositions(
+ CharSequence dataLocation, CloseableIterable<StructLike> deleteFile) {
+ return deletePositions(dataLocation, ImmutableList.of(deleteFile));
+ }
+
+ public static <T extends StructLike> CloseableIterable<Long> deletePositions(
+ CharSequence dataLocation, List<CloseableIterable<T>> deleteFiles) {
+ DataFileFilter<T> locationFilter = new DataFileFilter<>(dataLocation);
+ List<CloseableIterable<Long>> positions =
+ Lists.transform(
+ deleteFiles,
+ deletes ->
+ CloseableIterable.transform(
+ locationFilter.filter(deletes), row -> (Long)
POSITION_ACCESSOR.get(row)));
+
+ return new SortedMerge<>(Long::compare, positions);
+ }
+
+ private static class EqualitySetDeleteFilter<T> extends Filter<T> {
+ private final StructLikeSet deletes;
+ private final Function<T, StructLike> extractEqStruct;
+
+ protected EqualitySetDeleteFilter(Function<T, StructLike> extractEq,
StructLikeSet deletes) {
+ this.extractEqStruct = extractEq;
+ this.deletes = deletes;
+ }
+
+ @Override
+ protected boolean shouldKeep(T row) {
+ return !deletes.contains(extractEqStruct.apply(row));
+ }
+ }
+
+ private static class DataFileFilter<T extends StructLike> extends Filter<T> {
+ private final CharSequence dataLocation;
+
+ DataFileFilter(CharSequence dataLocation) {
+ this.dataLocation = dataLocation;
+ }
+
+ @Override
+ protected boolean shouldKeep(T posDelete) {
+ return Comparators.filePath()
+ .compare(dataLocation, (CharSequence)
FILENAME_ACCESSOR.get(posDelete)) ==
+ 0;
+ }
+ }
+}
