From 2a587822ab96c1cd156537a3a30ec0b1ae41fc41 Mon Sep 17 00:00:00 2001 From: Shruti Singhania Date: Fri, 3 Jul 2026 11:34:29 +0530 Subject: [PATCH 1/6] feat: Extract metadata from GCS read channel --- .../client/AbstractReadStrategy.java | 7 + .../client/AdaptiveReadStrategy.java | 7 + .../gcs/analyticscore/client/GcsFileInfo.java | 15 ++ .../analyticscore/client/GcsReadChannel.java | 56 ++++ .../GcsReadChannelMetadataExtractor.java | 164 +++++++++++ .../analyticscore/client/ReadStrategy.java | 4 + .../client/VectoredSeekableByteChannel.java | 7 + .../GcsReadChannelMetadataExtractorTest.java | 255 ++++++++++++++++++ .../client/GcsReadChannelTest.java | 42 +++ .../core/channel/SmartReadChannel.java | 7 + .../core/optimizer/GcsFooterOptimizer.java | 7 +- .../client/TrackingReadStrategy.java | 7 + 12 files changed, 577 insertions(+), 1 deletion(-) create mode 100644 client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java create mode 100644 client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AbstractReadStrategy.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AbstractReadStrategy.java index ba9c08b6..319e9a57 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AbstractReadStrategy.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AbstractReadStrategy.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.List; +import javax.annotation.Nullable; abstract class AbstractReadStrategy implements ReadStrategy { protected final GcsItemId itemId; @@ -45,6 +46,12 @@ abstract class AbstractReadStrategy implements ReadStrategy { this.itemInfo = itemInfo; } + @Override + @Nullable + public ReadChannel getSdkReadChannel() { + return channel; + } + @Override public void position(long newPosition) { this.position = newPosition; diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java index 876e1609..8b4c65cc 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java @@ -18,6 +18,7 @@ import com.google.cloud.ReadChannel; import com.google.cloud.storage.Storage; import java.io.IOException; +import javax.annotation.Nullable; class AdaptiveReadStrategy extends AbstractReadStrategy { private ReadStrategy currentStrategy; @@ -38,6 +39,12 @@ class AdaptiveReadStrategy extends AbstractReadStrategy { : new SequentialReadStrategy(storage, itemId, options, itemInfo); } + @Override + @Nullable + public ReadChannel getSdkReadChannel() { + return currentStrategy != null ? currentStrategy.getSdkReadChannel() : null; + } + @Override public ReadChannel getReadChannel(long requestedPosition, int bytesToRead) throws IOException { updateStrategy(requestedPosition); diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java index 293913ec..7ae7f26f 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java @@ -16,7 +16,9 @@ package com.google.cloud.gcs.analyticscore.client; import com.google.auto.value.AutoValue; +import com.google.cloud.storage.BlobId; import java.net.URI; +import java.util.Collections; import java.util.Map; /** @@ -47,6 +49,19 @@ public static Builder builder() { return new AutoValue_GcsFileInfo.Builder(); } + public static GcsFileInfo fromItemInfo(GcsItemInfo itemInfo) { + return builder() + .setItemInfo(itemInfo) + .setUri( + URI.create( + BlobId.of( + itemInfo.getItemId().getBucketName(), + itemInfo.getItemId().getObjectName().orElse("")) + .toGsUtilUri())) + .setAttributes(Collections.emptyMap()) + .build(); + } + /** Builder for {@link GcsFileInfo}. */ @AutoValue.Builder public abstract static class Builder { diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java index 01bc10d7..ebabbc8d 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java @@ -36,6 +36,7 @@ import java.util.List; import java.util.concurrent.ExecutorService; import java.util.function.IntFunction; +import javax.annotation.Nullable; class GcsReadChannel implements VectoredSeekableByteChannel { private Storage storage; @@ -124,6 +125,9 @@ public int read(ByteBuffer dst) throws IOException { private int readNextChunk(ByteBuffer dst) throws IOException { ReadChannel sdkChannel = strategy.getReadChannel(gcsReadChannelPosition, dst.remaining()); int bytesRead = sdkChannel.read(dst); + if (this.itemInfo == null) { + extractMetadataAfterRead(); + } if (bytesRead >= 0) { gcsReadChannelPosition += bytesRead; strategy.position(gcsReadChannelPosition); @@ -176,9 +180,18 @@ public long size() throws IOException { if (null != itemInfo) { return itemInfo.getSize(); } + if (extractMetadataAfterRead()) { + return itemInfo.getSize(); + } throw new IOException("Object metadata not initialized"); } + @Override + @Nullable + public GcsItemInfo getItemInfo() { + return itemInfo; + } + @Override public SeekableByteChannel truncate(long size) throws IOException { throw new UnsupportedOperationException("Cannot mutate read-only channel"); @@ -247,6 +260,9 @@ void readCombinedRange( int numOfBytesRead = 0; while (dataBuffer.hasRemaining()) { int bytesRead = channel.read(dataBuffer); + if (GcsReadChannel.this.itemInfo == null) { + extractMetadataAfterRead(readStrategy); + } if (bytesRead < 0) { // EOF reached. break; @@ -321,4 +337,44 @@ private void validatePosition(long position) throws IOException { "Invalid seek offset: position value (%d) must be >= 0 for '%s'", position, itemId)); } } + + private boolean extractMetadataAfterRead() { + return extractMetadataAfterRead(this.strategy); + } + + private synchronized boolean extractMetadataAfterRead(@Nullable ReadStrategy strategy) { + if (this.itemInfo != null) { + return true; + } + if (strategy == null) { + return false; + } + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(strategy.getSdkReadChannel()); + if (metadata == null) { + return false; + } + updateItemMetadata(metadata.getSize(), metadata.getGeneration()); + return true; + } + + private void updateItemMetadata(long extractedSize, long extractedGen) { + GcsItemId.Builder itemIdBuilder = + GcsItemId.builder().setBucketName(this.itemId.getBucketName()); + this.itemId.getObjectName().ifPresent(itemIdBuilder::setObjectName); + long genToSet = extractedGen > 0 ? extractedGen : this.itemId.getContentGeneration().orElse(0L); + if (genToSet > 0) { + itemIdBuilder.setContentGeneration(genToSet); + } + GcsItemId updatedItemId = itemIdBuilder.build(); + GcsItemInfo.Builder itemInfoBuilder = + GcsItemInfo.builder().setItemId(updatedItemId).setSize(extractedSize); + if (genToSet > 0) { + itemInfoBuilder.setContentGeneration(genToSet); + } else { + itemInfoBuilder.setContentGeneration(0L); + } + this.itemInfo = itemInfoBuilder.build(); + this.itemId = updatedItemId; + } } diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java new file mode 100644 index 00000000..943f781a --- /dev/null +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java @@ -0,0 +1,164 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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 com.google.cloud.gcs.analyticscore.client; + +import com.google.cloud.ReadChannel; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Utility class to extract metadata from GCS SDK {@link ReadChannel} instances using reflection. + */ +final class GcsReadChannelMetadataExtractor { + private static final Logger LOG = LoggerFactory.getLogger(GcsReadChannelMetadataExtractor.class); + + private GcsReadChannelMetadataExtractor() {} + + static final class ExtractedMetadata { + private final long size; + private final long generation; + + ExtractedMetadata(long size, long generation) { + this.size = size; + this.generation = generation; + } + + long getSize() { + return size; + } + + long getGeneration() { + return generation; + } + } + + @Nullable + static ExtractedMetadata extract(@Nullable ReadChannel sdkChannel) { + if (sdkChannel == null) { + return null; + } + Object resolvedMetadata = resolveMetadataObject(sdkChannel); + if (resolvedMetadata == null) { + return null; + } + long extractedSize = extractLongProperty(resolvedMetadata, "getSize", "size"); + if (extractedSize < 0) { + return null; + } + long extractedGen = extractLongProperty(resolvedMetadata, "getGeneration", "generation"); + return new ExtractedMetadata(extractedSize, extractedGen); + } + + @Nullable + private static Object resolveMetadataObject(ReadChannel sdkChannel) { + Class clazz = sdkChannel.getClass(); + while (clazz != null) { + for (String methodName : + new String[] { + "getObject", "getResolvedObject", "getBlobInfo", "getBlob", "getStorageObject" + }) { + try { + Method method = clazz.getDeclaredMethod(methodName); + method.setAccessible(true); + Object res = resolveFutureIfNeeded(method.invoke(sdkChannel)); + if (res != null) { + return res; + } + } catch (ReflectiveOperationException ignored) { + LOG.debug("Method {} not present on class {}", methodName, clazz.getName()); + } + } + for (String fieldName : new String[] {"storageObject", "blobInfo", "object", "result"}) { + try { + Field field = clazz.getDeclaredField(fieldName); + field.setAccessible(true); + Object val = resolveFutureIfNeeded(field.get(sdkChannel)); + if (val != null) { + return val; + } + } catch (ReflectiveOperationException ignored) { + LOG.debug("Field {} not present on class {}", fieldName, clazz.getName()); + } + } + clazz = clazz.getSuperclass(); + } + return null; + } + + @Nullable + private static Object resolveFutureIfNeeded(@Nullable Object obj) + throws ReflectiveOperationException { + if (!(obj instanceof Future)) { + return obj; + } + Future future = (Future) obj; + if (!future.isDone()) { + return null; + } + try { + return future.get(); + } catch (ExecutionException ignored) { + LOG.debug("Future execution failed or was cancelled", ignored); + return null; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } + + private static long extractLongProperty( + Object target, String primaryGetter, String fallbackGetter) { + for (Method m : target.getClass().getMethods()) { + if (m.getParameterCount() == 0 && m.getName().equals(primaryGetter)) { + long val = invokeLongGetter(target, m); + if (val >= 0) { + return val; + } + } + } + if (target instanceof Map) { + return -1L; + } + for (Method m : target.getClass().getMethods()) { + if (m.getParameterCount() == 0 && m.getName().equals(fallbackGetter)) { + long val = invokeLongGetter(target, m); + if (val >= 0) { + return val; + } + } + } + return -1L; + } + + private static long invokeLongGetter(Object target, Method method) { + try { + method.setAccessible(true); + Object val = method.invoke(target); + if (val instanceof Number) { + return ((Number) val).longValue(); + } + } catch (ReflectiveOperationException ignored) { + LOG.debug("Getter invocation failed for method {} on target {}", method.getName(), target); + } + return -1L; + } +} diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/ReadStrategy.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/ReadStrategy.java index 2ce84af5..1a6d9a77 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/ReadStrategy.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/ReadStrategy.java @@ -17,6 +17,7 @@ import com.google.cloud.ReadChannel; import java.io.IOException; +import javax.annotation.Nullable; /** Strategy for reading data from Google Cloud Storage. */ interface ReadStrategy { @@ -30,6 +31,9 @@ interface ReadStrategy { */ ReadChannel getReadChannel(long requestedPosition, int bytesToRead) throws IOException; + @Nullable + ReadChannel getSdkReadChannel(); + /** * Updates the strategy's current read position. * diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/VectoredSeekableByteChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/VectoredSeekableByteChannel.java index bbf6203c..f924024e 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/VectoredSeekableByteChannel.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/VectoredSeekableByteChannel.java @@ -20,6 +20,7 @@ import java.nio.channels.SeekableByteChannel; import java.util.List; import java.util.function.IntFunction; +import javax.annotation.Nullable; public interface VectoredSeekableByteChannel extends SeekableByteChannel { /** @@ -31,4 +32,10 @@ public interface VectoredSeekableByteChannel extends SeekableByteChannel { */ void readVectored(List ranges, IntFunction allocate) throws IOException; + + /** Returns the item metadata info if available, or null. */ + @Nullable + default GcsItemInfo getItemInfo() { + return null; + } } diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java new file mode 100644 index 00000000..b5c501e4 --- /dev/null +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java @@ -0,0 +1,255 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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 com.google.cloud.gcs.analyticscore.client; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.cloud.ReadChannel; +import com.google.cloud.RestorableState; +import com.google.cloud.storage.BlobInfo; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +class GcsReadChannelMetadataExtractorTest { + + @Test + void extract_nullChannel_returnsNull() { + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(null); + + assertThat(metadata).isNull(); + } + + @Test + void extract_unsupportedChannelClass_returnsNull() { + ReadChannel channel = Mockito.mock(ReadChannel.class); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + void extract_viaMethod_getBlobInfo_returnsMetadata() { + BlobInfo blobInfo = Mockito.mock(BlobInfo.class); + Mockito.when(blobInfo.getSize()).thenReturn(100L); + Mockito.when(blobInfo.getGeneration()).thenReturn(200L); + ReflectiveBlobInfoChannel channel = Mockito.mock(ReflectiveBlobInfoChannel.class); + Mockito.when(channel.getBlobInfo()).thenReturn(blobInfo); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNotNull(); + assertThat(metadata.getSize()).isEqualTo(100L); + assertThat(metadata.getGeneration()).isEqualTo(200L); + } + + @Test + void extract_viaField_result_returnsMetadata() { + BlobInfo blobInfo = Mockito.mock(BlobInfo.class); + Mockito.when(blobInfo.getSize()).thenReturn(300L); + Mockito.when(blobInfo.getGeneration()).thenReturn(400L); + ReadChannel channel = + new ReadChannel() { + private final BlobInfo result = blobInfo; + + @Override + public void close() {} + + @Override + public boolean isOpen() { + return true; + } + + @Override + public int read(ByteBuffer dst) { + return -1; + } + + @Override + public void seek(long position) {} + + @Override + public void setChunkSize(int chunkSize) {} + + @Override + public RestorableState capture() { + return null; + } + }; + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNotNull(); + assertThat(metadata.getSize()).isEqualTo(300L); + assertThat(metadata.getGeneration()).isEqualTo(400L); + } + + @Test + void extract_futureNotDone_returnsNull() { + CompletableFuture future = new CompletableFuture<>(); + ReflectiveFutureChannel channel = Mockito.mock(ReflectiveFutureChannel.class); + Mockito.when(channel.getObject()).thenReturn(future); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + void extract_futureDone_returnsMetadata() { + BlobInfo blobInfo = Mockito.mock(BlobInfo.class); + Mockito.when(blobInfo.getSize()).thenReturn(500L); + Mockito.when(blobInfo.getGeneration()).thenReturn(600L); + CompletableFuture future = CompletableFuture.completedFuture(blobInfo); + ReflectiveFutureChannel channel = Mockito.mock(ReflectiveFutureChannel.class); + Mockito.when(channel.getObject()).thenReturn(future); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNotNull(); + assertThat(metadata.getSize()).isEqualTo(500L); + assertThat(metadata.getGeneration()).isEqualTo(600L); + } + + @Test + @SuppressWarnings("FutureReturnValueIgnored") + void extract_futureExecutionException_returnsNull() throws Exception { + Future mockFuture = Mockito.mock(Future.class); + Mockito.when(mockFuture.isDone()).thenReturn(true); + Mockito.when(mockFuture.get()) + .thenThrow(new ExecutionException(new RuntimeException("failed"))); + ReflectiveFutureChannel channel = Mockito.mock(ReflectiveFutureChannel.class); + Mockito.when(channel.getObject()).thenReturn(mockFuture); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + void extract_fallbackGetter_returnsSize() throws IOException { + ByteArrayOutputStream stream = new ByteArrayOutputStream(); + stream.write(new byte[150]); + ReflectiveFallbackChannel channel = Mockito.mock(ReflectiveFallbackChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn(stream); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNotNull(); + assertThat(metadata.getSize()).isEqualTo(150L); + assertThat(metadata.getGeneration()).isEqualTo(-1L); + } + + @Test + void extract_targetIsMap_doesNotUseFallbackGetter_returnsNull() { + Map mapTarget = new HashMap<>(); + mapTarget.put("size", 1000L); + ReflectiveMapChannel channel = Mockito.mock(ReflectiveMapChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn(mapTarget); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + void extract_noSizeProperty_returnsNull() { + ReflectiveNoSizeChannel channel = Mockito.mock(ReflectiveNoSizeChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn("dummy-string-no-size-getter"); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + @SuppressWarnings("FutureReturnValueIgnored") + void extract_futureInterruptedException_returnsNull() throws Exception { + Future mockFuture = Mockito.mock(Future.class); + Mockito.when(mockFuture.isDone()).thenReturn(true); + Mockito.when(mockFuture.get()).thenThrow(new InterruptedException("interrupted")); + ReflectiveFutureChannel channel = Mockito.mock(ReflectiveFutureChannel.class); + Mockito.when(channel.getObject()).thenReturn(mockFuture); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + Thread.interrupted(); + } + + @Test + void extract_getterThrowsException_returnsNull() { + ThrowingModel throwingModel = new ThrowingModel(); + ReflectiveErrorChannel channel = Mockito.mock(ReflectiveErrorChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn(throwingModel); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + static class ThrowingModel { + public long getSize() { + throw new RuntimeException("size retrieval failed"); + } + } + + // Helper interfaces for Mockito stubbing of reflective getters + interface ReflectiveBlobInfoChannel extends ReadChannel { + BlobInfo getBlobInfo(); + } + + interface ReflectiveFutureChannel extends ReadChannel { + Future getObject(); + } + + interface ReflectiveFallbackChannel extends ReadChannel { + ByteArrayOutputStream getResolvedObject(); + } + + interface ReflectiveMapChannel extends ReadChannel { + Map getResolvedObject(); + } + + interface ReflectiveNoSizeChannel extends ReadChannel { + String getResolvedObject(); + } + + interface ReflectiveErrorChannel extends ReadChannel { + ThrowingModel getResolvedObject(); + } +} diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java index af6009b1..a495a85c 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java @@ -18,6 +18,7 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; +import com.google.api.services.storage.model.StorageObject; import com.google.cloud.ReadChannel; import com.google.cloud.gcs.analyticscore.common.GcsAnalyticsCoreTelemetryConstants.Metric; import com.google.cloud.gcs.analyticscore.common.telemetry.MetricKey; @@ -34,6 +35,7 @@ import com.google.common.collect.Lists; import java.io.EOFException; import java.io.IOException; +import java.math.BigInteger; import java.nio.ByteBuffer; import java.nio.channels.ClosedChannelException; import java.nio.charset.StandardCharsets; @@ -920,6 +922,42 @@ protected ReadStrategy createReadStrategy( } } + @Test + void read_withoutItemInfo_extractsMetadataViaReflection() throws IOException { + GcsItemId itemId = + GcsItemId.builder().setBucketName("test-bucket").setObjectName("test-object").build(); + String objectData = "hello world"; + StorageObject storageObject = + new StorageObject().setSize(BigInteger.valueOf(objectData.length())).setGeneration(123L); + Storage mockStorage = Mockito.mock(Storage.class); + ReflectiveReadChannel mockReadChannel = Mockito.mock(ReflectiveReadChannel.class); + Mockito.when( + mockStorage.reader( + Mockito.any(BlobId.class), Mockito.any(Storage.BlobSourceOption[].class))) + .thenReturn(mockReadChannel); + Mockito.when(mockReadChannel.isOpen()).thenReturn(true); + Mockito.when(mockReadChannel.read(Mockito.any(ByteBuffer.class))) + .thenAnswer( + invocation -> { + ByteBuffer buf = invocation.getArgument(0); + if (!buf.hasRemaining()) { + return -1; + } + int bytesToRead = Math.min(buf.remaining(), 5); + buf.position(buf.position() + bytesToRead); + return bytesToRead; + }); + Mockito.when(mockReadChannel.getStorageObject()).thenReturn(storageObject); + + GcsReadChannel gcsReadChannel = + new GcsReadChannel( + mockStorage, itemId, TEST_GCS_READ_OPTIONS, executorServiceSupplier, telemetry); + ByteBuffer buffer = ByteBuffer.allocate(5); + gcsReadChannel.read(buffer); + + assertThat(gcsReadChannel.size()).isEqualTo(objectData.length()); + } + private GcsItemInfo createItemInfoWith(long size) { GcsItemId itemId = GcsItemId.builder().setBucketName("test-bucket").setObjectName("test-object").build(); @@ -945,4 +983,8 @@ private String getGcsObjectRangeData(GcsObjectRange range) throws ExecutionException, InterruptedException { return StandardCharsets.UTF_8.decode(range.getByteBufferFuture().get()).toString(); } + + public interface ReflectiveReadChannel extends ReadChannel { + StorageObject getStorageObject(); + } } diff --git a/core/src/main/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannel.java b/core/src/main/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannel.java index b9adbedb..d8b562b6 100644 --- a/core/src/main/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannel.java +++ b/core/src/main/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannel.java @@ -21,6 +21,7 @@ import com.google.cloud.gcs.analyticscore.client.AnalyticsCacheManager; import com.google.cloud.gcs.analyticscore.client.GcsFileInfo; import com.google.cloud.gcs.analyticscore.client.GcsItemId; +import com.google.cloud.gcs.analyticscore.client.GcsItemInfo; import com.google.cloud.gcs.analyticscore.client.GcsObjectRange; import com.google.cloud.gcs.analyticscore.client.VectoredSeekableByteChannel; import com.google.cloud.gcs.analyticscore.core.optimizer.FormatOptimizer; @@ -144,6 +145,12 @@ public long size() throws IOException { return delegate.size(); } + @Override + @Nullable + public GcsItemInfo getItemInfo() { + return delegate.getItemInfo(); + } + @Override public VectoredSeekableByteChannel truncate(long size) throws IOException { delegate.truncate(size); diff --git a/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java b/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java index 27251c73..54f312f8 100644 --- a/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java +++ b/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java @@ -21,6 +21,7 @@ import com.google.cloud.gcs.analyticscore.client.AnalyticsCacheManager; import com.google.cloud.gcs.analyticscore.client.GcsFileInfo; import com.google.cloud.gcs.analyticscore.client.GcsItemId; +import com.google.cloud.gcs.analyticscore.client.GcsItemInfo; import com.google.cloud.gcs.analyticscore.client.GcsReadOptions; import com.google.cloud.gcs.analyticscore.client.VectoredSeekableByteChannel; import com.google.cloud.gcs.analyticscore.common.GcsAnalyticsCoreTelemetryConstants.Metric; @@ -81,7 +82,11 @@ public void onOpen(GcsFileInfo fileInfo, AnalyticsCacheManager cacheManager) { public int read(long position, ByteBuffer dst, VectoredSeekableByteChannel source) throws IOException { if (fileSize == -1) { - fileSize = source.size(); + GcsItemInfo info = source.getItemInfo(); + if (info == null) { + return 0; + } + fileSize = info.getSize(); prefetchSize = calculatePrefetchSize(fileSize, readOptions); } diff --git a/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/TrackingReadStrategy.java b/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/TrackingReadStrategy.java index b3d87b48..5bdb9105 100644 --- a/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/TrackingReadStrategy.java +++ b/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/TrackingReadStrategy.java @@ -18,6 +18,7 @@ import com.google.cloud.ReadChannel; import java.io.IOException; +import javax.annotation.Nullable; public class TrackingReadStrategy implements ReadStrategy { private final ReadStrategy delegate; @@ -31,6 +32,12 @@ public TrackingReadStrategy(ReadStrategy delegate) { this.delegate = delegate; } + @Override + @Nullable + public ReadChannel getSdkReadChannel() { + return delegate != null ? delegate.getSdkReadChannel() : null; + } + public void setEofAtCall(int eofAtCall) { this.eofAtCall = eofAtCall; } From 04e5b1158436affaed9b943ccaa12fb41a6ca72e Mon Sep 17 00:00:00 2001 From: Shruti Singhania Date: Fri, 3 Jul 2026 11:52:16 +0530 Subject: [PATCH 2/6] refactor: add getSdkReadChannel to GcsBidiReadChannel and remove unused GcsFileInfo.fromItemInfo method --- .../analyticscore/client/GcsBidiReadChannel.java | 8 +++++++- .../gcs/analyticscore/client/GcsFileInfo.java | 15 --------------- 2 files changed, 7 insertions(+), 16 deletions(-) diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannel.java index a31670c3..cd1b2090 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannel.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannel.java @@ -20,6 +20,7 @@ import com.google.api.core.ApiFuture; import com.google.api.core.ApiFutureCallback; import com.google.api.core.ApiFutures; +import com.google.cloud.ReadChannel; import com.google.cloud.gcs.analyticscore.common.telemetry.Telemetry; import com.google.cloud.storage.BlobId; import com.google.cloud.storage.BlobReadSession; @@ -104,7 +105,7 @@ protected ReadStrategy createReadStrategy( Storage storage, GcsItemId itemId, GcsReadOptions readOptions, GcsItemInfo itemInfo) { return new ReadStrategy() { @Override - public com.google.cloud.ReadChannel getReadChannel(long requestedPosition, int bytesToRead) { + public ReadChannel getReadChannel(long requestedPosition, int bytesToRead) { throw new UnsupportedOperationException( "Standard read is not supported on Bidi channel yet."); } @@ -122,6 +123,11 @@ public boolean isEof(long position) { return true; } + @Override + public ReadChannel getSdkReadChannel() { + return null; + } + @Override public void close() {} }; diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java index 7ae7f26f..293913ec 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileInfo.java @@ -16,9 +16,7 @@ package com.google.cloud.gcs.analyticscore.client; import com.google.auto.value.AutoValue; -import com.google.cloud.storage.BlobId; import java.net.URI; -import java.util.Collections; import java.util.Map; /** @@ -49,19 +47,6 @@ public static Builder builder() { return new AutoValue_GcsFileInfo.Builder(); } - public static GcsFileInfo fromItemInfo(GcsItemInfo itemInfo) { - return builder() - .setItemInfo(itemInfo) - .setUri( - URI.create( - BlobId.of( - itemInfo.getItemId().getBucketName(), - itemInfo.getItemId().getObjectName().orElse("")) - .toGsUtilUri())) - .setAttributes(Collections.emptyMap()) - .build(); - } - /** Builder for {@link GcsFileInfo}. */ @AutoValue.Builder public abstract static class Builder { From d977fd84a20bbcd0a2a6b18bfe07e13a8b571e96 Mon Sep 17 00:00:00 2001 From: Shruti Singhania Date: Fri, 3 Jul 2026 12:44:49 +0530 Subject: [PATCH 3/6] refactor: gemini comment and improve test coverage --- .../analyticscore/client/GcsReadChannel.java | 15 ++- .../GcsReadChannelMetadataExtractor.java | 6 +- .../client/GcsBidiReadChannelTest.java | 8 ++ .../GcsReadChannelMetadataExtractorTest.java | 35 ++++++ .../client/GcsReadChannelTest.java | 118 ++++++++++++++++++ .../core/channel/SmartReadChannelTest.java | 17 +++ .../optimizer/GcsFooterOptimizerTest.java | 12 ++ 7 files changed, 203 insertions(+), 8 deletions(-) diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java index 164e8011..27936563 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java @@ -46,8 +46,8 @@ public interface ItemInfoProvider { protected Storage storage; protected GcsReadOptions readOptions; - protected GcsItemInfo itemInfo; - protected GcsItemId itemId; + protected volatile GcsItemInfo itemInfo; + protected volatile GcsItemId itemId; private long gcsReadChannelPosition = 0; protected Supplier executorServiceSupplier; private static final ImmutableMap COMMON_ATTRIBUTES = @@ -153,7 +153,7 @@ public int read(ByteBuffer dst) throws IOException { private int readNextChunk(ByteBuffer dst) throws IOException { ReadChannel sdkChannel = strategy.getReadChannel(gcsReadChannelPosition, dst.remaining()); int bytesRead = sdkChannel.read(dst); - if (this.itemInfo == null) { + if (this.itemInfo == null && !metadataExtractionAttempted) { extractMetadataAfterRead(); } if (bytesRead >= 0) { @@ -297,7 +297,7 @@ void readCombinedRange( int numOfBytesRead = 0; while (dataBuffer.hasRemaining()) { int bytesRead = channel.read(dataBuffer); - if (GcsReadChannel.this.itemInfo == null) { + if (GcsReadChannel.this.itemInfo == null && !metadataExtractionAttempted) { extractMetadataAfterRead(readStrategy); } if (bytesRead < 0) { @@ -379,13 +379,14 @@ private boolean extractMetadataAfterRead() { return extractMetadataAfterRead(this.strategy); } - private synchronized boolean extractMetadataAfterRead(@Nullable ReadStrategy strategy) { + private synchronized boolean extractMetadataAfterRead(ReadStrategy strategy) { if (this.itemInfo != null) { return true; } - if (strategy == null) { + if (metadataExtractionAttempted) { return false; } + metadataExtractionAttempted = true; GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = GcsReadChannelMetadataExtractor.extract(strategy.getSdkReadChannel()); if (metadata == null) { @@ -414,4 +415,6 @@ private void updateItemMetadata(long extractedSize, long extractedGen) { this.itemInfo = itemInfoBuilder.build(); this.itemId = updatedItemId; } + + private volatile boolean metadataExtractionAttempted = false; } diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java index 943f781a..d37452aa 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java @@ -18,7 +18,9 @@ import com.google.cloud.ReadChannel; import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.util.Collection; import java.util.Map; +import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import javax.annotation.Nullable; @@ -116,7 +118,7 @@ private static Object resolveFutureIfNeeded(@Nullable Object obj) } try { return future.get(); - } catch (ExecutionException ignored) { + } catch (CancellationException | ExecutionException ignored) { LOG.debug("Future execution failed or was cancelled", ignored); return null; } catch (InterruptedException e) { @@ -135,7 +137,7 @@ private static long extractLongProperty( } } } - if (target instanceof Map) { + if (target instanceof Map || target instanceof Collection) { return -1L; } for (Method m : target.getClass().getMethods()) { diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java index df41f38a..6915ce46 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java @@ -340,4 +340,12 @@ void testClose_isIdempotent() throws Exception { // Verify that blobReadSession.close() was still called only once verify(blobReadSession, times(1)).close(); } + + @Test + void testDummyReadStrategy_getSdkReadChannel_returnsNull() { + ReadStrategy strategy = + reader.createReadStrategy(storage, itemId, GcsReadOptions.builder().build(), null); + + assertThat(strategy.getSdkReadChannel()).isNull(); + } } diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java index b5c501e4..ab6f9fa7 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java @@ -23,8 +23,11 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -222,6 +225,34 @@ void extract_getterThrowsException_returnsNull() { assertThat(metadata).isNull(); } + @Test + @SuppressWarnings("FutureReturnValueIgnored") + void extract_futureCancelledException_returnsNull() throws Exception { + Future mockFuture = Mockito.mock(Future.class); + Mockito.when(mockFuture.isDone()).thenReturn(true); + Mockito.when(mockFuture.get()).thenThrow(new CancellationException("cancelled")); + ReflectiveFutureChannel channel = Mockito.mock(ReflectiveFutureChannel.class); + Mockito.when(channel.getObject()).thenReturn(mockFuture); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + void extract_targetIsCollection_doesNotUseFallbackGetter_returnsNull() { + List collectionTarget = new ArrayList<>(); + collectionTarget.add("item"); + ReflectiveCollectionChannel channel = Mockito.mock(ReflectiveCollectionChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn(collectionTarget); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + static class ThrowingModel { public long getSize() { throw new RuntimeException("size retrieval failed"); @@ -245,6 +276,10 @@ interface ReflectiveMapChannel extends ReadChannel { Map getResolvedObject(); } + interface ReflectiveCollectionChannel extends ReadChannel { + List getResolvedObject(); + } + interface ReflectiveNoSizeChannel extends ReadChannel { String getResolvedObject(); } diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java index 5ca6cd2a..518c93cd 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelTest.java @@ -42,6 +42,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; @@ -985,6 +986,123 @@ void read_withoutItemInfo_extractsMetadataViaReflection() throws IOException { assertThat(gcsReadChannel.size()).isEqualTo(objectData.length()); } + @Test + void size_withoutItemInfo_extractsMetadataViaReflection() throws IOException { + GcsItemId itemId = + GcsItemId.builder().setBucketName("test-bucket").setObjectName("test-object").build(); + String objectData = "hello world"; + StorageObject storageObject = + new StorageObject().setSize(BigInteger.valueOf(objectData.length())).setGeneration(123L); + Storage mockStorage = Mockito.mock(Storage.class); + ReflectiveReadChannel mockReadChannel = Mockito.mock(ReflectiveReadChannel.class); + Mockito.when( + mockStorage.reader( + Mockito.any(BlobId.class), Mockito.any(Storage.BlobSourceOption[].class))) + .thenReturn(mockReadChannel); + Mockito.when(mockReadChannel.isOpen()).thenReturn(true); + Mockito.when(mockReadChannel.getStorageObject()).thenReturn(storageObject); + + GcsReadChannel gcsReadChannel = + new GcsReadChannel( + mockStorage, itemId, TEST_GCS_READ_OPTIONS, executorServiceSupplier, telemetry); + + assertThat(gcsReadChannel.size()).isEqualTo(objectData.length()); + assertThat(gcsReadChannel.getItemInfo().getContentGeneration()).isEqualTo(Optional.of(123L)); + } + + @Test + void size_withoutItemInfo_generationZero_extractsMetadataViaReflection() throws IOException { + GcsItemId itemId = + GcsItemId.builder().setBucketName("test-bucket").setObjectName("test-object").build(); + String objectData = "hello world"; + StorageObject storageObject = + new StorageObject().setSize(BigInteger.valueOf(objectData.length())).setGeneration(0L); + Storage mockStorage = Mockito.mock(Storage.class); + ReflectiveReadChannel mockReadChannel = Mockito.mock(ReflectiveReadChannel.class); + Mockito.when( + mockStorage.reader( + Mockito.any(BlobId.class), Mockito.any(Storage.BlobSourceOption[].class))) + .thenReturn(mockReadChannel); + Mockito.when(mockReadChannel.isOpen()).thenReturn(true); + Mockito.when(mockReadChannel.getStorageObject()).thenReturn(storageObject); + + GcsReadChannel gcsReadChannel = + new GcsReadChannel( + mockStorage, itemId, TEST_GCS_READ_OPTIONS, executorServiceSupplier, telemetry); + + assertThat(gcsReadChannel.size()).isEqualTo(objectData.length()); + assertThat(gcsReadChannel.getItemInfo().getContentGeneration()).isEqualTo(Optional.of(0L)); + } + + @Test + void readVectored_withoutItemInfo_extractsMetadataViaReflection() throws Exception { + GcsItemId itemId = + GcsItemId.builder().setBucketName("test-bucket").setObjectName("test-object").build(); + String objectData = "hello world"; + StorageObject storageObject = + new StorageObject().setSize(BigInteger.valueOf(objectData.length())).setGeneration(123L); + Storage mockStorage = Mockito.mock(Storage.class); + ReflectiveReadChannel mockReadChannel = Mockito.mock(ReflectiveReadChannel.class); + Mockito.when( + mockStorage.reader( + Mockito.any(BlobId.class), Mockito.any(Storage.BlobSourceOption[].class))) + .thenReturn(mockReadChannel); + Mockito.when(mockReadChannel.isOpen()).thenReturn(true); + Mockito.when(mockReadChannel.read(Mockito.any(ByteBuffer.class))) + .thenAnswer( + invocation -> { + ByteBuffer buf = invocation.getArgument(0); + if (!buf.hasRemaining()) { + return -1; + } + int bytesToRead = Math.min(buf.remaining(), objectData.length()); + buf.put(objectData.substring(0, bytesToRead).getBytes(StandardCharsets.UTF_8)); + return bytesToRead; + }); + Mockito.when(mockReadChannel.getStorageObject()).thenReturn(storageObject); + + GcsReadChannel gcsReadChannel = + new GcsReadChannel( + mockStorage, itemId, TEST_GCS_READ_OPTIONS, executorServiceSupplier, telemetry); + ImmutableList ranges = createRanges(ImmutableMap.of(0L, 5)); + + gcsReadChannel.readVectored(ranges, ByteBuffer::allocate); + + assertThat(getGcsObjectRangeData(ranges.get(0))).isEqualTo("hello"); + assertThat(gcsReadChannel.getItemInfo()).isNotNull(); + assertThat(gcsReadChannel.getItemInfo().getSize()).isEqualTo(objectData.length()); + } + + @Test + void readVectored_allocationReturnsNull_completesFuturesExceptionally() throws Exception { + GcsItemId itemId = + GcsItemId.builder().setBucketName("test-bucket").setObjectName("test-object").build(); + String objectData = "hello world"; + GcsItemInfo itemInfo = + GcsItemInfo.builder() + .setItemId(itemId) + .setSize(objectData.length()) + .setContentGeneration(0L) + .build(); + StorageTestUtils.createBlobInStorage( + storage, BlobId.of(itemId.getBucketName(), itemId.getObjectName().get(), 0L), objectData); + GcsReadChannel gcsReadChannel = + new GcsReadChannel( + storage, itemInfo, TEST_GCS_READ_OPTIONS, executorServiceSupplier, telemetry); + ImmutableList ranges = createRanges(ImmutableMap.of(0L, 5)); + + gcsReadChannel.readVectored(ranges, size -> null); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> ranges.get(0).getByteBufferFuture().get()); + assertThat(e).hasCauseThat().isInstanceOf(IOException.class); + assertThat(e).hasCauseThat().hasMessageThat().contains("Error while populating childRange"); + assertThat(e.getCause().getCause()).isInstanceOf(IllegalArgumentException.class); + assertThat(e.getCause().getCause()) + .hasMessageThat() + .contains("Buffer allocation returned null for combinedObjectRange"); + } + private GcsItemInfo createItemInfoWith(long size) { GcsItemId itemId = GcsItemId.builder().setBucketName("test-bucket").setObjectName("test-object").build(); diff --git a/core/src/test/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannelTest.java b/core/src/test/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannelTest.java index 1d22ece7..404e4ff1 100644 --- a/core/src/test/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannelTest.java +++ b/core/src/test/java/com/google/cloud/gcs/analyticscore/core/channel/SmartReadChannelTest.java @@ -696,4 +696,21 @@ void close_throwsError_rethrowsError() throws IOException { assertThat(exception).hasMessageThat().isEqualTo("Error close failed"); } + + @Test + void getItemInfo_delegatesToDelegate() throws IOException { + SmartReadChannel channel = + SmartReadChannel.builder() + .setDelegate(mockDelegate) + .setItemId(ITEM_ID) + .setCacheManager(mockCacheManager) + .build(); + GcsItemInfo expectedInfo = GcsItemInfo.builder().setItemId(ITEM_ID).build(); + when(mockDelegate.getItemInfo()).thenReturn(expectedInfo); + + GcsItemInfo actualInfo = channel.getItemInfo(); + + assertThat(actualInfo).isEqualTo(expectedInfo); + verify(mockDelegate).getItemInfo(); + } } diff --git a/core/src/test/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizerTest.java b/core/src/test/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizerTest.java index 6152e4cc..16eda641 100644 --- a/core/src/test/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizerTest.java +++ b/core/src/test/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizerTest.java @@ -332,4 +332,16 @@ void read_multipleReads_usesLocalBufferAndRecordsTelemetry() throws IOException verify(telemetry, times(1)).recordMetric(eq(Metric.FOOTER_CACHE_MISS), eq(1L), any()); verify(telemetry, times(1)).recordMetric(eq(Metric.FOOTER_PREFETCH_HIT), eq(1L), any()); } + + @Test + void read_fileSizeUninitialized_sourceGetItemInfoReturnsNull_returnsZero() throws IOException { + optimizer.onOpen(ITEM_ID, mockCacheManager); + VectoredSeekableByteChannel mockSource = mock(VectoredSeekableByteChannel.class); + when(mockSource.getItemInfo()).thenReturn(null); + ByteBuffer dst = ByteBuffer.allocate(10); + + int bytesRead = optimizer.read(990, dst, mockSource); + + assertThat(bytesRead).isEqualTo(0); + } } From 06914f4e293973694738f23f9475d9f97eb29bbf Mon Sep 17 00:00:00 2001 From: Shruti Singhania Date: Fri, 3 Jul 2026 13:17:52 +0530 Subject: [PATCH 4/6] Improve coverage --- .../client/AdaptiveReadStrategy.java | 2 +- .../analyticscore/client/GcsReadChannel.java | 6 - .../GcsReadChannelMetadataExtractorTest.java | 140 ++++++++++++++++++ 3 files changed, 141 insertions(+), 7 deletions(-) diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java index 8b4c65cc..6822ad26 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/AdaptiveReadStrategy.java @@ -42,7 +42,7 @@ class AdaptiveReadStrategy extends AbstractReadStrategy { @Override @Nullable public ReadChannel getSdkReadChannel() { - return currentStrategy != null ? currentStrategy.getSdkReadChannel() : null; + return currentStrategy.getSdkReadChannel(); } @Override diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java index 27936563..23d752fb 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java @@ -380,12 +380,6 @@ private boolean extractMetadataAfterRead() { } private synchronized boolean extractMetadataAfterRead(ReadStrategy strategy) { - if (this.itemInfo != null) { - return true; - } - if (metadataExtractionAttempted) { - return false; - } metadataExtractionAttempted = true; GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = GcsReadChannelMetadataExtractor.extract(strategy.getSdkReadChannel()); diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java index ab6f9fa7..a0a139f6 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java @@ -253,12 +253,142 @@ void extract_targetIsCollection_doesNotUseFallbackGetter_returnsNull() { assertThat(metadata).isNull(); } + @Test + void extract_fallbackGetterNegative_returnsNull() { + FallbackNegativeModel target = new FallbackNegativeModel(); + ReflectiveGenericChannel channel = Mockito.mock(ReflectiveGenericChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn(target); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + void extract_negativePrimaryPositiveFallback_returnsMetadata() { + NegativePrimaryPositiveFallbackModel target = new NegativePrimaryPositiveFallbackModel(); + ReflectiveGenericChannel channel = Mockito.mock(ReflectiveGenericChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn(target); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNotNull(); + assertThat(metadata.getSize()).isEqualTo(100L); + assertThat(metadata.getGeneration()).isEqualTo(-1L); + } + + @Test + void extract_bothPrimaryAndFallbackNegative_returnsNull() { + BothNegativeModel target = new BothNegativeModel(); + ReflectiveGenericChannel channel = Mockito.mock(ReflectiveGenericChannel.class); + Mockito.when(channel.getResolvedObject()).thenReturn(target); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNull(); + } + + @Test + void extract_fieldValueNull_fallsThroughToNextField() { + BlobInfo blobInfo = Mockito.mock(BlobInfo.class); + Mockito.when(blobInfo.getSize()).thenReturn(700L); + Mockito.when(blobInfo.getGeneration()).thenReturn(800L); + ReadChannel channel = new FakeChannelWithNullAndNonNullFields(blobInfo); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNotNull(); + assertThat(metadata.getSize()).isEqualTo(700L); + assertThat(metadata.getGeneration()).isEqualTo(800L); + } + + @Test + void extract_methodValueNull_fallsThroughToNextMethod() { + BlobInfo blobInfo = Mockito.mock(BlobInfo.class); + Mockito.when(blobInfo.getSize()).thenReturn(900L); + Mockito.when(blobInfo.getGeneration()).thenReturn(1000L); + ReflectiveNullAndNonNullMethodsChannel channel = + Mockito.mock(ReflectiveNullAndNonNullMethodsChannel.class); + Mockito.when(channel.getObject()).thenReturn(null); + Mockito.when(channel.getBlobInfo()).thenReturn(blobInfo); + + GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + GcsReadChannelMetadataExtractor.extract(channel); + + assertThat(metadata).isNotNull(); + assertThat(metadata.getSize()).isEqualTo(900L); + assertThat(metadata.getGeneration()).isEqualTo(1000L); + } + static class ThrowingModel { public long getSize() { throw new RuntimeException("size retrieval failed"); } } + static class FallbackNegativeModel { + public long size() { + return -5L; + } + } + + static class NegativePrimaryPositiveFallbackModel { + public long getSize() { + return -10L; + } + + public long size() { + return 100L; + } + } + + static class BothNegativeModel { + public long getSize() { + return -10L; + } + + public long size() { + return -5L; + } + } + + static class FakeChannelWithNullAndNonNullFields implements ReadChannel { + private final Object storageObject = null; + private final BlobInfo result; + + FakeChannelWithNullAndNonNullFields(BlobInfo result) { + this.result = result; + } + + @Override + public void close() {} + + @Override + public boolean isOpen() { + return true; + } + + @Override + public int read(ByteBuffer dst) { + return -1; + } + + @Override + public void seek(long position) {} + + @Override + public void setChunkSize(int chunkSize) {} + + @Override + public RestorableState capture() { + return null; + } + } + // Helper interfaces for Mockito stubbing of reflective getters interface ReflectiveBlobInfoChannel extends ReadChannel { BlobInfo getBlobInfo(); @@ -287,4 +417,14 @@ interface ReflectiveNoSizeChannel extends ReadChannel { interface ReflectiveErrorChannel extends ReadChannel { ThrowingModel getResolvedObject(); } + + interface ReflectiveGenericChannel extends ReadChannel { + Object getResolvedObject(); + } + + interface ReflectiveNullAndNonNullMethodsChannel extends ReadChannel { + Object getObject(); + + BlobInfo getBlobInfo(); + } } From 98a699f8deb39f7cebf669a5a7377cae5c7040ae Mon Sep 17 00:00:00 2001 From: Shruti Singhania Date: Tue, 14 Jul 2026 09:30:04 +0530 Subject: [PATCH 5/6] Implementing review comments --- .../analyticscore/client/GcsReadChannel.java | 36 ++++++++----------- .../GcsReadChannelMetadataExtractor.java | 31 ++++++++++------ .../core/optimizer/GcsFooterOptimizer.java | 7 +--- 3 files changed, 36 insertions(+), 38 deletions(-) diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java index 23d752fb..f540873d 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java @@ -18,6 +18,7 @@ import static com.google.common.base.Preconditions.checkNotNull; import com.google.cloud.ReadChannel; +import com.google.cloud.gcs.analyticscore.client.GcsReadChannelMetadataExtractor.ExtractedMetadata; import com.google.cloud.gcs.analyticscore.common.GcsAnalyticsCoreTelemetryConstants; import com.google.cloud.gcs.analyticscore.common.GcsAnalyticsCoreTelemetryConstants.Attribute; import com.google.cloud.gcs.analyticscore.common.GcsAnalyticsCoreTelemetryConstants.Metric; @@ -55,6 +56,7 @@ public interface ItemInfoProvider { protected final Telemetry telemetry; private final ReadStrategy strategy; private volatile boolean isGcsReadChannelOpen = true; + private volatile boolean metadataExtractionAttempted = false; protected final ItemInfoProvider itemInfoProvider; GcsReadChannel( @@ -153,9 +155,6 @@ public int read(ByteBuffer dst) throws IOException { private int readNextChunk(ByteBuffer dst) throws IOException { ReadChannel sdkChannel = strategy.getReadChannel(gcsReadChannelPosition, dst.remaining()); int bytesRead = sdkChannel.read(dst); - if (this.itemInfo == null && !metadataExtractionAttempted) { - extractMetadataAfterRead(); - } if (bytesRead >= 0) { gcsReadChannelPosition += bytesRead; strategy.position(gcsReadChannelPosition); @@ -297,7 +296,7 @@ void readCombinedRange( int numOfBytesRead = 0; while (dataBuffer.hasRemaining()) { int bytesRead = channel.read(dataBuffer); - if (GcsReadChannel.this.itemInfo == null && !metadataExtractionAttempted) { + if (itemInfo == null && !metadataExtractionAttempted) { extractMetadataAfterRead(readStrategy); } if (bytesRead < 0) { @@ -381,7 +380,7 @@ private boolean extractMetadataAfterRead() { private synchronized boolean extractMetadataAfterRead(ReadStrategy strategy) { metadataExtractionAttempted = true; - GcsReadChannelMetadataExtractor.ExtractedMetadata metadata = + ExtractedMetadata metadata = GcsReadChannelMetadataExtractor.extract(strategy.getSdkReadChannel()); if (metadata == null) { return false; @@ -391,24 +390,19 @@ private synchronized boolean extractMetadataAfterRead(ReadStrategy strategy) { } private void updateItemMetadata(long extractedSize, long extractedGen) { - GcsItemId.Builder itemIdBuilder = - GcsItemId.builder().setBucketName(this.itemId.getBucketName()); - this.itemId.getObjectName().ifPresent(itemIdBuilder::setObjectName); - long genToSet = extractedGen > 0 ? extractedGen : this.itemId.getContentGeneration().orElse(0L); - if (genToSet > 0) { + long genToSet = extractedGen >= 0 ? extractedGen : itemId.getContentGeneration().orElse(-1L); + + GcsItemId.Builder itemIdBuilder = GcsItemId.builder().setBucketName(itemId.getBucketName()); + itemId.getObjectName().ifPresent(itemIdBuilder::setObjectName); + + GcsItemInfo.Builder itemInfoBuilder = GcsItemInfo.builder().setSize(extractedSize); + + if (genToSet >= 0) { itemIdBuilder.setContentGeneration(genToSet); - } - GcsItemId updatedItemId = itemIdBuilder.build(); - GcsItemInfo.Builder itemInfoBuilder = - GcsItemInfo.builder().setItemId(updatedItemId).setSize(extractedSize); - if (genToSet > 0) { itemInfoBuilder.setContentGeneration(genToSet); - } else { - itemInfoBuilder.setContentGeneration(0L); } - this.itemInfo = itemInfoBuilder.build(); - this.itemId = updatedItemId; - } - private volatile boolean metadataExtractionAttempted = false; + itemId = itemIdBuilder.build(); + itemInfo = itemInfoBuilder.setItemId(itemId).build(); + } } diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java index d37452aa..4de85e93 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java @@ -16,6 +16,7 @@ package com.google.cloud.gcs.analyticscore.client; import com.google.cloud.ReadChannel; +import com.google.common.collect.ImmutableList; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.Collection; @@ -33,6 +34,13 @@ final class GcsReadChannelMetadataExtractor { private static final Logger LOG = LoggerFactory.getLogger(GcsReadChannelMetadataExtractor.class); + private static final ImmutableList METADATA_METHOD_NAMES = + ImmutableList.of( + "getObject", "getResolvedObject", "getBlobInfo", "getBlob", "getStorageObject"); + + private static final ImmutableList METADATA_FIELD_NAMES = + ImmutableList.of("storageObject", "blobInfo", "object", "result"); + private GcsReadChannelMetadataExtractor() {} static final class ExtractedMetadata { @@ -74,10 +82,7 @@ static ExtractedMetadata extract(@Nullable ReadChannel sdkChannel) { private static Object resolveMetadataObject(ReadChannel sdkChannel) { Class clazz = sdkChannel.getClass(); while (clazz != null) { - for (String methodName : - new String[] { - "getObject", "getResolvedObject", "getBlobInfo", "getBlob", "getStorageObject" - }) { + for (String methodName : METADATA_METHOD_NAMES) { try { Method method = clazz.getDeclaredMethod(methodName); method.setAccessible(true); @@ -85,11 +90,12 @@ private static Object resolveMetadataObject(ReadChannel sdkChannel) { if (res != null) { return res; } - } catch (ReflectiveOperationException ignored) { - LOG.debug("Method {} not present on class {}", methodName, clazz.getName()); + } catch (ReflectiveOperationException | RuntimeException ignored) { + LOG.debug( + "Method {} not present or inaccessible on class {}", methodName, clazz.getName()); } } - for (String fieldName : new String[] {"storageObject", "blobInfo", "object", "result"}) { + for (String fieldName : METADATA_FIELD_NAMES) { try { Field field = clazz.getDeclaredField(fieldName); field.setAccessible(true); @@ -97,8 +103,8 @@ private static Object resolveMetadataObject(ReadChannel sdkChannel) { if (val != null) { return val; } - } catch (ReflectiveOperationException ignored) { - LOG.debug("Field {} not present on class {}", fieldName, clazz.getName()); + } catch (ReflectiveOperationException | RuntimeException ignored) { + LOG.debug("Field {} not present or inaccessible on class {}", fieldName, clazz.getName()); } } clazz = clazz.getSuperclass(); @@ -158,8 +164,11 @@ private static long invokeLongGetter(Object target, Method method) { if (val instanceof Number) { return ((Number) val).longValue(); } - } catch (ReflectiveOperationException ignored) { - LOG.debug("Getter invocation failed for method {} on target {}", method.getName(), target); + } catch (ReflectiveOperationException | RuntimeException ignored) { + LOG.debug( + "Getter invocation failed or inaccessible for method {} on target {}", + method.getName(), + target); } return -1L; } diff --git a/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java b/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java index 54f312f8..27251c73 100644 --- a/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java +++ b/core/src/main/java/com/google/cloud/gcs/analyticscore/core/optimizer/GcsFooterOptimizer.java @@ -21,7 +21,6 @@ import com.google.cloud.gcs.analyticscore.client.AnalyticsCacheManager; import com.google.cloud.gcs.analyticscore.client.GcsFileInfo; import com.google.cloud.gcs.analyticscore.client.GcsItemId; -import com.google.cloud.gcs.analyticscore.client.GcsItemInfo; import com.google.cloud.gcs.analyticscore.client.GcsReadOptions; import com.google.cloud.gcs.analyticscore.client.VectoredSeekableByteChannel; import com.google.cloud.gcs.analyticscore.common.GcsAnalyticsCoreTelemetryConstants.Metric; @@ -82,11 +81,7 @@ public void onOpen(GcsFileInfo fileInfo, AnalyticsCacheManager cacheManager) { public int read(long position, ByteBuffer dst, VectoredSeekableByteChannel source) throws IOException { if (fileSize == -1) { - GcsItemInfo info = source.getItemInfo(); - if (info == null) { - return 0; - } - fileSize = info.getSize(); + fileSize = source.size(); prefetchSize = calculatePrefetchSize(fileSize, readOptions); } From a6c9a05d9a3778cbdc112bb78adbd68fd3acfe03 Mon Sep 17 00:00:00 2001 From: Shruti Singhania Date: Tue, 14 Jul 2026 09:37:30 +0530 Subject: [PATCH 6/6] fix spotless issue due to merge --- .../cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java index e0964b8b..b136c599 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java @@ -348,7 +348,7 @@ void testDummyReadStrategy_getSdkReadChannel_returnsNull() { assertThat(strategy.getSdkReadChannel()).isNull(); } - + @Test void read_success_returnsBytesReadAndUpdatesPosition() throws Exception { GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(20L).build();