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/07 15:37:45 UTC
[L] Change in asterixdb[master]: [NO ISSUE][CLOUD] Derive downloader paths from FileReference
From Hussain Towaileb <hu...@gmail.com>:
Hussain Towaileb has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21715?usp=email )
Change subject: [NO ISSUE][CLOUD] Derive downloader paths from FileReference
......................................................................
[NO ISSUE][CLOUD] Derive downloader paths from FileReference
The Azure and S3 sync parallel downloaders derived a file's local path by
searching the object key for the literal "storage", which only holds when
"storage" is the first segment under the configured prefix. A UDF library
lives at applications/library/storage/..., so the mapping dropped
"applications/library/" and the path no longer resolved, failing Python UDF
deployment on Azure Blob Storage with HYR0135.
Both now strip the configured prefix and resolve against the IO device of
the directory being downloaded, as the S3 transfer-manager and GCS
downloaders already do.
Also fixed: AzureParallelDownloader used the no-overwrite downloadToFile
overload, so no retry could replace a partial write; its error paths
discarded the exception and recorded the failure by re-resolving the
unstripped key, which threw again and escaped the handler; and
getPartitionNumFromRelativePath parsed from a negative index rather than
returning -1 for a path with no partition directory.
Added Python UDF cloud tests for Azure Blob Storage and GCS, which had no
coverage outside S3.
Ext-ref: MB-73763
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: I70c68187df28cde2dfa5255e69acb080fb9a4788
---
A asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonAzureTest.java
A asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonGCSTest.java
R asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonS3Test.java
A asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs-python.conf
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3SyncDownloader.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzureParallelDownloader.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
8 files changed, 435 insertions(+), 42 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/15/21715/1
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonAzureTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonAzureTest.java
new file mode 100644
index 0000000..fd526e4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonAzureTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.test.cloud_storage;
+
+import static org.apache.asterix.api.common.LocalCloudUtil.CLOUD_STORAGE_BUCKET;
+import static org.apache.asterix.api.common.LocalCloudUtilAdobeMock.fillConfigTemplate;
+import static org.apache.asterix.cloud.azure.LSMAzBlobStorageTest.AZURITE_CONTAINER_VER;
+import static org.apache.asterix.test.runtime.ExternalPythonFunctionIT.setNcEndpoints;
+import static org.apache.hyracks.util.file.FileUtil.joinPath;
+
+import java.net.URI;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.asterix.api.common.LocalCloudUtilAdobeMock;
+import org.apache.asterix.app.external.CloudUDFLibrarian;
+import org.apache.asterix.cloud.azure.LSMAzBlobStorageTest;
+import org.apache.asterix.common.config.GlobalConfig;
+import org.apache.asterix.test.common.TestExecutor;
+import org.apache.asterix.test.runtime.LangExecutionUtil;
+import org.apache.asterix.testframework.context.TestCaseContext;
+import org.apache.asterix.testframework.xml.Description;
+import org.apache.asterix.testframework.xml.TestCase;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.MethodSorters;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.testcontainers.azure.AzuriteContainer;
+import org.testcontainers.utility.MountableFile;
+
+import com.azure.core.http.netty.NettyAsyncHttpClientBuilder;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.blob.BlobServiceVersion;
+
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import reactor.netty.http.client.HttpClient;
+
+/**
+ * Run Python UDF tests in cloud deployment environment backed by Azure Blob Storage. This is the azblob
+ * counterpart of {@link CloudPythonS3Test}, which runs the same suite against S3.
+ */
+@RunWith(Parameterized.class)
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "azblob counterpart of CloudPythonS3Test; Python UDFs had no object-storage coverage outside S3")
+public class CloudPythonAzureTest {
+
+ private final TestCaseContext tcCtx;
+ private static final String SUITE_TESTS = "testsuite_it_python.xml";
+ private static final String ONLY_TESTS = "testsuite_cloud_storage_only.xml";
+ private static final String DELTA_RESULT_PATH = "results_cloud";
+ public static final String CONFIG_FILE_TEMPLATE = "src/test/resources/cc-cloud-storage-azblob.ftl";
+ public static final String CONFIG_FILE = "target/cc-cloud-storage-azblob-python.conf";
+ private static final String EXCLUDED_TESTS = "MP";
+ private static final String dsPath =
+ joinPath(System.getProperty("java.io.tmpdir"), "asterixdb_udf", "asterix_nc1_udf.sock");
+
+ private static AzuriteContainer azBlob;
+
+ public CloudPythonAzureTest(TestCaseContext tcCtx) {
+ this.tcCtx = tcCtx;
+ }
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ LocalCloudUtilAdobeMock.startS3CloudEnvironment(true, true);
+ LSMAzBlobStorageTest.generateSelfSignedTLS();
+ MountableFile azureCert = MountableFile.forHostPath("target/azure_test.pfx");
+ azBlob = new AzuriteContainer(AZURITE_CONTAINER_VER).withSsl(azureCert, "password");
+ azBlob.start();
+ SslContext insecureSslContext =
+ SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
+ BlobServiceClient blobServiceClient = new BlobServiceClientBuilder()
+ .serviceVersion(BlobServiceVersion.V2025_07_05).connectionString(azBlob.getConnectionString())
+ .httpClient(new NettyAsyncHttpClientBuilder(
+ HttpClient.create().secure(sslSpec -> sslSpec.sslContext(insecureSslContext).build())).build())
+ .buildClient();
+ URI blobStore = URI.create(blobServiceClient.getAccountUrl());
+ String endpoint = blobStore.getScheme() + "://" + blobStore.getAuthority();
+ fillConfigTemplate(endpoint, CONFIG_FILE_TEMPLATE, CONFIG_FILE);
+
+ cleanup(blobServiceClient);
+ initialize(blobServiceClient);
+
+ TestExecutor testExecutor = new TestExecutor(DELTA_RESULT_PATH);
+ testExecutor.executorId = "cloud";
+ testExecutor.stripSubstring = "//DB:";
+ LangExecutionUtil.setUp(CONFIG_FILE, testExecutor, false, false, new CloudUDFLibrarian(dsPath));
+ setNcEndpoints(testExecutor);
+ System.setProperty(GlobalConfig.CONFIG_FILE_PROPERTY, CONFIG_FILE);
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ LangExecutionUtil.tearDown();
+ LocalCloudUtilAdobeMock.shutdownSilently();
+ if (azBlob != null) {
+ azBlob.close();
+ azBlob.stop();
+ }
+ }
+
+ @Parameters(name = "CloudPythonAzureTest {index}: {0}")
+ public static Collection<Object[]> tests() throws Exception {
+ return LangExecutionUtil.tests(ONLY_TESTS, SUITE_TESTS);
+ }
+
+ @Test
+ public void test() throws Exception {
+ List<TestCase.CompilationUnit> cu = tcCtx.getTestCase().getCompilationUnit();
+ Assume.assumeTrue(cu.size() > 1 || !EXCLUDED_TESTS.equals(getText(cu.get(0).getDescription())));
+ LangExecutionUtil.test(tcCtx);
+ }
+
+ private static String getText(Description description) {
+ return description == null ? "" : description.getValue();
+ }
+
+ private static void cleanup(BlobServiceClient blobServiceClient) {
+ blobServiceClient.deleteBlobContainerIfExists(CLOUD_STORAGE_BUCKET);
+ }
+
+ private static void initialize(BlobServiceClient blobServiceClient) {
+ blobServiceClient.createBlobContainerIfNotExists(CLOUD_STORAGE_BUCKET);
+ }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonGCSTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonGCSTest.java
new file mode 100644
index 0000000..414a304
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonGCSTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.test.cloud_storage;
+
+import static org.apache.asterix.api.common.LocalCloudUtil.CLOUD_STORAGE_BUCKET;
+import static org.apache.asterix.api.common.LocalCloudUtil.MOCK_SERVER_REGION;
+import static org.apache.asterix.test.runtime.ExternalPythonFunctionIT.setNcEndpoints;
+import static org.apache.hyracks.util.file.FileUtil.joinPath;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.asterix.app.external.CloudUDFLibrarian;
+import org.apache.asterix.common.config.GlobalConfig;
+import org.apache.asterix.test.common.TestExecutor;
+import org.apache.asterix.test.runtime.LangExecutionUtil;
+import org.apache.asterix.testframework.context.TestCaseContext;
+import org.apache.asterix.testframework.xml.Description;
+import org.apache.asterix.testframework.xml.TestCase;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.MethodSorters;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageClass;
+import com.google.cloud.storage.StorageOptions;
+
+/**
+ * Run Python UDF tests in cloud deployment environment backed by Google Cloud Storage. This is the {@code gs}
+ * counterpart of {@link CloudPythonS3Test} and {@link CloudPythonAzureTest}, running the same suite against the
+ * fake-gcs-server started by the {@code gcs-tests} Maven profile.
+ * <p>
+ * This uses its own config rather than {@code cc-cloud-storage-gcs.conf}, which registers
+ * {@code UnstableStatementExecutorExtension} for {@link GCSCloudStorageUnstableTest} -- injected statement
+ * failures would make UDF deployment results meaningless here.
+ */
+@RunWith(Parameterized.class)
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "gs counterpart of CloudPythonS3Test; Python UDFs had no object-storage coverage outside S3")
+public class CloudPythonGCSTest {
+
+ private final TestCaseContext tcCtx;
+ private static final String SUITE_TESTS = "testsuite_it_python.xml";
+ private static final String ONLY_TESTS = "testsuite_cloud_storage_only.xml";
+ private static final String DELTA_RESULT_PATH = "results_cloud";
+ private static final String CONFIG_FILE_NAME = "src/test/resources/cc-cloud-storage-gcs-python.conf";
+ private static final String EXCLUDED_TESTS = "MP";
+ private static final String dsPath =
+ joinPath(System.getProperty("java.io.tmpdir"), "asterixdb_udf", "asterix_nc1_udf.sock");
+
+ public static final String MOCK_SERVER_HOSTNAME = "http://127.0.0.1:24443";
+ private static final String MOCK_SERVER_PROJECT_ID = "asterixdb-gcs-test-project-id";
+
+ public CloudPythonGCSTest(TestCaseContext tcCtx) {
+ this.tcCtx = tcCtx;
+ }
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ Storage storage = StorageOptions.newBuilder().setHost(MOCK_SERVER_HOSTNAME)
+ .setCredentials(NoCredentials.getInstance()).setProjectId(MOCK_SERVER_PROJECT_ID).build().getService();
+ cleanup(storage);
+ initialize(storage);
+ storage.close();
+
+ TestExecutor testExecutor = new TestExecutor(DELTA_RESULT_PATH);
+ testExecutor.executorId = "cloud";
+ testExecutor.stripSubstring = "//DB:";
+ LangExecutionUtil.setUp(CONFIG_FILE_NAME, testExecutor, false, false, new CloudUDFLibrarian(dsPath));
+ setNcEndpoints(testExecutor);
+ System.setProperty(GlobalConfig.CONFIG_FILE_PROPERTY, CONFIG_FILE_NAME);
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ LangExecutionUtil.tearDown();
+ }
+
+ @Parameters(name = "CloudPythonGCSTest {index}: {0}")
+ public static Collection<Object[]> tests() throws Exception {
+ return LangExecutionUtil.tests(ONLY_TESTS, SUITE_TESTS);
+ }
+
+ @Test
+ public void test() throws Exception {
+ List<TestCase.CompilationUnit> cu = tcCtx.getTestCase().getCompilationUnit();
+ Assume.assumeTrue(cu.size() > 1 || !EXCLUDED_TESTS.equals(getText(cu.get(0).getDescription())));
+ LangExecutionUtil.test(tcCtx);
+ }
+
+ private static String getText(Description description) {
+ return description == null ? "" : description.getValue();
+ }
+
+ private static void cleanup(Storage storage) {
+ try {
+ Iterable<Blob> blobs = storage.list(CLOUD_STORAGE_BUCKET).iterateAll();
+ blobs.forEach(Blob::delete);
+ storage.delete(CLOUD_STORAGE_BUCKET);
+ } catch (Exception ex) {
+ // ignore
+ }
+ }
+
+ private static void initialize(Storage storage) {
+ storage.create(BucketInfo.newBuilder(CLOUD_STORAGE_BUCKET).setStorageClass(StorageClass.STANDARD)
+ .setLocation(MOCK_SERVER_REGION).build());
+ }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonS3Test.java
similarity index 93%
rename from asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonTest.java
rename to asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonS3Test.java
index ad6f6b4..ed51726 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudPythonS3Test.java
@@ -54,7 +54,7 @@
*/
@RunWith(Parameterized.class)
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
-public class CloudPythonTest {
+public class CloudPythonS3Test {
private static final Logger LOGGER = LogManager.getLogger();
@@ -65,9 +65,10 @@
public static final String CONFIG_FILE_TEMPLATE = "src/test/resources/cc-cloud-storage.conf.ftl";
public static final String CONFIG_FILE = "target/cc-cloud-storage.conf";
private static final String EXCLUDED_TESTS = "MP";
- private static final String dsPath = joinPath("/", "tmp", "asterixdb_udf", "asterix_nc1_udf.sock");
+ private static final String dsPath =
+ joinPath(System.getProperty("java.io.tmpdir"), "asterixdb_udf", "asterix_nc1_udf.sock");
- public CloudPythonTest(TestCaseContext tcCtx) {
+ public CloudPythonS3Test(TestCaseContext tcCtx) {
this.tcCtx = tcCtx;
}
@@ -92,7 +93,7 @@
LocalCloudUtilAdobeMock.shutdownSilently();
}
- @Parameters(name = "CloudPythonTest {index}: {0}")
+ @Parameters(name = "CloudPythonS3Test {index}: {0}")
public static Collection<Object[]> tests() throws Exception {
return LangExecutionUtil.tests(ONLY_TESTS, SUITE_TESTS);
}
diff --git a/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs-python.conf b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs-python.conf
new file mode 100644
index 0000000..d1b4bd2
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs-python.conf
@@ -0,0 +1,76 @@
+; 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.
+
+[nc/asterix_nc1]
+txn.log.dir=target/tmp/asterix_nc1/txnlog
+core.dump.dir=target/tmp/asterix_nc1/coredump
+iodevices=target/tmp/asterix_nc1/iodevice1
+iodevices=../asterix-server/target/tmp/asterix_nc1/iodevice2
+nc.api.port=19004
+#jvm.args=-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5006
+
+[nc/asterix_nc2]
+ncservice.port=9091
+txn.log.dir=target/tmp/asterix_nc2/txnlog
+core.dump.dir=target/tmp/asterix_nc2/coredump
+iodevices=target/tmp/asterix_nc2/iodevice1,../asterix-server/target/tmp/asterix_nc2/iodevice2
+nc.api.port=19005
+#jvm.args=-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5007
+
+[nc]
+credential.file=src/test/resources/security/passwd
+python.cmd.autolocate=true
+python.env=FOO=BAR=BAZ,BAR=BAZ
+address=127.0.0.1
+command=asterixnc
+app.class=org.apache.asterix.hyracks.bootstrap.NCApplication
+jvm.args=-Xmx4096m -Dnode.Resolver="org.apache.asterix.external.util.IdentitiyResolverFactory"
+storage.buffercache.size=128MB
+storage.memorycomponent.globalbudget=512MB
+storage.max.columns.in.zeroth.segment=800
+
+[cc]
+address = 127.0.0.1
+app.class=org.apache.asterix.hyracks.bootstrap.CCApplication
+heartbeat.period=2000
+heartbeat.max.misses=25
+credential.file=src/test/resources/security/passwd
+
+[common]
+log.dir = logs/
+log.level = INFO
+compiler.framesize=32KB
+compiler.sortmemory=320KB
+compiler.groupmemory=160KB
+compiler.joinmemory=256KB
+compiler.textsearchmemory=160KB
+compiler.windowmemory=192KB
+compiler.ordered.fields=false
+compiler.internal.sanitycheck=true
+messaging.frame.size=4096
+messaging.frame.count=512
+cloud.deployment=true
+storage.buffercache.pagesize=32KB
+storage.partitioning=static
+cloud.storage.scheme=gs
+cloud.storage.bucket=cloud-storage-container
+cloud.storage.region=us-west2
+cloud.storage.endpoint=http://127.0.0.1:24443
+cloud.storage.anonymous.auth=true
+cloud.storage.cache.policy=selective
+cloud.max.write.requests.per.second=1000
+cloud.max.read.requests.per.second=5000
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
index 5f7ff0b..1c71b4c 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
@@ -24,7 +24,6 @@
import org.apache.hyracks.api.io.FileReference;
public interface IParallelDownloader extends AutoCloseable {
- String STORAGE_SUB_DIR = "storage";
/**
* Downloads files in all partitions
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3SyncDownloader.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3SyncDownloader.java
index 498c34b..898adb7 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3SyncDownloader.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/aws/s3/S3SyncDownloader.java
@@ -39,6 +39,7 @@
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.control.nc.io.IOManager;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.hyracks.util.annotations.ThreadSafe;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -54,7 +55,6 @@
private static final Logger LOGGER = LogManager.getLogger();
private final String bucket;
- private final IOManager ioManager;
private final S3Client s3Client;
private final S3ClientConfig config;
private final IRequestProfilerLimiter profiler;
@@ -62,7 +62,6 @@
S3SyncDownloader(String bucket, IOManager ioManager, S3ClientConfig config, IRequestProfilerLimiter profiler) {
this.bucket = bucket;
- this.ioManager = ioManager;
this.config = config;
this.profiler = profiler;
this.s3Client = (S3Client) S3CloudClient.buildClient(config).getConsumingClient();
@@ -139,18 +138,17 @@
List<Future<?>> downloads = new ArrayList<>();
int maxPending = config.getRequestsMaxPendingHttpConnections();
- List<S3Object> downloadObjects = new ArrayList<>();
+ List<FileReference> downloadTargets = new ArrayList<>();
for (FileReference fileReference : toDownload) {
profiler.objectMultipartDownload();
String prefix = config.getPrefix() + fileReference.getRelativePath();
List<S3Object> objects = S3ClientUtils.listS3Objects(s3Client, bucket, prefix);
- downloadObjects.addAll(objects);
+ for (S3Object s3Object : objects) {
+ downloadTargets.add(createDiskFileRef(fileReference, s3Object.key()));
+ }
}
- for (S3Object s3Object : downloadObjects) {
- String key = createDiskSubPath(s3Object.key());
- FileReference targetFile = ioManager.resolve(key);
-
+ for (FileReference targetFile : downloadTargets) {
FileUtils.createParentDirectories(targetFile.getFile());
Future<Void> future = executorService.submit(() -> {
@@ -160,8 +158,7 @@
} catch (IOException e) {
// Record failed file
failedFiles.add(targetFile);
- LOGGER.debug("Failed to download file using sync client: file {} having s3Key: {}", targetFile,
- s3Object.key(), e);
+ LOGGER.error("Failed to download file using sync client: {}", targetFile, e);
}
return null;
});
@@ -187,11 +184,17 @@
}
}
- private String createDiskSubPath(String objectName) {
- if (!objectName.startsWith(STORAGE_SUB_DIR)) {
- objectName = objectName.substring(objectName.indexOf(STORAGE_SUB_DIR));
- }
- return objectName;
+ /**
+ * Maps an object key onto its local file, on the same IO device as the directory the download was requested for.
+ * The key is the device-relative path of the file behind the configured prefix, so only that prefix is stripped
+ * -- the remainder is the local path verbatim, and {@link #downloadFile} re-adds the prefix to get the key back.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "MB-73763: replaces createDiskSubPath, which searched for 'storage' and so mangled "
+ + "applications/library/storage/... paths (and threw for a key without 'storage' at all)")
+ private FileReference createDiskFileRef(FileReference directory, String objectKey) {
+ String prefix = config.getPrefix();
+ String relativePath = objectKey.startsWith(prefix) ? objectKey.substring(prefix.length()) : objectKey;
+ return new FileReference(directory.getDeviceHandle(), relativePath);
}
@Override
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzureParallelDownloader.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzureParallelDownloader.java
index 57f044d..56981d5 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzureParallelDownloader.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/azure/blobstorage/AzureParallelDownloader.java
@@ -36,6 +36,9 @@
import org.apache.hyracks.api.util.ExceptionUtils;
import org.apache.hyracks.control.nc.io.IOManager;
import org.apache.hyracks.util.ExponentialRetryPolicy;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import com.azure.storage.blob.BlobAsyncClient;
import com.azure.storage.blob.BlobContainerAsyncClient;
@@ -47,7 +50,7 @@
import reactor.util.retry.Retry;
public class AzureParallelDownloader extends AbstractParallelDownloader {
- private final IOManager ioManager;
+ private static final Logger LOGGER = LogManager.getLogger();
private final BlobContainerAsyncClient blobContainerAsyncClient;
private final IRequestProfilerLimiter profiler;
private final AzBlobStorageClientConfig config;
@@ -55,7 +58,6 @@
public AzureParallelDownloader(IOManager ioManager, BlobContainerAsyncClient blobContainerAsyncClient,
IRequestProfilerLimiter profiler, AzBlobStorageClientConfig config) {
- this.ioManager = ioManager;
this.blobContainerAsyncClient = blobContainerAsyncClient;
this.profiler = profiler;
this.config = config;
@@ -85,7 +87,9 @@
BlobAsyncClient blobAsyncClient =
blobContainerAsyncClient.getBlobAsyncClient(config.getPrefix() + fileReference.getRelativePath());
- Mono<Void> downloadTask = blobAsyncClient.downloadToFile(absPath.toString()).then();
+ // overwrite: a download replaces whatever the local cache holds, and a retry must be able to
+ // replace a partially written file from the previous attempt
+ Mono<Void> downloadTask = blobAsyncClient.downloadToFile(absPath.toString(), true).then();
downloads.add(downloadTask);
if (maxConcurrent > 0 && downloads.size() >= maxConcurrent) {
@@ -112,7 +116,10 @@
List<Mono<Void>> directoryDownloads = new ArrayList<>();
for (FileReference directory : directories) {
- Mono<Void> directoryTask = downloadDirectoryAsync(directory, failedFiles).onErrorResume(e -> Mono.empty()); // Continue even if a directory fails
+ Mono<Void> directoryTask = downloadDirectoryAsync(directory, failedFiles).onErrorResume(e -> {
+ LOGGER.error("Failed to download directory {}", directory, e);
+ return Mono.empty(); // Continue even if a directory fails
+ });
directoryDownloads.add(directoryTask);
}
@@ -123,17 +130,21 @@
return failedFiles;
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "pass the requested directory down so the blob's local path is resolved against its device; log failures")
private Mono<Void> downloadDirectoryAsync(FileReference directory, Set<FileReference> failedFiles) {
return getBlobItems(directory).flatMap(blobItem -> {
profiler.objectGet();
- return downloadBlobAsync(blobItem, failedFiles);
- }, config.getRequestsMaxPendingHttpConnections()).then().doOnError(error -> failedFiles.add(directory)); // Record directory failure
+ return downloadBlobAsync(directory, blobItem, failedFiles);
+ }, config.getRequestsMaxPendingHttpConnections()).then().doOnError(error -> {
+ LOGGER.error("Failed to download blobs under {}", getListPrefix(directory), error);
+ failedFiles.add(directory); // Record directory failure
+ });
}
- private Mono<Void> downloadBlobAsync(BlobItem blobItem, Set<FileReference> failedFiles) {
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "MB-73763: derive the local path by stripping the configured prefix instead of searching for 'storage'")
+ private Mono<Void> downloadBlobAsync(FileReference directory, BlobItem blobItem, Set<FileReference> failedFiles) {
+ FileReference diskDestFile = createDiskFileRef(directory, blobItem.getName());
try {
- // Resolve destination path
- FileReference diskDestFile = ioManager.resolve(createDiskSubPath(blobItem.getName()));
Path absDiskBlobPath = getDiskDestPath(diskDestFile);
Path parentDiskPath = absDiskBlobPath.getParent();
@@ -141,22 +152,30 @@
BlobAsyncClient blobAsyncClient = blobContainerAsyncClient.getBlobAsyncClient(blobItem.getName());
- return blobAsyncClient.downloadToFile(absDiskBlobPath.toString()).doOnError(error -> {
- FileReference failedFile = ioManager.resolve(blobItem.getName());
- failedFiles.add(failedFile);
+ // overwrite: a download replaces whatever the local cache holds, and a retry must be able to
+ // replace a partially written file from the previous attempt
+ return blobAsyncClient.downloadToFile(absDiskBlobPath.toString(), true).doOnError(error -> {
+ LOGGER.error("Failed to download blob {} into {}", blobItem.getName(), diskDestFile, error);
+ failedFiles.add(diskDestFile);
}).then();
} catch (Exception e) {
- failedFiles.add(ioManager.resolve(blobItem.getName()));
+ LOGGER.error("Failed to prepare download of blob {} into {}", blobItem.getName(), diskDestFile, e);
+ failedFiles.add(diskDestFile);
return Mono.error(HyracksDataException.create(e));
}
}
- private String createDiskSubPath(String blobName) {
- int idx = blobName.indexOf(STORAGE_SUB_DIR);
- if (idx >= 0) {
- return blobName.substring(idx);
- }
- return blobName;
+ /**
+ * Maps a blob name onto its local file, on the same IO device as the directory the download was requested for.
+ * The blob name is the device-relative path of the file behind the configured prefix, so only that prefix is
+ * stripped -- the remainder is the local path verbatim.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "MB-73763: replaces createDiskSubPath, which searched for 'storage' and so mangled "
+ + "applications/library/storage/... paths")
+ private FileReference createDiskFileRef(FileReference directory, String blobName) {
+ String prefix = config.getPrefix();
+ String relativePath = blobName.startsWith(prefix) ? blobName.substring(prefix.length()) : blobName;
+ return new FileReference(directory.getDeviceHandle(), relativePath);
}
private void createDirectories(Path parentPath) throws HyracksDataException {
@@ -178,11 +197,14 @@
}
private Flux<BlobItem> getBlobItems(FileReference directoryToDownload) {
- ListBlobsOptions listBlobsOptions =
- new ListBlobsOptions().setPrefix(config.getPrefix() + directoryToDownload.getRelativePath());
+ ListBlobsOptions listBlobsOptions = new ListBlobsOptions().setPrefix(getListPrefix(directoryToDownload));
return blobContainerAsyncClient.listBlobs(listBlobsOptions);
}
+ private String getListPrefix(FileReference directoryToDownload) {
+ return config.getPrefix() + directoryToDownload.getRelativePath();
+ }
+
@Override
public void close() {
// Closing Azure Blob Clients is not required as the underlying netty connection pool
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
index 2fdfba4..d424788 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
@@ -115,8 +115,15 @@
return datasetName + File.separator + rebalanceCount + File.separator + idxName;
}
+ /**
+ * @return the partition number of the supplied path, or -1 if the path has no partition directory
+ */
public static int getPartitionNumFromRelativePath(String relativePath) {
- int startIdx = relativePath.lastIndexOf(PARTITION_DIR_PREFIX) + PARTITION_DIR_PREFIX.length();
+ int prefixIdx = relativePath.lastIndexOf(PARTITION_DIR_PREFIX);
+ if (prefixIdx < 0) {
+ return -1;
+ }
+ int startIdx = prefixIdx + PARTITION_DIR_PREFIX.length();
int partitionEndIdx = relativePath.indexOf(File.separatorChar, startIdx);
int idxEnd = partitionEndIdx != -1 ? partitionEndIdx : relativePath.length();
String partition = relativePath.substring(startIdx, idxEnd);
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21715?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: I70c68187df28cde2dfa5255e69acb080fb9a4788
Gerrit-Change-Number: 21715
Gerrit-PatchSet: 1
Gerrit-Owner: Hussain Towaileb <hu...@gmail.com>