-
Notifications
You must be signed in to change notification settings - Fork 15
feat: Extract metadata from GCS read channel #333
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
2a58782
31400f4
04e5b11
d977fd8
06914f4
bf0ab92
98be791
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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<ExecutorService> executorServiceSupplier; | ||
| private static final ImmutableMap<String, String> 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) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Drop "GcsReadChannel.this." |
||
| 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 = | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We can drop "this" Use "this" only if the reference is ambigious.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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 |
||
| 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 { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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; | ||
| } | ||
|
shrutisinghania marked this conversation as resolved.
|
||
|
|
||
| private volatile boolean metadataExtractionAttempted = false; | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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[] { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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"}) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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) { | ||
|
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; | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
we can drop this