You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/09/13 17:06:30 UTC
[M] Change in asterixdb[master]: [NO ISSUE][EXT]: Use S3 ListObjectsV2 paginator
From Hussain Towaileb <hu...@gmail.com>:
Hussain Towaileb has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21780?usp=email )
Change subject: [NO ISSUE][EXT]: Use S3 ListObjectsV2 paginator
......................................................................
[NO ISSUE][EXT]: Use S3 ListObjectsV2 paginator
Page through S3 ListObjectsV2 results with the AWS SDK's
ListObjectsV2Iterable instead of the hand-written continuation-token
loops, in both the external-dataset listing (S3Utils) and the
cloud-storage listing (S3ClientUtils). The iterable requests the next
page with the previous page's continuation token until the listing is
complete.
Added a unit test covering multi-page listing with chained continuation
tokens and directory-marker filtering on every page.
Ext-ref: MB-57193
Co-Authored-By: Claude Fable 5.1 <no...@anthropic.com>
Change-Id: Iad7db8d5467f71b20d1de34ed5e2fe6a1dbf35e9
---
M asterixdb/asterix-cloud/pom.xml
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtils.java
A asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtilsListObjectsTest.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3Utils.java
4 files changed, 168 insertions(+), 38 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/80/21780/1
diff --git a/asterixdb/asterix-cloud/pom.xml b/asterixdb/asterix-cloud/pom.xml
index 61ba8f4..1967f37 100644
--- a/asterixdb/asterix-cloud/pom.xml
+++ b/asterixdb/asterix-cloud/pom.xml
@@ -156,6 +156,10 @@
<artifactId>hyracks-cloud</artifactId>
</dependency>
<dependency>
+ <groupId>org.apache.hyracks</groupId>
+ <artifactId>hyracks-util</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.apache.asterix</groupId>
<artifactId>asterix-common</artifactId>
<version>${project.version}</version>
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtils.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtils.java
index 3aefc37..5d6d38f 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtils.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtils.java
@@ -18,6 +18,10 @@
*/
package org.apache.asterix.cloud.clients.aws.s3;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_FABLE_5_1;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.REFACTORED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
import java.io.File;
import java.net.URI;
import java.net.URISyntaxException;
@@ -28,11 +32,13 @@
import java.util.function.Predicate;
import org.apache.asterix.external.util.aws.s3.S3Utils;
+import org.apache.hyracks.util.annotations.AiProvenance;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
import software.amazon.awssdk.services.s3.model.S3Object;
+import software.amazon.awssdk.services.s3.paginators.ListObjectsV2Iterable;
public class S3ClientUtils {
@@ -40,30 +46,18 @@
throw new AssertionError("do not instantiate");
}
+ @AiProvenance(agent = CLAUDE_FABLE_5_1, tool = CLAUDE_CODE_UI, contributionKind = REFACTORED, notes = "Paginate with the SDK ListObjectsV2 iterable instead of a manual continuation-token loop")
public static List<S3Object> listS3Objects(S3Client s3Client, String bucket, String path) {
- String newMarker = null;
- ListObjectsV2Response listObjectsResponse;
- ListObjectsV2Request.Builder listObjectsBuilder = ListObjectsV2Request.builder().bucket(bucket);
- listObjectsBuilder.prefix(toCloudPrefix(path));
+ ListObjectsV2Request listObjectsRequest =
+ ListObjectsV2Request.builder().bucket(bucket).prefix(toCloudPrefix(path)).build();
List<S3Object> files = new ArrayList<>();
- while (true) {
- // List the objects from the start, or from the last marker in case of truncated result
- if (newMarker == null) {
- listObjectsResponse = s3Client.listObjectsV2(listObjectsBuilder.build());
- } else {
- listObjectsResponse = s3Client.listObjectsV2(listObjectsBuilder.continuationToken(newMarker).build());
- }
+ // the iterable requests the next page with the previous page's continuation token until the result is complete
+ ListObjectsV2Iterable listObjectsIterable = s3Client.listObjectsV2Paginator(listObjectsRequest);
+ for (ListObjectsV2Response listObjectsResponse : listObjectsIterable) {
// ignore objects ending with "/" since we don't create such objects.
// S3 can return folders as objects with size 0 and key ending with "/"
listObjectsResponse.contents().stream().filter(Predicate.not(S3Utils::isDirectory)).forEach(files::add);
-
- // Mark the flag as done if done, otherwise, get the marker of the previous response for the next request
- if (Boolean.FALSE.equals(listObjectsResponse.isTruncated())) {
- break;
- } else {
- newMarker = listObjectsResponse.nextContinuationToken();
- }
}
return files;
}
diff --git a/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtilsListObjectsTest.java b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtilsListObjectsTest.java
new file mode 100644
index 0000000..9600c17
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/test/java/org/apache/asterix/cloud/clients/aws/s3/S3ClientUtilsListObjectsTest.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.cloud.clients.aws.s3;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.apache.hyracks.util.annotations.AiProvenance.Agent;
+import org.apache.hyracks.util.annotations.AiProvenance.ContributionKind;
+import org.apache.hyracks.util.annotations.AiProvenance.Tool;
+import org.junit.Test;
+
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+/**
+ * Verifies that {@link S3ClientUtils#listS3Objects} walks every page of a truncated ListObjectsV2 result, chaining
+ * each request off the previous page's continuation token. A hand-rolled {@link S3Client} serves the pages, because
+ * the in-memory S3 mock used by the other tests returns every key in a single page and never truncates.
+ */
+@AiProvenance(agent = Agent.CLAUDE_FABLE_5_1, tool = Tool.CLAUDE_CODE_UI, contributionKind = ContributionKind.TEST_GENERATED)
+public class S3ClientUtilsListObjectsTest {
+
+ private static final String BUCKET = "bucket";
+ private static final String PREFIX = "storage/partition_0";
+
+ @Test
+ public void walksAllPagesChainingContinuationTokens() {
+ PagedS3Client client = new PagedS3Client(page(keys("a", 1000), "token-1"), page(keys("b", 1000), "token-2"),
+ page(keys("c", 999), null));
+
+ List<S3Object> listed = S3ClientUtils.listS3Objects(client, BUCKET, "/" + PREFIX);
+
+ assertEquals(2999, listed.size());
+ assertEquals(2999, listed.stream().map(S3Object::key).distinct().count());
+ assertEquals(Arrays.asList(null, "token-1", "token-2"), client.receivedTokens);
+ for (ListObjectsV2Request request : client.receivedRequests) {
+ assertEquals(BUCKET, request.bucket());
+ assertEquals(PREFIX, request.prefix());
+ }
+ }
+
+ @Test
+ public void singlePageIsRequestedOnce() {
+ PagedS3Client client = new PagedS3Client(page(keys("a", 3), null));
+
+ List<S3Object> listed = S3ClientUtils.listS3Objects(client, BUCKET, PREFIX);
+
+ assertEquals(3, listed.size());
+ assertEquals(1, client.receivedRequests.size());
+ assertNull(client.receivedTokens.get(0));
+ }
+
+ @Test
+ public void directoryMarkersAreDroppedOnEveryPage() {
+ List<String> firstPage = new ArrayList<>(keys("a", 2));
+ firstPage.add(PREFIX + "/dir-a/");
+ List<String> secondPage = new ArrayList<>(keys("b", 2));
+ secondPage.add(PREFIX + "/dir-b/");
+ PagedS3Client client = new PagedS3Client(page(firstPage, "token-1"), page(secondPage, null));
+
+ List<S3Object> listed = S3ClientUtils.listS3Objects(client, BUCKET, PREFIX);
+
+ assertEquals(4, listed.size());
+ assertEquals(0, listed.stream().filter(object -> object.key().endsWith("/")).count());
+ }
+
+ private static List<String> keys(String tag, int count) {
+ return IntStream.range(0, count).mapToObj(i -> PREFIX + "/" + tag + "-" + i).collect(Collectors.toList());
+ }
+
+ private static ListObjectsV2Response page(List<String> keys, String nextContinuationToken) {
+ List<S3Object> contents =
+ keys.stream().map(key -> S3Object.builder().key(key).size(1L).build()).collect(Collectors.toList());
+ return ListObjectsV2Response.builder().contents(contents).keyCount(contents.size())
+ .isTruncated(nextContinuationToken != null).nextContinuationToken(nextContinuationToken).build();
+ }
+
+ /**
+ * Serves a fixed sequence of pages; every operation other than ListObjectsV2 keeps the SDK's default behaviour
+ * of throwing, so the test fails loudly if the code under test starts calling anything else.
+ */
+ private static final class PagedS3Client implements S3Client {
+ private final List<ListObjectsV2Response> pages;
+ private final List<ListObjectsV2Request> receivedRequests = new ArrayList<>();
+ private final List<String> receivedTokens = new ArrayList<>();
+
+ private PagedS3Client(ListObjectsV2Response... pages) {
+ this.pages = Arrays.asList(pages);
+ }
+
+ @Override
+ public ListObjectsV2Response listObjectsV2(ListObjectsV2Request request) {
+ receivedRequests.add(request);
+ receivedTokens.add(request.continuationToken());
+ String token = request.continuationToken();
+ if (token == null) {
+ return pages.get(0);
+ }
+ for (int i = 0; i < pages.size() - 1; i++) {
+ if (token.equals(pages.get(i).nextContinuationToken())) {
+ return pages.get(i + 1);
+ }
+ }
+ throw new IllegalArgumentException("unknown continuation token " + token);
+ }
+
+ @Override
+ public String serviceName() {
+ return SERVICE_NAME;
+ }
+
+ @Override
+ public void close() {
+ // nothing to release
+ }
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3Utils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3Utils.java
index 212bbea..d2fd0ee 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3Utils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3Utils.java
@@ -75,7 +75,10 @@
import static org.apache.asterix.external.util.aws.s3.S3Constants.INPUT_STREAM_TYPE_FIELD_NAME;
import static org.apache.asterix.external.util.aws.s3.S3Constants.PATH_STYLE_ADDRESSING_FIELD_NAME;
import static org.apache.hyracks.api.util.ExceptionUtils.getMessageOrToString;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_FABLE_5_1;
import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_SONNET_4_6;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.REFACTORED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
import static org.apache.hyracks.util.annotations.AiProvenance.Tool.GITHUB_COPILOT;
import java.io.ByteArrayInputStream;
@@ -589,36 +592,22 @@
* @param prefix definition prefix
* @param includeExcludeMatcher include/exclude matchers to apply
*/
+ @AiProvenance(agent = CLAUDE_FABLE_5_1, tool = CLAUDE_CODE_UI, contributionKind = REFACTORED, notes = "Paginate with the SDK ListObjectsV2 iterable instead of a manual continuation-token loop")
private static List<S3Object> listS3Objects(S3Client s3Client, String container, String prefix,
AbstractExternalInputStreamFactory.IncludeExcludeMatcher includeExcludeMatcher,
ExternalDataPrefix externalDataPrefix, IExternalFilterEvaluator evaluator,
IWarningCollector warningCollector) throws HyracksDataException {
- String newMarker = null;
List<S3Object> filesOnly = new ArrayList<>();
+ ListObjectsV2Request listObjectsRequest =
+ ListObjectsV2Request.builder().bucket(container).prefix(prefix).build();
- ListObjectsV2Response listObjectsResponse;
- ListObjectsV2Request.Builder listObjectsBuilder = ListObjectsV2Request.builder().bucket(container);
- listObjectsBuilder.prefix(prefix);
-
- while (true) {
- // List the objects from the start, or from the last marker in case of truncated result
- if (newMarker == null) {
- listObjectsResponse = s3Client.listObjectsV2(listObjectsBuilder.build());
- } else {
- listObjectsResponse = s3Client.listObjectsV2(listObjectsBuilder.continuationToken(newMarker).build());
- }
-
+ // the iterable requests the next page with the previous page's continuation token until the result is complete
+ ListObjectsV2Iterable listObjectsIterable = s3Client.listObjectsV2Paginator(listObjectsRequest);
+ for (ListObjectsV2Response listObjectsResponse : listObjectsIterable) {
// Collect the paths to files only
collectAndFilterFiles(listObjectsResponse.contents(), includeExcludeMatcher.getPredicate(),
includeExcludeMatcher.getMatchersList(), filesOnly, externalDataPrefix, evaluator,
warningCollector);
-
- // Mark the flag as done if done, otherwise, get the marker of the previous response for the next request
- if (listObjectsResponse.isTruncated() != null && listObjectsResponse.isTruncated()) {
- newMarker = listObjectsResponse.nextContinuationToken();
- } else {
- break;
- }
}
return filesOnly;
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21780?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Iad7db8d5467f71b20d1de34ed5e2fe6a1dbf35e9
Gerrit-Change-Number: 21780
Gerrit-PatchSet: 1
Gerrit-Owner: Hussain Towaileb <hu...@gmail.com>