Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-5327576322 Hi @chamikaramj, thanks for taking the time to go through the PR and for all the detailed feedback. I acknowledge the concerns you raised around the filesystem abstraction, error handling, range reads, and performance at scale. I’ll go through each of these points and make the necessary changes. I’ll also run the 100GB performance test comparing the Beam FileSystem handler with the default implementation and share the results. Thanks again for the review and for pointing out these areas. I’ll work on the improvements ASAP. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
chamikaramj commented on code in PR #38873:
URL: https://github.com/apache/beam/pull/38873#discussion_r3789761978
##
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/BeamFileSystemClient.java:
##
@@ -0,0 +1,235 @@
+/*
+ * 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.beam.sdk.io.delta;
+
+import io.delta.kernel.engine.FileReadRequest;
+import io.delta.kernel.engine.FileSystemClient;
+import io.delta.kernel.utils.CloseableIterator;
+import io.delta.kernel.utils.FileStatus;
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
+import org.apache.beam.sdk.io.fs.MatchResult;
+import org.apache.beam.sdk.io.fs.MoveOptions;
+import org.apache.beam.sdk.io.fs.ResourceId;
+
+/** A Delta Kernel {@link FileSystemClient} backed by Beam's {@link
FileSystems}. */
+public class BeamFileSystemClient implements FileSystemClient {
+ @Override
+ public CloseableIterator listFrom(String path) throws
IOException {
Review Comment:
Also, please make sure that the Dataflow integration test suite passes for
this PR. You can do it by modifying following file as a part of your PR (just
set modifications to a different number).
File to modify:
https://github.com/apache/beam/blob/master/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json
Test suite runs:
https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml
##
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/BeamFileSystemClient.java:
##
@@ -0,0 +1,235 @@
+/*
+ * 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.beam.sdk.io.delta;
+
+import io.delta.kernel.engine.FileReadRequest;
+import io.delta.kernel.engine.FileSystemClient;
+import io.delta.kernel.utils.CloseableIterator;
+import io.delta.kernel.utils.FileStatus;
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
+import org.apache.beam.sdk.io.fs.MatchResult;
+import org.apache.beam.sdk.io.fs.MoveOptions;
+import org.apache.beam.sdk.io.fs.ResourceId;
+
+/** A Delta Kernel {@link FileSystemClient} backed by Beam's {@link
FileSystems}. */
+public class BeamFil
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-5036967980 Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-5036954769 R: @Abacn Hi, could you please take a look at my PR when you have a chance? Thanks! -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-5036941597 Reviewers are already assigned to this PR: @Abacn -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-5036929508 assign set of reviewers -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-5034230023 assign to next reviewer -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-5034134406 Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment `assign to next reviewer`: R: @Abacn for label java. Available commands: - `stop reviewer notifications` - opt out of the automated review tooling - `remind me after tests pass` - tag the comment author after tests pass - `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers) -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4980666084 Reminder, please take a look at this pr: @kennknowles @chamikaramj @ahmedabu98 -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4904075602 Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment `assign to next reviewer`: R: @kennknowles for label java. Available commands: - `stop reviewer notifications` - opt out of the automated review tooling - `remind me after tests pass` - tag the comment author after tests pass - `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers) -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4876485762 Reminder, please take a look at this pr: @Abacn @chamikaramj @ahmedabu98 -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4801805038 Assigning reviewers: R: @Abacn for label java. Note: If you would like to opt out of this review, comment `assign to next reviewer`. Available commands: - `stop reviewer notifications` - opt out of the automated review tooling - `remind me after tests pass` - tag the comment author after tests pass - `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers) The PR bot will only process comments in the main thread (not review comments). -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4801789424 assign set of reviewers -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4801785754 assign to next reviewer -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4768197404 Hi guys! Could you please take a look at my PR when you have a chance? Thanks! R: @ahmedabu98 R: @chamikaramj -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4720090983 No problem, thank you so much! -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
kennknowles commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4719985903 Hi @joaohmalves ! Sorry, I have been pretty swamped. I think @chamikaramj is perhaps the best reviewer since he designed https://s.apache.org/beam-delta-lake-source but I have also pinged @ahmedabu98 who is probably also a good choice. Either one will be better than me for this one. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
kennknowles commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4719987675 R: @ahmedabu98 R: @chamikaramj -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4719814190 Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4719799272 R: @kennknowles Hi, could you please take a look at my PR when you have a chance? Thanks! -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4711075035 assign to next reviewer -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4710569525 Reviewers are already assigned to this PR: @kennknowles -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4710557977 assign set of reviewers -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
gemini-code-assist[bot] commented on code in PR #38873:
URL: https://github.com/apache/beam/pull/38873#discussion_r3396122898
##
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/BeamFileSystemClient.java:
##
@@ -0,0 +1,234 @@
+/*
+ * 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.beam.sdk.io.delta;
+
+import io.delta.kernel.engine.FileReadRequest;
+import io.delta.kernel.engine.FileSystemClient;
+import io.delta.kernel.utils.CloseableIterator;
+import io.delta.kernel.utils.FileStatus;
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
+import org.apache.beam.sdk.io.fs.MatchResult;
+import org.apache.beam.sdk.io.fs.MoveOptions;
+import org.apache.beam.sdk.io.fs.ResourceId;
+
+/** A Delta Kernel {@link FileSystemClient} backed by Beam's {@link
FileSystems}. */
+public class BeamFileSystemClient implements FileSystemClient {
+ @Override
+ public CloseableIterator listFrom(String path) throws
IOException {
+String glob = globForSiblings(path);
+List statuses = new ArrayList<>();
+String normalizedInput = FileSystems.matchNewResource(path,
false).toString();
+for (MatchResult.Metadata metadata :
+ FileSystems.match(glob, EmptyMatchTreatment.ALLOW).metadata()) {
+ if (metadata.resourceId().isDirectory()) {
+continue;
+ }
+ String metadataPath = metadata.resourceId().toString();
+ if
(normalizeForOrdering(metadataPath).compareTo(normalizeForOrdering(normalizedInput))
>= 0) {
+statuses.add(toDeltaFileStatus(metadata));
+ }
+}
+statuses.sort(
+(first, second) ->
+normalizeForOrdering(first.getPath())
+.compareTo(normalizeForOrdering(second.getPath(;
+return closeableIterator(statuses.iterator());
+ }
+
+ @Override
+ public String resolvePath(String path) throws IOException {
+try {
+ return getFileStatus(path).getPath();
+} catch (IOException e) {
+ return FileSystems.matchNewResource(path, false).toString();
+}
+ }
+
+ @Override
+ public CloseableIterator readFiles(
+ CloseableIterator readRequests) {
+return new CloseableIterator() {
+ @Override
+ public boolean hasNext() {
+return readRequests.hasNext();
+ }
+
+ @Override
+ public ByteArrayInputStream next() {
+FileReadRequest request = readRequests.next();
+try {
+ return readRange(request.getPath(), request.getStartOffset(),
request.getReadLength());
+} catch (IOException e) {
+ throw new UncheckedIOException(
+ String.format(
+ "IOException reading from file %s at offset %s size %s",
+ request.getPath(), request.getStartOffset(),
request.getReadLength()),
+ e);
+}
+ }
+
+ @Override
+ public void close() throws IOException {
+readRequests.close();
+ }
+};
+ }
+
+ @Override
+ public boolean mkdirs(String path) throws IOException {
+if (isLocalPath(path)) {
+ Files.createDirectories(toLocalPath(path));
+}
+return true;
+ }
+
+ @Override
+ public boolean delete(String path) throws IOException {
+FileSystems.delete(
+Collections.singletonList(FileSystems.matchNewResource(path, false)),
+MoveOptions.StandardMoveOptions.IGNORE_MISSING_FILES);
+return true;
+ }
+
+ @Override
+ public FileStatus getFileStatus(String path) throws IOException {
+return toDeltaFileStatus(FileSystems.matchSingleFileSpec(path));
+ }
+
+ @Override
+ publi
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
joaohmalves commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4680896480 /gemini review -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4672344120 Assigning reviewers: R: @kennknowles for label java. Note: If you would like to opt out of this review, comment `assign to next reviewer`. Available commands: - `stop reviewer notifications` - opt out of the automated review tooling - `remind me after tests pass` - tag the comment author after tests pass - `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers) The PR bot will only process comments in the main thread (not review comments). -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
github-actions[bot] commented on PR #38873: URL: https://github.com/apache/beam/pull/38873#issuecomment-4664073347 Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Implement BeamFileSystemClient for Delta Lake connector [beam]
gemini-code-assist[bot] commented on code in PR #38873:
URL: https://github.com/apache/beam/pull/38873#discussion_r3382722269
##
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/BeamFileSystemClient.java:
##
@@ -0,0 +1,236 @@
+/*
+ * 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.beam.sdk.io.delta;
+
+import io.delta.kernel.engine.FileReadRequest;
+import io.delta.kernel.engine.FileSystemClient;
+import io.delta.kernel.utils.CloseableIterator;
+import io.delta.kernel.utils.FileStatus;
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
+import org.apache.beam.sdk.io.fs.MatchResult;
+import org.apache.beam.sdk.io.fs.MoveOptions;
+import org.apache.beam.sdk.io.fs.ResourceId;
+
+/** A Delta Kernel {@link FileSystemClient} backed by Beam's {@link
FileSystems}. */
+public class BeamFileSystemClient implements FileSystemClient {
+ @Override
+ public CloseableIterator listFrom(String path) throws
IOException {
+String glob = globForSiblings(path);
+List statuses = new ArrayList<>();
+String normalizedInput = FileSystems.matchNewResource(path,
false).toString();
+
+for (MatchResult.Metadata metadata :
+FileSystems.match(glob, EmptyMatchTreatment.ALLOW).metadata()) {
+ String metadataPath = metadata.resourceId().toString();
+ if
(normalizeForOrdering(metadataPath).compareTo(normalizeForOrdering(path)) >= 0)
{
+statuses.add(toDeltaFileStatus(metadata));
+ }
+}
Review Comment:
The `normalizedInput` variable is defined on line 55 but never used.
Instead, the raw `path` parameter is used in the comparison on line 60. This
can lead to incorrect filtering in `listFrom` if the input path and the matched
metadata paths have different normalization formats (e.g., `file:/` vs
`file:///`).
Additionally, `listFrom` should filter out directories to prevent downstream
components from attempting to read directories as files. We can achieve this by
checking `!metadata.resourceId().isDirectory()`.
```java
for (MatchResult.Metadata metadata :
FileSystems.match(glob, EmptyMatchTreatment.ALLOW).metadata()) {
if (metadata.resourceId().isDirectory()) {
continue;
}
String metadataPath = metadata.resourceId().toString();
if
(normalizeForOrdering(metadataPath).compareTo(normalizeForOrdering(normalizedInput))
>= 0) {
statuses.add(toDeltaFileStatus(metadata));
}
}
```
##
sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/BeamFileSystemClient.java:
##
@@ -0,0 +1,236 @@
+/*
+ * 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.beam.sdk.io.delta;
