You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "sungwy (via GitHub)" <gi...@apache.org> on 2026/09/12 19:07:42 UTC
[PR] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
sungwy opened a new pull request, #18079:
URL: https://github.com/apache/iceberg/pull/18079
Downscoped from @williamhyun's work in #17457. This PR supports the client reading pre-signed URLs if the server returns pre-signed URLs in place of `file-path` in a `planTableScan` response.
`PreSignedUrlInputFile` reads an `http` or `https` location as given, with a shared HTTP client.
Note for reviewers: A few assumptions were made regarding the intended user workflow for this initial scope:
1. An `http` or `https` location in a scan task, in place of the native location, is a pre-signed URL, signed for GET, and only read.
2. Expiry is communicated out of band; the client does not track it and uses a URL until it fails.
3. The URL replaces the location and flows through every path that treats `file_path` as a plain location, including the `files` metadata table and engine logs. I think that is acceptable for a first step; carrying the URL beside a native location would be a spec change to the task.
Out of scope: the client asking the server to pre-sign a location. That is a separate spec change and best addressed as a follow up change/implementation to this PR.
The change was initially drafted with Claude. The design, scope and review are mine.
In addition to the tests included in this PR, the read path was verified against S3 and GCS objects that were signed out of band.
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "sungwy (via GitHub)" <gi...@apache.org>.
sungwy commented on code in PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#discussion_r3997276143
##########
aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java:
##########
@@ -149,11 +150,15 @@ public S3FileIO(SerializableSupplier<S3Client> s3, SerializableSupplier<S3AsyncC
@Override
public InputFile newInputFile(String path) {
- return S3InputFile.fromLocation(path, clientForStoragePath(path), metrics);
+ return newInputFile(path, 0);
}
@Override
public InputFile newInputFile(String path, long length) {
+ if (PreSignedUrlInputFile.isHttpUrl(path)) {
+ return PreSignedUrlInputFile.of(path, length);
Review Comment:
as designed under the current assumption that http / https scheme is the discriminator that a location is presigned url as opposed to the native location
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "dillitz (via GitHub)" <gi...@apache.org>.
dillitz commented on PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#issuecomment-6194343696
> 3. The URL replaces the location and flows through every path that treats `file_path` as a plain location, including the `files` metadata table and engine logs. I think that is acceptable for a first step; carrying the URL beside a native location would be a spec change to the task.
I fear this extends a bit beyond the files table and logs. Besides being where we read a file from, the location is also often how other metadata refers to that file, and those references are compared as exact strings. So once the location becomes a URL, anything pointing at the file would need to point at that same URL too.
For example, a DV says which data file it belongs to through `referenced-data-file`, and the reader checks that it matches the data file's location exactly ([here](https://github.com/apache/iceberg/blob/c3239b8622e6a4a941d4e6f1d2e5d5d67230b4c4/data/src/main/java/org/apache/iceberg/data/BaseDeleteLoader.java#L279-L283)). So the server would also need to put the exact same signed string into every DV's `referenced-data-file`. Otherwise, the scan fails with `DV is expected to reference https://...` (+ the error message includes the signed URL). The DV test here handles this in the [test adapter](https://github.com/apache/iceberg/blob/ceafd63d95a60344f1da5f5e1f800ebc8fc3df78/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java#L1791), but a real server wouldn't necessarily know it has to.
DVs can probably be handled that way since `referenced-data-file` is part of the response, but I fear v2 position deletes can't: the data file paths are stored inside the delete file, so after reading it the client would compare native paths to the URL, and the deletes wouldn't apply. #17457 mentions positional deletes as unsupported, but as far as I can see nothing prevents a server from returning them.
Because of that, I wonder whether carrying the URL alongside the native location would be worth doing from the start, so the URL is only used to fetch bytes and the native path remains for everything that identifies the file. #17332 had a similar idea (a path → URL map in the plan response, with `file-path` left untouched).
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "sungwy (via GitHub)" <gi...@apache.org>.
sungwy commented on code in PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#discussion_r3997280966
##########
core/src/main/java/org/apache/iceberg/io/PreSignedUrlInputStream.java:
##########
@@ -0,0 +1,226 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Locale;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.io.ModalCloseable;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.metrics.MetricsContext.Unit;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * Reads an object through a URL with range requests. A seek beyond the bytes already read reopens
+ * the connection at the new offset.
+ */
+class PreSignedUrlInputStream extends SeekableInputStream {
+
+ private static final int SKIP_SIZE = 1024 * 1024;
+
+ private final CloseableHttpClient http;
+ private final String url;
+ private final Counter readBytes;
+ private final Counter readOperations;
+
+ private ClassicHttpResponse response = null;
+ private InputStream stream = null;
+ private boolean streamAtEof = false;
+ private long pos = 0;
+ private long next = 0;
+ private boolean closed = false;
+
+ PreSignedUrlInputStream(CloseableHttpClient http, String url, MetricsContext metrics) {
+ this.http = http;
+ this.url = url;
+ this.readBytes = metrics.counter(FileIOMetricsContext.READ_BYTES, Unit.BYTES);
+ this.readOperations = metrics.counter(FileIOMetricsContext.READ_OPERATIONS);
+ }
+
+ @Override
+ public long getPos() {
+ return next;
+ }
+
+ @Override
+ public void seek(long newPos) {
+ Preconditions.checkState(!closed, "Cannot seek: already closed");
+ Preconditions.checkArgument(newPos >= 0, "Invalid position (negative): %s", newPos);
+ this.next = newPos;
+ }
+
+ @Override
+ public int read() throws IOException {
+ Preconditions.checkState(!closed, "Cannot read: already closed");
+ positionStream();
+
+ int b = stream.read();
+ if (b >= 0) {
+ pos += 1;
+ next += 1;
+ readBytes.increment();
+ readOperations.increment();
+ } else {
+ streamAtEof = true;
+ }
+
+ return b;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ Preconditions.checkState(!closed, "Cannot read: already closed");
+ if (len == 0) {
+ return 0;
+ }
+
+ positionStream();
+
+ int n = stream.read(b, off, len);
+ if (n > 0) {
+ pos += n;
+ next += n;
+ readBytes.increment(n);
+ readOperations.increment();
+ } else if (n < 0) {
+ streamAtEof = true;
+ }
+
+ return n;
+ }
+
+ private void positionStream() throws IOException {
+ if (stream != null && next == pos) {
+ return;
+ }
+
+ if (stream != null && next > pos && next - pos <= SKIP_SIZE) {
+ long skip = next - pos;
+ try {
+ while (skip > 0) {
+ long skipped = stream.skip(skip);
+ if (skipped <= 0) {
+ break;
+ }
+ skip -= skipped;
+ }
+ } catch (IOException e) {
+ skip = -1;
+ }
+
+ if (skip == 0) {
+ pos = next;
+ return;
+ }
+ }
+
+ closeStream();
+ openStream(next);
+ pos = next;
+ }
+
+ private void openStream(long from) throws IOException {
+ HttpGet get = new HttpGet(url);
+ get.setHeader("Range", String.format(Locale.ROOT, "bytes=%d-", from));
+ ClassicHttpResponse opened = http.executeOpen(null, get, null);
+
+ int code = opened.getCode();
+ if (code == HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
+ // at or past the end of the object
+ discard(opened);
+ this.response = null;
+ this.stream = InputStream.nullInputStream();
+ return;
+ }
+
+ if (code != HttpStatus.SC_OK && code != HttpStatus.SC_PARTIAL_CONTENT) {
+ throw failure("Read", url, opened);
+ }
+
+ if (from > 0 && code == HttpStatus.SC_OK) {
+ // the server ignored the range; reading this as data from byte 0 would be silent corruption
+ discard(opened);
+ throw new IOException(
+ String.format(
+ Locale.ROOT, "Read of %s failed: server ignored Range bytes=%d-", url, from));
+ }
+
+ this.response = opened;
+ this.stream = opened.getEntity().getContent();
+ this.streamAtEof = false;
+ }
+
+ private void closeStream() throws IOException {
+ if (response != null) {
+ if (streamAtEof) {
+ response.close();
+ } else {
+ discard(response);
+ }
+
+ response = null;
+ }
+
+ stream = null;
+ streamAtEof = false;
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ closed = true;
+ closeStream();
+ }
+
+ /**
+ * Closes a response without reading the rest of its body. A graceful close drains the body to
+ * keep the connection reusable, which for an open-ended range means downloading the rest of the
+ * object.
+ */
+ static void discard(ClassicHttpResponse response) throws IOException {
+ if (response instanceof ModalCloseable) {
+ ((ModalCloseable) response).close(CloseMode.IMMEDIATE);
+ } else {
+ response.close();
+ }
+ }
+
+ /** The exception for a failed request; 404 is thrown. */
+ static IOException failure(String what, String url, ClassicHttpResponse response)
+ throws IOException {
+ try {
+ if (response.getCode() == HttpStatus.SC_NOT_FOUND) {
+ throw new NotFoundException("Location does not exist: %s", url);
+ }
+
+ return new IOException(
+ String.format(
+ Locale.ROOT, "%s of %s failed with HTTP %d", what, url, response.getCode()));
Review Comment:
same as above ^
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#issuecomment-6181476378
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#discussion_r3997267619
##########
aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java:
##########
@@ -149,11 +150,15 @@ public S3FileIO(SerializableSupplier<S3Client> s3, SerializableSupplier<S3AsyncC
@Override
public InputFile newInputFile(String path) {
- return S3InputFile.fromLocation(path, clientForStoragePath(path), metrics);
+ return newInputFile(path, 0);
}
@Override
public InputFile newInputFile(String path, long length) {
+ if (PreSignedUrlInputFile.isHttpUrl(path)) {
+ return PreSignedUrlInputFile.of(path, length);
Review Comment:
This short-circuit changes the existing S3FileIO contract for every `http(s)` location: `S3URI` deliberately supports arbitrary URI schemes for S3-compatible paths, and the class documentation says `https` locations are handled as native S3 paths. Such locations will now bypass the configured S3 client and fail unless they happen to be presigned GET URLs. Please preserve native handling for non-presigned locations (or introduce an explicit opt-in/marker) before routing them through this reader.
##########
core/src/main/java/org/apache/iceberg/io/PreSignedUrlInputFile.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link InputFile} addressed by a pre-signed URL: an {@code http} or {@code https} URL carrying
+ * its own authorization. {@link #location()} is the URL itself.
+ *
+ * <p>All instances share one HTTP client.
+ */
+public class PreSignedUrlInputFile implements InputFile {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PreSignedUrlInputFile.class);
+ private static final String DEFAULT_METRICS_IMPL =
+ "org.apache.iceberg.hadoop.HadoopMetricsContext";
+ private static final String METRICS_PREFIX = "presigned-url";
+
+ // pool size, in total and per host
+ private static final int MAX_CONNECTIONS = 100;
+ // the AWS SDK's HTTP defaults, as in S3FileIO; HttpClient's own socket timeout is unbounded
+ private static final long CONNECT_TIMEOUT_MS = 2_000;
+ private static final int SOCKET_TIMEOUT_MS = 30_000;
+
+ private static volatile CloseableHttpClient http;
+ private static volatile MetricsContext metrics;
+
+ private final String url;
+ private Long length;
+
+ private PreSignedUrlInputFile(String url, long length) {
+ this.url = url;
+ this.length = length > 0 ? length : null;
+ }
+
+ /** Whether {@code location} is an {@code http} or {@code https} URL. */
+ public static boolean isHttpUrl(String location) {
+ String lower = location.toLowerCase(Locale.ROOT);
+ return lower.startsWith("https://") || lower.startsWith("http://");
+ }
+
+ /**
+ * Returns an input file that reads {@code url} as given.
+ *
+ * @param url an {@code http} or {@code https} URL
+ * @param length the file length if known, otherwise {@code 0}
+ */
+ public static InputFile of(String url, long length) {
+ Preconditions.checkArgument(isHttpUrl(url), "Not an http or https URL: %s", url);
+ return new PreSignedUrlInputFile(url, length);
+ }
+
+ /**
+ * The length comes from the caller; {@code 0} means unknown, as in {@code S3InputFile}, and is
+ * then read from {@code Content-Range} on a single-byte range GET. HEAD is not an option, as the
+ * method is part of the signature and pre-signed URLs are signed for GET.
+ */
+ @Override
+ public long getLength() {
+ if (length == null) {
+ this.length = probeLength();
+ }
+
+ return length;
+ }
+
+ @Override
+ public SeekableInputStream newStream() {
+ return new PreSignedUrlInputStream(http(), url, metrics());
+ }
+
+ @Override
+ public String location() {
+ return url;
+ }
+
+ @Override
+ public boolean exists() {
+ try {
+ long probed = probeLength();
+ if (length == null) {
+ this.length = probed;
+ }
+
+ return true;
+ } catch (NotFoundException e) {
+ return false;
+ }
+ }
+
+ /** One single-byte range GET; the same request answers both the length and existence. */
+ private long probeLength() {
+ HttpGet get = new HttpGet(url);
+ get.setHeader("Range", "bytes=0-0");
+ try {
+ ClassicHttpResponse response = http().executeOpen(null, get, null);
+ int code = response.getCode();
+ if (code != HttpStatus.SC_PARTIAL_CONTENT
+ && code != HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
+ throw PreSignedUrlInputStream.failure("Length probe", url, response);
+ }
+
+ // Content-Range: bytes 0-0/<length>, or bytes */<length> for an empty object
+ Header contentRange = response.getFirstHeader("Content-Range");
+ PreSignedUrlInputStream.discard(response);
+ if (contentRange == null || !contentRange.getValue().contains("/")) {
+ throw new IOException("Length probe of " + url + " returned no Content-Range");
Review Comment:
When a length probe receives malformed metadata, this message exposes the complete pre-signed URL and its query signature. Redact the URL before placing it in the exception so diagnostics do not turn a failed read into credential disclosure.
##########
core/src/main/java/org/apache/iceberg/io/PreSignedUrlInputStream.java:
##########
@@ -0,0 +1,226 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Locale;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.io.ModalCloseable;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.metrics.Counter;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.metrics.MetricsContext.Unit;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * Reads an object through a URL with range requests. A seek beyond the bytes already read reopens
+ * the connection at the new offset.
+ */
+class PreSignedUrlInputStream extends SeekableInputStream {
+
+ private static final int SKIP_SIZE = 1024 * 1024;
+
+ private final CloseableHttpClient http;
+ private final String url;
+ private final Counter readBytes;
+ private final Counter readOperations;
+
+ private ClassicHttpResponse response = null;
+ private InputStream stream = null;
+ private boolean streamAtEof = false;
+ private long pos = 0;
+ private long next = 0;
+ private boolean closed = false;
+
+ PreSignedUrlInputStream(CloseableHttpClient http, String url, MetricsContext metrics) {
+ this.http = http;
+ this.url = url;
+ this.readBytes = metrics.counter(FileIOMetricsContext.READ_BYTES, Unit.BYTES);
+ this.readOperations = metrics.counter(FileIOMetricsContext.READ_OPERATIONS);
+ }
+
+ @Override
+ public long getPos() {
+ return next;
+ }
+
+ @Override
+ public void seek(long newPos) {
+ Preconditions.checkState(!closed, "Cannot seek: already closed");
+ Preconditions.checkArgument(newPos >= 0, "Invalid position (negative): %s", newPos);
+ this.next = newPos;
+ }
+
+ @Override
+ public int read() throws IOException {
+ Preconditions.checkState(!closed, "Cannot read: already closed");
+ positionStream();
+
+ int b = stream.read();
+ if (b >= 0) {
+ pos += 1;
+ next += 1;
+ readBytes.increment();
+ readOperations.increment();
+ } else {
+ streamAtEof = true;
+ }
+
+ return b;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ Preconditions.checkState(!closed, "Cannot read: already closed");
+ if (len == 0) {
+ return 0;
+ }
+
+ positionStream();
+
+ int n = stream.read(b, off, len);
+ if (n > 0) {
+ pos += n;
+ next += n;
+ readBytes.increment(n);
+ readOperations.increment();
+ } else if (n < 0) {
+ streamAtEof = true;
+ }
+
+ return n;
+ }
+
+ private void positionStream() throws IOException {
+ if (stream != null && next == pos) {
+ return;
+ }
+
+ if (stream != null && next > pos && next - pos <= SKIP_SIZE) {
+ long skip = next - pos;
+ try {
+ while (skip > 0) {
+ long skipped = stream.skip(skip);
+ if (skipped <= 0) {
+ break;
+ }
+ skip -= skipped;
+ }
+ } catch (IOException e) {
+ skip = -1;
+ }
+
+ if (skip == 0) {
+ pos = next;
+ return;
+ }
+ }
+
+ closeStream();
+ openStream(next);
+ pos = next;
+ }
+
+ private void openStream(long from) throws IOException {
+ HttpGet get = new HttpGet(url);
+ get.setHeader("Range", String.format(Locale.ROOT, "bytes=%d-", from));
+ ClassicHttpResponse opened = http.executeOpen(null, get, null);
+
+ int code = opened.getCode();
+ if (code == HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
+ // at or past the end of the object
+ discard(opened);
+ this.response = null;
+ this.stream = InputStream.nullInputStream();
+ return;
+ }
+
+ if (code != HttpStatus.SC_OK && code != HttpStatus.SC_PARTIAL_CONTENT) {
+ throw failure("Read", url, opened);
+ }
+
+ if (from > 0 && code == HttpStatus.SC_OK) {
+ // the server ignored the range; reading this as data from byte 0 would be silent corruption
+ discard(opened);
+ throw new IOException(
+ String.format(
+ Locale.ROOT, "Read of %s failed: server ignored Range bytes=%d-", url, from));
+ }
+
+ this.response = opened;
+ this.stream = opened.getEntity().getContent();
+ this.streamAtEof = false;
+ }
+
+ private void closeStream() throws IOException {
+ if (response != null) {
+ if (streamAtEof) {
+ response.close();
+ } else {
+ discard(response);
+ }
+
+ response = null;
+ }
+
+ stream = null;
+ streamAtEof = false;
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ closed = true;
+ closeStream();
+ }
+
+ /**
+ * Closes a response without reading the rest of its body. A graceful close drains the body to
+ * keep the connection reusable, which for an open-ended range means downloading the rest of the
+ * object.
+ */
+ static void discard(ClassicHttpResponse response) throws IOException {
+ if (response instanceof ModalCloseable) {
+ ((ModalCloseable) response).close(CloseMode.IMMEDIATE);
+ } else {
+ response.close();
+ }
+ }
+
+ /** The exception for a failed request; 404 is thrown. */
+ static IOException failure(String what, String url, ClassicHttpResponse response)
+ throws IOException {
+ try {
+ if (response.getCode() == HttpStatus.SC_NOT_FOUND) {
+ throw new NotFoundException("Location does not exist: %s", url);
+ }
+
+ return new IOException(
+ String.format(
+ Locale.ROOT, "%s of %s failed with HTTP %d", what, url, response.getCode()));
Review Comment:
This error message includes the complete pre-signed URL, including its query signature. Read failures are commonly logged by engines, so an expired or invalid request can expose a bearer credential; redact the URL before formatting exception messages (the request itself should still use the original URL).
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#issuecomment-5667074040
I thought https://github.com/apache/iceberg/pull/17457 is still active? Shouldn't we just finish that up first?
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "sungwy (via GitHub)" <gi...@apache.org>.
sungwy commented on code in PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#discussion_r3997277708
##########
core/src/main/java/org/apache/iceberg/io/PreSignedUrlInputFile.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link InputFile} addressed by a pre-signed URL: an {@code http} or {@code https} URL carrying
+ * its own authorization. {@link #location()} is the URL itself.
+ *
+ * <p>All instances share one HTTP client.
+ */
+public class PreSignedUrlInputFile implements InputFile {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PreSignedUrlInputFile.class);
+ private static final String DEFAULT_METRICS_IMPL =
+ "org.apache.iceberg.hadoop.HadoopMetricsContext";
+ private static final String METRICS_PREFIX = "presigned-url";
+
+ // pool size, in total and per host
+ private static final int MAX_CONNECTIONS = 100;
+ // the AWS SDK's HTTP defaults, as in S3FileIO; HttpClient's own socket timeout is unbounded
+ private static final long CONNECT_TIMEOUT_MS = 2_000;
+ private static final int SOCKET_TIMEOUT_MS = 30_000;
+
+ private static volatile CloseableHttpClient http;
+ private static volatile MetricsContext metrics;
+
+ private final String url;
+ private Long length;
+
+ private PreSignedUrlInputFile(String url, long length) {
+ this.url = url;
+ this.length = length > 0 ? length : null;
+ }
+
+ /** Whether {@code location} is an {@code http} or {@code https} URL. */
+ public static boolean isHttpUrl(String location) {
+ String lower = location.toLowerCase(Locale.ROOT);
+ return lower.startsWith("https://") || lower.startsWith("http://");
+ }
+
+ /**
+ * Returns an input file that reads {@code url} as given.
+ *
+ * @param url an {@code http} or {@code https} URL
+ * @param length the file length if known, otherwise {@code 0}
+ */
+ public static InputFile of(String url, long length) {
+ Preconditions.checkArgument(isHttpUrl(url), "Not an http or https URL: %s", url);
+ return new PreSignedUrlInputFile(url, length);
+ }
+
+ /**
+ * The length comes from the caller; {@code 0} means unknown, as in {@code S3InputFile}, and is
+ * then read from {@code Content-Range} on a single-byte range GET. HEAD is not an option, as the
+ * method is part of the signature and pre-signed URLs are signed for GET.
+ */
+ @Override
+ public long getLength() {
+ if (length == null) {
+ this.length = probeLength();
+ }
+
+ return length;
+ }
+
+ @Override
+ public SeekableInputStream newStream() {
+ return new PreSignedUrlInputStream(http(), url, metrics());
+ }
+
+ @Override
+ public String location() {
+ return url;
+ }
+
+ @Override
+ public boolean exists() {
+ try {
+ long probed = probeLength();
+ if (length == null) {
+ this.length = probed;
+ }
+
+ return true;
+ } catch (NotFoundException e) {
+ return false;
+ }
+ }
+
+ /** One single-byte range GET; the same request answers both the length and existence. */
+ private long probeLength() {
+ HttpGet get = new HttpGet(url);
+ get.setHeader("Range", "bytes=0-0");
+ try {
+ ClassicHttpResponse response = http().executeOpen(null, get, null);
+ int code = response.getCode();
+ if (code != HttpStatus.SC_PARTIAL_CONTENT
+ && code != HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
+ throw PreSignedUrlInputStream.failure("Length probe", url, response);
+ }
+
+ // Content-Range: bytes 0-0/<length>, or bytes */<length> for an empty object
+ Header contentRange = response.getFirstHeader("Content-Range");
+ PreSignedUrlInputStream.discard(response);
+ if (contentRange == null || !contentRange.getValue().contains("/")) {
+ throw new IOException("Length probe of " + url + " returned no Content-Range");
Review Comment:
this is in accordance with the current design assumptions, where a presigned url is treated like any other iceberg file location, and less like a credential.
Note to reviewers: does this make a case for moving away from the replace url in place approach we are exploring in this PR?
##########
core/src/main/java/org/apache/iceberg/io/PreSignedUrlInputFile.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link InputFile} addressed by a pre-signed URL: an {@code http} or {@code https} URL carrying
+ * its own authorization. {@link #location()} is the URL itself.
+ *
+ * <p>All instances share one HTTP client.
+ */
+public class PreSignedUrlInputFile implements InputFile {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PreSignedUrlInputFile.class);
+ private static final String DEFAULT_METRICS_IMPL =
+ "org.apache.iceberg.hadoop.HadoopMetricsContext";
+ private static final String METRICS_PREFIX = "presigned-url";
+
+ // pool size, in total and per host
+ private static final int MAX_CONNECTIONS = 100;
+ // the AWS SDK's HTTP defaults, as in S3FileIO; HttpClient's own socket timeout is unbounded
+ private static final long CONNECT_TIMEOUT_MS = 2_000;
+ private static final int SOCKET_TIMEOUT_MS = 30_000;
+
+ private static volatile CloseableHttpClient http;
+ private static volatile MetricsContext metrics;
+
+ private final String url;
+ private Long length;
+
+ private PreSignedUrlInputFile(String url, long length) {
+ this.url = url;
+ this.length = length > 0 ? length : null;
+ }
+
+ /** Whether {@code location} is an {@code http} or {@code https} URL. */
+ public static boolean isHttpUrl(String location) {
+ String lower = location.toLowerCase(Locale.ROOT);
+ return lower.startsWith("https://") || lower.startsWith("http://");
+ }
+
+ /**
+ * Returns an input file that reads {@code url} as given.
+ *
+ * @param url an {@code http} or {@code https} URL
+ * @param length the file length if known, otherwise {@code 0}
+ */
+ public static InputFile of(String url, long length) {
+ Preconditions.checkArgument(isHttpUrl(url), "Not an http or https URL: %s", url);
+ return new PreSignedUrlInputFile(url, length);
+ }
+
+ /**
+ * The length comes from the caller; {@code 0} means unknown, as in {@code S3InputFile}, and is
+ * then read from {@code Content-Range} on a single-byte range GET. HEAD is not an option, as the
+ * method is part of the signature and pre-signed URLs are signed for GET.
+ */
+ @Override
+ public long getLength() {
+ if (length == null) {
+ this.length = probeLength();
+ }
+
+ return length;
+ }
+
+ @Override
+ public SeekableInputStream newStream() {
+ return new PreSignedUrlInputStream(http(), url, metrics());
+ }
+
+ @Override
+ public String location() {
+ return url;
+ }
+
+ @Override
+ public boolean exists() {
+ try {
+ long probed = probeLength();
+ if (length == null) {
+ this.length = probed;
+ }
+
+ return true;
+ } catch (NotFoundException e) {
+ return false;
+ }
+ }
+
+ /** One single-byte range GET; the same request answers both the length and existence. */
+ private long probeLength() {
+ HttpGet get = new HttpGet(url);
+ get.setHeader("Range", "bytes=0-0");
+ try {
+ ClassicHttpResponse response = http().executeOpen(null, get, null);
+ int code = response.getCode();
+ if (code != HttpStatus.SC_PARTIAL_CONTENT
+ && code != HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
+ throw PreSignedUrlInputStream.failure("Length probe", url, response);
+ }
+
+ // Content-Range: bytes 0-0/<length>, or bytes */<length> for an empty object
+ Header contentRange = response.getFirstHeader("Content-Range");
+ PreSignedUrlInputStream.discard(response);
+ if (contentRange == null || !contentRange.getValue().contains("/")) {
+ throw new IOException("Length probe of " + url + " returned no Content-Range");
Review Comment:
this is in accordance with the current design assumptions, where a presigned url is treated like any other iceberg file location, and less like a credential.
Note to reviewers: does this make a case for moving away from the replace url in place (e.g. `file-path`) approach we are exploring in this PR?
--
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] Core, AWS, GCP: read pre-signed URLs returned by scan planning [iceberg]
Posted by "sungwy (via GitHub)" <gi...@apache.org>.
sungwy commented on code in PR #18079:
URL: https://github.com/apache/iceberg/pull/18079#discussion_r3997277708
##########
core/src/main/java/org/apache/iceberg/io/PreSignedUrlInputFile.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link InputFile} addressed by a pre-signed URL: an {@code http} or {@code https} URL carrying
+ * its own authorization. {@link #location()} is the URL itself.
+ *
+ * <p>All instances share one HTTP client.
+ */
+public class PreSignedUrlInputFile implements InputFile {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PreSignedUrlInputFile.class);
+ private static final String DEFAULT_METRICS_IMPL =
+ "org.apache.iceberg.hadoop.HadoopMetricsContext";
+ private static final String METRICS_PREFIX = "presigned-url";
+
+ // pool size, in total and per host
+ private static final int MAX_CONNECTIONS = 100;
+ // the AWS SDK's HTTP defaults, as in S3FileIO; HttpClient's own socket timeout is unbounded
+ private static final long CONNECT_TIMEOUT_MS = 2_000;
+ private static final int SOCKET_TIMEOUT_MS = 30_000;
+
+ private static volatile CloseableHttpClient http;
+ private static volatile MetricsContext metrics;
+
+ private final String url;
+ private Long length;
+
+ private PreSignedUrlInputFile(String url, long length) {
+ this.url = url;
+ this.length = length > 0 ? length : null;
+ }
+
+ /** Whether {@code location} is an {@code http} or {@code https} URL. */
+ public static boolean isHttpUrl(String location) {
+ String lower = location.toLowerCase(Locale.ROOT);
+ return lower.startsWith("https://") || lower.startsWith("http://");
+ }
+
+ /**
+ * Returns an input file that reads {@code url} as given.
+ *
+ * @param url an {@code http} or {@code https} URL
+ * @param length the file length if known, otherwise {@code 0}
+ */
+ public static InputFile of(String url, long length) {
+ Preconditions.checkArgument(isHttpUrl(url), "Not an http or https URL: %s", url);
+ return new PreSignedUrlInputFile(url, length);
+ }
+
+ /**
+ * The length comes from the caller; {@code 0} means unknown, as in {@code S3InputFile}, and is
+ * then read from {@code Content-Range} on a single-byte range GET. HEAD is not an option, as the
+ * method is part of the signature and pre-signed URLs are signed for GET.
+ */
+ @Override
+ public long getLength() {
+ if (length == null) {
+ this.length = probeLength();
+ }
+
+ return length;
+ }
+
+ @Override
+ public SeekableInputStream newStream() {
+ return new PreSignedUrlInputStream(http(), url, metrics());
+ }
+
+ @Override
+ public String location() {
+ return url;
+ }
+
+ @Override
+ public boolean exists() {
+ try {
+ long probed = probeLength();
+ if (length == null) {
+ this.length = probed;
+ }
+
+ return true;
+ } catch (NotFoundException e) {
+ return false;
+ }
+ }
+
+ /** One single-byte range GET; the same request answers both the length and existence. */
+ private long probeLength() {
+ HttpGet get = new HttpGet(url);
+ get.setHeader("Range", "bytes=0-0");
+ try {
+ ClassicHttpResponse response = http().executeOpen(null, get, null);
+ int code = response.getCode();
+ if (code != HttpStatus.SC_PARTIAL_CONTENT
+ && code != HttpStatus.SC_REQUESTED_RANGE_NOT_SATISFIABLE) {
+ throw PreSignedUrlInputStream.failure("Length probe", url, response);
+ }
+
+ // Content-Range: bytes 0-0/<length>, or bytes */<length> for an empty object
+ Header contentRange = response.getFirstHeader("Content-Range");
+ PreSignedUrlInputStream.discard(response);
+ if (contentRange == null || !contentRange.getValue().contains("/")) {
+ throw new IOException("Length probe of " + url + " returned no Content-Range");
Review Comment:
this is in accordance with the current design assumptions, where a presigned url is treated like any other iceberg file location, and less like a credential.
--
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