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<FileStatus> 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 BeamFileSystemClient implements FileSystemClient { + @Override + public CloseableIterator<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) + throws IOException { + ResourceId resourceId = FileSystems.matchNewResource(path, false); + try (ReadableByteChannel channel = FileSystems.open(resourceId)) { + byte[] data = new byte[readLength]; + if (channel instanceof SeekableByteChannel) { + ((SeekableByteChannel) channel).position(startOffset); + readFully(channel, ByteBuffer.wrap(data)); + } else { + try (InputStream stream = Channels.newInputStream(channel)) { + com.google.common.io.ByteStreams.skipFully(stream, startOffset); + com.google.common.io.ByteStreams.readFully(stream, data); + } + } + return new ByteArrayInputStream(data); + } + } + + private static void readFully(ReadableByteChannel channel, ByteBuffer buffer) throws IOException { + while (buffer.hasRemaining()) { + if (channel.read(buffer) < 0) { + throw new EOFException("Unexpected end of file"); + } + } + } + + private static FileStatus toDeltaFileStatus(MatchResult.Metadata metadata) { + return FileStatus.of( + metadata.resourceId().toString(), metadata.sizeBytes(), metadata.lastModifiedMillis()); + } + + private static String globForSiblings(String path) { + String normalized = path.replace('\\', '/'); + int lastSlash = normalized.lastIndexOf('/'); + if (lastSlash < 0) { + return "*"; + } + return normalized.substring(0, lastSlash + 1) + "*"; + } + + private static String normalizeForOrdering(String path) { + return path.replace('\\', '/'); + } + + private static boolean isLocalPath(String path) { + int schemeSeparator = path.indexOf(':'); Review Comment: Ditto. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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) { Review Comment: Catching a generic IOException here could mask real, transient filesystem issues (such as SocketTimeoutException, permission errors, or network partitions). Since getFileStatus relies on FileSystems.matchSingleFileSpec, you should catch FileNotFoundException specifically to handle the case where the file doesn't exist, and let other unexpected exceptions propagate. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); Review Comment: Please add a comment on why this was needed. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) + throws IOException { + ResourceId resourceId = FileSystems.matchNewResource(path, false); + try (ReadableByteChannel channel = FileSystems.open(resourceId)) { + byte[] data = new byte[readLength]; + if (channel instanceof SeekableByteChannel) { + ((SeekableByteChannel) channel).position(startOffset); + readFully(channel, ByteBuffer.wrap(data)); + } else { + try (InputStream stream = Channels.newInputStream(channel)) { + com.google.common.io.ByteStreams.skipFully(stream, startOffset); + com.google.common.io.ByteStreams.readFully(stream, data); + } + } + return new ByteArrayInputStream(data); + } + } + + private static void readFully(ReadableByteChannel channel, ByteBuffer buffer) throws IOException { + while (buffer.hasRemaining()) { + if (channel.read(buffer) < 0) { + throw new EOFException("Unexpected end of file"); + } + } + } + + private static FileStatus toDeltaFileStatus(MatchResult.Metadata metadata) { + return FileStatus.of( + metadata.resourceId().toString(), metadata.sizeBytes(), metadata.lastModifiedMillis()); + } + + private static String globForSiblings(String path) { + String normalized = path.replace('\\', '/'); + int lastSlash = normalized.lastIndexOf('/'); + if (lastSlash < 0) { + return "*"; + } + return normalized.substring(0, lastSlash + 1) + "*"; + } + + private static String normalizeForOrdering(String path) { + return path.replace('\\', '/'); + } + + private static boolean isLocalPath(String path) { + int schemeSeparator = path.indexOf(':'); + if (schemeSeparator < 0) { + return true; + } + String scheme = path.substring(0, schemeSeparator).toLowerCase(Locale.ROOT); + return scheme.length() == 1 || "file".equals(scheme); + } + + private static Path toLocalPath(String path) { + if (path.toLowerCase(Locale.ROOT).startsWith("file:")) { Review Comment: Ditto regarding paths. ########## 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 { Review Comment: Could you do some performance analysis to compare the BeamFileSystem handler vs the default one when reading from a large dataset (for example, 100GB) and publish results. You can use the DeltaIOIT integration test as a template (but run for a larger dataset). https://github.com/apache/beam/blob/master/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java 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,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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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: Please address this instead of resolving. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) + throws IOException { + ResourceId resourceId = FileSystems.matchNewResource(path, false); + try (ReadableByteChannel channel = FileSystems.open(resourceId)) { + byte[] data = new byte[readLength]; + if (channel instanceof SeekableByteChannel) { + ((SeekableByteChannel) channel).position(startOffset); + readFully(channel, ByteBuffer.wrap(data)); + } else { + try (InputStream stream = Channels.newInputStream(channel)) { + com.google.common.io.ByteStreams.skipFully(stream, startOffset); + com.google.common.io.ByteStreams.readFully(stream, data); + } + } + return new ByteArrayInputStream(data); + } + } + + private static void readFully(ReadableByteChannel channel, ByteBuffer buffer) throws IOException { + while (buffer.hasRemaining()) { + if (channel.read(buffer) < 0) { + throw new EOFException("Unexpected end of file"); + } + } + } + + private static FileStatus toDeltaFileStatus(MatchResult.Metadata metadata) { + return FileStatus.of( + metadata.resourceId().toString(), metadata.sizeBytes(), metadata.lastModifiedMillis()); + } + + private static String globForSiblings(String path) { + String normalized = path.replace('\\', '/'); + int lastSlash = normalized.lastIndexOf('/'); + if (lastSlash < 0) { + return "*"; + } + return normalized.substring(0, lastSlash + 1) + "*"; + } + + private static String normalizeForOrdering(String path) { + return path.replace('\\', '/'); Review Comment: Ditto. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + if (overwrite) { + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } else { + FileSystems.copy( + Collections.singletonList(srcResource), + Collections.singletonList(dstResource), + MoveOptions.StandardMoveOptions.SKIP_IF_DESTINATION_EXISTS); + } + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) + throws IOException { + ResourceId resourceId = FileSystems.matchNewResource(path, false); + try (ReadableByteChannel channel = FileSystems.open(resourceId)) { + byte[] data = new byte[readLength]; + if (channel instanceof SeekableByteChannel) { + ((SeekableByteChannel) channel).position(startOffset); + readFully(channel, ByteBuffer.wrap(data)); + } else { + try (InputStream stream = Channels.newInputStream(channel)) { + com.google.common.io.ByteStreams.skipFully(stream, startOffset); + com.google.common.io.ByteStreams.readFully(stream, data); + } + } + return new ByteArrayInputStream(data); + } + } + + private static void readFully(ReadableByteChannel channel, ByteBuffer buffer) throws IOException { + while (buffer.hasRemaining()) { + if (channel.read(buffer) < 0) { + throw new EOFException("Unexpected end of file"); + } + } + } + + private static FileStatus toDeltaFileStatus(MatchResult.Metadata metadata) { + return FileStatus.of( + metadata.resourceId().toString(), metadata.sizeBytes(), metadata.lastModifiedMillis()); + } + + private static String globForSiblings(String path) { + String normalized = path.replace('\\', '/'); + int lastSlash = normalized.lastIndexOf('/'); + if (lastSlash < 0) { + return "*"; + } + return path.substring(0, lastSlash + 1) + "*"; + } Review Comment: Please address this instead of resolving. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) Review Comment: Two concerns here: Memory: Allocating a byte[] based on readLength can cause OutOfMemoryErrors if the Delta engine requests a large chunk (e.g., parsing a large Parquet row group). Are we guaranteed by the Delta Kernel that readLength will remain relatively small? Performance: For non-seekable channels, ByteStreams.skipFully reads and discards bytes from the beginning of the file. On cloud object stores, this means pulling startOffset bytes over the network just to throw them away. This can be prohibitively expensive for large files. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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)); + } + } + 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + if (overwrite) { + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } else { + FileSystems.copy( + Collections.singletonList(srcResource), + Collections.singletonList(dstResource), + MoveOptions.StandardMoveOptions.SKIP_IF_DESTINATION_EXISTS); + } + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) + throws IOException { + ResourceId resourceId = FileSystems.matchNewResource(path, false); + try (ReadableByteChannel channel = FileSystems.open(resourceId)) { + byte[] data = new byte[readLength]; + if (channel instanceof SeekableByteChannel) { + ((SeekableByteChannel) channel).position(startOffset); + readFully(channel, ByteBuffer.wrap(data)); + } else { + try (InputStream stream = Channels.newInputStream(channel)) { + stream.skipNBytes(startOffset); + int read = stream.readNBytes(data, 0, readLength); + if (read != readLength) { + throw new EOFException( + String.format("Expected %s bytes from %s but read %s", readLength, path, read)); + } + } + } Review Comment: Please address instead of resolving. ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) + throws IOException { + ResourceId resourceId = FileSystems.matchNewResource(path, false); + try (ReadableByteChannel channel = FileSystems.open(resourceId)) { + byte[] data = new byte[readLength]; + if (channel instanceof SeekableByteChannel) { + ((SeekableByteChannel) channel).position(startOffset); + readFully(channel, ByteBuffer.wrap(data)); + } else { + try (InputStream stream = Channels.newInputStream(channel)) { + com.google.common.io.ByteStreams.skipFully(stream, startOffset); + com.google.common.io.ByteStreams.readFully(stream, data); + } + } + return new ByteArrayInputStream(data); + } + } + + private static void readFully(ReadableByteChannel channel, ByteBuffer buffer) throws IOException { + while (buffer.hasRemaining()) { + if (channel.read(buffer) < 0) { + throw new EOFException("Unexpected end of file"); + } + } + } + + private static FileStatus toDeltaFileStatus(MatchResult.Metadata metadata) { + return FileStatus.of( + metadata.resourceId().toString(), metadata.sizeBytes(), metadata.lastModifiedMillis()); + } + + private static String globForSiblings(String path) { Review Comment: Please address and make sure you use standard FileSystem methods (FileSystems.java), ResourceId etc. instead of manually parsing paths since this is expected to be used by all file systems supported by Beam (local, GCS, S3, Azure Blob etc.) ########## 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<FileStatus> listFrom(String path) throws IOException { + String glob = globForSiblings(path); + List<FileStatus> 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<ByteArrayInputStream> readFiles( + CloseableIterator<FileReadRequest> readRequests) { + return new CloseableIterator<ByteArrayInputStream>() { + @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 + public void copyFileAtomically(String src, String dst, boolean overwrite) throws IOException { + ResourceId srcResource = FileSystems.matchNewResource(src, false); + ResourceId dstResource = FileSystems.matchNewResource(dst, false); + MatchResult dstMatch = FileSystems.match(dst, EmptyMatchTreatment.ALLOW); + if (!overwrite + && dstMatch.status() == MatchResult.Status.OK + && !dstMatch.metadata().isEmpty()) { + throw new IOException("Destination already exists: " + dst); + } + + FileSystems.copy( + Collections.singletonList(srcResource), Collections.singletonList(dstResource)); + } + + private static ByteArrayInputStream readRange(String path, int startOffset, int readLength) + throws IOException { + ResourceId resourceId = FileSystems.matchNewResource(path, false); + try (ReadableByteChannel channel = FileSystems.open(resourceId)) { + byte[] data = new byte[readLength]; + if (channel instanceof SeekableByteChannel) { + ((SeekableByteChannel) channel).position(startOffset); + readFully(channel, ByteBuffer.wrap(data)); + } else { + try (InputStream stream = Channels.newInputStream(channel)) { + com.google.common.io.ByteStreams.skipFully(stream, startOffset); + com.google.common.io.ByteStreams.readFully(stream, data); + } + } + return new ByteArrayInputStream(data); + } + } + + private static void readFully(ReadableByteChannel channel, ByteBuffer buffer) throws IOException { + while (buffer.hasRemaining()) { + if (channel.read(buffer) < 0) { + throw new EOFException("Unexpected end of file"); + } + } + } + + private static FileStatus toDeltaFileStatus(MatchResult.Metadata metadata) { + return FileStatus.of( + metadata.resourceId().toString(), metadata.sizeBytes(), metadata.lastModifiedMillis()); + } + + private static String globForSiblings(String path) { + String normalized = path.replace('\\', '/'); + int lastSlash = normalized.lastIndexOf('/'); + if (lastSlash < 0) { + return "*"; + } + return normalized.substring(0, lastSlash + 1) + "*"; + } + + private static String normalizeForOrdering(String path) { + return path.replace('\\', '/'); + } + + private static boolean isLocalPath(String path) { + int schemeSeparator = path.indexOf(':'); + if (schemeSeparator < 0) { + return true; + } + String scheme = path.substring(0, schemeSeparator).toLowerCase(Locale.ROOT); + return scheme.length() == 1 || "file".equals(scheme); + } + + private static Path toLocalPath(String path) { + if (path.toLowerCase(Locale.ROOT).startsWith("file:")) { + try { + return Paths.get(new URI(path)); Review Comment: We shouldn't reinvent URI scheme parsing. You can determine the scheme cleanly via FileSystems.matchNewResource(path, false).getScheme(). Also, in mkdirs(), is there a reason we only create directories for local paths? hdfs:// is not local but still requires directory creation. Note that Beam's FileSystems generally handles parent directory creation automatically when creating files, so we might want to reconsider if explicit mkdirs handling is even needed for Beam, or if returning true unconditionally is safer. -- 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]
