chamikaramj commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3873571494
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java:
##########
@@ -398,18 +366,16 @@ private void outputResult(ProcessResult result,
MultiOutputReceiver output) {
@FinishBundle
public void finishBundle(FinishBundleContext context) throws Exception {
- // Block and wait for threads to finish their work
- int numErrors = 0;
- for (Future<ProcessResult> future : checkStateNotNull(activeTasks)) {
- ProcessResult result = future.get();
- if (result.errorRow != null) {
- context.output(ERRORS, result.errorRow, result.timestamp,
result.window);
- numErrors++;
- } else if (result.dataFile != null) {
- context.output(DATA_FILES, result.dataFile, result.timestamp,
result.window);
- }
+ checkStateNotNull(tasks).awaitAll(result -> outputAtFinish(result,
context));
+ }
+
+ private static void outputAtFinish(ProcessResult result,
FinishBundleContext context) {
+ if (result.errorRow != null) {
+ context.output(ERRORS, result.errorRow, result.timestamp,
result.window);
Review Comment:
Seems like we are ignoring the possibility of having some successful results
as well as errors or is that somehow not a concern here ?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/BoundedAsyncTasks.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.Consumer;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * Runs tasks on a fixed thread pool while bounding how many are in flight.
Results are handed to
+ * {@code onDone} on the caller's thread (from {@link #submit} and {@link
#awaitAll}), so a DoFn can
+ * emit them without a thread-safe output.
+ */
+class BoundedAsyncTasks<T> {
+ private final ExecutorService executor;
+ private final int maxInFlight;
+ private final Deque<Future<T>> active = new ArrayDeque<>();
+
+ BoundedAsyncTasks(int threads, int maxInFlight) {
+ Preconditions.checkArgument(threads > 0, "threads must be positive, got:
%s", threads);
+ Preconditions.checkArgument(
+ maxInFlight > 0, "maxInFlight must be positive, got: %s", maxInFlight);
+ this.executor =
+ Executors.newFixedThreadPool(
+ threads,
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("iceberg-async-task-%d")
+ .build());
+ this.maxInFlight = maxInFlight;
+ }
+
+ /**
+ * Submits a task, first delivering any finished results. Blocks while
{@code maxInFlight} tasks
+ * are outstanding. If a task failed, its exception is rethrown and every
other outstanding task
+ * is cancelled.
+ */
+ void submit(Callable<T> task, Consumer<T> onDone) throws Exception {
+ try {
+ drainFinished(onDone);
+ while (active.size() >= maxInFlight) {
+ Future<T> oldest = active.removeFirst();
+ onDone.accept(oldest.get()); // blocks until the oldest task completes
Review Comment:
Might be more efficient to wait for the completed tasks using
`ExecutorCompletionService ` instead of waiting for the oldest.
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java:
##########
@@ -807,7 +807,7 @@ public void testGetPartitionFromMetrics() throws
IOException, InterruptedExcepti
writer.close();
InputFile file = table.io().newInputFile(fileName);
- ParquetMetadata footer = AddFiles.readParquetFooter(fileName);
+ ParquetMetadata footer = ParquetFooters.read(fileName);
Review Comment:
Probably also add a seperate `ParquetFooters` test to test various scenarios
related to reading valid/invalid files ?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ParquetFooters.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.iceberg;
+
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+
+import java.io.IOException;
+import java.nio.channels.SeekableByteChannel;
+import org.apache.beam.sdk.io.Compression;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.ResourceId;
+import org.apache.beam.sdk.io.parquet.ParquetIO.ReadFiles.BeamParquetInputFile;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.io.InputFile;
+
+/** Reads Parquet footers through Beam's {@link FileSystems}, so no table or
FileIO is needed. */
+final class ParquetFooters {
+ private ParquetFooters() {}
+
+ static ParquetMetadata read(String filePath) throws IOException {
+ try (ParquetFileReader reader =
ParquetFileReader.open(inputFile(filePath))) {
+ return reader.getFooter();
Review Comment:
I assume this is able to efficiently read the footer without reading the
whole file ?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java:
##########
@@ -449,7 +415,7 @@ private Callable<ProcessResult> createProcessTask(
@Nullable ParquetMetadata parquetFooter = null;
if (format.equals(FileFormat.PARQUET)) {
try {
- parquetFooter = readParquetFooter(filePath);
+ parquetFooter = ParquetFooters.read(filePath);
Review Comment:
Reading footers is also parallelized across bundles, right ?
If not, I think reading Parquet footers from all files can be too expensive
for a single VM when we scale (for example, O(millions) of files).
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java:
##########
@@ -329,50 +314,33 @@ private static class ProcessResult {
@Setup
public void setup() {
- executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
+ tasks = new BoundedAsyncTasks<>(THREAD_POOL_SIZE, MAX_IN_FLIGHT_TASKS);
Review Comment:
For my knowledge, why did we introduce a new class instead of just sharing
the bounded executor ? Do we expect the pool size and max in flight tasks to be
different for future use-cases ?
If so shall we make this a more generic util instead of something specific
to Iceberg ? (can be a future update)
--
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]
