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 dc4a1ee8..d7a6ce99 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 @@ -25,6 +25,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; @@ -46,6 +47,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..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 @@ -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.getSdkReadChannel(); + } + @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/GcsBidiReadChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannel.java index 039dfe4a..b33dfe71 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.BlobInfo; @@ -101,7 +102,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."); } @@ -119,6 +120,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/GcsReadChannel.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannel.java index 589b69f2..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 @@ -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 { @FunctionalInterface @@ -45,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 = @@ -152,6 +153,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 && !metadataExtractionAttempted) { + extractMetadataAfterRead(); + } if (bytesRead >= 0) { gcsReadChannelPosition += bytesRead; strategy.position(gcsReadChannelPosition); @@ -204,6 +208,9 @@ public long size() throws IOException { if (itemInfo != null) { return itemInfo.getSize(); } + if (extractMetadataAfterRead()) { + return itemInfo.getSize(); + } if (itemInfoProvider == null) { throw new IOException("ItemInfo is not initialized and no ItemInfoProvider was provided."); } @@ -213,6 +220,12 @@ public long size() throws IOException { return itemInfo.getSize(); } + @Override + @Nullable + public GcsItemInfo getItemInfo() { + return itemInfo; + } + @Override public SeekableByteChannel truncate(long size) throws IOException { throw new UnsupportedOperationException("Cannot mutate read-only channel"); @@ -284,6 +297,9 @@ void readCombinedRange( int numOfBytesRead = 0; while (dataBuffer.hasRemaining()) { int bytesRead = channel.read(dataBuffer); + if (GcsReadChannel.this.itemInfo == null && !metadataExtractionAttempted) { + extractMetadataAfterRead(readStrategy); + } if (bytesRead < 0) { // EOF reached. break; @@ -358,4 +374,41 @@ 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(ReadStrategy strategy) { + metadataExtractionAttempted = true; + 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; + } + + 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 new file mode 100644 index 00000000..d37452aa --- /dev/null +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractor.java @@ -0,0 +1,166 @@ +/* + * 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.Collection; +import java.util.Map; +import java.util.concurrent.CancellationException; +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 (CancellationException | 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 || target instanceof Collection) { + 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/GcsBidiReadChannelTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsBidiReadChannelTest.java index 9b057912..e0964b8b 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 @@ -341,6 +341,14 @@ void close_calledMultipleTimes_isIdempotent() throws Exception { verify(blobReadSession, times(1)).close(); } + @Test + void testDummyReadStrategy_getSdkReadChannel_returnsNull() { + ReadStrategy strategy = + reader.createReadStrategy(storage, itemId, GcsReadOptions.builder().build(), null); + + assertThat(strategy.getSdkReadChannel()).isNull(); + } + @Test void read_success_returnsBytesReadAndUpdatesPosition() throws Exception { GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(20L).build(); 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..a0a139f6 --- /dev/null +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsReadChannelMetadataExtractorTest.java @@ -0,0 +1,430 @@ +/* + * 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.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; +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(); + } + + @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(); + } + + @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(); + } + + interface ReflectiveFutureChannel extends ReadChannel { + Future getObject(); + } + + interface ReflectiveFallbackChannel extends ReadChannel { + ByteArrayOutputStream getResolvedObject(); + } + + interface ReflectiveMapChannel extends ReadChannel { + Map getResolvedObject(); + } + + interface ReflectiveCollectionChannel extends ReadChannel { + List getResolvedObject(); + } + + interface ReflectiveNoSizeChannel extends ReadChannel { + String getResolvedObject(); + } + + interface ReflectiveErrorChannel extends ReadChannel { + ThrowingModel getResolvedObject(); + } + + interface ReflectiveGenericChannel extends ReadChannel { + Object getResolvedObject(); + } + + interface ReflectiveNullAndNonNullMethodsChannel extends ReadChannel { + Object getObject(); + + BlobInfo getBlobInfo(); + } +} 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 8eab6255..6231a831 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; @@ -35,12 +36,14 @@ 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; 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; @@ -944,6 +947,159 @@ 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()); + } + + @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(); @@ -969,4 +1125,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/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); + } } 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; }