You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "claudevdm (via GitHub)" <gi...@apache.org> on 2026/08/26 18:33:34 UTC
[PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
claudevdm opened a new pull request, #39896:
URL: https://github.com/apache/beam/pull/39896
ConvertToDataFile ran its per-file work (footer read, metrics, partition
resolution) on a private thread pool.
This change moves that queue into BoundedAsyncTasks and the footer read
into ParquetFooters so the upcoming schema pre-pass (a second DoFn that
reads every Parquet footer) can share both instead of copying them.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3873913155
##########
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:
Yes. Each DoFn instance runs up to 10 footer reads concurrently (THREAD_POOL_SIZE) and allows up to 100 files in flight (executing or queued) before process() blocks (MAX_IN_FLIGHT_TASKS). Each instance handles its own bundle of file paths.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3914384157
##########
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);
+ }
+
+ /** Clears anything left behind if the runner reuses this instance after a failed bundle. */
+ @StartBundle
+ public void startBundle() {
+
+ checkStateNotNull(tasks).cancelAll();
}
@Teardown
public void teardown() {
- if (executor != null) {
- executor.shutdownNow();
+ if (tasks != null) {
+ tasks.shutdown();
}
}
- @StartBundle
- public void startBundle() {
- activeTasks = Lists.newLinkedList();
- }
-
@ProcessElement
public void process(
@Element String filePath,
@Timestamp Instant timestamp,
BoundedWindow window,
PaneInfo paneInfo,
MultiOutputReceiver output)
- throws IOException, InterruptedException, ExecutionException {
- LinkedList<Future<ProcessResult>> activeTasks = checkStateNotNull(this.activeTasks);
-
- // start draining finished tasks, but don't block
- Iterator<Future<ProcessResult>> iterator = activeTasks.iterator();
- while (iterator.hasNext()) {
- Future<ProcessResult> future = iterator.next();
- if (future.isDone()) {
- outputResult(future.get(), output);
- iterator.remove();
- }
- }
-
- // if we have too many active tasks, wait until some finish
- while (activeTasks.size() >= MAX_IN_FLIGHT_TASKS) {
- Future<ProcessResult> oldestTask = activeTasks.removeFirst();
- outputResult(oldestTask.get(), output); // .get() blocks until the task completes
- }
-
- // create a new task for the current element and add to queue
+ throws Exception {
Callable<ProcessResult> task = createProcessTask(filePath, timestamp, window, paneInfo);
- activeTasks.add(checkStateNotNull(executor).submit(task));
+ checkStateNotNull(tasks).submit(task, result -> outputResult(result, output));
Review Comment:
@claudevdm you shouldn't output async here. OutputReceiver and ProcessContext are not thread safe.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3874055239
##########
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:
This is invoked once per file not per budnle. Each ProcessResult represents one datafile/ error row. Success is emitted to DATA_FILES, failure to ERRORS
https://github.com/claudevdm/beam/blob/f7edad1f06bc04dcd11b1eaee6dad3618b861048/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java#L346-L365
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3914936693
##########
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);
+ }
+
+ /** Clears anything left behind if the runner reuses this instance after a failed bundle. */
+ @StartBundle
+ public void startBundle() {
+
+ checkStateNotNull(tasks).cancelAll();
}
@Teardown
public void teardown() {
- if (executor != null) {
- executor.shutdownNow();
+ if (tasks != null) {
+ tasks.shutdown();
}
}
- @StartBundle
- public void startBundle() {
- activeTasks = Lists.newLinkedList();
- }
-
@ProcessElement
public void process(
@Element String filePath,
@Timestamp Instant timestamp,
BoundedWindow window,
PaneInfo paneInfo,
MultiOutputReceiver output)
- throws IOException, InterruptedException, ExecutionException {
- LinkedList<Future<ProcessResult>> activeTasks = checkStateNotNull(this.activeTasks);
-
- // start draining finished tasks, but don't block
- Iterator<Future<ProcessResult>> iterator = activeTasks.iterator();
- while (iterator.hasNext()) {
- Future<ProcessResult> future = iterator.next();
- if (future.isDone()) {
- outputResult(future.get(), output);
- iterator.remove();
- }
- }
-
- // if we have too many active tasks, wait until some finish
- while (activeTasks.size() >= MAX_IN_FLIGHT_TASKS) {
- Future<ProcessResult> oldestTask = activeTasks.removeFirst();
- outputResult(oldestTask.get(), output); // .get() blocks until the task completes
- }
-
- // create a new task for the current element and add to queue
+ throws Exception {
Callable<ProcessResult> task = createProcessTask(filePath, timestamp, window, paneInfo);
- activeTasks.add(checkStateNotNull(executor).submit(task));
+ checkStateNotNull(tasks).submit(task, result -> outputResult(result, output));
Review Comment:
The consumer is invoked synchronously inside submit() and awaitAll() on the calling thread
https://github.com/claudevdm/beam/blob/f7edad1f06bc04dcd11b1eaee6dad3618b861048/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/BoundedAsyncTasks.java#L62
https://github.com/claudevdm/beam/blob/f7edad1f06bc04dcd11b1eaee6dad3618b861048/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/BoundedAsyncTasks.java#L104-L110
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3874012934
##########
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:
I'd like to avoid making too many changes in this PR to how it works. Will consider this as a follow up.
Right now the all completed futures already drain before blocking, so this would only be an issue if there is tail latency on the oldest.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm merged PR #39896:
URL: https://github.com/apache/beam/pull/39896
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3873918389
##########
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:
Yes.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3873862963
##########
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:
The logic for back-pressure (process block when 100 tasks outstanding, handing back results on main thread, bookkeeping) was inline in the ConvertToDatafile DoFn. I extracted it to make it testable in isolation and reusable by other DoFn than ConvertToDatafile (it will also be needed in the schema prepass when reading footers).
Lets consider moving it to sdk/util in a follow up, I think that would require a refactor and some more thought.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: extract bounded async task plumbing and Parquet footer reads [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on code in PR #39896:
URL: https://github.com/apache/beam/pull/39896#discussion_r3914386821
##########
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));
Review Comment:
same as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org