You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2022/03/16 08:55:30 UTC

[GitHub] [iceberg] rajarshisarkar opened a new pull request #4342: AWS: Add S3 delete tagging

rajarshisarkar opened a new pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342


   This change adds [S3 Tags](https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-tagging.html) to the objects while deleting using `S3FileIO`. Users can pass their custom tags as part of the catalog properties. This would allow the users to [manage their storage lifecycle](https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html).
   
   Spark SQL launch command:
   ```
   sh spark-sql --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
       --conf spark.sql.catalog.my_catalog.warehouse=s3://iceberg-warehouse/s3-tagging \
       --conf spark.sql.catalog.my_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog \
       --conf spark.sql.catalog.my_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
       --conf spark.sql.catalog.my_catalog.s3.write.tags.my_key=my_val \
       --conf spark.sql.catalog.my_catalog.s3.write.tags.my_key2=my_val2 \
       --conf spark.sql.catalog.my_catalog.s3.delete.tags.my_key3=my_val3
   ```
   
   Tags in S3 after delete:
   
   ```
   aws s3api get-object-tagging --bucket iceberg-warehouse --key s3-tagging/metadata/00000-5d37f925-be01-44d0-87fd-15513606ff6b.metadata.json
   {
       "TagSet": [
           {
               "Key": "my_key2",
               "Value": "my_val2"
           },
           {
               "Key": "my_key3",
               "Value": "my_val3"
           },
           {
               "Key": "my_key",
               "Value": "my_val"
           }
       ]
   }
   ```
   Note: `my_key=my_val` and `my_key2=my_val2` are the tags when the object was written.
   
   ----
   
   cc: @jackye1995 @arminnajafi @singhpk234 @amogh-jahagirdar @xiaoxuandev @yyanyy


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] singhpk234 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

singhpk234 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r828872749



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -244,6 +244,15 @@
    */
   public static final String S3_WRITE_TAGS_PREFIX = "s3.write.tags.";
 
+  /**
+   * Used by {@link S3FileIO} to tag objects when deleting. To set, we can pass a catalog property.
+   * <p>
+   * For more details, see https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-tagging.html
+   * <p>
+   * Example in Spark: --conf spark.sql.catalog.my_catalog.s3.delete.tags.my_key=my_val

Review comment:
       can use  `s3.delete.tags.my_key=my_val` as example to make it engine agnostic.
   
   ref @rdblue [suggestion](https://github.com/apache/iceberg/pull/4334#discussion_r828195275) 
   ` 

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);

Review comment:
       [question] doing it one path at a time can slow stuff. As we will do (GET + PUT) for each one ... can we use multiThreading here ? 

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       [question] what if between our read via a GET and PUT some one updated the TAG ? in this case we might loose the tags (add + deleted) in between as it will be over-written by what my PUT is over-riding. Any thought's on this ?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831630989



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +207,31 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void doSoftDelete(String path) throws S3Exception {

Review comment:
       Or a generic (still helper) tagFile(String path, List<String> deleteTags). the caller, in this case deleteFiles, would be responsible for passing in the deletion tags.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831485921



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -254,6 +254,30 @@
    */
   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 instead of being hard-deleted. This is considered a

Review comment:
       based on the new behavior, the line here should say "objects are tagged with the configured key-value pairs before deletion"




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831492540



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -254,6 +254,30 @@
    */
   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 instead of being hard-deleted. 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_TAGS_THREADS = "s3.delete.tags.num-threads";

Review comment:
       I think there should not be a config that is sharing the prefix with a prefix config. This can just be named as `s3.delete.num-threads`




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rdblue commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rdblue commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073369730

@jackye1995, for case 1, I thought tags would remain on objects in versioned buckets that are deleted with a delete marker. So you could use a tag to control when the old version gets cleaned up, but logically delete it immediately. Isn't that a valid case?

For the question about `s3.delete-enabled` or `io.delete-enabled`, I would go with `s3.delete-enabled` although I could be convinced otherwise.

First, if the intent is to replace the S3 delete with a tag operation instead, then it seems like you wouldn't want to disable deletes across all storage for a table. If you have files stored across multiple schemes, then it doesn't make sense to disable deletes for other schemes.

Second, `io.delete-enabled` would address just use case 4. In that case, why delegate to `FileIO` to skip the delete rather than just not calling delete in the library? I don't think we would want `io.delete-enabled` because it would be impossible to make sure all of the `FileIO` implementations implement it consistently. Plus, you'd still have the problem where you probably don't want to turn it on in all object stores at once. You may have a different orphan file cleanup procedure for each one, or maybe for one you don't have a reliable prefix to list.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829670623



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can parallelize the fetching and updating of tags.
   
   ```      Tasks.foreach(paths)
             .noRetry()
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can parallelize the fetching and updating of tags.
   
   ```     
    Tasks.foreach(paths)
             .noRetry()
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829670623



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a threadpool here? So we can parallelize the fetching and updating of tags.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829672050



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +180,30 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void doSoftDelete(String path) throws S3Exception {
+    S3URI location = new S3URI(path);
+    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 != null && getObjectTaggingResponse.hasTagSet()) {
+      tags.addAll(getObjectTaggingResponse.tagSet());
+    }
+    tags.addAll(awsProperties.s3DeleteTags());

Review comment:
       nit: newline after if block




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073822100


   I have implemented the `s3.delete-enabled` flag which is different from the `s3.delete.tags` config. This should satisfy the "skip delete" and "tag before delete" use cases.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r837393505



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +190,31 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void tagFileToDelete(String path, Set<Tag> deleteTags) throws S3Exception {
+    S3URI location = new S3URI(path);

Review comment:
       Nice catch, missed on adding this after merging master.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829667438



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       Yes, the file is immutable. We add the write tags while creating the object only. So, I don't foresee tags changing (due to Iceberg).




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829671343



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -248,6 +248,15 @@
    */
   public static final String S3_WRITE_TAGS_PREFIX = "s3.write.tags.";
 
+  /**
+   * Used by {@link S3FileIO} to tag objects when deleting. To set, we can pass a catalog property.
+   * <p>
+   * For more details, see https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html

Review comment:
       it's a bit sudden to link this feature to lifecycle management. We should describe that, "when this config is set, objects are tagged with the configured key-value pairs instead of being hard-deleted. 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"




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831490420



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -65,6 +77,7 @@
   private transient S3Client client;
   private MetricsContext metrics = MetricsContext.nullMetrics();
   private final AtomicBoolean isResourceClosed = new AtomicBoolean(false);
+  private static volatile ExecutorService executorService;

Review comment:
       nit: static variable should be placed before instance variables




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831489874



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -94,6 +107,19 @@ public S3FileIO(SerializableSupplier<S3Client> s3) {
   public S3FileIO(SerializableSupplier<S3Client> s3, AwsProperties awsProperties) {
     this.s3 = s3;
     this.awsProperties = awsProperties;
+    if (executorService == null) {
+      synchronized (S3OutputStream.class) {

Review comment:
       should be S3FileIO.class




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] danielcweeks commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

danielcweeks commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1075425376

@jackye1995 I think we need to really clarify the use cases and make sure we're coming up with a solution that really addresses those. This may be a good topic for the next sync as well, since it's getting a little involved in terms of object lifecycle management in S3.

I would say the typical uses cases would be:

*Versioned Bucket:* Life-cycle policy for non-current versions+delete markers with direct delete as there's no requirement for tags.

*Non-Versioned Bucket:* Life-cycle policy for tagged objects and no direct delete (deleting after tagging doesn't really make sense).

I'm not sure if there's a use case for both versioned bucket and tagging for delete.

I would also question the use of life-cycle for transitioning tagged objects since there's a cost associated with the tagging, the transition and, depending on the tier, a minimum charge. It feels like that the use case where it becomes more cost effective gets narrow pretty quickly (as compared to just versioned bucket).

Given that there are these different strategies, we might actually want to separate this behavior into an interface for `DeletionStrategy` where we can provide implementations that are targeted to the use case. That would also allow us to isolate some things (like the executor service which is currently created whether it's used or not).

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r838135756



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -184,6 +245,19 @@ private S3Client client() {
     return client;
   }
 
+  private ExecutorService executorService() {

Review comment:
       I feel not closing is fine. We can keep it as static considering parity with `S3OutputStream`.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rdblue commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rdblue commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1083554671


   Thanks for getting this in, @amogh-jahagirdar and @jackye1995!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829666846



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       I have caught `S3Exception` in caller. 




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r838133144



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -184,6 +245,19 @@ private S3Client client() {
     return client;
   }
 
+  private ExecutorService executorService() {

Review comment:
       I think we cannot close it, because it's a static variable and shared across FileIOs? Which is similar to the one in `S3OutputStream`. We either not close it, or make it an instance variable instead. I think not closing it is fine given we are already doing that in S3OutputStream, any thoughts?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 edited a comment on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 edited a comment on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073306515

> It seems like we're trying to overlap use cases by highjacking delete behavior.

Thanks for the feedback Daniel! I think this is not trying to highjack the delete behavior. Based on what we've talked with some key customers, tagging an object is now the preferred S3 delete implementation in many companies. It is typically a sev0 in a corporation for data loss, so the data platform team prefers to gradually retire the delete files to different tiers and let S3 handle the lifecycle transition.

Note that this is not a discussion of table deletion/snapshot expiration with purge or not. For platforms that allow people to do table deletion with purge, I would imagine this feature to be preferred to avoid user accidentally running a DROP TABLE. And even for platform that does deletion without purge, eventually there is a team who does the "purge" part, and based on the customer discussions we have, this is typically what they do by tagging files and then define lifecycle policy so S3 takes care of the transition of files.

> This also requires that the configuration of the bucket aligns with the tags to actually achieve the desired behavior.

This can be explained in 3 parts:

1. ironically, to satisfy the use case i described, many teams today have to enable S3 object versioning to not hard-delete data files. However, it becomes a huge increase in cost and users have to implement custom workflows then to deal with versioned objects. So some sort of bucket configuration is always needed for enterprise data lake setup, and this tagged approach is going to save cost rather than increase cost for those users.

2. there are many features that won't work with the default S3 behavior and need bucket level settings, some other features we plan to add like S3 acceleration mode, S3 dual stack are all like that. It's mostly done for backwards compatibility. We see a strong adoption of using a lifecycle policy to soft-delete data instead of doing a direct s3 object deletion. There are recently also threads in similar products like Hudi and Delta for such use cases. I would consider this feature to fall into such category of "new features that users have to opt-in" on S3 side.

3. From Iceberg perspective, a file is no longer a part of the table if it is not a part of the Iceberg metadata tree. FileIO.delete is always used AFTER a table level commit that removes the file out of the metadata tree (in the full table deletion case it is also done after catalog operation that removes the table in catalog), so this semantic is still intact even if a file is not actually deleted for people would like to use such a feature.

> Overall, it feels like this is a bit of a reach in terms of functionality.

Let me know what do you think about this area of topic, if you have any better solutions than the proposed one. I think this is a small feature that could enable a lot of valuable use cases. Similar to the access point feature we are adding in S3FileIO, we consider that an intermediate solution which could be solved also at Iceberg level with larger spec change to support relative file path, I think there is value in offering this feature for people who knows how to use it, while we think of anything better that could solve such retention at Iceberg spec level.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 edited a comment on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 edited a comment on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073306515

> It seems like we're trying to overlap use cases by highjacking delete behavior.

Note that this is not a discussion of table deletion/snapshot expiration with purge or not. For platforms that allow people to do table deletion with purge, I would imagine this feature to be preferred to avoid accidental data lose. And even for platform that does deletion without purge, eventually there is a team who does the "purge" part, and based on the customer discussions we have, this is typically what they do by tagging files and then define lifecycle policy so S3 takes care of the transition of files.

> This also requires that the configuration of the bucket aligns with the tags to actually achieve the desired behavior.

This can be explained in 3 parts:

> Overall, it feels like this is a bit of a reach in terms of functionality.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831041772



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -248,6 +248,15 @@
    */
   public static final String S3_WRITE_TAGS_PREFIX = "s3.write.tags.";
 
+  /**
+   * Used by {@link S3FileIO} to tag objects when deleting. To set, we can pass a catalog property.
+   * <p>
+   * For more details, see https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html

Review comment:
       Thanks, for the feedback. I have made the changes.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073367370

> I can also imagine a situations where client-side deletes aren't allowed and the warehouse relies on an orphan files service. That means all of the combinations of "skip delete" and "tag before delete" are valid use cases, so we should separate those.

Yes totally agree. However, should it be `s3.delete-enabled`, or a broader `io.delete-enabled` feature? If a client-side delete is not allowed, it should be applied to all the FileIOs instead of just S3.

I thought about the 4 possible combinations, and I did not ask for an additional `-enabled` flag because:
1. delete enabled + tags set: this case does not really exist, because once the object is deleted in S3 there is no tag anymore.
2. delete disabled + tags set: this is the feature in the PR
3. delete enabled + no tag: this is the existing behavior
4. delete disabled + no tag: this is the case we agree that an orphan file service would take care of everything, and goes back to my question of should this be a feature flag across all FileIOs? If so we can publish a follow-up PR for that.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831047366



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       The document suggests:
   
   ```
   PUT Object tagging – Replaces tags on an object. You specify tags in the request body. There are two distinct scenarios of object tag management using this API.
   
   [1] Object has no tags – Using this API you can add a set of tags to an object (the object has no prior tags).
   [2] Object has a set of existing tags – To modify the existing tag set, you must first retrieve the existing tag set, modify it on the client side, and then use this API to replace the tag set.
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831488382



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -254,6 +254,30 @@
    */
   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 instead of being hard-deleted. 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_TAGS_THREADS = "s3.delete.tags.num-threads";
+
+  /**
+   * Used by {@link S3FileIO} to delete files.

Review comment:
       The description here is not accurate, should say something like "Determines if S3FileIO delete the object when io.delete() is called, default to true. Once disabled, users are expected to set tags through S3_DELETE_TAGS_PREFIX and manage deleted files through S3 lifecycle policy"




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829670796



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -108,6 +115,15 @@ public OutputFile newOutputFile(String path) {
 
   @Override
   public void deleteFile(String path) {
+    if (awsProperties.s3DeleteTags() != null && !awsProperties.s3DeleteTags().isEmpty()) {
+      try {
+        doSoftDelete(path);
+      } catch (S3Exception e) {
+        LOG.warn("Failed to add delete tags: {}", path, e);
+      }
+      return;

Review comment:
       nit: newline after try block




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829667926



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);

Review comment:
       Thanks, I have made the change.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829666775



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -111,11 +116,15 @@ public OutputFile newOutputFile(String path) {
 
   @Override
   public void deleteFile(String path) {
-    S3URI location = new S3URI(path);

Review comment:
       Sure, I have made the changes.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r837105489



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -303,6 +303,32 @@
    */
   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 S3FileIO delete the object when io.delete() is called, default to true. Once
+   * disabled, users are expected to set tags through S3_DELETE_TAGS_PREFIX and manage deleted files

Review comment:
       nit: `{@link #S3_DELETE_TAGS_PREFIX}`




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829670623



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a threadpool here? So we can parallelize the fetching and updating of tags.

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can parallelize the fetching and updating of tags.
   
   ```      Tasks.foreach(paths)
             .noRetry()
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can parallelize the fetching and updating of tags.
   
   ```     
    Tasks.foreach(paths)
             .noRetry()
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can parallelize the fetching and updating of tags. I've been thinking S3FileIO should implicitly have a threadpool but maybe it does not fit in the FileIO model
   
   ```     
    Tasks.foreach(paths)
             .noRetry()
             .executeWith(s3FileIoThreadPool)
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can get better throughput on the fetching and updating of tags. I've been thinking S3FileIO should implicitly have a threadpool but maybe it does not fit in the FileIO model
   
   ```     
    Tasks.foreach(paths)
             .noRetry()
             .executeWith(s3FileIoThreadPool)
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829666775



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -111,11 +116,15 @@ public OutputFile newOutputFile(String path) {
 
   @Override
   public void deleteFile(String path) {
-    S3URI location = new S3URI(path);

Review comment:
       Sure, I have made the changes.

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       I have caught `S3Exception` in caller. 

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       Yes, the file is immutable. We add the write tags while creating the object only. So, I don't foresee tags changing (due to Iceberg).

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());

Review comment:
       I have made the change.

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);

Review comment:
       Thanks, I have made the change.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829671949



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +180,30 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void doSoftDelete(String path) throws S3Exception {
+    S3URI location = new S3URI(path);
+    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 != null && getObjectTaggingResponse.hasTagSet()) {

Review comment:
       I don't think `getObjectTaggingResponse` would ever be null, we can at least trust AWS client with that :)




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r828370707



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());

Review comment:
       we can use the method `response.hasTagSet()` to avoid `tagSet` potentially being null




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r828372740



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -111,11 +116,15 @@ public OutputFile newOutputFile(String path) {
 
   @Override
   public void deleteFile(String path) {
-    S3URI location = new S3URI(path);

Review comment:
       to minimize diff, we can do the reverse logic instead to avoid a lot of indentation change, e.g.
   
   ```
   if (!deleteTags.isEmpty()) {
     doSoftDelete(path);
     return;
   }
   
   // original logic
   ...
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r832169409



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +207,31 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void doSoftDelete(String path) throws S3Exception {

Review comment:
       I have made the changes. Named it as `tagFileToDelete(String path, Set<Tag> deleteTags)`.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073306515

> It seems like we're trying to overlap use cases by highjacking delete behavior.

Note that this is not a discussion of table deletion/snapshot expiration with purge or not. For platforms that allow people to do table deletion with purge, I would imagine this feature to be preferred to avoid accidental data lose. And even for platform that does deletion without purge, eventually there is a team who does the "purge" part, and based on the customer discussions we have, this is typically what they do by tagging files and then define lifecycle policy so S3 takes care of the transition of files.

> This also requires that the configuration of the bucket aligns with the tags to actually achieve the desired behavior.

This can be explained in 3 parts:

1. ironically, to satisfy the use case i described, many teams have to enable S3 object versioning to not hard-delete data files. However, it becomes a huge increase in cost and users have to implement custom workflows then to deal with versioned objects. So some sort of bucket configuration is always needed for enterprise data lake setup, and this tagged approach is going to save cost rather than increase cost for those users.

> Overall, it feels like this is a bit of a reach in terms of functionality.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] danielcweeks commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

danielcweeks commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073298290


   I'm not sure I really follow why we would do this.  It seems like we're trying to overlap use cases by highjacking delete behavior.
   
   This also requires that the configuration of the bucket aligns with the tags to actually achieve the desired behavior.
   
   Overall, it feels like this is a bit of a reach in terms of functionality. 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1081465039


   @danielcweeks do you have any additional suggestions? If not I think the current approach that Ryan suggested and the current code path looks good to me.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831631757



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -94,6 +107,19 @@ public S3FileIO(SerializableSupplier<S3Client> s3) {
   public S3FileIO(SerializableSupplier<S3Client> s3, AwsProperties awsProperties) {
     this.s3 = s3;
     this.awsProperties = awsProperties;
+    if (executorService == null) {
+      synchronized (S3OutputStream.class) {
+        if (executorService == null) {
+          executorService = MoreExecutors.getExitingExecutorService(
+              (ThreadPoolExecutor) Executors.newFixedThreadPool(
+                  awsProperties.s3FileIoDeleteTagThreads(),
+                  new ThreadFactoryBuilder()
+                      .setDaemon(true)
+                      .setNameFormat("iceberg-s3fileio-deletetags-%d")
+                      .build()));
+        }
+      }
+    }

Review comment:
       I think there's a utility for this that we can leverage. ThreadPools.newWorkerPool(String namePrefix, int poolSize)
   
   https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/util/ThreadPools.java#L62




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831036008



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Thanks, I have introduced the executor service configurable via `s3.delete.tags.num-threads`.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831041364



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -248,6 +248,15 @@
    */
   public static final String S3_WRITE_TAGS_PREFIX = "s3.write.tags.";
 
+  /**
+   * Used by {@link S3FileIO} to tag objects when deleting. To set, we can pass a catalog property.
+   * <p>
+   * For more details, see https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html

Review comment:
       Thanks for the feedback. I have made the changes.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rdblue commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rdblue commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073347947


   @danielcweeks, @jackye1995, I think there's definitely value in being able to add tags to let lifecycle policies handle this. But I find it really confusing that adding a delete tag prevents the object from being deleted and instead just tags it.
   
   I think that tagging an object before deleting it and skipping the delete because the tag takes care of it should be separate configurations. It's reasonable for someone to replace deletes with tags, or tag an object and still delete it. I can also imagine a situations where client-side deletes aren't allowed and the warehouse relies on an orphan files service. That means all of the combinations of "skip delete" and "tag before delete" are valid use cases, so we should separate those.
   
   How about implementing this with a `s3.delete-enabled` flag and the `s3.delete.tags` config?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] singhpk234 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

singhpk234 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831255929



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -94,6 +107,19 @@ public S3FileIO(SerializableSupplier<S3Client> s3) {
   public S3FileIO(SerializableSupplier<S3Client> s3, AwsProperties awsProperties) {
     this.s3 = s3;
     this.awsProperties = awsProperties;
+    if (executorService == null) {
+      synchronized (S3OutputStream.class) {
+        if (executorService == null) {
+          executorService = MoreExecutors.getExitingExecutorService(
+              (ThreadPoolExecutor) Executors.newFixedThreadPool(
+                  awsProperties.s3FileIoDeleteTagThreads(),
+                  new ThreadFactoryBuilder()
+                      .setDaemon(true)
+                      .setNameFormat("iceberg-s3fileio-deletetags-%d")
+                      .build()));
+        }
+      }
+    }

Review comment:
       Can we shift this to a private function

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -65,6 +77,7 @@
   private transient S3Client client;
   private MetricsContext metrics = MetricsContext.nullMetrics();
   private final AtomicBoolean isResourceClosed = new AtomicBoolean(false);
+  private static volatile ExecutorService executorService;

Review comment:
       [question] Any reason, executor service need's to static ? 

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +163,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: {}", path, exc))
+          .run(this::doSoftDelete);
+      return;

Review comment:
       we should not return here as the flow will newer reach below if deleteTagging is enabled




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r837393861



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -303,6 +303,32 @@
    */
   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 S3FileIO delete the object when io.delete() is called, default to true. Once
+   * disabled, users are expected to set tags through S3_DELETE_TAGS_PREFIX and manage deleted files

Review comment:
       I have made the changes.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rdblue commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rdblue commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1075778273


   I talked with Dan about this and he made me question my take on `io.deletes-enabled` vs `s3.deletes-enabled`. I'm good either way.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] danielcweeks commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

danielcweeks commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r837674048



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -184,6 +245,19 @@ private S3Client client() {
     return client;
   }
 
+  private ExecutorService executorService() {

Review comment:
       We should probably shutdown the `ExecutorService` on close of the FileIO




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 merged pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 merged pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342


   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r828374563



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       this might throw limit exceeded exception because an object can have max 10 tags. But I guess that's fine, we are not catching any delete failures in the original `s3.deleteObject`, caller needs to catch `S3Exception` in general.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 edited a comment on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 edited a comment on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073306515

> It seems like we're trying to overlap use cases by highjacking delete behavior.

Note that this is not a discussion of table deletion/snapshot expiration with purge or not. For platforms that allow people to do table deletion with purge, I would imagine this feature to be preferred where users could accidentally run a DROP TABLE. And even for platform that does deletion without purge, eventually there is a team who does the "purge" part, and based on the customer discussions we have, this is typically what they do by tagging files and then define lifecycle policy so S3 takes care of the transition of files.

> This also requires that the configuration of the bucket aligns with the tags to actually achieve the desired behavior.

This can be explained in 3 parts:

> Overall, it feels like this is a bit of a reach in terms of functionality.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r830973721



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       Couldn't find one. The PUT request replaces the tags on the object: https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-tagging.html#tagging-apis




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831489611



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +207,31 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void doSoftDelete(String path) throws S3Exception {

Review comment:
       the method should now be named "tagFileToDelete"?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829653214



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       Yes there is a potential risk of race condition. However I think it is fine in most of the "proper" use of Iceberg table. The file should be immutable once written until deletion, I think the chance of people messing up with tagging is relatively low.
   
   In the worst case, a file will miss a tag but is guaranteed to contain a delete tag. It should still go through the lifecycle management based on the specific delete tag.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r828367632



##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -244,6 +244,15 @@
    */
   public static final String S3_WRITE_TAGS_PREFIX = "s3.write.tags.";
 
+  /**
+   * Used by {@link S3FileIO} to tag objects when deleting. To set, we can pass a catalog property.
+   * <p>
+   * For more details, see https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-tagging.html

Review comment:
       instead of this link, we should link to the lifecycle management https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829670623



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can parallelize the fetching and updating of tags. I've been thinking S3FileIO should implicitly have a threadpool but maybe it does not fit in the FileIO model
   
   ```     
    Tasks.foreach(paths)
             .noRetry()
             .executeWith(s3FileIoThreadPool)
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829670623



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       Should we have a configurable threadpool here? So we can get better throughput on the fetching and updating of tags. I've been thinking S3FileIO should implicitly have a threadpool but maybe it does not fit in the FileIO model
   
   ```     
    Tasks.foreach(paths)
             .noRetry()
             .executeWith(s3FileIoThreadPool)
             .suppressFailureWhenFinished()
             .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
             .run(this::doSoftDelete);
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] danielcweeks commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

danielcweeks commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1082103071


   @rajarshisarkar and @jackye1995 One small comment about proper shutdown above, but other than that I'm generally ok with this.  I think we can tackle refactoring some of this logic later (the S3FileIO is now mostly bulk/delete/tagging, so I think there's an opportunity to clean this up).
   
   It is important to note that for the bulk use cases, tagging will be quite a bit slower since we have to tag each object individually.  That may not scale particularly well, but at least the option to tag is there.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] singhpk234 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

singhpk234 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r837151207



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -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: {}", path, e);

Review comment:
       [minor] should we log both deleteTag as well as path : 
   ```suggestion
           LOG.warn("Failed to add delete tags: {} to {}", awsProperties.s3DeleteTags().toString(), path, e);
   ```

##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -303,6 +303,32 @@
    */
   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 S3FileIO delete the object when io.delete() is called, default to true. Once

Review comment:
       Nit: 
   ```suggestion
      * Determines if S3FileIO deletes the object when io.delete() is called, default to true. Once
   ```

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +190,31 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void tagFileToDelete(String path, Set<Tag> deleteTags) throws S3Exception {
+    S3URI location = new S3URI(path);

Review comment:
       can we wire in s3BucketToAccessPointMapping, to use access point in tag manipulation. 
   
   Adding Tag's via access point is an allowed operation : https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-points-usage-examples.html#add-tag-set-ap i.e both GetObjectTagging as well as PutObjectTagging operations are allowed




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1075969127

> I'm not sure if there's a use case for both versioned bucket and tagging for delete.

Yes, after doing some more research I think this is the right conclusion. I thought there was such a use case based on what @rdblue describe, but I've never heard of any.

Regarding having a `DeleteStrategy`, I don't have a strong opinion for the interface either. The original proposal of introducing `s3.delete.tags` is a minimum integration, but if some other approaches are easier for your integration use case we can totally go with that.

I don't know how this interface would look like though, do you have anything in your mind? Because unlike bulk operation, this is just conceptually a "delete" in the user's mind, which should ideally reuse the existing delete interface. So the strategy feels more like a config to me, where we can have `s3.delete.strategy=delete(default)|tagging` to choose from.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r838110750



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -184,6 +245,19 @@ private S3Client client() {
     return client;
   }
 
+  private ExecutorService executorService() {

Review comment:
       Thanks, I have made the changes.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r837393505



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +190,31 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void tagFileToDelete(String path, Set<Tag> deleteTags) throws S3Exception {
+    S3URI location = new S3URI(path);

Review comment:
       Nice catch, missed on added this after merging master.

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -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: {}", path, e);

Review comment:
       Made the changes.

##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -303,6 +303,32 @@
    */
   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 S3FileIO delete the object when io.delete() is called, default to true. Once

Review comment:
       I have made the changes.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1083462118


   This looks good to me, the only point with concern now is the executor service. I think we can merge this, and address the issue if there is a better way from @danielcweeks, otherwise the current implementation is consistent with what we had in `S3OutputStream`. Thanks for the work @rajarshisarkar, and thanks for the review @rdblue @danielcweeks @singhpk234 @amogh-jahagirdar !


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829684768



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       good point, we can follow what we did in https://github.com/apache/iceberg/blob/master/aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java#L110-L122 for this




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829667764



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());

Review comment:
       I have made the change.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r829653214



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       Yes there is a potential risk of race condition. However I think it is fine in most of the "proper" use of Iceberg table. The file should be immutable once written until deletion, I think the chance of people messing up with tagging is relatively low.
   
   In the worst case, a file will miss a tag but is guaranteed to contain a delete tag. It should still go through the lifecycle management based on the specific delete tag.

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -108,6 +115,15 @@ public OutputFile newOutputFile(String path) {
 
   @Override
   public void deleteFile(String path) {
+    if (awsProperties.s3DeleteTags() != null && !awsProperties.s3DeleteTags().isEmpty()) {
+      try {
+        doSoftDelete(path);
+      } catch (S3Exception e) {
+        LOG.warn("Failed to add delete tags: {}", path, e);
+      }
+      return;

Review comment:
       nit: newline after try block

##########
File path: aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java
##########
@@ -248,6 +248,15 @@
    */
   public static final String S3_WRITE_TAGS_PREFIX = "s3.write.tags.";
 
+  /**
+   * Used by {@link S3FileIO} to tag objects when deleting. To set, we can pass a catalog property.
+   * <p>
+   * For more details, see https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html

Review comment:
       it's a bit sudden to link this feature to lifecycle management. We should describe that, "when this config is set, objects are tagged with the configured key-value pairs instead of being hard-deleted. 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"

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +180,30 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void doSoftDelete(String path) throws S3Exception {
+    S3URI location = new S3URI(path);
+    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 != null && getObjectTaggingResponse.hasTagSet()) {

Review comment:
       I don't think `getObjectTaggingResponse` would ever be null, we can at least trust AWS client with that :)

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -155,6 +180,30 @@ public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureExcept
     }
   }
 
+  private void doSoftDelete(String path) throws S3Exception {
+    S3URI location = new S3URI(path);
+    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 != null && getObjectTaggingResponse.hasTagSet()) {
+      tags.addAll(getObjectTaggingResponse.tagSet());
+    }
+    tags.addAll(awsProperties.s3DeleteTags());

Review comment:
       nit: newline after if block

##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -125,6 +141,15 @@ 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()
+          .suppressFailureWhenFinished()
+          .onFailure((path, exc) -> LOG.warn("Failed to add delete tags: {}", path, exc))
+          .run(this::doSoftDelete);

Review comment:
       good point, we can follow what we did in https://github.com/apache/iceberg/blob/master/aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java#L110-L122 for this




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] amogh-jahagirdar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

amogh-jahagirdar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r831631757



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -94,6 +107,19 @@ public S3FileIO(SerializableSupplier<S3Client> s3) {
   public S3FileIO(SerializableSupplier<S3Client> s3, AwsProperties awsProperties) {
     this.s3 = s3;
     this.awsProperties = awsProperties;
+    if (executorService == null) {
+      synchronized (S3OutputStream.class) {
+        if (executorService == null) {
+          executorService = MoreExecutors.getExitingExecutorService(
+              (ThreadPoolExecutor) Executors.newFixedThreadPool(
+                  awsProperties.s3FileIoDeleteTagThreads(),
+                  new ThreadFactoryBuilder()
+                      .setDaemon(true)
+                      .setNameFormat("iceberg-s3fileio-deletetags-%d")
+                      .build()));
+        }
+      }
+    }

Review comment:
       I think there's a utility for this that we can leverage. ThreadPools.newWorkerPool(String namePrefix, int poolSize)




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] jackye1995 edited a comment on pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

jackye1995 edited a comment on pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#issuecomment-1073306515

> It seems like we're trying to overlap use cases by highjacking delete behavior.

Thanks for the feedback Daniel! I think this is not trying to highjack the delete behavior. Based on what we've heard from some key customers, tagging an object is now the preferred S3 delete implementation in many companies. It is typically a sev0 in a corporation for data loss, so the data platform team prefers to gradually retire the delete files to different tiers and let S3 handle the lifecycle transition.

Note that this is not a discussion of table deletion/snapshot expiration with purge or not. For platforms that allow people to do table deletion with purge, I would imagine this feature to be preferred where users could accidentally run a DROP TABLE. And even for platform that does deletion without purge, eventually there is a team who does the "purge" part, and based on the customer discussions we have, this is typically what they do by tagging files and then define lifecycle policy so S3 takes care of the transition of files.

> This also requires that the configuration of the bucket aligns with the tags to actually achieve the desired behavior.

This can be explained in 3 parts:

> Overall, it feels like this is a bit of a reach in terms of functionality.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rajarshisarkar commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rajarshisarkar commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r832170219



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -94,6 +107,19 @@ public S3FileIO(SerializableSupplier<S3Client> s3) {
   public S3FileIO(SerializableSupplier<S3Client> s3, AwsProperties awsProperties) {
     this.s3 = s3;
     this.awsProperties = awsProperties;
+    if (executorService == null) {
+      synchronized (S3OutputStream.class) {
+        if (executorService == null) {
+          executorService = MoreExecutors.getExitingExecutorService(
+              (ThreadPoolExecutor) Executors.newFixedThreadPool(
+                  awsProperties.s3FileIoDeleteTagThreads(),
+                  new ThreadFactoryBuilder()
+                      .setDaemon(true)
+                      .setNameFormat("iceberg-s3fileio-deletetags-%d")
+                      .build()));
+        }
+      }
+    }

Review comment:
       Thanks, for letting me know about the utility method.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [iceberg] rdblue commented on a change in pull request #4342: AWS: Add S3 delete tagging

Posted by GitBox <gi...@apache.org>.

rdblue commented on a change in pull request #4342:
URL: https://github.com/apache/iceberg/pull/4342#discussion_r830671915



##########
File path: aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
##########
@@ -128,36 +137,65 @@ public void deleteFile(String path) {
    */
   @Override
   public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
-    SetMultimap<String, String> bucketToObjects = Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
-    int numberOfFailedDeletions = 0;
-    for (String path : paths) {
-      S3URI location = new S3URI(path);
-      String bucket = location.bucket();
-      String objectKey = location.key();
-      Set<String> objectsInBucket = bucketToObjects.get(bucket);
-      if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
-        List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
-        numberOfFailedDeletions += failedDeletionsForBatch.size();
-        failedDeletionsForBatch.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-        bucketToObjects.removeAll(bucket);
+    if (deleteTags.isEmpty()) {
+      SetMultimap<String, String> bucketToObjects = Multimaps
+          .newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
+      int numberOfFailedDeletions = 0;
+      for (String path : paths) {
+        S3URI location = new S3URI(path);
+        String bucket = location.bucket();
+        String objectKey = location.key();
+        Set<String> objectsInBucket = bucketToObjects.get(bucket);
+        if (objectsInBucket.size() == awsProperties.s3FileIoDeleteBatchSize()) {
+          List<String> failedDeletionsForBatch = deleteObjectsInBucket(bucket, objectsInBucket);
+          numberOfFailedDeletions += failedDeletionsForBatch.size();
+          failedDeletionsForBatch
+              .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+          bucketToObjects.removeAll(bucket);
+        }
+        bucketToObjects.get(bucket).add(objectKey);
       }
-      bucketToObjects.get(bucket).add(objectKey);
-    }
 
-    // Delete the remainder
-    for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap().entrySet()) {
-      final String bucket = bucketToObjectsEntry.getKey();
-      final Collection<String> objects = bucketToObjectsEntry.getValue();
-      List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
-      failedDeletions.forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
-      numberOfFailedDeletions += failedDeletions.size();
-    }
+      // Delete the remainder
+      for (Map.Entry<String, Collection<String>> bucketToObjectsEntry : bucketToObjects.asMap()
+          .entrySet()) {
+        final String bucket = bucketToObjectsEntry.getKey();
+        final Collection<String> objects = bucketToObjectsEntry.getValue();
+        List<String> failedDeletions = deleteObjectsInBucket(bucket, objects);
+        failedDeletions
+            .forEach(failedPath -> LOG.warn("Failed to delete object at path {}", failedPath));
+        numberOfFailedDeletions += failedDeletions.size();
+      }
 
-    if (numberOfFailedDeletions > 0) {
-      throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      if (numberOfFailedDeletions > 0) {
+        throw new BulkDeletionFailureException(numberOfFailedDeletions);
+      }
+    } else {
+      paths.forEach(this::doSoftDelete);
     }
   }
 
+  private void doSoftDelete(String path) {
+    S3URI location = new S3URI(path);
+    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(getObjectTaggingResponse.tagSet());
+    tags.addAll(deleteTags);
+    PutObjectTaggingRequest putObjectTaggingRequest = PutObjectTaggingRequest.builder()

Review comment:
       Is there no way to set additional tags without touching existing tags?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org