KKcorps opened a new pull request, #19115:
URL: https://github.com/apache/pinot/pull/19115
## TL;DR
A partial upsert table with `postPartialUpsertTransformConfigs` can write
wrong values into its
derived columns. `PartialUpsertHandler` was built once per table and shared
by every partition, but
it holds record transformers that keep reusable evaluation state, and each
partition merges on its
own consumer thread. This PR moves handler creation down to the partition,
so each partition owns
its own handler.
## The problem
`BaseTableUpsertMetadataManager.init()` built **one** `PartialUpsertHandler`
and put it in the
shared `UpsertContext`. Every `BasePartitionUpsertMetadataManager` read that
same instance. Each
partition ingests on its own consumer thread, so N partitions on a server
meant N threads calling
`PartialUpsertHandler.merge()` at the same time.
`merge()` ends by running the post partial upsert transform chain. Those
transformers are not thread
safe. `ExpressionTransformer` holds `FunctionEvaluator` instances, and
`InbuiltFunctionEvaluator`'s
`FunctionExecutionNode` keeps a reusable `Object[] _arguments` array that it
refills on every
`execute()` call:
```java
for (int i = 0; i < _argumentNodes.length; i++) {
_arguments[i] = _argumentNodes[i].execute(row); // shared array,
refilled per row
}
...
return _functionInvoker.invoke(_arguments);
```
Two threads writing into that one array interleave. Thread A can fill in its
argument, get
descheduled, and then invoke the function after thread B has overwritten the
same slot. The derived
column for partition 0's row is then computed from partition 1's values.
Nothing throws. The row is
just wrong, and it gets persisted.
```mermaid
sequenceDiagram
participant P0 as Partition 0 consumer
participant P1 as Partition 1 consumer
participant N as FunctionExecutionNode<br/>(one shared _arguments[])
P0->>N: _arguments[0] = 100 (row A)
P1->>N: _arguments[0] = 999 (row B)
P0->>N: invoke(_arguments)
Note over P0,N: ❌ row A's derived column computed from 999
P1->>N: invoke(_arguments)
```
The root cause is ownership granularity. The handler was owned by the table,
but the mutable state
inside it can only be owned by one merging thread.
## The approach
Give each partition its own handler, which is the level the merge path
already works at.
1. `UpsertContext` now carries a `Supplier<PartialUpsertHandler>` instead of
a built instance.
`BaseTableUpsertMetadataManager` sets the supplier when the mode is
`PARTIAL`.
2. Each `BasePartitionUpsertMetadataManager` calls the supplier once in its
constructor and keeps
the result. One handler per partition, built when the partition manager
is built.
3. `getUpsertMode()` and `isTableTypeInconsistentDuringConsumption()` now
branch on whether the
supplier is set, which is exactly the signal the old handler-nullness
check gave them.
```mermaid
flowchart LR
subgraph Before["❌ Before"]
T[table manager] --> H[one handler]
H --> P0[Partition 0 thread]
H --> P1[Partition 1 thread]
end
subgraph After["✅ After"]
T2[table manager] --> S[handler supplier]
S --> H0[handler for P0] --> P2[Partition 0 thread]
S --> H1[handler for P1] --> P3[Partition 1 thread]
end
```
### Why per-partition is enough
`merge()` is only reachable from `doUpdateRecord()`, which already uses
per-partition unsynchronized
scratch state right next to the merge call:
```java
_reusePreviousRow.init(currentSegment, currentDocId);
_partialUpsertHandler.merge(_reusePreviousRow, record,
_reuseMergeResultHolder);
```
`_reusePreviousRow` and `_reuseMergeResultHolder` are plain fields on the
partition manager. Partial
upsert already depends on at most one thread running `doUpdateRecord()` per
partition at a time.
Consumption within a partition is serialized across segments, since a
segment stops consuming before
the next one starts. So putting the handler at the partition level relies on
an invariant the code
already relies on, rather than adding a new one. I have documented that
invariant on the classes
involved.
A `ThreadLocal` inside `PartialUpsertHandler` would also fix the race, but
it papers over the
ownership problem instead of fixing it, and leaves per-thread chains alive
for as long as the thread
lives. Building the handler where its mutable state belongs is the smaller
idea.
## Changes
| File | Change |
|---|---|
| `UpsertContext.java` | Carries `Supplier<PartialUpsertHandler>` instead of
a shared instance; `getPartialUpsertHandler()` →
`getPartialUpsertHandlerSupplier()`, `setPartialUpsertHandler()` →
`setPartialUpsertHandlerSupplier()` |
| `BaseTableUpsertMetadataManager.java` | Builds a supplier instead of one
handler |
| `BasePartitionUpsertMetadataManager.java` | Calls the supplier to build
the partition's own handler |
| `PartialUpsertHandler.java` | Javadoc: not thread safe, one instance per
partition |
| `ExpressionTransformer.java` | Javadoc: not thread safe |
| `InbuiltFunctionEvaluator.java` | Javadoc: not thread safe, function nodes
reuse one argument array |
No config, metric, REST, or wire format changes. Nothing is serialized or
sent across nodes, so
there is no mixed-version concern.
`UpsertContext` is an internal class in `pinot-segment-local`. The two
renamed methods had one
production caller each, both inside this module. Anyone maintaining an
out-of-tree
`TableUpsertMetadataManager` that reads
`UpsertContext.getPartialUpsertHandler()` would need the
one-line rename, so this is worth a `backward-incompat` glance even though
nothing in-tree breaks.
## Performance considerations
- One handler and transform chain per partition instead of per table.
Construction parses the
transform expressions, so this is real work, but it happens once when the
partition manager is
built and never on the ingestion path.
- Memory grows with the number of partitions the server hosts for the table.
Each chain is small.
- The steady-state merge path is unchanged. The chain is a plain field
again, so there is no
`ThreadLocal` lookup per merge.
## Testing
- `PartialUpsertHandlerTest`,
`ConcurrentMapPartitionUpsertMetadataManagerTest`,
`ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest`,
`BasePartitionUpsertMetadataManagerTest` and `ExpressionTransformerTest`
all pass (80 tests).
- `ConcurrentMapPartitionUpsertMetadataManagerTest` has three tests that
inject a mock handler
through the context. They now pass `() -> mockHandler`. That is the only
test change, and it is a
mechanical rename, not new coverage.
- **No new test in this PR.** A regression test for the original race needs
many threads hammering a
shared handler and asserting no cross-thread value leaks, which is timing
sensitive. I would rather
land the fix than gate it on a test that could turn flaky in CI. Happy to
add one if reviewers
want it here.
## Labels
`bugfix`
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
