This is an automated email from the ASF dual-hosted git repository.
peterxcli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 891cf0e86fe HDDS-14945. Implement Iceberg position delete file rewrite
for path migration (#10306)
891cf0e86fe is described below
commit 891cf0e86fe9b0bec15f86be60db28d1b553c1d3
Author: sreejasahithi <[email protected]>
AuthorDate: Tue Jun 2 21:31:00 2026 +0530
HDDS-14945. Implement Iceberg position delete file rewrite for path
migration (#10306)
---
hadoop-ozone/dist/src/main/license/bin/LICENSE.txt | 14 ++
hadoop-ozone/dist/src/main/license/jar-report.txt | 14 ++
hadoop-ozone/iceberg/pom.xml | 105 +++++++++++
.../ozone/iceberg/RewriteTablePathOzoneAction.java | 208 ++++++++++++++++++++-
.../iceberg/TestRewriteTablePathOzoneAction.java | 194 ++++++++++++++++++-
5 files changed, 526 insertions(+), 9 deletions(-)
diff --git a/hadoop-ozone/dist/src/main/license/bin/LICENSE.txt
b/hadoop-ozone/dist/src/main/license/bin/LICENSE.txt
index e3b97ca202a..736374acf2a 100644
--- a/hadoop-ozone/dist/src/main/license/bin/LICENSE.txt
+++ b/hadoop-ozone/dist/src/main/license/bin/LICENSE.txt
@@ -213,6 +213,7 @@ EDL 1.0
com.sun.activation:jakarta.activation
jakarta.activation:jakarta.activation-api
jakarta.xml.bind:jakarta.xml.bind-api
+ org.locationtech.jts:jts-core
EPL 2.0
@@ -398,6 +399,7 @@ Apache License 2.0
org.apache.hadoop:hadoop-common
org.apache.hadoop:hadoop-hdfs
org.apache.hadoop:hadoop-hdfs-client
+ org.apache.hadoop:hadoop-mapreduce-client-core
org.apache.hadoop:hadoop-shaded-guava
org.apache.hadoop:hadoop-shaded-protobuf_3_25
org.apache.httpcomponents:httpcore
@@ -405,6 +407,8 @@ Apache License 2.0
org.apache.iceberg:iceberg-bundled-guava
org.apache.iceberg:iceberg-common
org.apache.iceberg:iceberg-core
+ org.apache.iceberg:iceberg-orc
+ org.apache.iceberg:iceberg-parquet
org.apache.kerby:kerb-admin
org.apache.kerby:kerb-client
org.apache.kerby:kerb-common
@@ -422,6 +426,16 @@ Apache License 2.0
org.apache.kerby:token-provider
org.apache.logging.log4j:log4j-api
org.apache.logging.log4j:log4j-core
+ org.apache.orc:orc-core
+ org.apache.orc:orc-shims
+ org.apache.parquet:parquet-avro
+ org.apache.parquet:parquet-column
+ org.apache.parquet:parquet-common
+ org.apache.parquet:parquet-encoding
+ org.apache.parquet:parquet-format-structures
+ org.apache.parquet:parquet-hadoop
+ org.apache.parquet:parquet-jackson
+ org.apache.parquet:parquet-variant
org.apache.ranger:ranger-audit-core
org.apache.ranger:ranger-authz-api
org.apache.ranger:ranger-intg
diff --git a/hadoop-ozone/dist/src/main/license/jar-report.txt
b/hadoop-ozone/dist/src/main/license/jar-report.txt
index fc3bdf32e84..8209c270f07 100644
--- a/hadoop-ozone/dist/src/main/license/jar-report.txt
+++ b/hadoop-ozone/dist/src/main/license/jar-report.txt
@@ -65,6 +65,7 @@ share/ozone/lib/hadoop-auth.jar
share/ozone/lib/hadoop-common.jar
share/ozone/lib/hadoop-hdfs-client.jar
share/ozone/lib/hadoop-hdfs.jar
+share/ozone/lib/hadoop-mapreduce-client-core.jar
share/ozone/lib/hadoop-shaded-guava.jar
share/ozone/lib/hadoop-shaded-protobuf_3_25.jar
share/ozone/lib/hdds-cli-common.jar
@@ -91,6 +92,8 @@ share/ozone/lib/iceberg-api.jar
share/ozone/lib/iceberg-bundled-guava.jar
share/ozone/lib/iceberg-common.jar
share/ozone/lib/iceberg-core.jar
+share/ozone/lib/iceberg-orc.jar
+share/ozone/lib/iceberg-parquet.jar
share/ozone/lib/istack-commons-runtime.jar
share/ozone/lib/j2objc-annotations.jar
share/ozone/lib/jackson-annotations.jar
@@ -166,6 +169,7 @@ share/ozone/lib/json-simple.jar
share/ozone/lib/jsp-api.jar
share/ozone/lib/jspecify.jar
share/ozone/lib/jsr311-api.jar
+share/ozone/lib/jts-core.jar
share/ozone/lib/jtokkit.jar
share/ozone/lib/kerb-core.jar
share/ozone/lib/kerb-crypto.jar
@@ -226,6 +230,8 @@ share/ozone/lib/opentelemetry-sdk-logs.jar
share/ozone/lib/opentelemetry-sdk-metrics.jar
share/ozone/lib/opentelemetry-sdk-trace.jar
share/ozone/lib/opentelemetry-sdk.jar
+share/ozone/lib/orc-core-nohive.jar
+share/ozone/lib/orc-shims.jar
share/ozone/lib/osgi-resource-locator.jar
share/ozone/lib/ozone-cli-admin.jar
share/ozone/lib/ozone-cli-debug.jar
@@ -254,6 +260,14 @@ share/ozone/lib/ozone-s3-secret-store.jar
share/ozone/lib/ozone-s3gateway.jar
share/ozone/lib/ozone-tools.jar
share/ozone/lib/ozone-vapor.jar
+share/ozone/lib/parquet-avro.jar
+share/ozone/lib/parquet-column.jar
+share/ozone/lib/parquet-common.jar
+share/ozone/lib/parquet-encoding.jar
+share/ozone/lib/parquet-format-structures.jar
+share/ozone/lib/parquet-hadoop.jar
+share/ozone/lib/parquet-jackson.jar
+share/ozone/lib/parquet-variant.jar
share/ozone/lib/perfmark-api.jar
share/ozone/lib/picocli-shell-jline3.jar
share/ozone/lib/picocli.jar
diff --git a/hadoop-ozone/iceberg/pom.xml b/hadoop-ozone/iceberg/pom.xml
index d7b822c3860..988f7bc6f66 100644
--- a/hadoop-ozone/iceberg/pom.xml
+++ b/hadoop-ozone/iceberg/pom.xml
@@ -32,6 +32,11 @@
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <version>1.12.0</version>
+ </dependency>
<!-- Iceberg dependencies -->
<dependency>
@@ -64,11 +69,111 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-orc</artifactId>
+ <version>${iceberg.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-parquet</artifactId>
+ <version>${iceberg.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-pool</groupId>
+ <artifactId>commons-pool</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.orc</groupId>
+ <artifactId>orc-core</artifactId>
+ <version>1.9.7</version>
+ <classifier>nohive</classifier>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.threeten</groupId>
+ <artifactId>threeten-extra</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-column</artifactId>
+ <version>1.16.0</version>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>runtime</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>com.github.pjfanning</groupId>
+ <artifactId>jersey-json</artifactId>
+ </exclusion>
+
+ <!-- Jersey -->
+ <exclusion>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-guice</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-servlet</artifactId>
+ </exclusion>
+
+ <!-- JAXB -->
+ <exclusion>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ </exclusion>
+ <!-- Hadoop YARN -->
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-common</artifactId>
+ </exclusion>
+
+ <!-- Jetty websocket/client -->
+ <exclusion>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty.websocket</groupId>
+ <artifactId>websocket-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty.websocket</groupId>
+ <artifactId>websocket-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty.websocket</groupId>
+ <artifactId>websocket-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
diff --git
a/hadoop-ozone/iceberg/src/main/java/org/apache/hadoop/ozone/iceberg/RewriteTablePathOzoneAction.java
b/hadoop-ozone/iceberg/src/main/java/org/apache/hadoop/ozone/iceberg/RewriteTablePathOzoneAction.java
index 0327b7a5fd6..4dc50434a3c 100644
---
a/hadoop-ozone/iceberg/src/main/java/org/apache/hadoop/ozone/iceberg/RewriteTablePathOzoneAction.java
+++
b/hadoop-ozone/iceberg/src/main/java/org/apache/hadoop/ozone/iceberg/RewriteTablePathOzoneAction.java
@@ -46,18 +46,33 @@
import org.apache.iceberg.PartitionStatisticsFile;
import org.apache.iceberg.RewriteTablePathUtil;
import org.apache.iceberg.RewriteTablePathUtil.RewriteResult;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableMetadata.MetadataLogEntry;
import org.apache.iceberg.TableMetadataParser;
import org.apache.iceberg.actions.ImmutableRewriteTablePath;
import org.apache.iceberg.actions.RewriteTablePath;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.avro.DataReader;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.data.orc.GenericOrcReader;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.DeleteSchemaUtil;
import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.util.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -228,8 +243,6 @@ private boolean versionInFilePath(String path, String
version) {
}
private String rebuildMetadata() {
- // TODO: position delete file entries in rewriteManifestResult.copyPlan()
reference staging paths
- // that are never written, exclude them until position delete
rewriting is implemented.
TableMetadata startMetadata = startVersionName != null
? new StaticTableOperations(startVersionName, table.io()).current()
: null;
@@ -255,6 +268,13 @@ private String rebuildMetadata() {
RewriteContentFileResult rewriteManifestResult =
rewriteManifests(deltaSnapshotIds, endMetadata,
rewriteManifestListResult.toRewrite());
+ Set<DeleteFile> deleteFiles =
+ rewriteManifestResult.toRewrite().stream()
+ .filter(e -> e instanceof DeleteFile)
+ .map(e -> (DeleteFile) e)
+ .collect(Collectors.toSet());
+ rewritePositionDeletes(deleteFiles);
+
Set<Pair<String, String>> copyPlan = new HashSet<>();
copyPlan.addAll(rewriteVersionResult.copyPlan());
copyPlan.addAll(rewriteManifestListResult.copyPlan());
@@ -675,4 +695,188 @@ private static RewriteResult<DeleteFile>
writeDeleteManifest(
throw new RuntimeIOException(e);
}
}
+
+ static class OzonePositionDeleteReaderWriter implements
RewriteTablePathUtil.PositionDeleteReaderWriter {
+ @Override
+ public CloseableIterable<Record> reader(
+ InputFile inputFile, FileFormat format, PartitionSpec spec) {
+ return positionDeletesReader(inputFile, format, spec);
+ }
+
+ @Override
+ public PositionDeleteWriter<Record> writer(
+ OutputFile outputFile,
+ FileFormat format,
+ PartitionSpec spec,
+ StructLike partition,
+ Schema rowSchema)
+ throws IOException {
+ return positionDeletesWriter(outputFile, format, spec, partition,
rowSchema);
+ }
+ }
+
+ private void rewritePositionDeletes(Set<DeleteFile> toRewrite) {
+ /*
+ * NOTE: Rewriting position delete files updates embedded data file paths,
which changes the
+ * resulting file size. This causes a metadata mismatch in the manifests:
+ *
+ * 1. Dependency: Manifests MUST be rewritten first because they are the
source of truth used to identify which
+ * position delete files exist and need processing.
+ * 2. Issue: Because manifests are written before the delete files are
updated, the 'file_size_in_bytes' field
+ * in the manifest reflects the original size, not the new size.
+ * 3. Impact: Some catalogs (e.g., REST catalogs like Polaris) will fail
to read these files as the reader uses
+ * the stale size from the manifest.
+ *
+ * This is a known Iceberg limitation being addressed by the Iceberg
community. Once that fix is available
+ * in the Iceberg core, this action should be updated accordingly.
+ */
+ if (toRewrite.isEmpty()) {
+ return;
+ }
+
+ RewriteTablePathUtil.PositionDeleteReaderWriter posDeleteReaderWriter =
new OzonePositionDeleteReaderWriter();
+ int maxInFlight = parallelism * MAX_INFLIGHT_MULTIPLIER;
+ Semaphore semaphore = new Semaphore(maxInFlight);
+ ExecutorCompletionService<Void> completionService = new
ExecutorCompletionService<>(executorService);
+ int submittedTasks = 0;
+ int completedTasks = 0;
+
+ try {
+ for (DeleteFile deleteFile : toRewrite) {
+ semaphore.acquire();
+ boolean taskSubmitted = false;
+ try {
+ completionService.submit(() -> {
+ try {
+ rewritePositionDelete(deleteFile, table, sourcePrefix,
targetPrefix, stagingDir, posDeleteReaderWriter);
+ return null;
+ } finally {
+ semaphore.release();
+ }
+ });
+ taskSubmitted = true;
+ submittedTasks++;
+ } finally {
+ if (!taskSubmitted) {
+ semaphore.release();
+ }
+ }
+
+ Future<Void> done;
+ while ((done = completionService.poll()) != null) {
+ done.get();
+ completedTasks++;
+ }
+ }
+
+ while (completedTasks < submittedTasks) {
+ completionService.take().get();
+ completedTasks++;
+ }
+
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ executorService.shutdownNow();
+ throw new RuntimeException("Interrupted while rewriting position delete
files", e);
+
+ } catch (ExecutionException e) {
+ executorService.shutdownNow();
+ throw new RuntimeException("Failed to rewrite position delete file",
e.getCause());
+ }
+ }
+
+ private static void rewritePositionDelete(
+ DeleteFile deleteFile,
+ Table table,
+ String sourcePrefixArg,
+ String targetPrefixArg,
+ String stagingLocationArg,
+ RewriteTablePathUtil.PositionDeleteReaderWriter posDeleteReaderWriter) {
+ try {
+ FileIO io = table.io();
+ String newPath =
+ RewriteTablePathUtil.stagingPath(
+ deleteFile.location(), sourcePrefixArg, stagingLocationArg);
+ OutputFile outputFile = io.newOutputFile(newPath);
+ PartitionSpec spec = table.specs().get(deleteFile.specId());
+ RewriteTablePathUtil.rewritePositionDeleteFile(
+ deleteFile,
+ outputFile,
+ io,
+ spec,
+ sourcePrefixArg,
+ targetPrefixArg,
+ posDeleteReaderWriter);
+ } catch (IOException e) {
+ LOG.error("Failed to rewrite position delete file: {}",
+ deleteFile.location(), e);
+ throw new RuntimeIOException(e);
+ }
+ }
+
+ static CloseableIterable<Record> positionDeletesReader(
+ InputFile inputFile, FileFormat format, PartitionSpec spec) {
+ Schema deleteSchema = DeleteSchemaUtil.posDeleteReadSchema(spec.schema());
+ switch (format) {
+ case AVRO:
+ return Avro.read(inputFile)
+ .project(deleteSchema)
+ .reuseContainers()
+ .createReaderFunc(DataReader::create)
+ .build();
+
+ case PARQUET:
+ return Parquet.read(inputFile)
+ .project(deleteSchema)
+ .reuseContainers()
+ .createReaderFunc(
+ fileSchema -> GenericParquetReaders.buildReader(deleteSchema,
fileSchema))
+ .build();
+
+ case ORC:
+ return ORC.read(inputFile)
+ .project(deleteSchema)
+ .createReaderFunc(fileSchema ->
GenericOrcReader.buildReader(deleteSchema, fileSchema))
+ .build();
+
+ default:
+ LOG.error("Unsupported file format: {} for input file: {}", format,
inputFile.location());
+ throw new UnsupportedOperationException("Unsupported file format: " +
format);
+ }
+ }
+
+ static PositionDeleteWriter<Record> positionDeletesWriter(
+ OutputFile outputFile,
+ FileFormat format,
+ PartitionSpec spec,
+ StructLike partition,
+ Schema rowSchema)
+ throws IOException {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile)
+ .createWriterFunc(DataWriter::create)
+ .withPartition(partition)
+ .rowSchema(rowSchema)
+ .withSpec(spec)
+ .buildPositionWriter();
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile)
+ .createWriterFunc(GenericParquetWriter::create)
+ .withPartition(partition)
+ .rowSchema(rowSchema)
+ .withSpec(spec)
+ .buildPositionWriter();
+ case ORC:
+ return ORC.writeDeletes(outputFile)
+ .createWriterFunc(GenericOrcWriter::buildWriter)
+ .withPartition(partition)
+ .rowSchema(rowSchema)
+ .withSpec(spec)
+ .buildPositionWriter();
+ default:
+ LOG.error("Unsupported file format: {} for output file: {}", format,
outputFile.location());
+ throw new UnsupportedOperationException("Unsupported file format: " +
format);
+ }
+ }
}
diff --git
a/hadoop-ozone/iceberg/src/test/java/org/apache/hadoop/ozone/iceberg/TestRewriteTablePathOzoneAction.java
b/hadoop-ozone/iceberg/src/test/java/org/apache/hadoop/ozone/iceberg/TestRewriteTablePathOzoneAction.java
index 87c9c665ff8..5b11edc1a11 100644
---
a/hadoop-ozone/iceberg/src/test/java/org/apache/hadoop/ozone/iceberg/TestRewriteTablePathOzoneAction.java
+++
b/hadoop-ozone/iceberg/src/test/java/org/apache/hadoop/ozone/iceberg/TestRewriteTablePathOzoneAction.java
@@ -33,6 +33,7 @@
import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.DataFile;
@@ -48,7 +49,11 @@
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionStatisticsFile;
import org.apache.iceberg.RewriteTablePathUtil;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
@@ -57,14 +62,27 @@
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableMetadata.MetadataLogEntry;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.actions.RewriteTablePath;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.Pair;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.Mockito;
/**
@@ -92,7 +110,7 @@ class TestRewriteTablePathOzoneAction {
private Path stagingDir;
@BeforeEach
- public void setupTableLocation() {
+ public void setupTableLocation() throws IOException {
String tableLocation =
tableDir.toUri().toString().replaceFirst("^file:///", "file:/") + TABLE_NAME;
this.table = createTable(tableLocation + "/");
this.sourcePrefix = tableLocation;
@@ -101,7 +119,7 @@ public void setupTableLocation() {
@Test
void fullTablePathRewrite() throws Exception {
- RewriteTablePath.Result result = new RewriteTablePathOzoneAction(table)
+ RewriteTablePath.Result result = new RewriteTablePathOzoneAction(table, 2)
.rewriteLocationPrefix(sourcePrefix, targetPrefix)
.stagingLocation(stagingDir.toString() + "/")
.execute();
@@ -305,9 +323,18 @@ void endVersionRejectsDeletedVersionFile() {
assertThat(exception).hasMessageContaining("does not exist");
}
+ @Test
+ void usesCurrentMetadataIfEndVersionNotProvided() {
+ String currentMetadata = ((HasTableOperations)
table).operations().current().metadataFileLocation();
+ RewriteTablePathOzoneAction action = new
RewriteTablePathOzoneAction(table);
+ action.rewriteLocationPrefix(sourcePrefix,
targetPrefix).stagingLocation(stagingDir + "/");
+ RewriteTablePath.Result result = action.execute();
+
assertThat(result.latestVersion()).isEqualTo(RewriteTablePathUtil.fileName(currentMetadata));
+ }
+
+ @Test
void defaultStagingDirIsUnderTableMetadataLocation() {
String metadataLocation =
RewriteTablePathOzoneUtils.getMetadataLocation(table);
-
RewriteTablePath.Result result = new RewriteTablePathOzoneAction(table)
.rewriteLocationPrefix(sourcePrefix, targetPrefix)
.execute();
@@ -366,6 +393,110 @@ void statsFileCopyPlanReturnsBeforeToAfterPathPairs() {
Pair.of("before-1.stats", "after-1.stats"),
Pair.of("before-2.stats", "after-2.stats")), copyPlan);
}
+
+ @Test
+ void rejectsTablesWithPartitionStatistics() {
+ TableMetadata baseMetadata = ((HasTableOperations)
table).operations().current();
+ long snapshotId = baseMetadata.currentSnapshot().snapshotId();
+ PartitionStatisticsFile statsFile =
Mockito.mock(PartitionStatisticsFile.class);
+ Mockito.when(statsFile.snapshotId()).thenReturn(snapshotId);
+ Mockito.when(statsFile.path()).thenReturn(sourcePrefix +
"/metadata/dummy.stats");
+ Mockito.when(statsFile.fileSizeInBytes()).thenReturn(100L);
+ TableMetadata metadataWithStats = TableMetadata.buildFrom(baseMetadata)
+ .setPartitionStatistics(statsFile)
+ .build();
+
+ TableOperations ops = ((HasTableOperations) table).operations();
+ ops.commit(baseMetadata, metadataWithStats);
+
+ RewriteTablePath action = new RewriteTablePathOzoneAction(table)
+ .rewriteLocationPrefix(sourcePrefix, targetPrefix)
+ .stagingLocation(stagingDir + "/");
+
+ IllegalArgumentException exception =
assertThrows(IllegalArgumentException.class, action::execute);
+ assertThat(exception).hasMessageContaining("Partition statistics files are
not supported yet.");
+ }
+
+ @Test
+ public void positionDeletesReaderUnsupportedFormat() {
+ InputFile mockInput = Mockito.mock(InputFile.class);
+ Mockito.when(mockInput.location()).thenReturn("s3://bucket/test.txt");
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ FileFormat mockUnsupportedFormat = Mockito.mock(FileFormat.class);
+ Mockito.when(mockUnsupportedFormat.toString()).thenReturn("txt");
+
+ UnsupportedOperationException exception =
assertThrows(UnsupportedOperationException.class,
+ () -> RewriteTablePathOzoneAction.positionDeletesReader(mockInput,
mockUnsupportedFormat, spec));
+
+ assertThat(exception).hasMessageContaining("Unsupported file format: txt");
+ }
+
+ @Test
+ public void positionDeletesWriterUnsupportedFormat() {
+ OutputFile mockOutput = Mockito.mock(OutputFile.class);
+ Mockito.when(mockOutput.location()).thenReturn("s3://bucket/test.txt");
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ FileFormat mockUnsupportedFormat = Mockito.mock(FileFormat.class);
+ Mockito.when(mockUnsupportedFormat.toString()).thenReturn("txt");
+
+ UnsupportedOperationException exception =
assertThrows(UnsupportedOperationException.class,
+ () -> RewriteTablePathOzoneAction.positionDeletesWriter(
+ mockOutput, mockUnsupportedFormat, spec, null, null));
+
+ assertThat(exception).hasMessageContaining("Unsupported file format: txt");
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = FileFormat.class, names = {"AVRO", "ORC", "PARQUET"})
+ void positionDeletesAvroAndOrcRoundTrip(FileFormat format, @TempDir Path
temp) throws IOException {
+ String extension = format.name().toLowerCase();
+ String path = temp.resolve("test." + extension).toUri().toString();
+ OutputFile outputFile = table.io().newOutputFile(path);
+ PartitionSpec spec = table.spec();
+
+ try (PositionDeleteWriter<Record> writer =
RewriteTablePathOzoneAction.positionDeletesWriter(
+ outputFile, format, spec, new PartitionData(spec.partitionType()),
SCHEMA)) {
+
+ GenericRecord row = GenericRecord.create(SCHEMA);
+ row.setField("c1", 42);
+ row.setField("c2", format.name() + "-test");
+
+ writer.write(PositionDelete.<Record>create().set("data.parquet", 100L,
row));
+ }
+
+ try (CloseableIterable<Record> reader =
RewriteTablePathOzoneAction.positionDeletesReader(
+ table.io().newInputFile(path), format, spec)) {
+
+ List<Record> results = new ArrayList<>();
+ reader.forEach(results::add);
+
+ assertThat(results).hasSize(1);
+ Record record = results.get(0);
+
+
assertThat(record.getField("file_path").toString()).isEqualTo("data.parquet");
+ assertThat(record.getField("pos")).isEqualTo(100L);
+
+ Record rowResult = (Record) record.getField("row");
+ assertThat(rowResult.getField("c1")).isEqualTo(42);
+ assertThat(rowResult.getField("c2")).isEqualTo(format.name() + "-test");
+ }
+ }
+
+ @Test
+ void manifestsToRewriteRejectsMissingManifestList() {
+ Snapshot snapshot = table.currentSnapshot();
+ String manifestListLocation = snapshot.manifestListLocation();
+ table.io().deleteFile(manifestListLocation);
+
+ RewriteTablePath action = new RewriteTablePathOzoneAction(table)
+ .rewriteLocationPrefix(sourcePrefix, targetPrefix)
+ .stagingLocation(stagingDir + "/");
+
+ RuntimeException exception = assertThrows(RuntimeException.class,
action::execute);
+ assertThat(exception).hasMessageContaining("Failed to collect manifests to
rewrite");
+ assertThat(exception.getCause()).hasMessageContaining("Failed to read
manifests for snapshot " +
+ snapshot.snapshotId());
+ }
/**
* For every staged file in the CSV copy plan, asserts that internal paths
are rewritten
@@ -375,8 +506,8 @@ void statsFileCopyPlanReturnsBeforeToAfterPathPairs() {
* manifest-list references all start with target.</li>
* <li><b>snap-*.avro (manifest-list)</b>: target path starts with target,
and every
* manifest entry path inside the staged file starts with target.</li>
- * <li><b>*.avro (manifest)</b>: target path starts with target (content
rewrite
- * is not yet implemented).</li>
+ * <li><b>*.avro (manifest)</b>: target path starts with target and the
content inside it.</li>
+ * <li><b>deletes.parquet(position delete file)</b>: target path starts
with target and the content inside it.</li>
* </ul>
*/
private void assertAllInternalPathsRewritten(Set<Pair<String, String>>
csvPairs, String target) throws Exception {
@@ -392,6 +523,8 @@ private void
assertAllInternalPathsRewritten(Set<Pair<String, String>> csvPairs,
} else if (RewriteTablePathUtil.fileName(stagingPath).endsWith(".avro"))
{
assertTrue(targetPath.startsWith(target),
"Manifest file target path should start with target prefix: " +
targetPath);
+ } else if (stagingPath.endsWith("deletes.parquet")) {
+ assertStagedDeleteFileInternalPathsRewritten(table, stagingPath,
target);
}
}
}
@@ -538,6 +671,27 @@ private void
assertDeleteManifestPathsRewritten(ManifestFile staged, ManifestFil
"Rewritten delete manifest should reference the same delete files (by
name) as the original");
}
+ private static void assertStagedDeleteFileInternalPathsRewritten(
+ Table tbl, String stagedPath, String targetPrefix) throws IOException {
+ Schema readSchema = DeleteSchemaUtil.pathPosSchema();
+ String pathColumn = MetadataColumns.DELETE_FILE_PATH.name();
+ int rowCount = 0;
+ try (CloseableIterable<Record> rows =
+ Parquet.read(tbl.io().newInputFile(stagedPath))
+ .project(readSchema)
+ .createReaderFunc(fileSchema ->
GenericParquetReaders.buildReader(readSchema, fileSchema))
+ .build()) {
+ for (Record row : rows) {
+ Object path = row.getField(pathColumn);
+ assertTrue(
+ path.toString().startsWith(targetPrefix),
+ stagedPath + " row " + rowCount + ": path '" + path
+ + "' must start with '" + targetPrefix + "'");
+ rowCount++;
+ }
+ }
+ }
+
private static List<String> metadataLogEntryPaths(Table tbl) {
TableMetadata meta = ((HasTableOperations) tbl).operations().current();
List<String> paths = new ArrayList<>();
@@ -569,13 +723,19 @@ private static Set<Pair<String, String>>
readCsvPairs(Table tbl, String fileList
return pairs;
}
- private Table createTable(String location) {
+ private Table createTable(String location) throws IOException {
HadoopTables tables = new HadoopTables(new Configuration());
Table tbl = tables.create(SCHEMA, PartitionSpec.unpartitioned(), new
HashMap<>(), location);
for (int i = 0; i < COMMITS; i++) {
- String dataPath = location + "/data/batch-" + i + ".parquet";
+ String dataPath = location + "data/batch-" + i + ".parquet";
tbl.newAppend().appendFile(dummyDataFile(dataPath)).commit();
}
+
+ for (int i = 0; i < 2; i++) {
+ String dataPath = location + "data/batch-" + i + ".parquet";
+ DeleteFile df = writePositionDeleteFile(tbl, dataPath);
+ tbl.newRowDelta().addDeletes(df).commit();
+ }
return tables.load(location);
}
@@ -588,6 +748,26 @@ private DataFile dummyDataFile(String dataPath) {
.build();
}
+ private DeleteFile writePositionDeleteFile(Table tbl, String
referencedDataPath)
+ throws IOException {
+ String deleteUri = RewriteTablePathUtil.combinePaths(
+ tbl.location(), "data/" + UUID.randomUUID() + "-deletes.parquet");
+ PositionDeleteWriter<Record> writer =
+ Parquet.writeDeletes(tbl.io().newOutputFile(deleteUri))
+ .createWriterFunc(GenericParquetWriter::create)
+ .withSpec(tbl.spec())
+ .withPartition(new PartitionData(tbl.spec().partitionType()))
+ .metricsConfig(MetricsConfig.forPositionDelete(tbl))
+ .overwrite()
+ .buildPositionWriter();
+ try {
+ writer.write(PositionDelete.<Record>create().set(referencedDataPath,
0L));
+ } finally {
+ writer.close();
+ }
+ return writer.toDeleteFile();
+ }
+
private static StatisticsFile statisticsFile(String path, long
fileSizeInBytes) {
return new GenericStatisticsFile(1L, path, fileSizeInBytes, 0L, List.of());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
