Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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.");
}
Expand All @@ -119,6 +120,11 @@ public boolean isEof(long position) {
return true;
}

@Override
public ReadChannel getSdkReadChannel() {
return null;
}

@Override
public void close() {}
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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<ExecutorService> executorServiceSupplier;
private static final ImmutableMap<String, String> COMMON_ATTRIBUTES =
Expand Down Expand Up @@ -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) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we can drop this

extractMetadataAfterRead();
}
if (bytesRead >= 0) {
gcsReadChannelPosition += bytesRead;
strategy.position(gcsReadChannelPosition);
Expand Down Expand Up @@ -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.");
}
Expand All @@ -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");
Expand Down Expand Up @@ -284,6 +297,9 @@ void readCombinedRange(
int numOfBytesRead = 0;
while (dataBuffer.hasRemaining()) {
int bytesRead = channel.read(dataBuffer);
if (GcsReadChannel.this.itemInfo == null && !metadataExtractionAttempted) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Drop "GcsReadChannel.this."

extractMetadataAfterRead(readStrategy);
}
if (bytesRead < 0) {
// EOF reached.
break;
Expand Down Expand Up @@ -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 =

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any reason for keeping this type qualified ?

GcsReadChannelMetadataExtractor.extract(strategy.getSdkReadChannel());
if (metadata == null) {
return false;
}
updateItemMetadata(metadata.getSize(), metadata.getGeneration());
return true;
}

private void updateItemMetadata(long extractedSize, long extractedGen) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can drop "this"

Use "this" only if the reference is ambigious.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: A bit confusing to read. Can we reorganize as follows.

extract generation
newline
itemId creation
newline
iteminfo creation
newline
set itemId and itemInfo

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 {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is redundant. we should leave this unset. Please remove.

itemInfoBuilder.setContentGeneration(0L);
}
this.itemInfo = itemInfoBuilder.build();
this.itemId = updatedItemId;
}
Comment thread
shrutisinghania marked this conversation as resolved.

private volatile boolean metadataExtractionAttempted = false;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please format. Move this to variable declaration section.

}
Original file line number Diff line number Diff line change
@@ -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[] {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you please create these as static constants instead.

"getObject", "getResolvedObject", "getBlobInfo", "getBlob", "getStorageObject"
}) {
try {
Method method = clazz.getDeclaredMethod(methodName);
method.setAccessible(true);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This can throw SecurityException or ObjectInaccesible exception as well. Please modify the catch blocks to handle all RunTimeExceptions to be sure.

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"}) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you please create these as static constants

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) {
Comment thread
shrutisinghania marked this conversation as resolved.
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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -30,6 +31,9 @@ interface ReadStrategy {
*/
ReadChannel getReadChannel(long requestedPosition, int bytesToRead) throws IOException;

@Nullable
ReadChannel getSdkReadChannel();

/**
* Updates the strategy's current read position.
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
/**
Expand All @@ -31,4 +32,10 @@ public interface VectoredSeekableByteChannel extends SeekableByteChannel {
*/
void readVectored(List<GcsObjectRange> ranges, IntFunction<ByteBuffer> allocate)
throws IOException;

/** Returns the item metadata info if available, or null. */
@Nullable
default GcsItemInfo getItemInfo() {
return null;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
Loading
Loading