This is an automated email from the ASF dual-hosted git repository.
deniskuzZ 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 4e9caba7c02 HIVE-30127: Iceberg: Upgrade iceberg version to 1.12.0
(#6856)
4e9caba7c02 is described below
commit 4e9caba7c02a947a39c60de427acb7189c8c1fe8
Author: Raghav Aggarwal <[email protected]>
AuthorDate: Tue Oct 13 14:08:48 2026 +0530
HIVE-30127: Iceberg: Upgrade iceberg version to 1.12.0 (#6856)
---
.../iceberg/hive/HMSTablePropertyHelper.java | 8 +-
.../java/org/apache/iceberg/hive/HiveCatalog.java | 56 ++++-
.../apache/iceberg/hive/HiveOperationsBase.java | 7 +-
.../apache/iceberg/hive/HiveTableOperations.java | 36 +--
.../apache/iceberg/hive/HiveViewOperations.java | 12 +-
iceberg/iceberg-handler/pom.xml | 4 +
.../apache/iceberg/mr/hive/IcebergAcidUtil.java | 2 +-
.../mr/hive/writer/HiveFileWriterFactory.java | 140 +++++-------
.../writer/HiveIcebergCopyOnWriteRecordWriter.java | 2 +-
.../mr/hive/writer/HiveIcebergDeleteWriter.java | 4 +-
.../iceberg/mr/hive/writer/WriterBuilder.java | 7 +-
.../org/apache/iceberg/parquet/VariantUtil.java | 8 +-
.../java/org/apache/iceberg/data/FileHelpers.java | 207 ++++++++++--------
.../apache/iceberg/data/GenericAppenderHelper.java | 102 ++++++---
.../iceberg/mr/hive/TestHiveIcebergCRUD.java | 64 ++++--
.../mr/hive/test/utils/HiveIcebergTestUtils.java | 40 ++--
.../src/main/java/org/apache/iceberg/DVUtil.java | 214 ------------------
.../apache/iceberg/deletes/BaseDVFileWriter.java | 240 --------------------
.../java/org/apache/iceberg/deletes/Deletes.java | 242 ---------------------
iceberg/pom.xml | 2 +-
itests/qtest-iceberg/pom.xml | 5 +
pom.xml | 9 +-
.../org/apache/iceberg/rest/HMSCatalogAdapter.java | 8 +-
.../rest/TestHMSCatalogAdapterSnapshotMode.java | 176 +++++++++++++++
standalone-metastore/pom.xml | 2 +-
25 files changed, 597 insertions(+), 1000 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 0639e50debe..81840148cc7 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
@@ -122,7 +122,7 @@ public static void updateHmsTableForIcebergTable(
obsoleteProps,
currentLocation,
parameters,
-
BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.toUpperCase(Locale.ENGLISH),
+
BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.toUpperCase(Locale.ROOT),
metadata.schema(),
maxHiveTablePropertySize);
setStorageHandler(parameters, hiveEngineEnabled);
@@ -168,7 +168,7 @@ public static void updateHmsTableForIcebergView(
obsoleteProps,
currentLocation,
parameters,
- HiveOperationsBase.ICEBERG_VIEW_TYPE_VALUE.toUpperCase(Locale.ENGLISH),
+ HiveOperationsBase.ICEBERG_VIEW_TYPE_VALUE.toUpperCase(Locale.ROOT),
metadata.schema(),
maxHiveTablePropertySize);
parameters.put(hive_metastoreConstants.META_TABLE_STORAGE,
HIVE_ICEBERG_STORAGE_HANDLER);
@@ -335,8 +335,8 @@ static void verifyMetadataHash(TableMetadata metadata,
String metadataHashFromHM
}
private static byte[] hashOf(TableMetadata tableMetadata) {
- try (HashWriter hashWriter = new HashWriter("SHA-256",
StandardCharsets.UTF_8)) {
- JsonGenerator generator = JsonUtil.factory().createGenerator(hashWriter);
+ 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();
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 95749370c82..123971e64aa 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
@@ -21,6 +21,7 @@
import java.io.IOException;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -36,6 +37,7 @@
import org.apache.hadoop.hive.metastore.api.PrincipalType;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.iceberg.BaseMetastoreTableOperations;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.CatalogUtil;
@@ -57,6 +59,7 @@
import org.apache.iceberg.exceptions.NoSuchViewException;
import org.apache.iceberg.exceptions.NotFoundException;
import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.CloseableGroup;
import org.apache.iceberg.io.FileIO;
import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
@@ -108,6 +111,7 @@ public class HiveCatalog extends BaseMetastoreViewCatalog
private boolean listAllTables = false;
private boolean uniqueTableLocation;
private Map<String, String> catalogProperties;
+ private CloseableGroup closeableGroup;
public HiveCatalog() {
}
@@ -157,6 +161,12 @@ public void initialize(String inputName, Map<String,
String> properties) {
}
this.clients = new CachedClientPool(conf, properties);
+
+ this.closeableGroup = new CloseableGroup();
+ closeableGroup.addCloseable(fileIO);
+ closeableGroup.addCloseable(keyManagementClient);
+ closeableGroup.addCloseable(metricsReporter());
+ closeableGroup.setSuppressCloseFailure(true);
}
@Override
@@ -176,17 +186,18 @@ public List<TableIdentifier> listTables(Namespace
namespace) {
String database = namespace.level(0);
try {
- List<String> tableNames = clients.run(client ->
client.getAllTables(database));
List<TableIdentifier> tableIdentifiers;
if (listAllTables) {
+ List<String> tableNames = clients.run(client ->
client.getAllTables(database));
tableIdentifiers = tableNames.stream()
.map(t -> TableIdentifier.of(namespace, t))
.collect(Collectors.toList());
} else {
- tableIdentifiers = listIcebergTables(
- tableNames, namespace,
BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE);
-
+ // Retrieving only the matching table names from HMS via server-side
filter
+ tableIdentifiers =
+ listIcebergTablesByFilter(
+ namespace,
BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE);
}
LOG.debug("Listing of namespace: {} resulted in the following tables:
{}", namespace, tableIdentifiers);
@@ -352,6 +363,37 @@ private List<TableIdentifier> listIcebergTables(
.collect(Collectors.toList());
}
+ /**
+ * Lists the {@link TableIdentifier} of all Iceberg tables under the given
database.
+ *
+ * @param namespace the namespace corresponding to the database
+ * @param tableTypeValue the value of {@code parameters.table_type} to
filter on.
+ */
+ private List<TableIdentifier> listIcebergTablesByFilter(
+ Namespace namespace, String tableTypeValue) throws TException,
InterruptedException {
+ String database = namespace.level(0);
+ // HMS normalizes the `table_type` parameter value to upper case when
persisting it
+ // (e.g. "iceberg" is stored as "ICEBERG"), so the filter value must be
upper-cased to match.
+ // Note: `like` (rather than `=`) is used intentionally. Some HMS backends
(e.g. Derby and
+ // Oracle, where PARAM_VALUE is a CLOB) do not support `=` comparison on
property values and
+ // fail the filter query (see HIVE-21614); newer HMS versions internally
rewrite `=` to `like`
+ // for this reason. As the value carries no `%`/`_` wildcards, `like` is
equivalent to `=` here.
+ String filter =
+ hive_metastoreConstants.HIVE_FILTER_FIELD_PARAMS +
+ BaseMetastoreTableOperations.TABLE_TYPE_PROP +
+ " like \"" +
+ tableTypeValue.toUpperCase(Locale.ROOT) +
+ "\"";
+
+ List<String> icebergTableNames =
+ clients.run(
+ client -> client.listTableNamesByFilter(database, filter, (short)
-1 /* no limit */));
+
+ return icebergTableNames.stream()
+ .map(tableName -> TableIdentifier.of(namespace, tableName))
+ .collect(Collectors.toList());
+ }
+
@SuppressWarnings("checkstyle:CyclomaticComplexity")
private void renameTableOrView(
TableIdentifier from,
@@ -830,10 +872,8 @@ public Map<String, String> properties() {
@Override
public void close() throws IOException {
- super.close();
-
- if (keyManagementClient != null) {
- keyManagementClient.close();
+ if (closeableGroup != null) {
+ closeableGroup.close();
}
}
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveOperationsBase.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveOperationsBase.java
index bf2684daa3f..4913a091a46 100644
---
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveOperationsBase.java
+++
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveOperationsBase.java
@@ -221,15 +221,16 @@ static void cleanupMetadataAndUnlock(
default Table newHmsTable(String hmsTableOwner) {
Preconditions.checkNotNull(hmsTableOwner, "'hmsOwner' parameter can't be
null");
- final long currentTimeMillis = System.currentTimeMillis();
+ // epoch seconds; HMS Thrift stores these as i32
+ final int currentTimeSeconds = (int) (System.currentTimeMillis() / 1000L);
Table newTable =
new Table(
table(),
database(),
hmsTableOwner,
- (int) currentTimeMillis / 1000,
- (int) currentTimeMillis / 1000,
+ currentTimeSeconds,
+ currentTimeSeconds,
Integer.MAX_VALUE,
null,
Collections.emptyList(),
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 6d3e3cc3635..c240463f68d 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
@@ -49,7 +49,6 @@
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;
@@ -221,6 +220,7 @@ protected void doRefresh() {
.map(Lists::newLinkedList)
.orElseGet(Lists::newLinkedList);
+ // Active transactions may still need keys that are not in committed
metadata.
if (encryptionManager != null) {
Set<String> keyIdsFromMetadata =
encryptedKeys.stream().map(EncryptedKey::keyId).collect(Collectors.toSet());
@@ -246,28 +246,14 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
boolean newTable = base == null;
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);
+ String newMetadataLocation = writeNewMetadataIfRequired(newTable,
metadata);
+ boolean hiveEngineEnabled = hiveEngineEnabled(metadata, conf);
BaseMetastoreOperations.CommitStatus commitStatus =
BaseMetastoreOperations.CommitStatus.FAILURE;
boolean updateHiveTable = false;
- HiveLock lock = lockObject(base != null ? base : tableMetadata);
+ HiveLock lock = lockObject(base != null ? base : metadata);
try {
lock.lock();
@@ -287,7 +273,7 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
LOG.debug("Committing existing table: {}", fullName);
} else {
tbl = newHmsTable(
- tableMetadata.property(HiveCatalog.HMS_TABLE_OWNER,
HiveHadoopUtil.currentUser())
+ metadata.property(HiveCatalog.HMS_TABLE_OWNER,
HiveHadoopUtil.currentUser())
);
LOG.debug("Committing new table: {}", fullName);
}
@@ -304,8 +290,8 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
tbl.setSd(
HiveOperationsBase.storageDescriptor(
- tableMetadata.schema(),
- tableMetadata.location(),
+ metadata.schema(),
+ metadata.location(),
hiveEngineEnabled, // set to pickup any schema changes
bucketCols,
numBuckets));
@@ -324,7 +310,7 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
if (base != null) {
removedProps =
base.properties().keySet().stream()
- .filter(key -> !tableMetadata.properties().containsKey(key))
+ .filter(key -> !metadata.properties().containsKey(key))
.collect(Collectors.toSet());
Preconditions.checkArgument(
@@ -341,7 +327,7 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
HMSTablePropertyHelper.updateHmsTableForIcebergTable(
newMetadataLocation,
tbl,
- tableMetadata,
+ metadata,
removedProps,
hiveEngineEnabled,
maxHiveTablePropertySize,
@@ -395,7 +381,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,
tableMetadata);
+ commitStatus = checkCommitStatusStrict(newMetadataLocation,
metadata);
if (commitStatus == BaseMetastoreOperations.CommitStatus.FAILURE) {
throw new CommitFailedException(
e, "The table %s.%s has been modified concurrently", database,
tableName);
@@ -406,7 +392,7 @@ protected void doCommit(TableMetadata base, TableMetadata
metadata) {
database,
tableName,
e);
- commitStatus = checkCommitStatus(newMetadataLocation, tableMetadata);
+ commitStatus = checkCommitStatus(newMetadataLocation, metadata);
}
switch (commitStatus) {
diff --git
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java
index 7cbe4e0c51a..49e86046960 100644
---
a/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java
+++
b/iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java
@@ -275,6 +275,11 @@ public void doCommit(ViewMetadata base, ViewMetadata
metadata) {
*/
private boolean checkCurrentMetadataLocation(String newMetadataLocation) {
ViewMetadata metadata = refresh();
+ if (metadata == null) {
+ // View creation may not have registered metadata yet.
+ return false;
+ }
+
return newMetadataLocation.equals(metadata.metadataFileLocation());
}
@@ -283,7 +288,8 @@ private static boolean hiveLockEnabled(Configuration conf) {
}
private Table newHMSView(ViewMetadata metadata) {
- final long currentTimeMillis = System.currentTimeMillis();
+ // epoch seconds; HMS Thrift stores these as i32
+ final int currentTimeSeconds = (int) (System.currentTimeMillis() / 1000L);
String hmsTableOwner =
PropertyUtil.propertyAsString(
metadata.properties(), HiveCatalog.HMS_TABLE_OWNER,
HiveHadoopUtil.currentUser());
@@ -293,8 +299,8 @@ private Table newHMSView(ViewMetadata metadata) {
table(),
database(),
hmsTableOwner,
- (int) currentTimeMillis / 1000,
- (int) currentTimeMillis / 1000,
+ currentTimeSeconds,
+ currentTimeSeconds,
Integer.MAX_VALUE,
null,
Collections.emptyList(),
diff --git a/iceberg/iceberg-handler/pom.xml b/iceberg/iceberg-handler/pom.xml
index c991f807094..b82ce9983a9 100644
--- a/iceberg/iceberg-handler/pom.xml
+++ b/iceberg/iceberg-handler/pom.xml
@@ -171,6 +171,10 @@
<artifactId>junit-vintage-engine</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache5-client</artifactId>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergAcidUtil.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergAcidUtil.java
index b881e07b2eb..a28423cdd9f 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergAcidUtil.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergAcidUtil.java
@@ -138,7 +138,7 @@ public static PositionDelete<Record>
getPositionDelete(Record rec, Record rowDat
rowData.set(i - dataOffset, rec.get(i));
}
- positionDelete.set(filePath, ObjectUtils.defaultIfNull(filePosition, 0L),
rowData);
+ positionDelete.set(filePath, ObjectUtils.defaultIfNull(filePosition, 0L));
return positionDelete;
}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveFileWriterFactory.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveFileWriterFactory.java
index 8923d825159..c72989264aa 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveFileWriterFactory.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveFileWriterFactory.java
@@ -19,47 +19,49 @@
package org.apache.iceberg.mr.hive.writer;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Map;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
-import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
-import org.apache.iceberg.avro.Avro;
-import org.apache.iceberg.data.BaseFileWriterFactory;
+import org.apache.iceberg.data.GenericFileWriterFactory;
import org.apache.iceberg.data.Record;
-import org.apache.iceberg.data.avro.DataWriter;
-import org.apache.iceberg.data.orc.GenericOrcWriter;
import org.apache.iceberg.data.parquet.GenericParquetWriter;
-import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.parquet.VariantUtil;
-class HiveFileWriterFactory extends BaseFileWriterFactory<Record> {
+class HiveFileWriterFactory implements FileWriterFactory<Record> {
+ private final GenericFileWriterFactory delegate;
+ private final Table table;
+ private final FileFormat dataFileFormat;
+ private final Schema dataSchema;
private final Map<String, String> properties;
private Record sampleRecord = null;
HiveFileWriterFactory(
- Table table,
- FileFormat dataFileFormat,
- Schema dataSchema,
- SortOrder dataSortOrder,
- FileFormat deleteFileFormat,
- int[] equalityFieldIds,
- Schema equalityDeleteRowSchema,
- SortOrder equalityDeleteSortOrder,
- Schema positionDeleteRowSchema) {
- super(
- table,
- dataFileFormat,
- dataSchema,
- dataSortOrder,
- deleteFileFormat,
- equalityFieldIds,
- equalityDeleteRowSchema,
- equalityDeleteSortOrder,
- positionDeleteRowSchema);
- properties = table.properties();
+ Table table, FileFormat dataFileFormat, Schema dataSchema, FileFormat
deleteFileFormat) {
+ this.table = table;
+ this.dataFileFormat = dataFileFormat;
+ this.dataSchema = dataSchema;
+ this.properties = table.properties();
+
+ GenericFileWriterFactory.Builder builder =
+ new GenericFileWriterFactory.Builder(table)
+ .dataFileFormat(dataFileFormat)
+ .dataSchema(dataSchema)
+ .deleteFileFormat(deleteFileFormat);
+
+ this.delegate = builder.build();
}
static Builder builderFor(Table table) {
@@ -67,52 +69,41 @@ static Builder builderFor(Table table) {
}
@Override
- protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
- }
-
- @Override
- protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
-
- }
-
- @Override
- protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
- }
-
- @Override
- protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
- // Configure variant shredding if enabled and a sample record is available
- if (VariantUtil.shouldUseVariantShredding(properties, dataSchema())) {
-
builder.variantShreddingFunc(VariantUtil.variantShreddingFunc(sampleRecord,
dataSchema()));
+ public DataWriter<Record> newDataWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ // Iceberg's GenericAppenderFactory/RegistryBasedFileWriterFactory does
not currently support
+ // injecting a custom variantShreddingFunc, so we must instantiate
Parquet.writeData directly
+ // when variant shredding is enabled for the table.
+ if (VariantUtil.shouldUseVariantShredding(dataFileFormat, properties,
dataSchema)) {
+ try {
+ return Parquet.writeData(file)
+ .schema(dataSchema)
+ .createWriterFunc(GenericParquetWriter::create)
+ .setAll(properties)
+ .metricsConfig(MetricsConfig.forTable(table))
+ .withSpec(spec)
+ .withPartition(partition)
+ .withKeyMetadata(file.keyMetadata())
+
.variantShreddingFunc(VariantUtil.variantShreddingFunc(sampleRecord,
dataSchema))
+ .overwrite()
+ .build();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
}
+ return delegate.newDataWriter(file, spec, partition);
}
@Override
- protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
-
- }
-
- @Override
- protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
- }
-
- @Override
- protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(GenericOrcWriter::buildWriter);
- }
-
- @Override
- protected void configureEqualityDelete(ORC.DeleteWriteBuilder
deleteWriteBuilder) {
-
+ public EqualityDeleteWriter<Record> newEqualityDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ return delegate.newEqualityDeleteWriter(file, spec, partition);
}
@Override
- protected void configurePositionDelete(ORC.DeleteWriteBuilder
deleteWriteBuilder) {
- deleteWriteBuilder.createWriterFunc(GenericOrcWriter::buildWriter);
+ public PositionDeleteWriter<Record> newPositionDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ return delegate.newPositionDeleteWriter(file, spec, partition);
}
static class Builder {
@@ -120,7 +111,6 @@ static class Builder {
private FileFormat dataFileFormat;
private Schema dataSchema;
private FileFormat deleteFileFormat;
- private Schema positionDeleteRowSchema;
Builder(Table table) {
this.table = table;
@@ -141,22 +131,8 @@ Builder deleteFileFormat(FileFormat newDeleteFileFormat) {
return this;
}
- Builder positionDeleteRowSchema(Schema newPositionDeleteRowSchema) {
- this.positionDeleteRowSchema = newPositionDeleteRowSchema;
- return this;
- }
-
HiveFileWriterFactory build() {
- return new HiveFileWriterFactory(
- table,
- dataFileFormat,
- dataSchema,
- null,
- deleteFileFormat,
- null,
- null,
- null,
- positionDeleteRowSchema);
+ return new HiveFileWriterFactory(table, dataFileFormat, dataSchema,
deleteFileFormat);
}
}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergCopyOnWriteRecordWriter.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergCopyOnWriteRecordWriter.java
index cdb03e1a89c..cd058bbd1f5 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergCopyOnWriteRecordWriter.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergCopyOnWriteRecordWriter.java
@@ -55,7 +55,7 @@ class HiveIcebergCopyOnWriteRecordWriter extends
HiveIcebergDefaultWriter {
public void write(Writable row) throws IOException {
Record record = ((Container<Record>) row).get();
PositionDelete<Record> positionDelete =
IcebergAcidUtil.getPositionDelete(record, rowDataTemplate, false);
- Record rowData = positionDelete.row();
+ Record rowData = rowDataTemplate;
if (positionDelete.pos() < 0) {
int specId = IcebergAcidUtil.parseSpecId(record);
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergDeleteWriter.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergDeleteWriter.java
index 6970556fc46..f4526189e86 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergDeleteWriter.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/HiveIcebergDeleteWriter.java
@@ -61,10 +61,10 @@ public void write(Writable row) throws IOException {
PositionDelete<Record> positionDelete =
IcebergAcidUtil.getPositionDelete(rec, rowDataTemplate, isMergeTask);
int specId = IcebergAcidUtil.parseSpecId(rec);
PartitionKey partitionKey = isMergeTask ?
IcebergAcidUtil.parsePartitionKey(rec) :
- partition(positionDelete.row(), specId);
+ partition(rowDataTemplate, specId);
if (skipRowData) {
// Set null as the row data as we intend to avoid writing the actual row
data in the delete file.
- positionDelete.set(positionDelete.path(), positionDelete.pos(), null);
+ positionDelete.set(positionDelete.path(), positionDelete.pos());
}
writer.write(positionDelete, specs.get(specId), partitionKey);
}
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/WriterBuilder.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/WriterBuilder.java
index dbe10bc4401..20386be3ffe 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/WriterBuilder.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/writer/WriterBuilder.java
@@ -132,9 +132,6 @@ public HiveIcebergWriter build() {
.dataFileFormat(context.dataFileFormat())
.dataSchema(shouldAddRowLineageColumns ?
MetadataColumns.schemaWithRowLineage(table.schema()) : table.schema())
.deleteFileFormat(context.deleteFileFormat())
- .positionDeleteRowSchema(context.skipRowData() ||
!context.inputOrdered() ?
- // SortingPositionOnlyDeleteWriter doesn't support rawData in
delete schema
- null : table.schema())
.build();
HiveIcebergWriter writer;
@@ -234,11 +231,11 @@ static class Context {
Context(Map<String, String> properties, UnaryOperator<String> ops, String
tableName) {
String dataFileFormatName =
properties.getOrDefault(DEFAULT_FILE_FORMAT,
DEFAULT_FILE_FORMAT_DEFAULT);
- this.dataFileFormat =
FileFormat.valueOf(dataFileFormatName.toUpperCase(Locale.ENGLISH));
+ this.dataFileFormat =
FileFormat.valueOf(dataFileFormatName.toUpperCase(Locale.ROOT));
String deleteFileFormatName =
properties.getOrDefault(DELETE_DEFAULT_FILE_FORMAT,
dataFileFormatName);
- this.deleteFileFormat =
FileFormat.valueOf(deleteFileFormatName.toUpperCase(Locale.ENGLISH));
+ this.deleteFileFormat =
FileFormat.valueOf(deleteFileFormatName.toUpperCase(Locale.ROOT));
this.targetDataFileSize = PropertyUtil.propertyAsLong(properties,
TableProperties.WRITE_TARGET_FILE_SIZE_BYTES,
TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/VariantUtil.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/VariantUtil.java
index 736a39b895c..fa9a41feb8a 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/VariantUtil.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/VariantUtil.java
@@ -22,6 +22,7 @@
import java.util.List;
import java.util.Map;
import org.apache.iceberg.Accessor;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.data.Record;
@@ -89,8 +90,11 @@ private static List<VariantField>
variantFieldsForShredding(Schema schema) {
return results;
}
- public static boolean shouldUseVariantShredding(Map<String, String>
properties, Schema schema) {
- return isVariantShreddingEnabled(properties) && hasVariantFields(schema);
+ public static boolean shouldUseVariantShredding(
+ FileFormat fileFormat, Map<String, String> properties, Schema schema) {
+ return fileFormat == FileFormat.PARQUET &&
+ isVariantShreddingEnabled(properties) &&
+ hasVariantFields(schema);
}
private static boolean hasVariantFields(Schema schema) {
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/FileHelpers.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/FileHelpers.java
index 5f071f5b885..563363f2f26 100644
---
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/FileHelpers.java
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/FileHelpers.java
@@ -22,14 +22,16 @@
import java.io.Closeable;
import java.io.IOException;
import java.util.List;
-import java.util.Map;
import org.apache.iceberg.DataFile;
-import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionData;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
import org.apache.iceberg.deletes.BaseDVFileWriter;
import org.apache.iceberg.deletes.DVFileWriter;
import org.apache.iceberg.deletes.EqualityDeleteWriter;
@@ -38,98 +40,98 @@
import org.apache.iceberg.encryption.EncryptedFiles;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.DeleteSchemaUtil;
import org.apache.iceberg.io.FileAppender;
-import org.apache.iceberg.io.FileAppenderFactory;
import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.Pair;
-import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
-import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
-
public class FileHelpers {
private FileHelpers() {
}
- public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(Table table,
OutputFile out,
-
List<Pair<CharSequence, Long>> deletes)
- throws IOException {
- return writeDeleteFile(table, out, null, deletes);
- }
-
- public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(Table table,
OutputFile out, StructLike partition,
-
List<Pair<CharSequence, Long>> deletes)
+ public static DeleteFile testOnlyPosDeleteFileWithRow(
+ Table table, OutputFile out, StructLike partition, List<Record>
posDeleteRecords)
throws IOException {
- FileFormat format = defaultFormat(table.properties());
- FileAppenderFactory<Record> factory = new
GenericAppenderFactory(table.schema(), table.spec());
+ Schema posDeleteSchema = DeleteSchemaUtil.posDeleteSchema(table.schema());
+ FileAppender<Record> appender =
+ Parquet.write(out)
+ .schema(posDeleteSchema)
+ .createWriterFunc(GenericParquetWriter::create)
+ .metricsConfig(MetricsConfig.forPositionDelete())
+ .overwrite()
+ .build();
+ try (Closeable ignored = appender) {
+ posDeleteRecords.forEach(appender::add);
+ }
- PositionDeleteWriter<Record> writer =
- factory.newPosDeleteWriter(encrypt(out), format, partition);
- PositionDelete<Record> posDelete = PositionDelete.create();
- try (Closeable toClose = writer) {
- for (Pair<CharSequence, Long> delete : deletes) {
- writer.write(posDelete.set(delete.first(), delete.second(), null));
- }
+ FileMetadata.Builder builder =
+ FileMetadata.deleteFileBuilder(table.spec())
+ .ofPositionDeletes()
+ .withFormat(FileFormat.PARQUET)
+ .withPath(out.location())
+ .withFileSizeInBytes(appender.length())
+ .withMetrics(appender.metrics());
+ if (partition != null) {
+ builder.withPartition(partition);
}
- return Pair.of(writer.toDeleteFile(), writer.referencedDataFiles());
+ return builder.build();
}
- public static DeleteFile writeDeleteFile(Table table, OutputFile out,
List<Record> deletes, Schema deleteRowSchema)
- throws IOException {
- return writeDeleteFile(table, out, null, deletes, deleteRowSchema);
+ public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
+ Table table, OutputFile out, List<Pair<CharSequence, Long>> deletes)
throws IOException {
+ return writeDeleteFile(table, out, null, deletes);
}
- public static DeleteFile writeDeleteFile(Table table, OutputFile out,
StructLike partition,
- List<Record> deletes, Schema
deleteRowSchema)
+ public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
+ Table table, OutputFile out, List<Pair<CharSequence, Long>> deletes, int
formatVersion)
throws IOException {
- FileFormat format = defaultFormat(table.properties());
- int[] equalityFieldIds =
deleteRowSchema.columns().stream().mapToInt(Types.NestedField::fieldId).toArray();
- FileAppenderFactory<Record> factory = new
GenericAppenderFactory(table.schema(), table.spec(),
- equalityFieldIds, deleteRowSchema, null);
-
- EqualityDeleteWriter<Record> writer =
factory.newEqDeleteWriter(encrypt(out), format, partition);
- try (Closeable toClose = writer) {
- writer.write(deletes);
- }
+ return writeDeleteFile(table, out, null, deletes, formatVersion);
+ }
- return writer.toDeleteFile();
+ public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
+ Table table, OutputFile out, StructLike partition,
List<Pair<CharSequence, Long>> deletes)
+ throws IOException {
+ return writeDeleteFile(table, out, partition, deletes, 2);
}
public static Pair<DeleteFile, CharSequenceSet> writeDeleteFile(
- Table table,
- OutputFile out,
- StructLike partition,
- List<Pair<CharSequence, Long>> deletes,
- int formatVersion)
- throws IOException {
+ Table table,
+ OutputFile out,
+ StructLike partition,
+ List<Pair<CharSequence, Long>> deletes,
+ int formatVersion)
+ throws IOException {
if (formatVersion >= 3) {
OutputFileFactory fileFactory =
- OutputFileFactory.builderFor(table, 1,
1).format(FileFormat.PUFFIN).build();
+ OutputFileFactory.builderFor(table, 1,
1).format(FileFormat.PUFFIN).build();
DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null);
try (DVFileWriter closeableWriter = writer) {
for (Pair<CharSequence, Long> delete : deletes) {
closeableWriter.delete(
- delete.first().toString(), delete.second(), table.spec(),
partition);
+ delete.first().toString(), delete.second(), table.spec(),
partition);
}
}
return Pair.of(
- Iterables.getOnlyElement(writer.result().deleteFiles()),
- writer.result().referencedDataFiles());
+ Iterables.getOnlyElement(writer.result().deleteFiles()),
+ writer.result().referencedDataFiles());
} else {
FileWriterFactory<Record> factory =
GenericFileWriterFactory.builderFor(table).build();
PositionDeleteWriter<Record> writer =
- factory.newPositionDeleteWriter(encrypt(out), table.spec(),
partition);
+ factory.newPositionDeleteWriter(encrypt(out), table.spec(),
partition);
PositionDelete<Record> posDelete = PositionDelete.create();
try (Closeable toClose = writer) {
for (Pair<CharSequence, Long> delete : deletes) {
- writer.write(posDelete.set(delete.first(), delete.second(), null));
+ writer.write(posDelete.set(delete.first(), delete.second()));
}
}
@@ -137,60 +139,81 @@ public static Pair<DeleteFile, CharSequenceSet>
writeDeleteFile(
}
}
- public static DataFile writeDataFile(Table table, OutputFile out,
List<Record> rows) throws IOException {
- FileFormat format = defaultFormat(table.properties());
- GenericAppenderFactory factory = new
GenericAppenderFactory(table.schema());
+ public static DeleteFile writeDeleteFile(
+ Table table, OutputFile out, List<Record> deletes, Schema
deleteRowSchema)
+ throws IOException {
+ return writeDeleteFile(table, out, null, deletes, deleteRowSchema);
+ }
- FileAppender<Record> writer = factory.newAppender(out, format);
+ public static DeleteFile writeDeleteFile(
+ Table table,
+ OutputFile out,
+ StructLike partition,
+ List<Record> deletes,
+ Schema deleteRowSchema)
+ throws IOException {
+ int[] equalityFieldIds =
+
deleteRowSchema.columns().stream().mapToInt(Types.NestedField::fieldId).toArray();
+ FileWriterFactory<Record> factory =
+ GenericFileWriterFactory.builderFor(table)
+ .equalityDeleteRowSchema(deleteRowSchema)
+ .equalityFieldIds(equalityFieldIds)
+ .build();
+
+ EqualityDeleteWriter<Record> writer =
+ factory.newEqualityDeleteWriter(encrypt(out), table.spec(), partition);
try (Closeable toClose = writer) {
- writer.addAll(rows);
+ writer.write(deletes);
}
- return DataFiles.builder(table.spec())
- .withFormat(format)
- .withPath(out.location())
- .withFileSizeInBytes(writer.length())
- .withSplitOffsets(writer.splitOffsets())
- .withMetrics(writer.metrics())
- .build();
+ return writer.toDeleteFile();
}
- public static DataFile writeDataFile(Table table, OutputFile out, StructLike
partition, List<Record> rows)
+ public static DataFile writeDataFile(Table table, OutputFile out,
List<Record> rows)
throws IOException {
- FileFormat format = defaultFormat(table.properties());
- GenericAppenderFactory factory = new
GenericAppenderFactory(table.schema(), table.spec());
+ return writeDataFile(table, out, rows, null);
+ }
+
+ public static DataFile writeDataFile(
+ Table table, OutputFile out, List<Record> rows, PartitionData partition)
throws IOException {
+ FileWriterFactory<Record> factory =
GenericFileWriterFactory.builderFor(table).build();
+
+ DataWriter<Record> writer = factory.newDataWriter(encrypt(out),
table.spec(), partition);
+ try (Closeable toClose = writer) {
+ writer.write(rows);
+ }
+
+ return writer.toDataFile();
+ }
+
+ public static DataFile writeDataFile(
+ Table table, OutputFile out, StructLike partition, List<Record> rows)
throws IOException {
+ FileWriterFactory<Record> factory =
GenericFileWriterFactory.builderFor(table).build();
- FileAppender<Record> writer = factory.newAppender(out, format);
+ DataWriter<Record> writer = factory.newDataWriter(encrypt(out),
table.spec(), partition);
try (Closeable toClose = writer) {
- writer.addAll(rows);
+ writer.write(rows);
}
- return DataFiles.builder(table.spec())
- .withFormat(format)
- .withPath(out.location())
- .withPartition(partition)
- .withFileSizeInBytes(writer.length())
- .withSplitOffsets(writer.splitOffsets())
- .withMetrics(writer.metrics())
- .build();
+ return writer.toDataFile();
}
public static DeleteFile writePosDeleteFile(
- Table table, OutputFile out, StructLike partition,
List<PositionDelete<?>> deletes)
- throws IOException {
+ Table table, OutputFile out, StructLike partition,
List<PositionDelete<?>> deletes)
+ throws IOException {
return writePosDeleteFile(table, out, partition, deletes, 2);
}
public static DeleteFile writePosDeleteFile(
- Table table,
- OutputFile out,
- StructLike partition,
- List<PositionDelete<?>> deletes,
- int formatVersion)
- throws IOException {
+ Table table,
+ OutputFile out,
+ StructLike partition,
+ List<PositionDelete<?>> deletes,
+ int formatVersion)
+ throws IOException {
if (formatVersion >= 3) {
OutputFileFactory fileFactory =
- OutputFileFactory.builderFor(table, 1,
1).format(FileFormat.PUFFIN).build();
+ OutputFileFactory.builderFor(table, 1,
1).format(FileFormat.PUFFIN).build();
DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null);
try (DVFileWriter closeableWriter = writer) {
for (PositionDelete<?> delete : deletes) {
@@ -200,13 +223,10 @@ public static DeleteFile writePosDeleteFile(
return Iterables.getOnlyElement(writer.result().deleteFiles());
} else {
- FileWriterFactory<Record> factory =
- GenericFileWriterFactory.builderFor(table)
- .positionDeleteRowSchema(table.schema())
- .build();
+ FileWriterFactory<Record> factory =
GenericFileWriterFactory.builderFor(table).build();
PositionDeleteWriter<?> writer =
- factory.newPositionDeleteWriter(encrypt(out), table.spec(),
partition);
+ factory.newPositionDeleteWriter(encrypt(out), table.spec(),
partition);
try (Closeable toClose = writer) {
for (PositionDelete delete : deletes) {
writer.write(delete);
@@ -217,12 +237,7 @@ public static DeleteFile writePosDeleteFile(
}
}
- private static EncryptedOutputFile encrypt(OutputFile out) {
+ public static EncryptedOutputFile encrypt(OutputFile out) {
return EncryptedFiles.encryptedOutput(out, EncryptionKeyMetadata.EMPTY);
}
-
- private static FileFormat defaultFormat(Map<String, String> properties) {
- String formatString = properties.getOrDefault(DEFAULT_FILE_FORMAT,
DEFAULT_FILE_FORMAT_DEFAULT);
- return FileFormat.fromString(formatString);
- }
}
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
index c1313041802..902943e9a8d 100644
---
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
@@ -21,43 +21,58 @@
import java.io.File;
import java.io.IOException;
+import java.nio.file.Path;
import java.util.List;
+import java.util.UUID;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.DataFile;
-import org.apache.iceberg.DataFiles;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Files;
+import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
-import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.junit.Assert;
import org.junit.rules.TemporaryFolder;
-/**
- * Helper for appending {@link DataFile} to a table or appending {@link
Record}s to a table.
- */
+import static org.apache.iceberg.data.FileHelpers.encrypt;
+
+/** Helper for appending {@link DataFile} to a table or appending {@link
Record}s to a table. */
public class GenericAppenderHelper {
private static final String ORC_CONFIG_PREFIX = "^orc.*";
+ private static final String PARQUET_CONFIG_PATTERN = ".*parquet.*";
private final Table table;
private final FileFormat fileFormat;
- private final TemporaryFolder tmp;
+ private final Path temp;
private final Configuration conf;
public GenericAppenderHelper(Table table, FileFormat fileFormat,
TemporaryFolder tmp, Configuration conf) {
+ this(table, fileFormat, tmp != null ? tmp.getRoot().toPath() : null, conf);
+ }
+
+ public GenericAppenderHelper(Table table, FileFormat fileFormat,
TemporaryFolder tmp) {
+ this(table, fileFormat, tmp, null);
+ }
+
+ public GenericAppenderHelper(Table table, FileFormat fileFormat, Path temp,
Configuration conf) {
this.table = table;
this.fileFormat = fileFormat;
- this.tmp = tmp;
+ this.temp = temp;
this.conf = conf;
}
- public void appendToTable(DataFile... dataFiles) {
+ public GenericAppenderHelper(Table table, FileFormat fileFormat, Path temp) {
+ this(table, fileFormat, temp, null);
+ }
+
+ public void appendToTable(String branch, DataFile... dataFiles) {
Preconditions.checkNotNull(table, "table not set");
- AppendFiles append = table.newAppend();
+ AppendFiles append =
+ table.newAppend().toBranch(branch != null ? branch :
SnapshotRef.MAIN_BRANCH);
for (DataFile dataFile : dataFiles) {
append = append.appendFile(dataFile);
@@ -66,45 +81,68 @@ public void appendToTable(DataFile... dataFiles) {
append.commit();
}
+ public void appendToTable(DataFile... dataFiles) {
+ appendToTable(null, dataFiles);
+ }
+
public void appendToTable(List<Record> records) throws IOException {
- appendToTable(null, records);
+ appendToTable(null, null, records);
+ }
+
+ public void appendToTable(String branch, List<Record> records) throws
IOException {
+ appendToTable(null, branch, records);
+ }
+
+ public void appendToTable(StructLike partition, String branch, List<Record>
records)
+ throws IOException {
+ appendToTable(branch, writeFile(partition, records));
}
public void appendToTable(StructLike partition, List<Record> records) throws
IOException {
appendToTable(writeFile(partition, records));
}
+ public DataFile writeFile(List<Record> records) throws IOException {
+ Preconditions.checkNotNull(table, "table not set");
+ File file = temp.resolve("generic-appender-test-" +
UUID.randomUUID()).toFile();
+ return appendToLocalFile(table, file, fileFormat, null, records, conf);
+ }
+
public DataFile writeFile(StructLike partition, List<Record> records) throws
IOException {
Preconditions.checkNotNull(table, "table not set");
- File file = tmp.newFile();
- Assert.assertTrue(file.delete());
+ File file =
+ temp.resolve("generic-appender-partition-test-" +
UUID.randomUUID()).toFile();
return appendToLocalFile(table, file, fileFormat, partition, records,
conf);
}
- private static DataFile appendToLocalFile(Table table, File file, FileFormat
format, StructLike partition,
- List<Record> records, Configuration conf)
+ private static DataFile appendToLocalFile(
+ Table table,
+ File file,
+ FileFormat format,
+ StructLike partition,
+ List<Record> records,
+ Configuration conf)
throws IOException {
- GenericAppenderFactory appenderFactory = new
GenericAppenderFactory(table.schema());
+ GenericFileWriterFactory.Builder builder =
+ new GenericFileWriterFactory.Builder(table).dataFileFormat(format);
+
+ // Push down ORC related settings to appender if there are any
+ if (FileFormat.ORC.equals(format) && conf != null) {
+ builder.writerProperties(conf.getValByRegex(ORC_CONFIG_PREFIX));
+ }
- // Push down ORC related settings to appender
- if (FileFormat.ORC.equals(format)) {
- appenderFactory.setAll(conf.getValByRegex(ORC_CONFIG_PREFIX));
+ if (FileFormat.PARQUET.equals(format) && conf != null) {
+ builder.writerProperties(conf.getValByRegex(PARQUET_CONFIG_PATTERN));
}
- FileAppender<Record> appender = appenderFactory.newAppender(
- Files.localOutput(file), format);
- try (FileAppender<Record> fileAppender = appender) {
- fileAppender.addAll(records);
+ DataWriter<Record> writer =
+ builder.build().newDataWriter(encrypt(Files.localOutput(file)),
table.spec(), partition);
+ try (writer) {
+ for (Record r : records) {
+ writer.write(r);
+ }
}
- return DataFiles.builder(table.spec())
- .withRecordCount(records.size())
- .withFileSizeInBytes(file.length())
- .withPath(Files.localInput(file).location())
- .withMetrics(appender.metrics())
- .withFormat(format)
- .withPartition(partition)
- .withSplitOffsets(appender.splitOffsets())
- .build();
+ return writer.toDataFile();
}
}
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java
index 1826d241a01..8b3fbfbc464 100644
---
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java
@@ -21,17 +21,23 @@
import java.io.IOException;
import java.util.List;
+import java.util.UUID;
import java.util.stream.StreamSupport;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.FileHelpers;
+import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.mr.TestHelper;
import org.apache.iceberg.mr.hive.test.TestTables.TestTableType;
import
org.apache.iceberg.mr.hive.test.utils.HiveIcebergStorageHandlerTestUtils;
@@ -142,9 +148,7 @@ public void
testReadAndWriteFormatV2Unpartitioned_PosDelete() throws IOException
DataFile dataFile =
StreamSupport.stream(tbl.currentSnapshot().addedDataFiles(tbl.io()).spliterator(),
false)
.findFirst()
.orElseThrow(() -> new RuntimeException("Did not find any data files
for test table"));
- List<PositionDelete<Record>> deletes = ImmutableList.of(positionDelete(
- dataFile.path(), 2L,
HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS.get(2))
- );
+ List<PositionDelete<Record>> deletes =
ImmutableList.of(positionDelete(dataFile.path(), 2L));
DeleteFile deleteFile =
HiveIcebergTestUtils.createPositionalDeleteFile(tbl, "dummyPath",
fileFormat, null, deletes);
tbl.newRowDelta().addDeletes(deleteFile).commit();
@@ -158,7 +162,7 @@ public void
testReadAndWriteFormatV2Unpartitioned_PosDelete() throws IOException
}
@Test
- public void testReadAndWriteFormatV2Partitioned_PosDelete_RowNotSupplied()
throws IOException {
+ public void testReadAndWriteFormatV2Partitioned_PosDelete() throws
IOException {
Assume.assumeTrue(formatVersion == 2);
PartitionSpec spec =
PartitionSpec.builderFor(HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA)
@@ -171,15 +175,15 @@ public void
testReadAndWriteFormatV2Partitioned_PosDelete_RowNotSupplied() throw
"(0, 'Blake', 'Blue')");
tbl.refresh();
- // delete the first and third rows from the newly-added data file - with
row supplied
+ // delete the first and third rows from the newly-added data file
DataFile dataFile =
StreamSupport.stream(tbl.currentSnapshot().addedDataFiles(tbl.io()).spliterator(),
false)
.filter(file -> file.partition().get(0, Long.class) == 0L)
.filter(file -> file.recordCount() == 3)
.findAny()
.orElseThrow(() -> new RuntimeException("Did not find the desired data
file in the test table"));
List<PositionDelete<Record>> deletes = ImmutableList.of(
- positionDelete(dataFile.path(), 0L, null),
- positionDelete(dataFile.path(), 2L, null)
+ positionDelete(dataFile.path(), 0L),
+ positionDelete(dataFile.path(), 2L)
);
DeleteFile deleteFile =
HiveIcebergTestUtils.createPositionalDeleteFile(tbl, "dummyPath",
fileFormat, ImmutableMap.of("customer_id", 0L), deletes);
@@ -194,9 +198,15 @@ public void
testReadAndWriteFormatV2Partitioned_PosDelete_RowNotSupplied() throw
Assert.assertArrayEquals(new Object[] {2L, "Trudy", "Pink"},
objects.get(3));
}
+ /**
+ * Tests reading/consuming legacy Iceberg V2 tables containing position
delete files with embedded
+ * row data. Iceberg 1.12.0 removed writer support for row data in position
deletes, but Hive
+ * readers must remain backward compatible and safely ignore the embedded
row field when
+ * projecting deleted positions.
+ */
@Test
- public void testReadAndWriteFormatV2Partitioned_PosDelete_RowSupplied()
throws IOException {
- Assume.assumeTrue(formatVersion == 2);
+ public void testConsumeLegacyFormatV2Partitioned_PosDelete_RowSupplied()
throws IOException {
+ Assume.assumeTrue(formatVersion == 2 && fileFormat == FileFormat.PARQUET);
PartitionSpec spec =
PartitionSpec.builderFor(HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA)
.identity("customer_id").build();
@@ -214,18 +224,35 @@ public void
testReadAndWriteFormatV2Partitioned_PosDelete_RowSupplied() throws I
.filter(file -> file.recordCount() == 3)
.findAny()
.orElseThrow(() -> new RuntimeException("Did not find the desired data
file in the test table"));
+
+ // Construct a legacy position delete file with embedded row data (fields:
file_path, pos, row).
+ // FileHelpers.testOnlyPosDeleteFileWithRow writes this directly without
using the standard delete writer,
+ // which disallows writing row data in Iceberg 1.12.0+.
+ Schema posDeleteSchema = DeleteSchemaUtil.posDeleteSchema(tbl.schema());
List<Record> rowsToDel =
TestHelper.RecordsBuilder.newInstance(HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA)
- .add(0L, "Laura", "Yellow").add(0L, "Blake", "Blue").build();
- List<PositionDelete<Record>> deletes = ImmutableList.of(
- positionDelete(dataFile.path(), 0L, rowsToDel.get(0)),
- positionDelete(dataFile.path(), 2L, rowsToDel.get(1))
- );
- DeleteFile deleteFile =
HiveIcebergTestUtils.createPositionalDeleteFile(tbl, "dummyPath",
- fileFormat, ImmutableMap.of("customer_id", 0L), deletes);
+ .add(0L, "Laura", "Yellow")
+ .add(0L, "Blake", "Blue")
+ .build();
+
+ Record delRecord0 = GenericRecord.create(posDeleteSchema);
+ delRecord0.setField(MetadataColumns.DELETE_FILE_PATH.name(),
dataFile.path().toString());
+ delRecord0.setField(MetadataColumns.DELETE_FILE_POS.name(), 0L);
+ delRecord0.setField(MetadataColumns.DELETE_FILE_ROW_FIELD_NAME,
rowsToDel.get(0));
+
+ Record delRecord2 = GenericRecord.create(posDeleteSchema);
+ delRecord2.setField(MetadataColumns.DELETE_FILE_PATH.name(),
dataFile.path().toString());
+ delRecord2.setField(MetadataColumns.DELETE_FILE_POS.name(), 2L);
+ delRecord2.setField(MetadataColumns.DELETE_FILE_ROW_FIELD_NAME,
rowsToDel.get(1));
+
+ OutputFile deleteOutputFile = tbl.io().newOutputFile(
+ tbl.location() + "/data/dummyLegacyPosDelete_" + UUID.randomUUID() +
".parquet");
+ DeleteFile deleteFile = FileHelpers.testOnlyPosDeleteFileWithRow(
+ tbl, deleteOutputFile, dataFile.partition(),
ImmutableList.of(delRecord0, delRecord2));
tbl.newRowDelta().addDeletes(deleteFile).commit();
List<Object[]> objects = shell.executeStatement("SELECT * FROM customers
ORDER BY customer_id, first_name");
+ // Verify that Hive readers successfully filter out deleted rows (0:
Alice, 0: John, 1: Bob, 2: Trudy)
Assert.assertEquals(4, objects.size());
Assert.assertArrayEquals(new Object[] {0L, "Alice", "Brown"},
objects.get(0));
Assert.assertArrayEquals(new Object[] {0L, "John", "Green"},
objects.get(1));
@@ -620,8 +647,9 @@ public void testMultiInsert() {
Assert.assertEquals(1, res.size());
}
- private static <T> PositionDelete<T> positionDelete(CharSequence path, long
pos, T row) {
+ private static <T> PositionDelete<T> positionDelete(CharSequence path, long
pos) {
PositionDelete<T> positionDelete = PositionDelete.create();
- return positionDelete.set(path, pos, row);
+ positionDelete.set(path, pos);
+ return positionDelete;
}
}
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/utils/HiveIcebergTestUtils.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/utils/HiveIcebergTestUtils.java
index 2e3d2106874..87033a3a709 100644
---
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/utils/HiveIcebergTestUtils.java
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/test/utils/HiveIcebergTestUtils.java
@@ -68,7 +68,7 @@
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
-import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.GenericFileWriterFactory;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.IcebergGenerics;
import org.apache.iceberg.data.Record;
@@ -78,7 +78,7 @@
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.hadoop.HadoopOutputFile;
import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.mr.hive.HiveTableUtil;
import org.apache.iceberg.mr.hive.test.TestHiveShell;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -364,8 +364,12 @@ public static DeleteFile createEqualityDeleteFile(Table
table, PartitionSpec spe
.toList();
Schema eqDeleteRowSchema =
table.schema().select(equalityFields.toArray(new String[]{}));
- FileAppenderFactory<Record> appenderFactory = new
GenericAppenderFactory(table.schema(), spec,
- ArrayUtil.toIntArray(equalityFieldIds), eqDeleteRowSchema, null);
+ FileWriterFactory<Record> writerFactory =
+ new GenericFileWriterFactory.Builder(table)
+ .deleteFileFormat(fileFormat)
+ .equalityFieldIds(ArrayUtil.toIntArray(equalityFieldIds))
+ .equalityDeleteRowSchema(eqDeleteRowSchema)
+ .build();
EncryptedOutputFile outputFile =
table.encryption().encrypt(HadoopOutputFile.fromPath(
new org.apache.hadoop.fs.Path(table.location(), deleteFilePath), new
Configuration()));
@@ -374,9 +378,10 @@ public static DeleteFile createEqualityDeleteFile(Table
table, PartitionSpec spe
part = new PartitionKey(spec, eqDeleteRowSchema);
part.partition(rowsToDelete.get(0));
}
- EqualityDeleteWriter<Record> eqWriter =
appenderFactory.newEqDeleteWriter(outputFile, fileFormat, part);
- try (EqualityDeleteWriter<Record> writer = eqWriter) {
- writer.write(rowsToDelete);
+ EqualityDeleteWriter<Record> eqWriter =
+ writerFactory.newEqualityDeleteWriter(outputFile, spec, part);
+ try (eqWriter) {
+ eqWriter.write(rowsToDelete);
}
return eqWriter.toDeleteFile();
}
@@ -394,9 +399,8 @@ public static DeleteFile createEqualityDeleteFile(Table
table, PartitionSpec spe
public static DeleteFile createPositionalDeleteFile(Table table, String
deleteFilePath, FileFormat fileFormat,
Map<String, Object> partitionValues, List<PositionDelete<Record>>
deletes) throws IOException {
- Schema posDeleteRowSchema = deletes.get(0).row() == null ? null :
table.schema();
- FileAppenderFactory<Record> appenderFactory = new
GenericAppenderFactory(table.schema(), table.spec(),
- null, null, posDeleteRowSchema);
+ FileWriterFactory<Record> writerFactory =
+ new
GenericFileWriterFactory.Builder(table).deleteFileFormat(fileFormat).build();
EncryptedOutputFile outputFile =
table.encryption().encrypt(HadoopOutputFile.fromPath(
new org.apache.hadoop.fs.Path(table.location(), deleteFilePath), new
Configuration()));
@@ -407,13 +411,15 @@ public static DeleteFile createPositionalDeleteFile(Table
table, String deleteFi
partitionKey.partition(record);
}
- PositionDeleteWriter<Record> posWriter =
appenderFactory.newPosDeleteWriter(outputFile, fileFormat, partitionKey);
- try (PositionDeleteWriter<Record> writer = posWriter) {
- deletes.forEach(del -> {
- PositionDelete positionDelete = PositionDelete.create();
- positionDelete.set(del.path(), del.pos(), del.row());
- writer.write(positionDelete);
- });
+ PositionDeleteWriter<Record> posWriter =
+ writerFactory.newPositionDeleteWriter(outputFile, table.spec(),
partitionKey);
+ try (posWriter) {
+ deletes.forEach(
+ del -> {
+ PositionDelete positionDelete = PositionDelete.create();
+ positionDelete.set(del.path(), del.pos());
+ posWriter.write(positionDelete);
+ });
}
return posWriter.toDeleteFile();
}
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
deleted file mode 100644
index 5f67d57e540..00000000000
--- a/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/DVUtil.java
+++ /dev/null
@@ -1,214 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.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
deleted file mode 100644
index ab8c85b50eb..00000000000
---
a/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java
+++ /dev/null
@@ -1,240 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.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
deleted file mode 100644
index 127c8046f9e..00000000000
---
a/iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/deletes/Deletes.java
+++ /dev/null
@@ -1,242 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.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;
- }
- }
-}
diff --git a/iceberg/pom.xml b/iceberg/pom.xml
index 7fcb6387e4f..543436192d2 100644
--- a/iceberg/pom.xml
+++ b/iceberg/pom.xml
@@ -41,7 +41,7 @@
<assertj.version>3.27.3</assertj.version>
<immutables.value.version>2.10.0</immutables.value.version>
<validate.skip>false</validate.skip>
- <caffeine.version>2.9.3</caffeine.version>
+ <caffeine.version>3.2.4</caffeine.version>
</properties>
<modules>
<module>patched-iceberg-api</module>
diff --git a/itests/qtest-iceberg/pom.xml b/itests/qtest-iceberg/pom.xml
index 23e68ce4e01..5243eaa64fd 100644
--- a/itests/qtest-iceberg/pom.xml
+++ b/itests/qtest-iceberg/pom.xml
@@ -558,6 +558,11 @@
<artifactId>jaxb-runtime</artifactId>
<version>${jaxb-runtime.version}</version>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache5-client</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git a/pom.xml b/pom.xml
index a887c6333b8..e774827e5c3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -159,7 +159,7 @@
<httpcomponents.client.version>4.5.13</httpcomponents.client.version>
<httpcomponents.core.version>4.4.13</httpcomponents.core.version>
<!-- Upgrade roaringbit version whenever upgrading iceberg version -->
- <iceberg.version>1.11.0</iceberg.version>
+ <iceberg.version>1.12.0</iceberg.version>
<immutables.version>2.9.2</immutables.version>
<httpcomponents5.core.version>5.3.4</httpcomponents5.core.version>
<httpcomponents5.client.version>5.5</httpcomponents5.client.version>
@@ -211,7 +211,7 @@
<protoc-jar-maven-plugin.version>3.11.4</protoc-jar-maven-plugin.version>
<protoc.path>${env.PROTOC_PATH}</protoc.path>
<rat.version>0.16.1</rat.version>
- <roaringbit.version>1.3.0</roaringbit.version>
+ <roaringbit.version>1.6.23</roaringbit.version>
<stax.version>1.0.1</stax.version>
<slf4j.version>1.7.30</slf4j.version>
<ST4.version>4.0.4</ST4.version>
@@ -288,6 +288,11 @@
<version>${aws-java-sdk.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache5-client</artifactId>
+ <version>${aws-java-sdk.version}</version>
+ </dependency>
<dependency>
<groupId>com.amazonaws.secretsmanager</groupId>
<artifactId>aws-secretsmanager-caching-java</artifactId>
diff --git
a/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java
b/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java
index 859a8ebe601..6357fb4b080 100644
---
a/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java
+++
b/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java
@@ -26,6 +26,7 @@
import java.time.Clock;
import java.util.Arrays;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import javax.servlet.http.HttpServletResponse;
import org.apache.iceberg.BaseTable;
@@ -378,7 +379,12 @@ private RESTResponse tableExists(Map<String, String> vars)
{
private LoadTableResponse loadTable(Map<String, String> vars) {
TableIdentifier ident = identFromPathVars(vars);
- return CatalogHandlers.loadTable(catalog, ident);
+ String snapshots =
vars.get(RESTCatalogProperties.SNAPSHOTS_QUERY_PARAMETER);
+ RESTCatalogProperties.SnapshotMode mode =
RESTCatalogProperties.SNAPSHOT_LOADING_MODE_DEFAULT;
+ if (snapshots != null) {
+ mode =
RESTCatalogProperties.SnapshotMode.valueOf(snapshots.toUpperCase(Locale.ROOT));
+ }
+ return CatalogHandlers.loadTable(catalog, ident, mode);
}
private LoadTableResponse registerTable(Map<String, String> vars, Object
body) {
diff --git
a/standalone-metastore/metastore-rest-catalog/src/test/java/org/apache/iceberg/rest/TestHMSCatalogAdapterSnapshotMode.java
b/standalone-metastore/metastore-rest-catalog/src/test/java/org/apache/iceberg/rest/TestHMSCatalogAdapterSnapshotMode.java
new file mode 100644
index 00000000000..fdd0ad5773a
--- /dev/null
+++
b/standalone-metastore/metastore-rest-catalog/src/test/java/org/apache/iceberg/rest/TestHMSCatalogAdapterSnapshotMode.java
@@ -0,0 +1,176 @@
+/*
+ * 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.rest;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Collections;
+import java.util.Map;
+import javax.servlet.http.HttpServletResponse;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.rest.HTTPRequest.HTTPMethod;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class TestHMSCatalogAdapterSnapshotMode {
+
+ private static final String TABLE_PATH =
"v1/namespaces/accounting/tables/my_table";
+
+ private HMSCatalogAdapter adapter;
+ private HttpServletResponse response;
+ private StringWriter stringWriter;
+
+ @BeforeEach
+ void setup() throws Exception {
+ Catalog catalog =
+ Mockito.mock(
+ Catalog.class,
+ Mockito.withSettings().extraInterfaces(SupportsNamespaces.class,
ViewCatalog.class));
+
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+
+ Snapshot snap1 = Mockito.mock(Snapshot.class);
+ Mockito.when(snap1.snapshotId()).thenReturn(1L);
+ Mockito.when(snap1.sequenceNumber()).thenReturn(1L);
+ Mockito.when(snap1.timestampMillis()).thenReturn(1000L);
+
+ Snapshot snap2 = Mockito.mock(Snapshot.class);
+ Mockito.when(snap2.snapshotId()).thenReturn(2L);
+ Mockito.when(snap2.sequenceNumber()).thenReturn(2L);
+ Mockito.when(snap2.timestampMillis()).thenReturn(2000L);
+
+ TableMetadata initialMetadata =
+ TableMetadata.buildFromEmpty(2)
+ .setLocation("file:///tmp/table")
+ .addSchema(schema)
+ .setDefaultPartitionSpec(PartitionSpec.unpartitioned())
+ .setDefaultSortOrder(SortOrder.unsorted())
+ .addSnapshot(snap1)
+ .addSnapshot(snap2)
+ .setBranchSnapshot(2L, SnapshotRef.MAIN_BRANCH)
+ .discardChanges()
+ .withMetadataLocation("file:///tmp/table/metadata.json")
+ .build();
+
+ BaseTable table = Mockito.mock(BaseTable.class);
+ TableOperations operations = Mockito.mock(TableOperations.class);
+ Mockito.when(table.operations()).thenReturn(operations);
+ Mockito.when(operations.current()).thenReturn(initialMetadata);
+
Mockito.when(catalog.loadTable(Mockito.any(TableIdentifier.class))).thenReturn(table);
+
+ adapter = new HMSCatalogAdapter("test_catalog", catalog, null,
Collections.emptyList());
+
+ response = Mockito.mock(HttpServletResponse.class);
+ stringWriter = new StringWriter();
+ Mockito.when(response.getWriter()).thenReturn(new
PrintWriter(stringWriter));
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (adapter != null) {
+ adapter.close();
+ }
+ }
+
+ private LoadTableResponse executeLoadTable(String snapshotMode) throws
Exception {
+ Map<String, String> queryParams =
+ snapshotMode != null
+ ? ImmutableMap.of(RESTCatalogProperties.SNAPSHOTS_QUERY_PARAMETER,
snapshotMode)
+ : null;
+ return adapter.execute(HTTPMethod.GET, TABLE_PATH, queryParams, null,
response);
+ }
+
+ @Test
+ void testDefaultSnapshotMode() throws Exception {
+ // When snapshots query parameter is omitted, mode defaults to
SnapshotMode.ALL
+ LoadTableResponse res = executeLoadTable(null);
+
+ Assertions.assertNotNull(res, "Response should not be null");
+ Assertions.assertNotNull(res.tableMetadata(), "TableMetadata should not be
null");
+ Assertions.assertEquals(
+ 2,
+ res.tableMetadata().snapshots().size(),
+ "Default mode (ALL) should preserve all snapshots including historical
ones");
+ Assertions.assertNotNull(res.tableMetadata().snapshot(1L), "Snapshot 1
should be retained");
+ Assertions.assertNotNull(res.tableMetadata().snapshot(2L), "Snapshot 2
should be retained");
+ }
+
+ @Test
+ void testSnapshotModeAll() throws Exception {
+ LoadTableResponse res = executeLoadTable("all");
+
+ Assertions.assertNotNull(res, "Response should not be null");
+ Assertions.assertNotNull(res.tableMetadata(), "TableMetadata should not be
null");
+ Assertions.assertEquals(
+ 2, res.tableMetadata().snapshots().size(), "Mode ALL should return all
snapshots");
+ Assertions.assertNotNull(
+ res.tableMetadata().snapshot(1L), "Historical snapshot 1 should be
present");
+ Assertions.assertNotNull(
+ res.tableMetadata().snapshot(2L), "Branch snapshot 2 should be
present");
+ }
+
+ @Test
+ void testSnapshotModeRefs() throws Exception {
+ LoadTableResponse res = executeLoadTable("refs");
+
+ Assertions.assertNotNull(res, "Response should not be null");
+ Assertions.assertNotNull(res.tableMetadata(), "TableMetadata should not be
null");
+ Assertions.assertEquals(
+ 1,
+ res.tableMetadata().snapshots().size(),
+ "Mode REFS should suppress historical snapshots");
+ Assertions.assertNull(
+ res.tableMetadata().snapshot(1L),
+ "Historical snapshot 1 should be suppressed in REFS mode");
+ Assertions.assertNotNull(
+ res.tableMetadata().snapshot(2L), "Branch snapshot 2 should be
retained in REFS mode");
+ }
+
+ @Test
+ void testInvalidSnapshotMode() throws Exception {
+ LoadTableResponse res = executeLoadTable("invalid");
+
+ Assertions.assertNull(res, "Response should be null for an invalid
snapshot mode");
+ Mockito.verify(response).setStatus(400);
+ Assertions.assertTrue(
+ stringWriter.toString().contains("IllegalArgumentException"),
+ "Error response should report IllegalArgumentException");
+ Assertions.assertTrue(
+ stringWriter.toString().contains("INVALID"),
+ "Error response message should mention the invalid mode");
+ }
+}
diff --git a/standalone-metastore/pom.xml b/standalone-metastore/pom.xml
index f532004b138..bc0e48099ed 100644
--- a/standalone-metastore/pom.xml
+++ b/standalone-metastore/pom.xml
@@ -93,7 +93,7 @@
<guava.version>22.0</guava.version>
<hadoop.version>3.4.2</hadoop.version>
<hikaricp.version>4.0.3</hikaricp.version>
- <iceberg.version>1.11.0</iceberg.version>
+ <iceberg.version>1.12.0</iceberg.version>
<jexl.version>3.7.0</jexl.version>
<jackson.version>2.21.5</jackson.version>
<javolution.version>5.5.1</javolution.version>
