Abacn commented on PR #39867:
URL: https://github.com/apache/beam/pull/39867#issuecomment-5413636734
Please check the following
1. Beam Batch Transforms Permit Updating / Triggered Side Inputs
- In Apache Beam’s unified model, running in batch mode
(`--streaming=false` or bounded inputs) does **not** prohibit non-default
windowing or triggering on side inputs. A bounded PCollection used as a side
input may specify early firings, element-count triggers, or repeated triggers
(e.g. `Repeatedly.forever(AfterPane.elementCountAtLeast(N))`).
- When multiple panes are produced for a window, downstream transforms
expect to receive updated view values as new panes arrive.
- While this change attempts to handle updates by calling
`SideInputCache.invalidate(...)` in `DoFnOperator.addSideInputValue`, this
invalidation is racy across parallel subtasks in the same TaskManager JVM:
- `SideInputCache` is a **`static` process-wide cache** shared across
all threads/slots in a TaskManager.
- When an updated side input element is broadcast, Subtask 0 might
process `processElement2` and invalidate the cache entry.
- Concurrently, Subtask 1 (on another thread) is processing main input
elements in `processElement1` and hasn't yet processed `processElement2`.
Subtask 1 calls `CachedSideInputReader.get()`, encounters a cache miss due to
Subtask 0's invalidation, evaluates its delegate
[`SideInputHandler`](/runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputHandler.java#L156-L178),
reads the **old** value from Subtask 1's local state, and re-inserts the
**stale** value back into the shared JVM cache.
- Subtask 0 (which already updated its state) will now read the stale
value from the cache.
2. Violation of PTransform Encapsulation (Leaky Public API)
- Beam pipelines are built by composing modular, reusable `PTransform`s
(from Beam core, connectors, or third-party libraries).
- An end user authoring a pipeline cannot and should not be expected to
know whether every internal DoFn inside every nested composite transform uses a
single-pane materialized view, an updating view, or mutates elements in the
materialized collection.
- Introducing `--cacheSideInputMaterialization` as a global
[`FlinkPipelineOptions`](/apache/beam/tree/release-2.76/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java)
flag forces users to know the implementation details of all transforms in
their pipeline. Enabling it to optimize one transform could silently break
semantic correctness in another.
- An opt-in flag does not resolve the issue: optimizations in Beam
runners must preserve Beam model semantics.
3. **Cleaner / Safer Alternatives**
- **Inspect Triggering Automatically**: The runner can inspect
`view.getWindowingStrategyInternal()`. If the side input uses `DefaultTrigger`
and bounded data, it is guaranteed to emit at most once per window, making
caching per `(view, window)` safe without needing any user-facing flag.
- **Scope the Cache Safely**: Instead of a process-wide `static` cache
across slots, scope the cache per-subtask instance, per-bundle, or
per-partition (similar to [Spark Structured Streaming's
`CachedSideInputReader`](/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/CachedSideInputReader.java#L45-L95)
and FnApi bundle cache tokens).
--
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]
