AWS: Add S3 delete tagging

aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java

@@ -303,6 +303,32 @@ public class AwsProperties implements Serializable {
    */
   public static final String S3_WRITE_TAGS_PREFIX = "s3.write.tags.";
 
+  /**
+   * Used by {@link S3FileIO} to tag objects when deleting. When this config is set, objects are
+   * tagged with the configured key-value pairs before deletion. This is considered a soft-delete,
+   * because users are able to configure tag-based object lifecycle policy at bucket level to
+   * transition objects to different tiers.
+   * <p>
+   * For more details, see https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html
+   * <p>
+   * Example: s3.delete.tags.my_key=my_val
+   */
+  public static final String S3_DELETE_TAGS_PREFIX = "s3.delete.tags.";
+
+  /**
+   * Number of threads to use for adding delete tags to S3 objects, default to {@link
+   * Runtime#availableProcessors()}
+   */
+  public static final String S3FILEIO_DELETE_THREADS = "s3.delete.num-threads";
+
+  /**
+   * Determines if {@link S3FileIO} deletes the object when io.delete() is called, default to true. Once
+   * disabled, users are expected to set tags through {@link #S3_DELETE_TAGS_PREFIX} and manage
+   * deleted files through S3 lifecycle policy.
+   */
+  public static final String S3_DELETE_ENABLED = "s3.delete-enabled";
+  public static final boolean S3_DELETE_ENABLED_DEFAULT = true;
+
   /**
    * Used by {@link S3FileIO}, prefix used for bucket access point configuration.
    * To set, we can pass a catalog property.
@@ -342,6 +368,9 @@ public class AwsProperties implements Serializable {
   private ObjectCannedACL s3FileIoAcl;
   private boolean isS3ChecksumEnabled;
   private final Set<Tag> s3WriteTags;
+  private final Set<Tag> s3DeleteTags;
+  private int s3FileIoDeleteThreads;
+  private boolean isS3DeleteEnabled;
   private final Map<String, String> s3BucketToAccessPointMapping;
 
   private String glueCatalogId;
@@ -363,6 +392,9 @@ public AwsProperties() {
     this.s3fileIoStagingDirectory = System.getProperty("java.io.tmpdir");
     this.isS3ChecksumEnabled = S3_CHECKSUM_ENABLED_DEFAULT;
     this.s3WriteTags = Sets.newHashSet();
+    this.s3DeleteTags = Sets.newHashSet();
+    this.s3FileIoDeleteThreads = Runtime.getRuntime().availableProcessors();
+    this.isS3DeleteEnabled = S3_DELETE_ENABLED_DEFAULT;
     this.s3BucketToAccessPointMapping = ImmutableMap.of();
 
     this.glueCatalogId = null;
@@ -427,6 +459,10 @@ public AwsProperties(Map<String, String> properties) {
         String.format("Deletion batch size must be between 1 and %s", S3FILEIO_DELETE_BATCH_SIZE_MAX));
 
     this.s3WriteTags = toTags(properties, S3_WRITE_TAGS_PREFIX);
+    this.s3DeleteTags = toTags(properties, S3_DELETE_TAGS_PREFIX);
+    this.s3FileIoDeleteThreads = PropertyUtil.propertyAsInt(properties, S3FILEIO_DELETE_THREADS,
+            Runtime.getRuntime().availableProcessors());
+    this.isS3DeleteEnabled = PropertyUtil.propertyAsBoolean(properties, S3_DELETE_ENABLED, S3_DELETE_ENABLED_DEFAULT);
     this.s3BucketToAccessPointMapping = PropertyUtil.propertiesWithPrefix(properties, S3_ACCESS_POINTS_PREFIX);
 
     this.dynamoDbTableName = PropertyUtil.propertyAsString(properties, DYNAMODB_TABLE_NAME,
@@ -549,6 +585,26 @@ public Set<Tag> s3WriteTags() {
     return s3WriteTags;
   }
 
+  public Set<Tag> s3DeleteTags() {
+    return s3DeleteTags;
+  }
+
+  public int s3FileIoDeleteThreads() {
+    return s3FileIoDeleteThreads;
+  }
+
+  public void setS3FileIoDeleteThreads(int threads) {
+    this.s3FileIoDeleteThreads = threads;
+  }
+
+  public boolean isS3DeleteEnabled() {
+    return isS3DeleteEnabled;
+  }
+
+  public void setS3DeleteEnabled(boolean s3DeleteEnabled) {
+    this.isS3DeleteEnabled = s3DeleteEnabled;
+  }
+
   private Set<Tag> toTags(Map<String, String> properties, String prefix) {
     return PropertyUtil.propertiesWithPrefix(properties, prefix)
         .entrySet().stream()

aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import org.apache.iceberg.aws.AwsClientFactories;
@@ -40,14 +41,22 @@
 import org.apache.iceberg.relocated.com.google.common.collect.SetMultimap;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.SerializableSupplier;
+import org.apache.iceberg.util.Tasks;
+import org.apache.iceberg.util.ThreadPools;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.Delete;
 import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
 import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
 import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
+import software.amazon.awssdk.services.s3.model.GetObjectTaggingRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectTaggingResponse;
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
+import software.amazon.awssdk.services.s3.model.PutObjectTaggingRequest;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.Tag;
+import software.amazon.awssdk.services.s3.model.Tagging;
 
 /**
  * FileIO implementation backed by S3.
@@ -59,6 +68,7 @@
 public class S3FileIO implements FileIO, SupportsBulkOperations {
   private static final Logger LOG = LoggerFactory.getLogger(S3FileIO.class);
   private static final String DEFAULT_METRICS_IMPL = "org.apache.iceberg.hadoop.HadoopMetricsContext";
+  private static volatile ExecutorService executorService;
 
   private SerializableSupplier<S3Client> s3;
   private AwsProperties awsProperties;
@@ -108,6 +118,18 @@ public OutputFile newOutputFile(String path) {
 
   @Override
   public void deleteFile(String path) {
+    if (awsProperties.s3DeleteTags() != null && !awsProperties.s3DeleteTags().isEmpty()) {
+      try {
+        tagFileToDelete(path, awsProperties.s3DeleteTags());
+      } catch (S3Exception e) {
+        LOG.warn("Failed to add delete tags: {} to {}", awsProperties.s3DeleteTags(), path, e);
+      }
+    }
+
+    if (!awsProperties.isS3DeleteEnabled()) {
+      return;
+    }
+
     S3URI location = new S3URI(path, awsProperties.s3BucketToAccessPointMapping());
     DeleteObjectRequest deleteRequest =
         DeleteObjectRequest.builder().bucket(location.bucket()).key(location.key()).build();
@@ -125,6 +147,20 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
+    if (awsProperties.s3DeleteTags() != null && !awsProperties.s3DeleteTags().isEmpty()) {
+      Tasks.foreach(paths)
+          .noRetry()
+          .executeWith(executorService())
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {} to {}",
+              awsProperties.s3DeleteTags(), path, exc))
+          .run(path -> tagFileToDelete(path, awsProperties.s3DeleteTags()));
+    }
+
+    if (!awsProperties.isS3DeleteEnabled()) {
+      return;
+    }
+
     SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
     int numberOfFailedDeletions = 0;
     for (String path : paths) {
@@ -155,6 +191,31 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void tagFileToDelete(String path, Set<Tag> deleteTags) throws S3Exception {
+    S3URI location = new S3URI(path, awsProperties.s3BucketToAccessPointMapping());
+    String bucket = location.bucket();
+    String objectKey = location.key();
+    GetObjectTaggingRequest getObjectTaggingRequest = GetObjectTaggingRequest.builder()
+        .bucket(bucket)
+        .key(objectKey)
+        .build();
+    GetObjectTaggingResponse getObjectTaggingResponse = client()
+        .getObjectTagging(getObjectTaggingRequest);
+    // Get existing tags, if any and then add the delete tags
+    Set<Tag> tags = Sets.newHashSet();
+    if (getObjectTaggingResponse.hasTagSet()) {
+      tags.addAll(getObjectTaggingResponse.tagSet());
+    }
+
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()
+        .bucket(bucket)
+        .key(objectKey)
+        .tagging(Tagging.builder().tagSet(tags).build())
+        .build();
+    client().putObjectTagging(putObjectTaggingRequest);
+  }
+
   private List<String> deleteObjectsInBucket(String bucket, Collection<String> objects) {
     if (!objects.isEmpty()) {
       List<ObjectIdentifier> objectIds = objects
@@ -184,6 +245,19 @@ private S3Client client() {
     return client;
   }
 
+  private ExecutorService executorService() {
+    if (executorService == null) {
+      synchronized (S3FileIO.class) {
+        if (executorService == null) {
+          executorService = ThreadPools.newWorkerPool(
+              "iceberg-s3fileio-delete", awsProperties.s3FileIoDeleteThreads());
+        }
+      }
+    }
+
+    return executorService;
+  }
+
   @Override
   public void initialize(Map<String, String> properties) {
     this.awsProperties = new AwsProperties(properties);

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3OutputStream.java

@@ -92,7 +92,8 @@ public class TestS3OutputStream {
       AwsProperties.S3FILEIO_MULTIPART_SIZE, Integer.toString(5 * 1024 * 1024),
       AwsProperties.S3FILEIO_STAGING_DIRECTORY, tmpDir.toString(),
       "s3.write.tags.abc", "123",
-      "s3.write.tags.def", "789"));
+      "s3.write.tags.def", "789",
+      "s3.delete.tags.xyz", "456"));
 
   public TestS3OutputStream() throws IOException {
   }