This is an automated email from the ASF dual-hosted git repository.
clintropolis pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new d11b956bd6c feat: google cloud storage SegmentRangeReader for partial
loads (#20454)
d11b956bd6c is described below
commit d11b956bd6c39472c0225a4c17497a4224e0670d
Author: Clint Wylie <[email protected]>
AuthorDate: Wed Sep 30 20:41:01 2026 -0700
feat: google cloud storage SegmentRangeReader for partial loads (#20454)
---
extensions-core/azure-extensions/pom.xml | 7 +
.../storage/azure/AzureDataSegmentPusherTest.java | 55 +++--
.../storage/google/GoogleDataSegmentPusher.java | 20 +-
.../druid/storage/google/GoogleLoadSpec.java | 41 ++++
.../storage/google/GoogleSegmentRangeReader.java | 132 ++++++++++++
.../google/GoogleDataSegmentPusherTest.java | 96 ++++++++-
.../druid/storage/google/GoogleLoadSpecTest.java | 91 +++++++++
.../google/GoogleSegmentRangeReaderTest.java | 223 +++++++++++++++++++++
.../druid/storage/s3/S3DataSegmentPusherTest.java | 58 +++++-
9 files changed, 700 insertions(+), 23 deletions(-)
diff --git a/extensions-core/azure-extensions/pom.xml
b/extensions-core/azure-extensions/pom.xml
index 4147e7f8ce2..dd91b981a3c 100644
--- a/extensions-core/azure-extensions/pom.xml
+++ b/extensions-core/azure-extensions/pom.xml
@@ -132,6 +132,13 @@
<scope>provided</scope>
</dependency>
<!-- Tests -->
+ <dependency>
+ <groupId>org.apache.druid</groupId>
+ <artifactId>druid-processing</artifactId>
+ <version>${project.parent.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
diff --git
a/extensions-core/azure-extensions/src/test/java/org/apache/druid/storage/azure/AzureDataSegmentPusherTest.java
b/extensions-core/azure-extensions/src/test/java/org/apache/druid/storage/azure/AzureDataSegmentPusherTest.java
index 6bc3f316647..d8a32f634e1 100644
---
a/extensions-core/azure-extensions/src/test/java/org/apache/druid/storage/azure/AzureDataSegmentPusherTest.java
+++
b/extensions-core/azure-extensions/src/test/java/org/apache/druid/storage/azure/AzureDataSegmentPusherTest.java
@@ -24,9 +24,13 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.io.Files;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.MapUtils;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.loading.DeepStorageSegmentConfig;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.partition.LinearShardSpec;
@@ -47,6 +51,7 @@ import java.util.regex.Pattern;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -292,30 +297,56 @@ public class AzureDataSegmentPusherTest extends
EasyMockSupport
AzureDataSegmentPusher pusher =
new AzureDataSegmentPusher(azureStorage, azureAccountConfig,
segmentConfigWithPrefix, NO_ZIP_CONFIG);
- // version.bin = [0, 0, 0, 0x0A] → IndexIO.V10_VERSION, a single
range-readable druid.segment
- Files.write(new byte[]{0x0, 0x0, 0x0, 0x0A},
tempPath.resolve("version.bin").toFile());
+ // A real V10 segment rather than a hand-made file, so the layout under
test is whatever IndexMergerV10 actually
+ // writes: one druid.segment whose leading byte is the version
getVersionFromDir reads.
+ final File segmentDir = buildV10Segment(tempPath);
+ final File[] segmentFiles = segmentDir.listFiles();
+ assertNotNull(segmentFiles);
final String expectedDir = PREFIX + "/" +
pusher.getStorageDir(SEGMENT_TO_PUSH, false);
- azureStorage.uploadBlockBlob(
- EasyMock.anyObject(File.class),
- EasyMock.eq(CONTAINER_NAME),
- EasyMock.eq(expectedDir + "/version.bin"),
- EasyMock.eq(MAX_TRIES)
- );
- EasyMock.expectLastCall();
+ final ImmutableList.Builder<String> listing = ImmutableList.builder();
+ for (final File file : segmentFiles) {
+ azureStorage.uploadBlockBlob(
+ EasyMock.anyObject(File.class),
+ EasyMock.eq(CONTAINER_NAME),
+ EasyMock.eq(expectedDir + "/" + file.getName()),
+ EasyMock.eq(MAX_TRIES)
+ );
+ EasyMock.expectLastCall();
+ listing.add(expectedDir + "/" + file.getName());
+ }
+ // everything under the path was written by this push, so there is nothing
to clean up
EasyMock.expect(azureStorage.listBlobs(CONTAINER_NAME, expectedDir + "/",
null, MAX_TRIES))
- .andReturn(ImmutableList.of(expectedDir + "/version.bin"));
+ .andReturn(listing.build());
replayAll();
- DataSegment segment = pusher.push(tempPath.toFile(), SEGMENT_TO_PUSH,
false);
+ DataSegment segment = pusher.push(segmentDir, SEGMENT_TO_PUSH, false);
- assertEquals(10, (int) segment.getBinaryVersion());
+ // a V10 segment is a single file, and that is what makes it range-readable
+ assertEquals(1, segmentFiles.length);
+ assertEquals(IndexIO.V10_FILE_NAME, segmentFiles[0].getName());
+ assertEquals(IndexIO.V10_VERSION, (int) segment.getBinaryVersion());
assertEquals(Boolean.TRUE, segment.getLoadSpec().get("rangeable"));
verifyAll();
}
+ private static File buildV10Segment(Path tempPath)
+ {
+ return IndexBuilder.create()
+ .useV10()
+ .tmpDir(tempPath.resolve("v10").toFile())
+ .rows(ImmutableList.of(
+ new MapBasedInputRow(
+ DateTimes.of("2015-01-01"),
+ ImmutableList.of("dim"),
+ ImmutableMap.of("dim", "a")
+ )
+ ))
+ .buildMMappedIndexFile();
+ }
+
@Test
public void test_pushZip_doesNotStampRangeable(@TempDir Path tempPath)
throws Exception
{
diff --git
a/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleDataSegmentPusher.java
b/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleDataSegmentPusher.java
index 144e43a1331..f98c2fa3f22 100644
---
a/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleDataSegmentPusher.java
+++
b/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleDataSegmentPusher.java
@@ -28,6 +28,7 @@ import org.apache.druid.java.util.common.IOE;
import org.apache.druid.java.util.common.RE;
import org.apache.druid.java.util.common.RetryUtils;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.SegmentUtils;
import org.apache.druid.segment.loading.DataSegmentPusher;
import org.apache.druid.segment.loading.DeepStorageSegmentConfig;
@@ -181,8 +182,11 @@ public class GoogleDataSegmentPusher implements
DataSegmentPusher
deleteStaleObjects(dirPath, pushedPaths);
+ // V10 unzipped is rangeable: a single druid.segment with a range-readable
header. V9 unzipped is a directory of
+ // separate smoosh files the range-read path can't consume.
+ final boolean rangeable = version == IndexIO.V10_VERSION;
return segment.withSize(size)
- .withLoadSpec(makeLoadSpec(config.getBucket(), dirPath))
+ .withLoadSpec(makeLoadSpec(config.getBucket(), dirPath,
rangeable))
.withBinaryVersion(version);
}
@@ -246,4 +250,18 @@ public class GoogleDataSegmentPusher implements
DataSegmentPusher
);
}
+ /**
+ * Variant that stamps {@link GoogleLoadSpec#RANGEABLE} so {@link
GoogleLoadSpec#openRangeReader()} can decide
+ * range-read eligibility. Used by the unzipped push path where the binary
version is known at write time.
+ */
+ private Map<String, Object> makeLoadSpec(String bucket, String path, boolean
rangeable)
+ {
+ return ImmutableMap.of(
+ "type", GoogleStorageDruidModule.SCHEME,
+ "bucket", bucket,
+ "path", path,
+ GoogleLoadSpec.RANGEABLE, rangeable
+ );
+ }
+
}
diff --git
a/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleLoadSpec.java
b/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleLoadSpec.java
index 4629762a7ba..3a3d95a3f4f 100644
---
a/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleLoadSpec.java
+++
b/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleLoadSpec.java
@@ -21,29 +21,43 @@ package org.apache.druid.storage.google;
import com.fasterxml.jackson.annotation.JacksonInject;
import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
import com.google.common.base.Preconditions;
import org.apache.druid.segment.loading.LoadSpec;
import org.apache.druid.segment.loading.SegmentLoadingException;
+import org.apache.druid.segment.loading.SegmentRangeReader;
+import org.apache.druid.utils.CompressionUtils;
+import javax.annotation.Nullable;
import java.io.File;
@JsonTypeName(GoogleStorageDruidModule.SCHEME)
public class GoogleLoadSpec implements LoadSpec
{
+ static final String RANGEABLE = "rangeable";
+
@JsonProperty
private final String bucket;
@JsonProperty
private final String path;
+ /**
+ * Stamped at push time when {@link GoogleDataSegmentPusher} writes a
segment in a layout that supports byte-range
+ * reads. Only {@code Boolean.TRUE} enables {@link #openRangeReader()};
absence or {@code false} means full-download.
+ */
+ @Nullable
+ private final Boolean rangeable;
+
private final GoogleDataSegmentPuller puller;
@JsonCreator
public GoogleLoadSpec(
@JsonProperty("bucket") String bucket,
@JsonProperty("path") String path,
+ @JsonProperty(RANGEABLE) @Nullable Boolean rangeable,
@JacksonInject GoogleDataSegmentPuller puller
)
{
@@ -51,6 +65,7 @@ public class GoogleLoadSpec implements LoadSpec
Preconditions.checkNotNull(path);
this.bucket = bucket;
this.path = path;
+ this.rangeable = rangeable;
this.puller = puller;
}
@@ -59,4 +74,30 @@ public class GoogleLoadSpec implements LoadSpec
{
return new LoadSpecResult(puller.getSegmentFiles(bucket, path,
file).size());
}
+
+ /**
+ * Returns a {@link GoogleSegmentRangeReader} when the segment was stamped
{@link #rangeable} {@code true} at push
+ * time and isn't a zip; otherwise {@code null}.
+ */
+ @Nullable
+ @Override
+ public SegmentRangeReader openRangeReader()
+ {
+ if (CompressionUtils.isZip(path) || !Boolean.TRUE.equals(rangeable)) {
+ return null;
+ }
+ return new GoogleSegmentRangeReader(puller.storage, bucket, path);
+ }
+
+ /**
+ * Returns the range-reads-supported flag stamped at push time, or {@code
null} for legacy segments pushed before
+ * this field existed (which will load via the full-download path).
+ */
+ @JsonProperty(RANGEABLE)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @Nullable
+ public Boolean getRangeable()
+ {
+ return rangeable;
+ }
}
diff --git
a/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleSegmentRangeReader.java
b/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleSegmentRangeReader.java
new file mode 100644
index 00000000000..a42311ff01f
--- /dev/null
+++
b/extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleSegmentRangeReader.java
@@ -0,0 +1,132 @@
+/*
+ * 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.druid.storage.google;
+
+import com.google.common.base.Preconditions;
+import org.apache.druid.data.input.impl.RetryingInputStream;
+import org.apache.druid.data.input.impl.prefetch.ObjectOpenFunction;
+import org.apache.druid.segment.loading.SegmentRangeReader;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * {@link SegmentRangeReader} backed by Google Cloud Storage range reads. The
segment is expected to be stored as raw
+ * (unzipped) objects under a common path, i.e. the layout produced by {@code
GoogleDataSegmentPusher.pushNoZip} where
+ * each segment file is uploaded as {@code pathPrefix + file.getName()}. Each
{@link #readRange} call resolves the
+ * target object as {@code pathPrefix + filename} and opens a read channel
limited to {@code [offset, offset + length)}.
+ * <p>
+ * The returned stream is wrapped in a {@link RetryingInputStream} with the
{@link GoogleUtils#GOOGLE_RETRY} predicate,
+ * the same retry policy {@link GoogleDataSegmentPuller} uses for full-segment
downloads. The GCS client retries a
+ * failed request on its own, but only the {@code RetryingInputStream} can
resume a read that failed partway through:
+ * it reopens at the byte offset already consumed, so a transient mid-stream
error becomes a fresh range read for the
+ * remaining bytes rather than a restart of the whole range.
+ */
+public class GoogleSegmentRangeReader implements SegmentRangeReader
+{
+ private final GoogleStorage storage;
+ private final String bucket;
+ private final String pathPrefix;
+
+ public GoogleSegmentRangeReader(GoogleStorage storage, String bucket, String
pathPrefix)
+ {
+ this.storage = Preconditions.checkNotNull(storage, "storage");
+ this.bucket = Preconditions.checkNotNull(bucket, "bucket");
+ this.pathPrefix = Preconditions.checkNotNull(pathPrefix, "pathPrefix");
+ }
+
+ @Override
+ public InputStream readRange(String filename, long offset, long length)
throws IOException
+ {
+ Preconditions.checkNotNull(filename, "filename");
+ Preconditions.checkArgument(offset >= 0, "offset must be non-negative, got
[%s]", offset);
+ Preconditions.checkArgument(length >= 0, "length must be non-negative, got
[%s]", length);
+
+ if (length == 0) {
+ // SegmentFileBuilderV10 allows zero-length internal-file entries,
short-circuit
+ return new ByteArrayInputStream(new byte[0]);
+ }
+ return new RetryingInputStream<>(
+ new RangeRequest(pathPrefix + filename, offset, length),
+ new RangeOpenFunction(storage, bucket),
+ GoogleUtils.GOOGLE_RETRY,
+ null
+ );
+ }
+
+ /**
+ * Immutable description of a range read. Held as the {@code object} of
{@link RetryingInputStream} so retries can
+ * reopen with knowledge of the original offset and length without
rebuilding the request from scratch.
+ */
+ private static final class RangeRequest
+ {
+ final String objectPath;
+ final long offset;
+ final long length;
+
+ RangeRequest(String objectPath, long offset, long length)
+ {
+ this.objectPath = objectPath;
+ this.offset = offset;
+ this.length = length;
+ }
+ }
+
+ /**
+ * Opens (or reopens, on retry) a GCS range read for a {@link RangeRequest}.
The {@code start} argument is the number
+ * of bytes already successfully consumed from the logical stream, so the
range still owed is
+ * {@code [request.offset + start, request.offset + request.length)}.
+ */
+ private static final class RangeOpenFunction implements
ObjectOpenFunction<RangeRequest>
+ {
+ private final GoogleStorage storage;
+ private final String bucket;
+
+ RangeOpenFunction(GoogleStorage storage, String bucket)
+ {
+ this.storage = storage;
+ this.bucket = bucket;
+ }
+
+ @Override
+ public InputStream open(RangeRequest request) throws IOException
+ {
+ return open(request, 0L);
+ }
+
+ @Override
+ public InputStream open(RangeRequest request, long start) throws
IOException
+ {
+ final long remaining = request.length - start;
+ if (remaining <= 0) {
+ // Logically nothing left to read, only reachable if a retry fires
after the consumer drained the entire
+ // range successfully, which shouldn't happen in practice. Returning
empty keeps us robust either way.
+ return new ByteArrayInputStream(new byte[0]);
+ }
+ // A StorageException is deliberately left to propagate as it is.
GoogleUtils.isRetryable inspects only the
+ // throwable it is handed (it does not walk the cause chain) and answers
`instanceof IOException` for anything
+ // it does not recognize, so wrapping this would report every failure as
retryable and spend ten backing-off
+ // attempts on a permission error. Raw, the predicate calls
StorageException#isRetryable and decides correctly;
+ // RetryingInputStream hands the caller an IOException around it either
way.
+ return storage.getInputStream(bucket, request.objectPath, request.offset
+ start, remaining);
+ }
+ }
+}
diff --git
a/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleDataSegmentPusherTest.java
b/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleDataSegmentPusherTest.java
index 012c4c3c1c8..8d9e7af6e69 100644
---
a/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleDataSegmentPusherTest.java
+++
b/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleDataSegmentPusherTest.java
@@ -22,7 +22,11 @@ package org.apache.druid.storage.google;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Files;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.loading.DeepStorageSegmentConfig;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
@@ -148,17 +152,105 @@ public class GoogleDataSegmentPusherTest extends
EasyMockSupport
DataSegment segment = pusher.push(tempFolder, segmentToPush, false);
- // the trailing slash is what marks the path as a directory of files
rather than a single object
+ // the trailing slash is what marks the path as a directory of files
rather than a single object, and V1 (the test
+ // fixture) is not V10, so rangeable is stamped false
Assertions.assertEquals(ImmutableMap.of(
"type", GoogleStorageDruidModule.SCHEME,
"bucket", BUCKET,
- "path", expectedDir + "/"
+ "path", expectedDir + "/",
+ "rangeable", false
), segment.getLoadSpec());
Assertions.assertEquals(2 * data.length, segment.getSize());
verifyAll();
}
+ @Test
+ public void testPushNoZipV10StampsRangeableTrue() throws Exception
+ {
+ // A real V10 segment rather than a hand-made file, so the layout under
test is whatever IndexMergerV10 actually
+ // writes: one druid.segment whose leading byte is the version
getVersionFromDir reads.
+ final File segmentDir = IndexBuilder.create()
+ .useV10()
+ .tmpDir(new File(tempFolder, "v10"))
+ .rows(ImmutableList.of(
+ new MapBasedInputRow(
+ DateTimes.of("2015-01-01"),
+ ImmutableList.of("dim"),
+ ImmutableMap.of("dim", "a")
+ )
+ ))
+ .buildMMappedIndexFile();
+
+ final File[] segmentFiles = segmentDir.listFiles();
+ Assertions.assertNotNull(segmentFiles);
+ long segmentSize = 0;
+ for (final File file : segmentFiles) {
+ segmentSize += file.length();
+ }
+
+ DataSegment segmentToPush = newSegmentToPush(segmentSize);
+
+ GoogleDataSegmentPusher pusher =
createMockBuilder(GoogleDataSegmentPusher.class)
+ .withConstructor(storage, googleAccountConfig, INPUT_DATA_CONFIG,
NO_ZIP_CONFIG)
+ .addMockedMethod("insert", File.class, String.class, String.class)
+ .createMock();
+
+ final String expectedDir = PREFIX + "/" +
pusher.getStorageDir(segmentToPush, false);
+ final ImmutableList.Builder<GoogleStorageObjectMetadata> listing =
ImmutableList.builder();
+ for (final File file : segmentFiles) {
+ pusher.insert(
+ EasyMock.anyObject(File.class),
+ EasyMock.anyString(),
+ EasyMock.eq(expectedDir + "/" + file.getName())
+ );
+ EasyMock.expectLastCall();
+ listing.add(new GoogleStorageObjectMetadata(BUCKET, expectedDir + "/" +
file.getName(), file.length(), 0L));
+ }
+ // everything under the path was written by this push, so there is nothing
to clean up
+ EasyMock.expect(storage.list(EasyMock.eq(BUCKET), EasyMock.eq(expectedDir
+ "/"), EasyMock.anyObject(), EasyMock.anyObject()))
+ .andReturn(new GoogleStorageObjectPage(listing.build(), null));
+
+ replayAll();
+
+ DataSegment segment = pusher.push(segmentDir, segmentToPush, false);
+
+ // a V10 segment is a single file, and that is what makes it range-readable
+ Assertions.assertEquals(1, segmentFiles.length);
+ Assertions.assertEquals(IndexIO.V10_FILE_NAME, segmentFiles[0].getName());
+ Assertions.assertEquals(IndexIO.V10_VERSION, (int)
segment.getBinaryVersion());
+ Assertions.assertEquals(Boolean.TRUE,
segment.getLoadSpec().get("rangeable"));
+
+ verifyAll();
+ }
+
+ @Test
+ public void testPushZipDoesNotStampRangeable() throws Exception
+ {
+ // The zip path uses the no-flag makeLoadSpec overload; openRangeReader
returns null on the zip short circuit
+ // regardless, but the loadSpec stays compact for zipped segments by
omitting the field entirely.
+ final byte[] data = new byte[]{0x0, 0x0, 0x0, 0x1};
+ Files.write(data, new File(tempFolder, "version.bin"));
+
+ DataSegment segmentToPush = newSegmentToPush(data.length);
+
+ GoogleDataSegmentPusher pusher =
createMockBuilder(GoogleDataSegmentPusher.class)
+ .withConstructor(storage, googleAccountConfig, INPUT_DATA_CONFIG,
ZIP_CONFIG)
+ .addMockedMethod("insert", File.class, String.class, String.class)
+ .createMock();
+
+ pusher.insert(EasyMock.anyObject(File.class),
EasyMock.eq("application/zip"), EasyMock.anyString());
+ EasyMock.expectLastCall();
+
+ replayAll();
+
+ DataSegment segment = pusher.push(tempFolder, segmentToPush, false);
+
+ Assertions.assertFalse(segment.getLoadSpec().containsKey("rangeable"));
+
+ verifyAll();
+ }
+
@Test
public void testPushNoZipRemovesObjectsLeftByPreviousPush() throws Exception
{
diff --git
a/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleLoadSpecTest.java
b/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleLoadSpecTest.java
new file mode 100644
index 00000000000..2bd2747cc73
--- /dev/null
+++
b/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleLoadSpecTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.druid.storage.google;
+
+import org.apache.druid.segment.loading.SegmentRangeReader;
+import org.easymock.EasyMockSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class GoogleLoadSpecTest extends EasyMockSupport
+{
+ private static final String BUCKET = "test-bucket";
+ private static final String RAW_PATH = "path/to/segment/";
+ private static final String ZIP_PATH = "path/to/index.zip";
+
+ private GoogleStorage storage;
+
+ @BeforeEach
+ public void setUp()
+ {
+ storage = createMock(GoogleStorage.class);
+ // opening a range reader must not touch storage, so any call fails these
tests
+ replayAll();
+ }
+
+ @Test
+ public void testOpenRangeReaderReturnsReaderWhenRangeableTrue()
+ {
+ final SegmentRangeReader reader = newLoadSpec(RAW_PATH,
true).openRangeReader();
+ assertNotNull(reader);
+ assertInstanceOf(GoogleSegmentRangeReader.class, reader);
+ verifyAll();
+ }
+
+ @Test
+ public void testOpenRangeReaderReturnsNullForZipPathEvenWhenRangeableTrue()
+ {
+ // Defensive: a zip object can't be range-read; the zip check wins over
the flag even if hand-crafted input claims
+ // the layout is rangeable.
+ assertNull(newLoadSpec(ZIP_PATH, true).openRangeReader());
+ verifyAll();
+ }
+
+ @Test
+ public void testOpenRangeReaderReturnsNullWhenRangeableFalse()
+ {
+ assertNull(newLoadSpec(RAW_PATH, false).openRangeReader());
+ verifyAll();
+ }
+
+ @Test
+ public void testOpenRangeReaderReturnsNullForLegacySegmentWithoutFlag()
+ {
+ // Legacy segment (pushed before this field existed) → null flag →
full-download path.
+ assertNull(newLoadSpec(RAW_PATH, null).openRangeReader());
+ verifyAll();
+ }
+
+ private GoogleLoadSpec newLoadSpec(String path, @Nullable Boolean rangeable)
+ {
+ return new GoogleLoadSpec(
+ BUCKET,
+ path,
+ rangeable,
+ new GoogleDataSegmentPuller(storage, new GoogleInputDataConfig())
+ );
+ }
+}
diff --git
a/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleSegmentRangeReaderTest.java
b/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleSegmentRangeReaderTest.java
new file mode 100644
index 00000000000..e35078cef5a
--- /dev/null
+++
b/extensions-core/google-extensions/src/test/java/org/apache/druid/storage/google/GoogleSegmentRangeReaderTest.java
@@ -0,0 +1,223 @@
+/*
+ * 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.druid.storage.google;
+
+import com.google.cloud.storage.StorageException;
+import com.google.common.io.ByteStreams;
+import org.apache.druid.segment.loading.SegmentRangeReader;
+import org.easymock.Capture;
+import org.easymock.CaptureType;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class GoogleSegmentRangeReaderTest extends EasyMockSupport
+{
+ private static final String BUCKET = "test-bucket";
+ private static final String PATH_PREFIX =
"ds/2024-01-01T00:00:00.000Z_2024-01-02T00:00:00.000Z/0/0/";
+
+ private GoogleStorage storage;
+ private GoogleSegmentRangeReader reader;
+
+ @BeforeEach
+ public void setUp()
+ {
+ storage = createMock(GoogleStorage.class);
+ reader = new GoogleSegmentRangeReader(storage, BUCKET, PATH_PREFIX);
+ }
+
+ @Test
+ public void testReadRangeOpensPathPrefixPlusFilenameAtOffsetAndLength()
throws IOException
+ {
+ EasyMock.expect(storage.getInputStream(BUCKET, PATH_PREFIX +
"druid.segment", 100L, 250L))
+ .andReturn(new ByteArrayInputStream(new byte[0]));
+
+ replayAll();
+
+ try (InputStream ignored = reader.readRange("druid.segment", 100, 250)) {
+ // open is performed in the RetryingInputStream constructor; the read
channel should have already been opened.
+ }
+
+ verifyAll();
+ }
+
+ @Test
+ public void testReadRangeBuildsDifferentPathsForDifferentFilenames() throws
IOException
+ {
+ final Capture<String> pathCapture = Capture.newInstance(CaptureType.ALL);
+ EasyMock.expect(storage.getInputStream(EasyMock.eq(BUCKET),
EasyMock.capture(pathCapture), EasyMock.eq(0L), EasyMock.eq(16L)))
+ .andReturn(new ByteArrayInputStream(new byte[0]))
+ .times(2);
+
+ replayAll();
+
+ reader.readRange("file-a", 0, 16).close();
+ reader.readRange("file-b", 0, 16).close();
+
+ verifyAll();
+
+ assertEquals(PATH_PREFIX + "file-a", pathCapture.getValues().get(0));
+ assertEquals(PATH_PREFIX + "file-b", pathCapture.getValues().get(1));
+ }
+
+ @Test
+ public void testReadRangeFailsFastOnNonRetryableStorageException() throws
IOException
+ {
+ // 403 isn't retryable per StorageException#isRetryable, so open fails
once and RetryingInputStream surfaces it as
+ // IOException(StorageException). The exception must reach the predicate
unwrapped for that to hold:
+ // GoogleUtils.isRetryable inspects only the throwable it is handed and
answers `instanceof IOException` for
+ // anything it doesn't recognize, so an IOException wrapper here would
report a permission error as retryable and
+ // spend ten backing-off attempts on it.
+ EasyMock.expect(storage.getInputStream(BUCKET, PATH_PREFIX + "f", 0L, 1L))
+ .andThrow(new StorageException(403, "denied"));
+
+ replayAll();
+
+ final IOException thrown = assertThrows(IOException.class, () ->
reader.readRange("f", 0, 1));
+ assertSame(StorageException.class, thrown.getCause().getClass());
+
+ verifyAll();
+ }
+
+ @Test
+ public void testReadRangeRetriesMidStreamFromBytesAlreadyConsumed() throws
IOException
+ {
+ // First range read delivers the first 4 bytes then errors mid-stream with
a retryable IOException.
+ // RetryingInputStream should reopen for the bytes still owed, exercising
the offset math in
+ // RangeOpenFunction.open(request, start). NOTE: this test sleeps ~1s
because RetryingInputStream's first retry
+ // uses RetryUtils.BASE_SLEEP_MILLIS exponential backoff (no
@VisibleForTesting hook is accessible from here).
+ final byte[] firstChunk = {0x01, 0x02, 0x03, 0x04};
+ final byte[] secondChunk = {0x05, 0x06, 0x07, 0x08, 0x09, 0x0A};
+ final byte[] all = new byte[firstChunk.length + secondChunk.length];
+ System.arraycopy(firstChunk, 0, all, 0, firstChunk.length);
+ System.arraycopy(secondChunk, 0, all, firstChunk.length,
secondChunk.length);
+
+ // First: the full requested range
+ EasyMock.expect(storage.getInputStream(BUCKET, PATH_PREFIX + "f", 100L,
10L))
+ .andReturn(failingAfter(firstChunk));
+ // Retry: resume at (offset + bytes already consumed), asking only for
what is still owed
+ EasyMock.expect(storage.getInputStream(BUCKET, PATH_PREFIX + "f", 104L,
6L))
+ .andReturn(new ByteArrayInputStream(secondChunk));
+
+ replayAll();
+
+ final byte[] read;
+ try (InputStream stream = reader.readRange("f", 100, 10)) {
+ read = ByteStreams.toByteArray(stream);
+ }
+ assertArrayEquals(all, read);
+
+ verifyAll();
+ }
+
+ @Test
+ public void
testReadRangeReturnsEmptyStreamForZeroLengthWithoutContactingStorage() throws
IOException
+ {
+ // SegmentFileBuilderV10 allows zero-length internal-file entries;
readRange must accept length=0 and return an
+ // empty stream without opening a read channel. Replaying with no
expectations fails the test on any call.
+ replayAll();
+
+ try (InputStream stream = reader.readRange("f", 100, 0)) {
+ assertEquals(-1, stream.read());
+ }
+
+ verifyAll();
+ }
+
+ @Test
+ public void testReadRangeRejectsNegativeOffset()
+ {
+ replayAll();
+ assertThrows(IllegalArgumentException.class, () -> reader.readRange("f",
-1, 16));
+ verifyAll();
+ }
+
+ @Test
+ public void testReadRangeRejectsNegativeLength()
+ {
+ replayAll();
+ assertThrows(IllegalArgumentException.class, () -> reader.readRange("f",
0, -1));
+ verifyAll();
+ }
+
+ @Test
+ public void testRetryPredicateCannotSeeThroughAnIOExceptionWrapper()
+ {
+ // Guards the reason RangeOpenFunction hands StorageException to
RetryingInputStream unwrapped. If isRetryable ever
+ // starts walking the cause chain, wrapping becomes safe and the comment
at that throw site should be revisited.
+ final StorageException nonRetryable = new StorageException(403, "denied");
+ assertFalse(GoogleUtils.isRetryable(nonRetryable));
+ assertTrue(GoogleUtils.isRetryable(new IOException(nonRetryable)));
+ }
+
+ @Test
+ public void testImplementsSegmentRangeReader()
+ {
+ // Compile-time and runtime guard: ensure the implements relationship
survives refactors so callers can rely on
+ // returning GoogleSegmentRangeReader from openRangeReader().
+ final SegmentRangeReader downcast = reader;
+ assertSame(reader, downcast);
+ }
+
+ /**
+ * Returns a stream that delivers the given {@code head} bytes successfully,
then raises a bare {@link IOException}
+ * (no cause) on the next read — which {@link GoogleUtils#isRetryable}
treats as retryable, so
+ * {@link org.apache.druid.data.input.impl.RetryingInputStream} will close
this delegate and call the open function
+ * again with {@code start = head.length}.
+ */
+ private static InputStream failingAfter(byte[] head)
+ {
+ return new FilterInputStream(new ByteArrayInputStream(head))
+ {
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException
+ {
+ final int read = super.read(b, off, len);
+ if (read < 0) {
+ throw new IOException("mid-stream failure");
+ }
+ return read;
+ }
+
+ @Override
+ public int read() throws IOException
+ {
+ final int read = super.read();
+ if (read < 0) {
+ throw new IOException("mid-stream failure");
+ }
+ return read;
+ }
+ };
+ }
+}
diff --git
a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPusherTest.java
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPusherTest.java
index 5e89f5a3897..c92f28c40f0 100644
---
a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPusherTest.java
+++
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPusherTest.java
@@ -19,11 +19,17 @@
package org.apache.druid.storage.s3;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.io.Files;
+import org.apache.druid.data.input.MapBasedInputRow;
import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.loading.DeepStorageSegmentConfig;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.partition.NoneShardSpec;
@@ -156,16 +162,41 @@ public class S3DataSegmentPusherTest
config.setBucket("bucket");
config.setBaseKey("key");
- // version.bin = [0, 0, 0, 0x0A] → IndexIO.V10_VERSION
+ // A real V10 segment rather than a hand-made file, so the layout under
test is whatever IndexMergerV10 actually
+ // writes: one druid.segment whose leading byte is the version
getVersionFromDir reads.
+ final File segmentDir = IndexBuilder.create()
+ .useV10()
+ .tmpDir(new File(tempFolder, "v10"))
+ .rows(ImmutableList.of(
+ new MapBasedInputRow(
+ DateTimes.of("2015-01-01"),
+ ImmutableList.of("dim"),
+ ImmutableMap.of("dim", "a")
+ )
+ ))
+ .buildMMappedIndexFile();
+
+ final File[] segmentFiles = segmentDir.listFiles();
+ Assertions.assertNotNull(segmentFiles);
+ long segmentSize = 0;
+ for (final File file : segmentFiles) {
+ segmentSize += file.length();
+ }
+
DataSegment segment = validate(
false,
"key/foo/2015-01-01T00:00:00\\.000Z_2016-01-01T00:00:00\\.000Z/0/0/",
s3Client,
config,
false,
- new byte[]{0x0, 0x0, 0x0, 0x0A}
+ segmentDir,
+ segmentSize
);
- Assertions.assertEquals(10, (int) segment.getBinaryVersion());
+
+ // a V10 segment is a single file, and that is what makes it range-readable
+ Assertions.assertEquals(1, segmentFiles.length);
+ Assertions.assertEquals(IndexIO.V10_FILE_NAME, segmentFiles[0].getName());
+ Assertions.assertEquals(IndexIO.V10_VERSION, (int)
segment.getBinaryVersion());
Assertions.assertEquals(Boolean.TRUE,
segment.getLoadSpec().get("rangeable"));
}
@@ -378,13 +409,24 @@ public class S3DataSegmentPusherTest
byte[] versionBytes
) throws IOException
{
- S3DataSegmentPusher pusher = new S3DataSegmentPusher(s3Client, config, new
DeepStorageSegmentConfig(zip));
-
// Create a mock segment on disk
File tmp = new File(tempFolder, "version.bin");
-
Files.write(versionBytes, tmp);
- final long size = versionBytes.length;
+
+ return validate(useUniquePath, matcher, s3Client, config, zip, tempFolder,
versionBytes.length);
+ }
+
+ private DataSegment validate(
+ boolean useUniquePath,
+ String matcher,
+ ServerSideEncryptingAmazonS3 s3Client,
+ S3DataSegmentPusherConfig config,
+ boolean zip,
+ File segmentDir,
+ long size
+ ) throws IOException
+ {
+ S3DataSegmentPusher pusher = new S3DataSegmentPusher(s3Client, config, new
DeepStorageSegmentConfig(zip));
DataSegment segmentToPush = new DataSegment(
"foo",
@@ -398,7 +440,7 @@ public class S3DataSegmentPusherTest
size
);
- DataSegment segment = pusher.push(tempFolder, segmentToPush,
useUniquePath);
+ DataSegment segment = pusher.push(segmentDir, segmentToPush,
useUniquePath);
Assertions.assertEquals(segmentToPush.getSize(), segment.getSize());
Assertions.assertEquals("bucket", segment.getLoadSpec().get("bucket"));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
