You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "prudhvimaharishi (via GitHub)" <gi...@apache.org> on 2025/10/15 02:53:53 UTC
[PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
prudhvimaharishi opened a new pull request, #14333:
URL: https://github.com/apache/iceberg/pull/14333
closes #14326
A new GoogleCloudStorageInputStreamWrapper is created which delegates the calls to input stream provided by GCS Analytics Core Library. The implementation is hidden behind the flag - "gcs.analytics.core.enabled".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2519941041
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
Review Comment:
Does this need to be public? Feels like it's ok at package-level protection.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] GCS: Integrate GCSAnalyticsCore Library [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2604714236
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
Review Comment:
I guess adding metrics for the vectored path is something we need to look at, but we can address that separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "ajayky-os (via GitHub)" <gi...@apache.org>.
ajayky-os commented on PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#issuecomment-3628508658
> > Ran TPC-DS and TP-CH suite on a spark cluster to validate the functionality.
>
> @prudhvimaharishi could you share the results if possible?
We are working on verifying the scan time improvements, micro benchmark we have documented now at https://github.com/GoogleCloudPlatform/gcs-analytics-core?tab=readme-ov-file#micro-benchmarks.
For end to end query execution time, we did 2 set of benchmarks:
1. (gcs-analytics-core enabled, vectored read enabled) vs (gcs-analytics-core disabled and vectored read disabled)
2. (gcs-analytics-core enabled, vectored read enabled) vs (gcs-analytics-core disabled and vectored read enabled)
E2E benchmarking setup:
- 3 iterations of TPCDS and TPCH queries were run in spark job, sparkMeasure was used to capture stats like CPU time, gc time, etc.
- E2E is end to end time for running the spark sql query.
- -ve percentage change means gcs-analytics-core performed better.
**(gcs-analytics-core enabled, vectored read enabled) vs (gcs-analytics-core disabled and vectored read disabled)**
Median of 3 iterations:
|Schema Size|% Change E2E Time|% Change Executor Cpu Time|% Change Jvm Gc Time|% Change Shuffle Fetch Wait Time|
|---|---|---|---|---|
|tpcds_sf10|-4.81|5.50|28.31|36.51|
|tpcds_sf100|-4.33|38.60|78.18|60.40|
|tpcds_sf1000|-6.59|9.63|95.48|204.39|
|tpch_sf10|-1.76|-4.68|49.11|42.96|
|tpch_sf100|-2.73|-5.27|113.64|73.02|
|tpch_sf1000|-7.87|-4.08|64.41|89.27|
Average of 3 iterations:
|Schema Size|% Change E2E Time|% Change Executor Cpu Time|% Change Jvm Gc Time|% Change Shuffle Fetch Wait Time|
|---|---|---|---|---|
|tpcds_sf10|-4.14|5.49|40.84|150.38|
|tpcds_sf100|-3.45|6.67|85.25|140.36|
|tpcds_sf1000|-4.92|8.46|100.96|113.59|
|tpch_sf10|-2.34|-4.53|41.88|95.81|
|tpch_sf100|-2.68|-5.10|104.57|298.36|
|tpch_sf1000|-6.79|-4.66|84.59|102.29|
**(gcs-analytics-core enabled, vectored read enabled) vs (gcs-analytics-core disabled and vectored read enabled)**
Median of 3 iterations
|Schema Size|% Change E2E Time|% Change Executor Cpu Time|% Change Jvm Gc Time|% Change Shuffle Fetch Wait Time|
|---|---|---|---|---|
|tpcds_sf10|-7.66|34.07|5.94|-27.07|
|tpcds_sf100|-0.81|4701.41|20.89|15.47|
|tpcds_sf1000|-8.93|38.85|33.71|35.10|
|tpch_sf10|-2.21|-8.04|-17.37|-3.11|
|tpch_sf100|-1.40|-5.30|9.73|-3.21|
|tpch_sf1000|6.73|-6.28|17.29|647.98|
Average of 3 iterations:
|Schema Size|% Change E2E Time|% Change Executor Cpu Time|% Change Jvm Gc Time|% Change Shuffle Fetch Wait Time|
|---|---|---|---|---|
|tpcds_sf10|-8.14|28.59|18.44|56.34|
|tpcds_sf100|-1.28|4051.78|48.83|89.41|
|tpcds_sf1000|-4.10|30301.87|6431.37|-9.43|
|tpch_sf10|-2.15|-7.66|-20.96|-27.98|
|tpch_sf100|-3.14|-5.11|10.50|169.43|
|tpch_sf1000|7.89|-5.54|17.97|318.64|
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567574240
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new com.github.dockerjava.api.model.HostConfig()
+ .withPortBindings(
+ new com.github.dockerjava.api.model.PortBinding(
+ com.github.dockerjava.api.model.Ports.Binding.bindPort(
+ GCS_EMULATOR_PORT),
+ new com.github.dockerjava.api.model.ExposedPort(
+ GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
+ String endpoint = String.format("http://localhost:%d", GCS_EMULATOR_PORT);
+ StorageOptions options =
+ StorageOptions.newBuilder()
+ .setProjectId(PROJECT_ID)
+ .setHost(endpoint)
+ .setCredentials(NoCredentials.getInstance())
+ .build();
+ storage = options.getService();
+ storage.create(BucketInfo.of(BUCKET));
+ }
+
+ @AfterAll
+ public static void afterClass() {
+ if (storage != null) {
+ storage.delete(BUCKET);
+ }
+ GCS_EMULATOR.stop();
+ }
+
+ @BeforeEach
+ public void before() {
+ fileIO = new GCSFileIO(() -> storage);
+ fileIO.initialize(ImmutableMap.of());
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @AfterEach
+ public void after() {
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @Test
+ public void testNewInputFileGcsAnalyticsCoreDisabled() throws IOException {
+ String location = String.format("gs://%s/path/to/file.txt", BUCKET);
+ byte[] expected = new byte[1024 * 1024];
+ random.nextBytes(expected);
+ storage.create(BlobInfo.newBuilder(BlobId.fromGsUtilUri(location)).build(), expected);
+ InputFile in = fileIO.newInputFile(location);
+ byte[] actual = new byte[1024 * 1024];
+
+ try (InputStream is = in.newStream()) {
+ IOUtil.readFully(is, actual, 0, expected.length);
+ }
+
+ assertThat(actual).isEqualTo(expected);
+ }
+
+ @Test
+ public void testNewInputFileGcsAnalyticsCoreEnabled() throws IOException {
+ String location = String.format("gs://%s/path/to/file.txt", BUCKET);
+ byte[] expected = new byte[1024 * 1024];
+ random.nextBytes(expected);
+ storage.create(BlobInfo.newBuilder(BlobId.fromGsUtilUri(location)).build(), expected);
+ fileIO.initialize(
+ ImmutableMap.of(
+ "gcs.analytics-core.enabled", "true",
Review Comment:
we should be able to use the constant defined in `GCPProperties`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567561181
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new com.github.dockerjava.api.model.HostConfig()
Review Comment:
can we replace all of those qualified imports with normal ones?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567603940
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
Review Comment:
can we make this package-private?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#issuecomment-3584655849
> Ran TPC-DS and TP-CH suite on a spark cluster to validate the functionality.
@prudhvimaharishi could you share the results if possible?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567607451
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
+ implements RangeReadable {
+ private GoogleCloudStorageInputStream googleCloudStorageInputStream;
+
+ public GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream googleCloudStorageInputStream) {
+ this.googleCloudStorageInputStream = googleCloudStorageInputStream;
Review Comment:
maybe there should be a `Preconditions.checkArgument(null != stream, "Invalid stream: null");` check here to make sure this is never null
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2519927619
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +78,33 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ GcsFileInfo fileInfo = getGcsFileInfo();
+ return new GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream.create(gcsFileSystem(), fileInfo));
+ } catch (IOException e) {
+ LOG.error("Failed to create GCS analytics core input stream.", e);
+ throw new RuntimeIOException(
+ e, "Failed to create GCS analytics core input stream for: %s", blobId().toGsUtilUri());
+ }
+ }
+
return new GCSInputStream(storage(), blobId(), blobSize, gcpProperties(), metrics());
}
+
+ GcsFileInfo getGcsFileInfo() {
+ BlobId blobId = blobId();
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(getLength()).build();
Review Comment:
I'm a little concerned about the behavior of the `getLength` call since we may be introducing an extra call just to populate the size. Is this a required field? Can we provide it only if we know the length as opposed to always fetching?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "ajayky-os (via GitHub)" <gi...@apache.org>.
ajayky-os commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2576418282
##########
gcp-bundle/build.gradle:
##########
@@ -28,6 +28,7 @@ project(":iceberg-gcp-bundle") {
implementation "com.google.cloud:google-cloud-storage"
implementation "com.google.cloud:google-cloud-bigquery"
implementation "com.google.cloud:google-cloud-core"
+ implementation "com.google.cloud.gcs.analytics:gcs-analytics-core:1.1.2"
Review Comment:
Updated.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +78,33 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ GcsFileInfo fileInfo = getGcsFileInfo();
+ return new GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream.create(gcsFileSystem(), fileInfo));
+ } catch (IOException e) {
+ LOG.error("Failed to create GCS analytics core input stream.", e);
+ throw new RuntimeIOException(
+ e, "Failed to create GCS analytics core input stream for: %s", blobId().toGsUtilUri());
+ }
+ }
+
return new GCSInputStream(storage(), blobId(), blobSize, gcpProperties(), metrics());
}
+
+ GcsFileInfo getGcsFileInfo() {
+ BlobId blobId = blobId();
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(getLength()).build();
Review Comment:
Thank you for pointing the issue. Updated the implementation to avoid calling `getLength` and use length only when it is available. Now `gcsFileInfo()` is only called when `blobSize != null`.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
+ GcsFileSystemOptions fileSystemOptions = gcsAnalyticsCoreOptions.getGcsFileSystemOptions();
Review Comment:
Currently all the config/options specific to read operation are supported by GcsFileSystem, and GcsFileSystem is being used for read operations only.
https://github.com/GoogleCloudPlatform/gcs-analytics-core/blob/main/CONFIGURATION.md
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
Review Comment:
Update visibility to package private.
##########
build.gradle:
##########
@@ -717,6 +717,7 @@ project(':iceberg-gcp') {
compileOnly platform(libs.google.libraries.bom)
compileOnly "com.google.cloud:google-cloud-storage"
compileOnly "com.google.cloud:google-cloud-kms"
+ compileOnly "com.google.cloud.gcs.analytics:gcs-analytics-core:1.1.2"
Review Comment:
Done.
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/TestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new com.github.dockerjava.api.model.HostConfig()
Review Comment:
Done.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +78,33 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ GcsFileInfo fileInfo = getGcsFileInfo();
+ return new GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream.create(gcsFileSystem(), fileInfo));
+ } catch (IOException e) {
+ LOG.error("Failed to create GCS analytics core input stream.", e);
+ throw new RuntimeIOException(
+ e, "Failed to create GCS analytics core input stream for: %s", blobId().toGsUtilUri());
+ }
+ }
+
return new GCSInputStream(storage(), blobId(), blobSize, gcpProperties(), metrics());
}
+
+ GcsFileInfo getGcsFileInfo() {
Review Comment:
Done.
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new com.github.dockerjava.api.model.HostConfig()
+ .withPortBindings(
+ new com.github.dockerjava.api.model.PortBinding(
+ com.github.dockerjava.api.model.Ports.Binding.bindPort(
+ GCS_EMULATOR_PORT),
+ new com.github.dockerjava.api.model.ExposedPort(
+ GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
+ String endpoint = String.format("http://localhost:%d", GCS_EMULATOR_PORT);
+ StorageOptions options =
+ StorageOptions.newBuilder()
+ .setProjectId(PROJECT_ID)
+ .setHost(endpoint)
+ .setCredentials(NoCredentials.getInstance())
+ .build();
+ storage = options.getService();
+ storage.create(BucketInfo.of(BUCKET));
+ }
+
+ @AfterAll
+ public static void afterClass() {
+ if (storage != null) {
+ storage.delete(BUCKET);
+ }
+ GCS_EMULATOR.stop();
+ }
+
+ @BeforeEach
+ public void before() {
+ fileIO = new GCSFileIO(() -> storage);
+ fileIO.initialize(ImmutableMap.of());
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @AfterEach
+ public void after() {
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @Test
+ public void testNewInputFileGcsAnalyticsCoreDisabled() throws IOException {
Review Comment:
Updated the test cases to avoid this.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
+ implements RangeReadable {
+ private GoogleCloudStorageInputStream googleCloudStorageInputStream;
+
+ public GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream googleCloudStorageInputStream) {
+ this.googleCloudStorageInputStream = googleCloudStorageInputStream;
Review Comment:
Done.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
+ implements RangeReadable {
+ private GoogleCloudStorageInputStream googleCloudStorageInputStream;
Review Comment:
Done
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -37,6 +49,7 @@ static GCSInputFile fromLocation(
String location, long length, PrefixedStorage storage, MetricsContext metrics) {
return new GCSInputFile(
storage.storage(),
+ storage.gcsFileSystem(),
Review Comment:
Some points I had to not pass `storage` directly are:
- Passing individual service in constructor explicitly declares services needed by these classes to function.
- PrefixedStorage has implementation details of creating and configuring `Storage` & `GcsFileSystem`.`GCSInputFile` and `GCSOutputFile` are not coupled to it, which makes it easier to test.
Let me know of your perspective.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSOutputFile.java:
##########
@@ -34,12 +35,20 @@ class GCSOutputFile extends BaseGCSFile implements OutputFile {
static GCSOutputFile fromLocation(
String location, PrefixedStorage storage, MetricsContext metrics) {
return new GCSOutputFile(
- storage.storage(), BlobId.fromGsUtilUri(location), storage.gcpProperties(), metrics);
+ storage.storage(),
+ storage.gcsFileSystem(),
Review Comment:
Currently `GcsFileSystem` is added in `GCSOutputFile` to support the `toInputFile()` used for reads.
We do plan to add features (GRPC support, Zonal Buckets, etc.) and optimizations in write behavior and are working on updating the documentation.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
+ GcsFileSystemOptions fileSystemOptions = gcsAnalyticsCoreOptions.getGcsFileSystemOptions();
+ if (this.closeableGroup == null) {
+ this.closeableGroup = new CloseableGroup();
+ }
+ Credentials credentials = getCredentials(properties, closeableGroup);
Review Comment:
Credential being passed to the storage in default implementation and to GcsFileSystem are identical. IIUC credential refresh used by REST implementation is eventually handled at [OAuth2RefreshCredentialsHandler.java](https://github.com/apache/iceberg/blob/main/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java) which is still common in both implementation.
We also tested the implementation with credential vending enabled for biglake catalog which is REST implementation https://docs.cloud.google.com/biglake/docs/blms-rest-catalog#configure-catalog.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
Review Comment:
`GCSInputStream` seems to be capturing metrics only for `read(...)` methods, added those metrics in the wrapper implementation.
Default implementation of [readVectored](https://github.com/apache/iceberg/blob/a2ac14180ecb8e38bb8172e715172b26194b763d/api/src/main/java/org/apache/iceberg/io/RangeReadable.java#L105) uses `readFully()` where we are not capturing these metrics, same with `readTail` method. It seems the existing metrics are not updated consistently. We can take this up separately to make these metrics consistent.
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
Review Comment:
Done
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new com.github.dockerjava.api.model.HostConfig()
+ .withPortBindings(
+ new com.github.dockerjava.api.model.PortBinding(
+ com.github.dockerjava.api.model.Ports.Binding.bindPort(
+ GCS_EMULATOR_PORT),
+ new com.github.dockerjava.api.model.ExposedPort(
+ GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
+ String endpoint = String.format("http://localhost:%d", GCS_EMULATOR_PORT);
+ StorageOptions options =
+ StorageOptions.newBuilder()
+ .setProjectId(PROJECT_ID)
+ .setHost(endpoint)
+ .setCredentials(NoCredentials.getInstance())
+ .build();
+ storage = options.getService();
+ storage.create(BucketInfo.of(BUCKET));
+ }
+
+ @AfterAll
+ public static void afterClass() {
+ if (storage != null) {
+ storage.delete(BUCKET);
+ }
+ GCS_EMULATOR.stop();
+ }
+
+ @BeforeEach
+ public void before() {
+ fileIO = new GCSFileIO(() -> storage);
+ fileIO.initialize(ImmutableMap.of());
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @AfterEach
+ public void after() {
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @Test
+ public void testNewInputFileGcsAnalyticsCoreDisabled() throws IOException {
+ String location = String.format("gs://%s/path/to/file.txt", BUCKET);
+ byte[] expected = new byte[1024 * 1024];
+ random.nextBytes(expected);
+ storage.create(BlobInfo.newBuilder(BlobId.fromGsUtilUri(location)).build(), expected);
+ InputFile in = fileIO.newInputFile(location);
+ byte[] actual = new byte[1024 * 1024];
+
+ try (InputStream is = in.newStream()) {
+ IOUtil.readFully(is, actual, 0, expected.length);
+ }
+
+ assertThat(actual).isEqualTo(expected);
+ }
+
+ @Test
+ public void testNewInputFileGcsAnalyticsCoreEnabled() throws IOException {
+ String location = String.format("gs://%s/path/to/file.txt", BUCKET);
+ byte[] expected = new byte[1024 * 1024];
+ random.nextBytes(expected);
+ storage.create(BlobInfo.newBuilder(BlobId.fromGsUtilUri(location)).build(), expected);
+ fileIO.initialize(
+ ImmutableMap.of(
+ "gcs.analytics-core.enabled", "true",
Review Comment:
Done.
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
Review Comment:
Done.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
Review Comment:
Done
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +78,33 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ GcsFileInfo fileInfo = getGcsFileInfo();
+ return new GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream.create(gcsFileSystem(), fileInfo));
+ } catch (IOException e) {
+ LOG.error("Failed to create GCS analytics core input stream.", e);
+ throw new RuntimeIOException(
Review Comment:
SGTM, updated the implementation to fallback to GCSInputStream.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
Review Comment:
Thanks for pointing this out, it should have been `"gcs."` on place of empty string, added unit test for the same.
In iceberg config keys are prefixed with "gcs." eg: `gcs.project-id`, in hadoop connector similar flags are prefixed with "fs.gs" ex: `fs.gs.project-id`. We intend to use the same implementation in both the places. Hence this prefix is used to avoid repeating handing of similar flags.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
Review Comment:
Done.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
+ GcsFileSystemOptions fileSystemOptions = gcsAnalyticsCoreOptions.getGcsFileSystemOptions();
+ if (this.closeableGroup == null) {
+ this.closeableGroup = new CloseableGroup();
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567640193
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -37,6 +49,7 @@ static GCSInputFile fromLocation(
String location, long length, PrefixedStorage storage, MetricsContext metrics) {
return new GCSInputFile(
storage.storage(),
+ storage.gcsFileSystem(),
Review Comment:
you might as well just pass `storage` directly instead of passing `storage.storage()` and `storage.gcsFileSystem()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567609918
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
Review Comment:
we don't use `get` prefixes in the codebase. Can you please fix this one and any other method prefixed with `get`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567552881
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
Review Comment:
```suggestion
public class TestGCSFileIO {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2519974430
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
Review Comment:
I'm a little concerned about this wrapper specifically because we aren't implementing `GCSInputStream::readVectored` (and that input stream implementation as a whole) and putting the implementation in this wrapper class means we're losing features implemented there like metrics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] GCS: Integrate GCSAnalyticsCore Library [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks merged PR #14333:
URL: https://github.com/apache/iceberg/pull/14333
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "ajayky-os (via GitHub)" <gi...@apache.org>.
ajayky-os commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2599740848
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/TestGcsFileIO.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.github.dockerjava.api.model.ExposedPort;
+import com.github.dockerjava.api.model.HostConfig;
+import com.github.dockerjava.api.model.PortBinding;
+import com.github.dockerjava.api.model.Ports.Binding;
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class TestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private static final Random RANDOM = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new HostConfig()
+ .withPortBindings(
+ new PortBinding(
+ Binding.bindPort(GCS_EMULATOR_PORT),
+ new ExposedPort(GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
Review Comment:
[Yes](https://javadoc.io/doc/org.testcontainers/junit-jupiter/latest/org/testcontainers/junit/jupiter/Testcontainers.html), thanks for pointing out.. removed manual start and stop.
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/TestGcsFileIO.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.github.dockerjava.api.model.ExposedPort;
+import com.github.dockerjava.api.model.HostConfig;
+import com.github.dockerjava.api.model.PortBinding;
+import com.github.dockerjava.api.model.Ports.Binding;
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class TestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private static final Random RANDOM = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new HostConfig()
+ .withPortBindings(
+ new PortBinding(
+ Binding.bindPort(GCS_EMULATOR_PORT),
+ new ExposedPort(GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
+ String endpoint = String.format("http://localhost:%d", GCS_EMULATOR_PORT);
+ StorageOptions options =
+ StorageOptions.newBuilder()
+ .setProjectId(PROJECT_ID)
+ .setHost(endpoint)
+ .setCredentials(NoCredentials.getInstance())
+ .build();
+ storage = options.getService();
+ storage.create(BucketInfo.of(BUCKET));
+ }
+
+ @AfterAll
+ public static void afterClass() {
+ if (storage != null) {
+ storage.delete(BUCKET);
+ }
+ GCS_EMULATOR.stop();
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2519943350
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
+ GcsFileSystemOptions fileSystemOptions = gcsAnalyticsCoreOptions.getGcsFileSystemOptions();
Review Comment:
Are all of the configs/options supported in `GCSProperties` supported with the `GcsFileSystem` implementation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2519907178
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSOutputFile.java:
##########
@@ -34,12 +35,20 @@ class GCSOutputFile extends BaseGCSFile implements OutputFile {
static GCSOutputFile fromLocation(
String location, PrefixedStorage storage, MetricsContext metrics) {
return new GCSOutputFile(
- storage.storage(), BlobId.fromGsUtilUri(location), storage.gcpProperties(), metrics);
+ storage.storage(),
+ storage.gcsFileSystem(),
Review Comment:
Do we really need to go through this path for writes? What improvements or changes is it making to the write behavior that Storage doesn't already provide? This isn't clear from the [documentation](https://github.com/GoogleCloudPlatform/gcs-analytics-core) on the library, which feels a little light on details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "ajayky-os (via GitHub)" <gi...@apache.org>.
ajayky-os commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2599742754
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +77,45 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ return newGoogleCloudStorageInputStream();
+ } catch (IOException e) {
+ LOG.error(
+ "Failed to create GCS analytics core input stream for {}, falling back to default.",
+ uri(),
+ e);
+ }
+ }
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567570170
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new com.github.dockerjava.api.model.HostConfig()
+ .withPortBindings(
+ new com.github.dockerjava.api.model.PortBinding(
+ com.github.dockerjava.api.model.Ports.Binding.bindPort(
+ GCS_EMULATOR_PORT),
+ new com.github.dockerjava.api.model.ExposedPort(
+ GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
+ String endpoint = String.format("http://localhost:%d", GCS_EMULATOR_PORT);
+ StorageOptions options =
+ StorageOptions.newBuilder()
+ .setProjectId(PROJECT_ID)
+ .setHost(endpoint)
+ .setCredentials(NoCredentials.getInstance())
+ .build();
+ storage = options.getService();
+ storage.create(BucketInfo.of(BUCKET));
+ }
+
+ @AfterAll
+ public static void afterClass() {
+ if (storage != null) {
+ storage.delete(BUCKET);
+ }
+ GCS_EMULATOR.stop();
+ }
+
+ @BeforeEach
+ public void before() {
+ fileIO = new GCSFileIO(() -> storage);
+ fileIO.initialize(ImmutableMap.of());
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @AfterEach
+ public void after() {
+ for (Blob blob : storage.list(BUCKET).iterateAll()) {
+ storage.delete(blob.getBlobId());
+ }
+ }
+
+ @Test
+ public void testNewInputFileGcsAnalyticsCoreDisabled() throws IOException {
Review Comment:
we typically don't use `test` prefixes when introducing new tests as those are from JUnit3 and don't add any value
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567626491
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
Review Comment:
what is the prefix here used for where we're passing an empty string?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567609918
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
Review Comment:
we don't use `get` prefixes in the codebase. Can you please fix this one and any other method prefixed with `get`. Also I think this method can be private
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2519946563
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
+ GcsFileSystemOptions fileSystemOptions = gcsAnalyticsCoreOptions.getGcsFileSystemOptions();
+ if (this.closeableGroup == null) {
+ this.closeableGroup = new CloseableGroup();
+ }
+ Credentials credentials = getCredentials(properties, closeableGroup);
Review Comment:
It's not clear to me whether the credential refresh used by REST implementations is compatible with this? Can you make sure that using this client doesn't break refresh?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "ajayky-os (via GitHub)" <gi...@apache.org>.
ajayky-os commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2599761102
##########
gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGcsInputFile.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.gcs.analyticscore.client.GcsFileInfo;
+import com.google.cloud.gcs.analyticscore.client.GcsFileSystem;
+import com.google.cloud.gcs.analyticscore.client.GcsItemId;
+import com.google.cloud.gcs.analyticscore.client.GcsItemInfo;
+import com.google.cloud.gcs.analyticscore.core.GcsAnalyticsCoreOptions;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+public class TestGcsInputFile {
+
+ private static final String TEST_BUCKET = "TEST_BUCKET";
+ private static final String KEY = "file/path/a.dat";
+ private static final String LOCATION = "gs://" + TEST_BUCKET + "/" + KEY;
+ private static final long FILE_SIZE = 1024L;
+
+ private Storage storage;
+ private GcsFileSystem gcsFileSystem;
+ private GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions;
+ private PrefixedStorage prefixedStorage;
+ private GCPProperties gcpProperties;
+ private MetricsContext metricsContext;
+ private Blob blob;
+
+ @BeforeEach
+ public void before() {
+ storage = mock(Storage.class);
+ gcsFileSystem = mock(GcsFileSystem.class);
+ prefixedStorage = mock(PrefixedStorage.class);
+ gcsAnalyticsCoreOptions = new GcsAnalyticsCoreOptions("", Collections.emptyMap());
+ gcpProperties = new GCPProperties();
+ metricsContext = MetricsContext.nullMetrics();
+ blob = mock(Blob.class);
+ when(prefixedStorage.storage()).thenReturn(storage);
+ when(prefixedStorage.gcsFileSystem()).thenReturn(gcsFileSystem);
+ when(prefixedStorage.gcpProperties()).thenReturn(gcpProperties);
+ when(gcsFileSystem.getFileSystemOptions())
+ .thenReturn(gcsAnalyticsCoreOptions.getGcsFileSystemOptions());
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocation() {
+ GCSInputFile inputFile = GCSInputFile.fromLocation(LOCATION, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocationWithLength() {
+ GCSInputFile inputFile =
+ GCSInputFile.fromLocation(LOCATION, FILE_SIZE, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLength() {
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLengthCached() {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabled() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(FILE_SIZE).build();
+ GcsFileInfo gcsFileInfo =
+ GcsFileInfo.builder()
+ .setItemInfo(itemInfo)
+ .setUri(URI.create(LOCATION))
+ .setAttributes(ImmutableMap.of())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, gcsFileInfo))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabledObjectSizeNull() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreDisabled() throws IOException {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ try (MockedConstruction<GCSInputStream> mocked =
+ mockConstruction(
+ GCSInputStream.class,
+ (mock, context) -> {
+ assertThat(context.arguments()).hasSize(5);
+ assertThat(context.arguments().get(0)).isEqualTo(storage);
+ assertThat(context.arguments().get(1)).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(context.arguments().get(2)).isEqualTo(FILE_SIZE);
+ assertThat(context.arguments().get(3)).isEqualTo(gcpProperties);
+ assertThat(context.arguments().get(4)).isEqualTo(metricsContext);
+ })) {
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GCSInputStream.class);
+ assertThat(mocked.constructed()).hasSize(1);
+ }
+ }
+ }
+
+ @Test
+ public void newStream_analyticsCoreInitializationFailed() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenThrow(new IOException("GCS connector failed"));
Review Comment:
[GcsInputFile](https://github.com/apache/iceberg/pull/14333/files#diff-115d5cd9b3751551797c0b45afb5603f47327588159829b3e3118cd9d2ce1af9R83) captures the exception and fallbacks to default InputStream implementation on place of throwing it.. added an assertion on `mocked` that it is called, not sure we can assert on the exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2601384133
##########
gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGcsInputFile.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.gcs.analyticscore.client.GcsFileInfo;
+import com.google.cloud.gcs.analyticscore.client.GcsFileSystem;
+import com.google.cloud.gcs.analyticscore.client.GcsItemId;
+import com.google.cloud.gcs.analyticscore.client.GcsItemInfo;
+import com.google.cloud.gcs.analyticscore.core.GcsAnalyticsCoreOptions;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+public class TestGcsInputFile {
+
+ private static final String TEST_BUCKET = "TEST_BUCKET";
+ private static final String KEY = "file/path/a.dat";
+ private static final String LOCATION = "gs://" + TEST_BUCKET + "/" + KEY;
+ private static final long FILE_SIZE = 1024L;
+
+ private Storage storage;
+ private GcsFileSystem gcsFileSystem;
+ private GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions;
+ private PrefixedStorage prefixedStorage;
+ private GCPProperties gcpProperties;
+ private MetricsContext metricsContext;
+ private Blob blob;
+
+ @BeforeEach
+ public void before() {
+ storage = mock(Storage.class);
+ gcsFileSystem = mock(GcsFileSystem.class);
+ prefixedStorage = mock(PrefixedStorage.class);
+ gcsAnalyticsCoreOptions = new GcsAnalyticsCoreOptions("", Collections.emptyMap());
+ gcpProperties = new GCPProperties();
+ metricsContext = MetricsContext.nullMetrics();
+ blob = mock(Blob.class);
+ when(prefixedStorage.storage()).thenReturn(storage);
+ when(prefixedStorage.gcsFileSystem()).thenReturn(gcsFileSystem);
+ when(prefixedStorage.gcpProperties()).thenReturn(gcpProperties);
+ when(gcsFileSystem.getFileSystemOptions())
+ .thenReturn(gcsAnalyticsCoreOptions.getGcsFileSystemOptions());
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocation() {
+ GCSInputFile inputFile = GCSInputFile.fromLocation(LOCATION, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocationWithLength() {
+ GCSInputFile inputFile =
+ GCSInputFile.fromLocation(LOCATION, FILE_SIZE, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLength() {
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLengthCached() {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabled() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(FILE_SIZE).build();
+ GcsFileInfo gcsFileInfo =
+ GcsFileInfo.builder()
+ .setItemInfo(itemInfo)
+ .setUri(URI.create(LOCATION))
+ .setAttributes(ImmutableMap.of())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, gcsFileInfo))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabledObjectSizeNull() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreDisabled() throws IOException {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ try (MockedConstruction<GCSInputStream> mocked =
+ mockConstruction(
+ GCSInputStream.class,
+ (mock, context) -> {
+ assertThat(context.arguments()).hasSize(5);
+ assertThat(context.arguments().get(0)).isEqualTo(storage);
+ assertThat(context.arguments().get(1)).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(context.arguments().get(2)).isEqualTo(FILE_SIZE);
+ assertThat(context.arguments().get(3)).isEqualTo(gcpProperties);
+ assertThat(context.arguments().get(4)).isEqualTo(metricsContext);
+ })) {
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GCSInputStream.class);
+ assertThat(mocked.constructed()).hasSize(1);
+ }
+ }
+ }
+
+ @Test
+ public void newStream_analyticsCoreInitializationFailed() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenThrow(new IOException("GCS connector failed"));
Review Comment:
can you please add a comment to the test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2519919359
##########
gcp-bundle/build.gradle:
##########
@@ -28,6 +28,7 @@ project(":iceberg-gcp-bundle") {
implementation "com.google.cloud:google-cloud-storage"
implementation "com.google.cloud:google-cloud-bigquery"
implementation "com.google.cloud:google-cloud-core"
+ implementation "com.google.cloud.gcs.analytics:gcs-analytics-core:1.1.2"
Review Comment:
We need to update the LICENSE/NOTICE files for this additional dependency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567553982
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/IntegrationTestGcsFileIO.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class IntegrationTestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private final Random random = new Random(1);
Review Comment:
this can probably be static as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567550794
##########
build.gradle:
##########
@@ -717,6 +717,7 @@ project(':iceberg-gcp') {
compileOnly platform(libs.google.libraries.bom)
compileOnly "com.google.cloud:google-cloud-storage"
compileOnly "com.google.cloud:google-cloud-kms"
+ compileOnly "com.google.cloud.gcs.analytics:gcs-analytics-core:1.1.2"
Review Comment:
please define the version in libs.versions.toml
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567587865
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +78,33 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ GcsFileInfo fileInfo = getGcsFileInfo();
+ return new GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream.create(gcsFileSystem(), fileInfo));
+ } catch (IOException e) {
+ LOG.error("Failed to create GCS analytics core input stream.", e);
+ throw new RuntimeIOException(
+ e, "Failed to create GCS analytics core input stream for: %s", blobId().toGsUtilUri());
+ }
+ }
+
return new GCSInputStream(storage(), blobId(), blobSize, gcpProperties(), metrics());
}
+
+ GcsFileInfo getGcsFileInfo() {
Review Comment:
```suggestion
private GcsFileInfo gcsFileInfo() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567604523
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+
+public class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream
+ implements RangeReadable {
+ private GoogleCloudStorageInputStream googleCloudStorageInputStream;
Review Comment:
```suggestion
private final GoogleCloudStorageInputStream stream;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567592889
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +78,33 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ GcsFileInfo fileInfo = getGcsFileInfo();
+ return new GoogleCloudStorageInputStreamWrapper(
+ GoogleCloudStorageInputStream.create(gcsFileSystem(), fileInfo));
+ } catch (IOException e) {
+ LOG.error("Failed to create GCS analytics core input stream.", e);
+ throw new RuntimeIOException(
Review Comment:
what about falling back to the GCSInputStream here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2567633495
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/PrefixedStorage.java:
##########
@@ -117,5 +126,54 @@ public void close() {
// GCS Storage does not appear to be closable, so release the reference
storage = null;
}
+
+ if (null != gcsFileSystem) {
+ gcsFileSystem.close();
+ gcsFileSystem = null;
+ }
+ }
+
+ static Credentials getCredentials(Map<String, String> properties, CloseableGroup closeableGroup) {
+ GCPProperties gcpProperties = new GCPProperties(properties);
+ if (gcpProperties.oauth2Token().isPresent()) {
+ return GCPAuthUtils.oauth2CredentialsFromGcpProperties(
+ new GCPProperties(properties), closeableGroup);
+ } else if (gcpProperties.noAuth()) {
+ return NoCredentials.getInstance();
+ } else {
+ return null;
+ }
+ }
+
+ public GcsFileSystem gcsFileSystem() {
+ if (gcsFileSystem == null) {
+ synchronized (this) {
+ if (gcsFileSystem == null) {
+ this.gcsFileSystem = gcsFileSystemSupplier.get();
+ }
+ }
+ }
+ return this.gcsFileSystem;
+ }
+
+ private SerializableSupplier<GcsFileSystem> getGcsFileSystemSupplier(
+ Map<String, String> properties) {
+ ImmutableMap.Builder<String, String> propertiesWithUserAgent =
+ new ImmutableMap.Builder<String, String>()
+ .putAll(properties)
+ .put("user-agent", GCS_FILE_IO_USER_AGENT);
+ GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions =
+ new GcsAnalyticsCoreOptions("", propertiesWithUserAgent.build());
+ GcsFileSystemOptions fileSystemOptions = gcsAnalyticsCoreOptions.getGcsFileSystemOptions();
+ if (this.closeableGroup == null) {
+ this.closeableGroup = new CloseableGroup();
Review Comment:
we should probably just init this always in the constructor. Also you probably want to add `gcsFileSystem` to the closeable group after `gcsFileSystem` was initialized (then you wouldn't need to call close on it explicitly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "prudhvimaharishi (via GitHub)" <gi...@apache.org>.
prudhvimaharishi commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2446681906
##########
gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java:
##########
@@ -128,6 +132,8 @@ public GCPProperties(Map<String, String> properties) {
gcsDeleteBatchSize =
PropertyUtil.propertyAsInt(
properties, GCS_DELETE_BATCH_SIZE, GCS_DELETE_BATCH_SIZE_DEFAULT);
+ gcsAnalyticsCoreEnabled =
Review Comment:
This is a first iteration for the library, we dint want to change default functionality in this iteration.
Our end goal is to host many more optimizations, production test this on real world analytical workloads and enable this by default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "mrcnc (via GitHub)" <gi...@apache.org>.
mrcnc commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2445410868
##########
gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java:
##########
@@ -128,6 +132,8 @@ public GCPProperties(Map<String, String> properties) {
gcsDeleteBatchSize =
PropertyUtil.propertyAsInt(
properties, GCS_DELETE_BATCH_SIZE, GCS_DELETE_BATCH_SIZE_DEFAULT);
+ gcsAnalyticsCoreEnabled =
Review Comment:
Any reason why we would not want this enabled by default?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2597837215
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/TestGcsFileIO.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.github.dockerjava.api.model.ExposedPort;
+import com.github.dockerjava.api.model.HostConfig;
+import com.github.dockerjava.api.model.PortBinding;
+import com.github.dockerjava.api.model.Ports.Binding;
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class TestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private static final Random RANDOM = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new HostConfig()
+ .withPortBindings(
+ new PortBinding(
+ Binding.bindPort(GCS_EMULATOR_PORT),
+ new ExposedPort(GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
Review Comment:
container start should be already handled by the annotations I believe
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java:
##########
@@ -64,6 +77,45 @@ public long getLength() {
@Override
public SeekableInputStream newStream() {
+ if (gcpProperties().isGcsAnalyticsCoreEnabled()) {
+ try {
+ return newGoogleCloudStorageInputStream();
+ } catch (IOException e) {
+ LOG.error(
+ "Failed to create GCS analytics core input stream for {}, falling back to default.",
+ uri(),
+ e);
+ }
+ }
Review Comment:
nit: newline after } here and in other places
##########
gcp/src/integration/java/org/apache/iceberg/gcp/gcs/TestGcsFileIO.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.github.dockerjava.api.model.ExposedPort;
+import com.github.dockerjava.api.model.HostConfig;
+import com.github.dockerjava.api.model.PortBinding;
+import com.github.dockerjava.api.model.Ports.Binding;
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.IOUtil;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class TestGcsFileIO {
+
+ private static final String BUCKET = "test-bucket";
+ private static final String PROJECT_ID = "test-project";
+ private static final int GCS_EMULATOR_PORT = 4443;
+ private static final Random RANDOM = new Random(1);
+
+ @Container
+ private static final GenericContainer<?> GCS_EMULATOR =
+ new GenericContainer<>("fsouza/fake-gcs-server:latest")
+ .withExposedPorts(GCS_EMULATOR_PORT)
+ .withCreateContainerCmdModifier(
+ cmd ->
+ cmd.withHostConfig(
+ new HostConfig()
+ .withPortBindings(
+ new PortBinding(
+ Binding.bindPort(GCS_EMULATOR_PORT),
+ new ExposedPort(GCS_EMULATOR_PORT)))))
+ .withCommand(
+ "-scheme",
+ "http",
+ "-external-url",
+ String.format("http://localhost:%d", GCS_EMULATOR_PORT))
+ .waitingFor(
+ new HttpWaitStrategy()
+ .forPort(GCS_EMULATOR_PORT)
+ .forPath("/storage/v1/b")
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+
+ private GCSFileIO fileIO;
+ private static Storage storage;
+
+ @BeforeAll
+ public static void beforeClass() {
+ GCS_EMULATOR.start();
+ String endpoint = String.format("http://localhost:%d", GCS_EMULATOR_PORT);
+ StorageOptions options =
+ StorageOptions.newBuilder()
+ .setProjectId(PROJECT_ID)
+ .setHost(endpoint)
+ .setCredentials(NoCredentials.getInstance())
+ .build();
+ storage = options.getService();
+ storage.create(BucketInfo.of(BUCKET));
+ }
+
+ @AfterAll
+ public static void afterClass() {
+ if (storage != null) {
+ storage.delete(BUCKET);
+ }
+ GCS_EMULATOR.stop();
Review Comment:
stopping should also be already handled
##########
gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntFunction;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class TestGoogleCloudStorageInputStreamWrapper {
Review Comment:
```suggestion
public class TestGCSInputStreamWrapper {
```
##########
gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGcsInputFile.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.gcs.analyticscore.client.GcsFileInfo;
+import com.google.cloud.gcs.analyticscore.client.GcsFileSystem;
+import com.google.cloud.gcs.analyticscore.client.GcsItemId;
+import com.google.cloud.gcs.analyticscore.client.GcsItemInfo;
+import com.google.cloud.gcs.analyticscore.core.GcsAnalyticsCoreOptions;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+public class TestGcsInputFile {
+
+ private static final String TEST_BUCKET = "TEST_BUCKET";
+ private static final String KEY = "file/path/a.dat";
+ private static final String LOCATION = "gs://" + TEST_BUCKET + "/" + KEY;
+ private static final long FILE_SIZE = 1024L;
+
+ private Storage storage;
+ private GcsFileSystem gcsFileSystem;
+ private GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions;
+ private PrefixedStorage prefixedStorage;
+ private GCPProperties gcpProperties;
+ private MetricsContext metricsContext;
+ private Blob blob;
+
+ @BeforeEach
+ public void before() {
+ storage = mock(Storage.class);
+ gcsFileSystem = mock(GcsFileSystem.class);
+ prefixedStorage = mock(PrefixedStorage.class);
+ gcsAnalyticsCoreOptions = new GcsAnalyticsCoreOptions("", Collections.emptyMap());
+ gcpProperties = new GCPProperties();
+ metricsContext = MetricsContext.nullMetrics();
+ blob = mock(Blob.class);
+ when(prefixedStorage.storage()).thenReturn(storage);
+ when(prefixedStorage.gcsFileSystem()).thenReturn(gcsFileSystem);
+ when(prefixedStorage.gcpProperties()).thenReturn(gcpProperties);
+ when(gcsFileSystem.getFileSystemOptions())
+ .thenReturn(gcsAnalyticsCoreOptions.getGcsFileSystemOptions());
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocation() {
+ GCSInputFile inputFile = GCSInputFile.fromLocation(LOCATION, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocationWithLength() {
+ GCSInputFile inputFile =
+ GCSInputFile.fromLocation(LOCATION, FILE_SIZE, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLength() {
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLengthCached() {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabled() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(FILE_SIZE).build();
+ GcsFileInfo gcsFileInfo =
+ GcsFileInfo.builder()
+ .setItemInfo(itemInfo)
+ .setUri(URI.create(LOCATION))
+ .setAttributes(ImmutableMap.of())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, gcsFileInfo))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabledObjectSizeNull() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreDisabled() throws IOException {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ try (MockedConstruction<GCSInputStream> mocked =
+ mockConstruction(
+ GCSInputStream.class,
+ (mock, context) -> {
+ assertThat(context.arguments()).hasSize(5);
+ assertThat(context.arguments().get(0)).isEqualTo(storage);
+ assertThat(context.arguments().get(1)).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(context.arguments().get(2)).isEqualTo(FILE_SIZE);
+ assertThat(context.arguments().get(3)).isEqualTo(gcpProperties);
+ assertThat(context.arguments().get(4)).isEqualTo(metricsContext);
+ })) {
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GCSInputStream.class);
+ assertThat(mocked.constructed()).hasSize(1);
+ }
+ }
+ }
+
+ @Test
+ public void newStream_analyticsCoreInitializationFailed() throws IOException {
Review Comment:
```suggestion
public void newStreamAnalyticsCoreInitializationFailed() throws IOException {
```
##########
gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGcsInputFile.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.gcs.analyticscore.client.GcsFileInfo;
+import com.google.cloud.gcs.analyticscore.client.GcsFileSystem;
+import com.google.cloud.gcs.analyticscore.client.GcsItemId;
+import com.google.cloud.gcs.analyticscore.client.GcsItemInfo;
+import com.google.cloud.gcs.analyticscore.core.GcsAnalyticsCoreOptions;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+public class TestGcsInputFile {
+
+ private static final String TEST_BUCKET = "TEST_BUCKET";
+ private static final String KEY = "file/path/a.dat";
+ private static final String LOCATION = "gs://" + TEST_BUCKET + "/" + KEY;
+ private static final long FILE_SIZE = 1024L;
+
+ private Storage storage;
+ private GcsFileSystem gcsFileSystem;
+ private GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions;
+ private PrefixedStorage prefixedStorage;
+ private GCPProperties gcpProperties;
+ private MetricsContext metricsContext;
+ private Blob blob;
+
+ @BeforeEach
+ public void before() {
+ storage = mock(Storage.class);
+ gcsFileSystem = mock(GcsFileSystem.class);
+ prefixedStorage = mock(PrefixedStorage.class);
+ gcsAnalyticsCoreOptions = new GcsAnalyticsCoreOptions("", Collections.emptyMap());
+ gcpProperties = new GCPProperties();
+ metricsContext = MetricsContext.nullMetrics();
+ blob = mock(Blob.class);
+ when(prefixedStorage.storage()).thenReturn(storage);
+ when(prefixedStorage.gcsFileSystem()).thenReturn(gcsFileSystem);
+ when(prefixedStorage.gcpProperties()).thenReturn(gcpProperties);
+ when(gcsFileSystem.getFileSystemOptions())
+ .thenReturn(gcsAnalyticsCoreOptions.getGcsFileSystemOptions());
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocation() {
+ GCSInputFile inputFile = GCSInputFile.fromLocation(LOCATION, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocationWithLength() {
+ GCSInputFile inputFile =
+ GCSInputFile.fromLocation(LOCATION, FILE_SIZE, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLength() {
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLengthCached() {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabled() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(FILE_SIZE).build();
+ GcsFileInfo gcsFileInfo =
+ GcsFileInfo.builder()
+ .setItemInfo(itemInfo)
+ .setUri(URI.create(LOCATION))
+ .setAttributes(ImmutableMap.of())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, gcsFileInfo))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabledObjectSizeNull() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreDisabled() throws IOException {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ try (MockedConstruction<GCSInputStream> mocked =
+ mockConstruction(
+ GCSInputStream.class,
+ (mock, context) -> {
+ assertThat(context.arguments()).hasSize(5);
+ assertThat(context.arguments().get(0)).isEqualTo(storage);
+ assertThat(context.arguments().get(1)).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(context.arguments().get(2)).isEqualTo(FILE_SIZE);
+ assertThat(context.arguments().get(3)).isEqualTo(gcpProperties);
+ assertThat(context.arguments().get(4)).isEqualTo(metricsContext);
+ })) {
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GCSInputStream.class);
+ assertThat(mocked.constructed()).hasSize(1);
+ }
+ }
+ }
+
+ @Test
+ public void newStream_analyticsCoreInitializationFailed() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenThrow(new IOException("GCS connector failed"));
Review Comment:
shouldn't there be a check that verifies that this error is thrown? Something like `assertThatThrownBy(..).isInstanceOf(...).hasMessage(..)`
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.api.client.util.Preconditions;
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileIOMetricsContext;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+
+class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream implements RangeReadable {
+ private final Counter readBytesCounter;
Review Comment:
```suggestion
private final Counter readBytes;
```
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.api.client.util.Preconditions;
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileIOMetricsContext;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+
+class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream implements RangeReadable {
Review Comment:
```suggestion
class GCSInputStreamWrapper extends SeekableInputStream implements RangeReadable {
```
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.api.client.util.Preconditions;
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileIOMetricsContext;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+
+class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream implements RangeReadable {
+ private final Counter readBytesCounter;
+ private final Counter readOperationsCounter;
Review Comment:
```suggestion
private final Counter readOperations;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "ajayky-os (via GitHub)" <gi...@apache.org>.
ajayky-os commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2599766542
##########
gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.IntFunction;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class TestGoogleCloudStorageInputStreamWrapper {
Review Comment:
Done.
##########
gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGcsInputFile.java:
##########
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.gcs.analyticscore.client.GcsFileInfo;
+import com.google.cloud.gcs.analyticscore.client.GcsFileSystem;
+import com.google.cloud.gcs.analyticscore.client.GcsItemId;
+import com.google.cloud.gcs.analyticscore.client.GcsItemInfo;
+import com.google.cloud.gcs.analyticscore.core.GcsAnalyticsCoreOptions;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+public class TestGcsInputFile {
+
+ private static final String TEST_BUCKET = "TEST_BUCKET";
+ private static final String KEY = "file/path/a.dat";
+ private static final String LOCATION = "gs://" + TEST_BUCKET + "/" + KEY;
+ private static final long FILE_SIZE = 1024L;
+
+ private Storage storage;
+ private GcsFileSystem gcsFileSystem;
+ private GcsAnalyticsCoreOptions gcsAnalyticsCoreOptions;
+ private PrefixedStorage prefixedStorage;
+ private GCPProperties gcpProperties;
+ private MetricsContext metricsContext;
+ private Blob blob;
+
+ @BeforeEach
+ public void before() {
+ storage = mock(Storage.class);
+ gcsFileSystem = mock(GcsFileSystem.class);
+ prefixedStorage = mock(PrefixedStorage.class);
+ gcsAnalyticsCoreOptions = new GcsAnalyticsCoreOptions("", Collections.emptyMap());
+ gcpProperties = new GCPProperties();
+ metricsContext = MetricsContext.nullMetrics();
+ blob = mock(Blob.class);
+ when(prefixedStorage.storage()).thenReturn(storage);
+ when(prefixedStorage.gcsFileSystem()).thenReturn(gcsFileSystem);
+ when(prefixedStorage.gcpProperties()).thenReturn(gcpProperties);
+ when(gcsFileSystem.getFileSystemOptions())
+ .thenReturn(gcsAnalyticsCoreOptions.getGcsFileSystemOptions());
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocation() {
+ GCSInputFile inputFile = GCSInputFile.fromLocation(LOCATION, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void fromLocationWithLength() {
+ GCSInputFile inputFile =
+ GCSInputFile.fromLocation(LOCATION, FILE_SIZE, prefixedStorage, metricsContext);
+
+ assertThat(inputFile.blobId()).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLength() {
+ when(blob.getSize()).thenReturn(FILE_SIZE);
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void getLengthCached() {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ assertThat(inputFile.getLength()).isEqualTo(FILE_SIZE);
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabled() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ GcsItemInfo itemInfo = GcsItemInfo.builder().setItemId(itemId).setSize(FILE_SIZE).build();
+ GcsFileInfo gcsFileInfo =
+ GcsFileInfo.builder()
+ .setItemInfo(itemInfo)
+ .setUri(URI.create(LOCATION))
+ .setAttributes(ImmutableMap.of())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, gcsFileInfo))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreEnabledObjectSizeNull() throws IOException {
+ GCPProperties enabledGcpProperties =
+ new GCPProperties(ImmutableMap.of("gcs.analytics-core.enabled", "true"));
+ BlobId blobId = BlobId.fromGsUtilUri(LOCATION);
+ GcsItemId itemId =
+ GcsItemId.builder()
+ .setBucketName(blobId.getBucket())
+ .setObjectName(blobId.getName())
+ .build();
+ try (MockedStatic<GoogleCloudStorageInputStream> mocked =
+ mockStatic(GoogleCloudStorageInputStream.class)) {
+ mocked
+ .when(() -> GoogleCloudStorageInputStream.create(gcsFileSystem, itemId))
+ .thenReturn(mock(GoogleCloudStorageInputStream.class));
+
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ null,
+ enabledGcpProperties,
+ metricsContext);
+
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GoogleCloudStorageInputStreamWrapper.class);
+ }
+ }
+ }
+
+ @Test
+ public void newStreamGcsAnalyticsCoreDisabled() throws IOException {
+ GCSInputFile inputFile =
+ new GCSInputFile(
+ storage,
+ gcsFileSystem,
+ BlobId.fromGsUtilUri(LOCATION),
+ FILE_SIZE,
+ gcpProperties,
+ metricsContext);
+
+ try (MockedConstruction<GCSInputStream> mocked =
+ mockConstruction(
+ GCSInputStream.class,
+ (mock, context) -> {
+ assertThat(context.arguments()).hasSize(5);
+ assertThat(context.arguments().get(0)).isEqualTo(storage);
+ assertThat(context.arguments().get(1)).isEqualTo(BlobId.fromGsUtilUri(LOCATION));
+ assertThat(context.arguments().get(2)).isEqualTo(FILE_SIZE);
+ assertThat(context.arguments().get(3)).isEqualTo(gcpProperties);
+ assertThat(context.arguments().get(4)).isEqualTo(metricsContext);
+ })) {
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ assertThat(stream).isInstanceOf(GCSInputStream.class);
+ assertThat(mocked.constructed()).hasSize(1);
+ }
+ }
+ }
+
+ @Test
+ public void newStream_analyticsCoreInitializationFailed() throws IOException {
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [GCS] Integrating GCSAnalyticsCore Library into iceberg-gcp. [iceberg]
Posted by "ajayky-os (via GitHub)" <gi...@apache.org>.
ajayky-os commented on code in PR #14333:
URL: https://github.com/apache/iceberg/pull/14333#discussion_r2599765951
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.api.client.util.Preconditions;
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileIOMetricsContext;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+
+class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream implements RangeReadable {
Review Comment:
Done.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.api.client.util.Preconditions;
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileIOMetricsContext;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+
+class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream implements RangeReadable {
+ private final Counter readBytesCounter;
Review Comment:
Done.
##########
gcp/src/main/java/org/apache/iceberg/gcp/gcs/GoogleCloudStorageInputStreamWrapper.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.gcp.gcs;
+
+import com.google.api.client.util.Preconditions;
+import com.google.cloud.gcs.analyticscore.client.GcsObjectRange;
+import com.google.cloud.gcs.analyticscore.core.GoogleCloudStorageInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Collectors;
+import org.apache.iceberg.io.FileIOMetricsContext;
+import org.apache.iceberg.io.FileRange;
+import org.apache.iceberg.io.RangeReadable;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+
+class GoogleCloudStorageInputStreamWrapper extends SeekableInputStream implements RangeReadable {
+ private final Counter readBytesCounter;
+ private final Counter readOperationsCounter;
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org