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 43098c9913a5..cebf11e984ff 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) { @@ -295,4 +304,39 @@ private static void setField( private static boolean exposeInHmsProperties(long maxHiveTablePropertySize) { return maxHiveTablePropertySize > 0; } + + @VisibleForTesting + static void setMetadataHash(TableMetadata metadata, Map 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 58f6ad903ff3..48a3bea889ee 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; @@ -47,6 +48,8 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.encryption.EncryptionUtil; +import org.apache.iceberg.encryption.KeyManagementClient; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchIcebergViewException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; @@ -97,6 +100,7 @@ public class HiveCatalog extends BaseMetastoreViewCatalog private String name; private Configuration conf; private FileIO fileIO; + private KeyManagementClient keyManagementClient; private ClientPool clients; private boolean listAllTables = false; private Map catalogProperties; @@ -136,6 +140,11 @@ public void initialize(String inputName, Map properties) { } else { this.fileIO = CatalogUtil.loadFileIO(fileIOImpl, properties, conf); } + if (catalogProperties.containsKey(CatalogProperties.ENCRYPTION_KMS_TYPE) || + catalogProperties.containsKey(CatalogProperties.ENCRYPTION_KMS_IMPL)) { + this.keyManagementClient = EncryptionUtil.createKmsClient(properties); + } + this.clients = new CachedClientPool(conf, properties); } @@ -677,7 +686,7 @@ private boolean isValidateNamespace(Namespace namespace) { public TableOperations newTableOps(TableIdentifier tableIdentifier) { String dbName = tableIdentifier.namespace().level(0); String tableName = tableIdentifier.name(); - return new HiveTableOperations(conf, clients, fileIO, name, dbName, tableName); + return new HiveTableOperations(conf, clients, fileIO, keyManagementClient, name, dbName, tableName); } @Override @@ -810,6 +819,15 @@ public Map 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 104dd6584080..af93c578beb8 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 metaClients; + private EncryptionManager encryptionManager; + private EncryptingFileIO encryptingFileIO; + private String tableKeyId; + private int encryptionDekLength; + + private List encryptedKeys = List.of(); + protected HiveTableOperations( Configuration conf, ClientPool 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 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,15 +206,61 @@ 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 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 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; @@ -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 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 properties, Configurat ConfigProperties.LOCK_HIVE_ENABLED, TableProperties.HIVE_LOCK_ENABLED_DEFAULT); } + private void encryptionPropsFromMetadata(Map 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 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 c61bde44d654..7715b1226770 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 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 d36f41b191ea..fb08127683b0 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/IcebergTableUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java index 0240135c92b7..e94015da059c 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java @@ -20,6 +20,7 @@ package org.apache.iceberg.mr.hive; import java.io.IOException; +import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.time.ZoneId; import java.util.Collection; @@ -767,7 +768,7 @@ static Map getPartitionStats(Table table, Map pa .filter(evaluator::eval) .transform(IcebergTableUtil::recordToPartitionStats) .stream().reduce((left, right) -> { - left.appendStats(right); + appendStats(left, right); return left; }).orElse(null); } @@ -796,6 +797,28 @@ private static PartitionStats recordToPartitionStats(StructLike record) { return stats; } + private static final Method APPEND_STATS_METHOD; + + static { + try { + APPEND_STATS_METHOD = PartitionStats.class.getDeclaredMethod("appendStats", PartitionStats.class); + APPEND_STATS_METHOD.setAccessible(true); + } catch (NoSuchMethodException e) { + throw new RuntimeException("Failed to access PartitionStats.appendStats", e); + } + } + + /** + * Invokes the package-private {@code PartitionStats.appendStats} via reflection. + */ + private static void appendStats(PartitionStats target, PartitionStats other) { + try { + APPEND_STATS_METHOD.invoke(target, other); + } catch (ReflectiveOperationException e) { + throw new RuntimeException("Failed to invoke PartitionStats.appendStats", e); + } + } + public static PartitionSpec getPartitionSpec(Table icebergTable, String partitionPath) throws MetaException, HiveException { if (icebergTable == null || partitionPath == null || partitionPath.isEmpty()) { 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 4189b3aea209..bfc7cc1fc5ba 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 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 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 orcRecordReader(Jo private static RecordReader parquetRecordReader(JobConf job, Reporter reporter, FileScanTask task, Path path, long start, long length, SyntheticFileId fileId, - Map initialColumnDefaults, Expression residual) throws IOException { + Map 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 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 96f8705a3f7e..635a311a6c00 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 open(FileScanTask currentTask, Schema readSchema) { DeleteFilter 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 000000000000..0b97927e9e42 --- /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 64eae4ccfafb..adea4cfc06d8 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 000000000000..753113c89b23 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveCatalogEncryption.java @@ -0,0 +1,373 @@ +/* + * 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.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 static final String ENCRYPTION_PROPS = String.format( + "TBLPROPERTIES ('%s'='%s', '%s'='3')", + TableProperties.ENCRYPTION_TABLE_KEY, UnitestKMS.MASTER_KEY_NAME1, + TableProperties.FORMAT_VERSION); + + @Parameters(name = "fileFormat={0}, catalog={1}, isVectorized={2}, formatVersion={3}") + public static Collection parameters() { + return HiveIcebergStorageHandlerWithEngineBase.getParameters(p -> + p.testTableType() == TestTableType.HIVE_CATALOG && + p.formatVersion() == 3 && p.isVectorized() && p.fileFormat() == FileFormat.PARQUET); + } + + @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(), ENCRYPTION_PROPS)); + + // 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 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 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(), ENCRYPTION_PROPS)); + 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(), ENCRYPTION_PROPS)); + + 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(), ENCRYPTION_PROPS)); + + 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(), ENCRYPTION_PROPS)); + + 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(), ENCRYPTION_PROPS)); + 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(), ENCRYPTION_PROPS)); + 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(), ENCRYPTION_PROPS)); + + 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(), ENCRYPTION_PROPS)); + + 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() { + 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(), ENCRYPTION_PROPS)); + 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 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 9d073aa81f15..23771129e867 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/parquet/TestParquetFooterInputFromCache.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/parquet/TestParquetFooterInputFromCache.java new file mode 100644 index 000000000000..ca79fd948ed9 --- /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 000000000000..5f67d57e5402 --- /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 mergeAndWriteDVsIfRequired( + Map> dvsByReferencedFile, + String mergedOutputLocation, + FileIO fileIO, + Map specs, + ExecutorService pool) { + List finalDVs = Lists.newArrayList(); + Multimap duplicates = + Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList); + Map> partitions = Maps.newHashMap(); + + for (Map.Entry> 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 deletes = + readAndMergeDVs(duplicates.values().toArray(DeleteFile[]::new), fileIO, pool); + + finalDVs.addAll(writeDVs(deletes, fileIO, mergedOutputLocation, partitions)); + return finalDVs; + } + + private static void validateCanMerge( + Multimap duplicates, + Map> partitions) { + Map> comparatorsBySpecId = Maps.newHashMap(); + for (Map.Entry> entry : duplicates.asMap().entrySet()) { + String referencedFile = entry.getKey(); + + // validate that each file matches the expected partition + Pair partition = partitions.get(referencedFile); + Long sequenceNumber = Iterables.getFirst(entry.getValue(), null).dataSequenceNumber(); + PartitionSpec spec = partition.first(); + StructLike tuple = partition.second(); + Comparator 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 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 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 writeDVs( + Map mergedIndexByFile, + FileIO fileIO, + String dvOutputLocation, + Map> 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 entry : mergedIndexByFile.entrySet()) { + String referencedLocation = entry.getKey(); + PositionDeleteIndex mergedPositions = entry.getValue(); + Pair 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 000000000000..ab8c85b50eb7 --- /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 dvOutputFile; + private final Function loadPreviousDeletes; + private final Map deletesByPath = Maps.newHashMap(); + private final Map blobsByPath = Maps.newHashMap(); + private DeleteWriteResult result = null; + + public BaseDVFileWriter( + OutputFileFactory fileFactory, Function loadPreviousDeletes) { + this(loadPreviousDeletes, fileFactory::newOutputFile); + } + + public BaseDVFileWriter( + Supplier dvOutputFile, + Function loadPreviousDeletes) { + this(loadPreviousDeletes, () -> EncryptedFiles.plainAsEncryptedOutput(dvOutputFile.get())); + } + + BaseDVFileWriter( + Function loadPreviousDeletes, + Supplier 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 dvs = Lists.newArrayList(); + CharSequenceSet referencedDataFiles = CharSequenceSet.empty(); + List 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 000000000000..127c8046f9e5 --- /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 FILENAME_ACCESSOR = + POSITION_DELETE_SCHEMA.accessorForField(MetadataColumns.DELETE_FILE_PATH.fieldId()); + private static final Accessor POSITION_ACCESSOR = + POSITION_DELETE_SCHEMA.accessorForField(MetadataColumns.DELETE_FILE_POS.fieldId()); + + private Deletes() { + } + + public static DVFileWriter writeDVs( + EncryptedOutputFile outputFile, Function loadPreviousDeletes) { + return new BaseDVFileWriter(loadPreviousDeletes, () -> outputFile); + } + + public static CloseableIterable filter( + CloseableIterable rows, Function rowToDeleteKey, StructLikeSet deleteSet) { + if (deleteSet.isEmpty()) { + return rows; + } + + EqualitySetDeleteFilter 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 CloseableIterable markDeleted( + CloseableIterable rows, Predicate isDeleted, Consumer 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 CloseableIterable filterDeleted( + CloseableIterable rows, Predicate isDeleted, DeleteCounter counter) { + Filter remainingRowsFilter = + new Filter() { + @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 eqDeletes, Types.StructType eqType) { + try (CloseableIterable 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 CharSequenceMap toPositionIndexes( + CloseableIterable posDeletes) { + return toPositionIndexes(posDeletes, null /* unknown delete file */); + } + + /** + * Builds a map of position delete indexes by path. + * + *

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 CharSequenceMap toPositionIndexes( + CloseableIterable posDeletes, DeleteFile file) { + CharSequenceMap indexes = CharSequenceMap.create(); + + try (CloseableIterable 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 PositionDeleteIndex toPositionIndex( + CharSequence dataLocation, CloseableIterable posDeletes, DeleteFile file) { + CloseableIterable positions = extractPositions(dataLocation, posDeletes); + List files = ImmutableList.of(file); + return toPositionIndex(positions, files); + } + + private static CloseableIterable extractPositions( + CharSequence dataLocation, CloseableIterable rows) { + DataFileFilter filter = new DataFileFilter<>(dataLocation); + CloseableIterable filteredRows = filter.filter(rows); + return CloseableIterable.transform(filteredRows, row -> (Long) POSITION_ACCESSOR.get(row)); + } + + public static PositionDeleteIndex toPositionIndex(CloseableIterable posDeletes) { + return toPositionIndex(posDeletes, ImmutableList.of()); + } + + private static PositionDeleteIndex toPositionIndex( + CloseableIterable posDeletes, List files) { + try (CloseableIterable 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 deletePositions( + CharSequence dataLocation, CloseableIterable deleteFile) { + return deletePositions(dataLocation, ImmutableList.of(deleteFile)); + } + + public static CloseableIterable deletePositions( + CharSequence dataLocation, List> deleteFiles) { + DataFileFilter locationFilter = new DataFileFilter<>(dataLocation); + List> 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 extends Filter { + private final StructLikeSet deletes; + private final Function extractEqStruct; + + protected EqualitySetDeleteFilter(Function 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 extends Filter { + 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; + } + } +} diff --git a/iceberg/pom.xml b/iceberg/pom.xml index 3f314174a0a7..26bc249797db 100644 --- a/iceberg/pom.xml +++ b/iceberg/pom.xml @@ -26,7 +26,7 @@ .. . - 1.10.1 + 1.11.0 4.0.3 5.2.0 1.12.0 diff --git a/itests/pom.xml b/itests/pom.xml index 566bd2fde3dd..6750cf069f36 100644 --- a/itests/pom.xml +++ b/itests/pom.xml @@ -26,7 +26,7 @@ .. 2.32.0 - 1.10.1 + 1.11.0 3.27.3 diff --git a/standalone-metastore/metastore-rest-catalog/pom.xml b/standalone-metastore/metastore-rest-catalog/pom.xml index d987f7cce972..09cfd3725c8c 100644 --- a/standalone-metastore/metastore-rest-catalog/pom.xml +++ b/standalone-metastore/metastore-rest-catalog/pom.xml @@ -23,7 +23,7 @@ .. UTF-8 false - 1.10.1 + 1.11.0