hansva opened a new issue, #9328:
URL: https://github.com/apache/hop/issues/9328
### Apache Hop version?
2.21.0-SNAPSHOT
### Java version?
21
### Operating system
Other
### What happened?
*Copy rows to result* (`RowsToResult`) has three correctness problems and
one performance problem, all in the few lines of `processRow()`. They will be
fixed together, because each fix changes the same lines.
**1. Result rows are stored by reference.** `processRow()` keeps the
incoming row array in its result buffer and passes the same array on with
`putRow()`. A transform after Copy rows to result that changes values in place,
such as If null, changes the stored result row too. The workflow then receives
the values as they were after that transform, not as they reached Copy rows to
result. Each row is also stored with the live input row metadata instead of a
snapshot.
**2. Copies merge into a plain `ArrayList` without a lock.** At the end of
its input, each copy calls `getResultRows().addAll()` on the pipeline's
`ArrayList`. With *Number of copies* above 1, or two Copy rows to result
transforms in one pipeline, two copies can finish at the same moment. Rows are
then lost, or `addAll()` throws an `ArrayIndexOutOfBoundsException`. This
depends on thread timing.
**3. No result rows from the single-threaded engine.** The buffered rows are
handed over only when `getRow()` returns null.
`SingleThreadedPipelineExecutor.oneIteration()` calls `processRow()` exactly
once per row waiting on the input, then `batchComplete()`, so `getRow()` never
returns null there. `RowsToResult` has no `batchComplete()`, so a pipeline run
single-threaded, or a Simple mapping in that mode, returns 0 result rows.
**4. The row metadata is cloned for every row.** For every row,
`processRow()` clones the input row metadata and calls `meta.getFields()` on
it, although `RowsToResultMeta.getFields()` changes nothing. Profiling a
2,000,000-row pipeline that ends in Copy rows to result shows 96% of the
transform's CPU in that clone, mostly `ValueMetaBase.clone()` rebuilding number
formats. The transform runs at about 70k rows/s. It is also the critical path
of a Pipeline executor whose child pipeline ends in Copy rows to result.
Points 1 to 3 come from reading the code on main (3a70efa9b5); the tests
below reproduce them.
### Proposed fix
- Build the output row metadata once, on the first row: clone the input row
metadata, apply `getFields()` and keep the result in `RowsToResultData`. Every
stored row and every `putRow()` uses it. This fixes 4 and gives each stored row
a metadata snapshot (1).
- Store a copy of each row (`IRowMeta.cloneRow()`), so later transforms
can't change it (1). The copy costs far less than the metadata clone it
replaces. The benchmark below shows whether a plain array copy is needed
instead.
- Hand the buffered rows over to the pipeline's result rows under a lock,
both at the end of input and in a new `batchComplete()`. Clear the buffer after
each hand-over so no row is added twice (2, 3).
- A new `RowsToResultTest` covers four cases:
- a later in-place change does not reach the result rows
- two copies lose no rows
- the single-threaded executor returns every row
- the output row metadata is built once
- Measure rows/s on the 2,000,000-row pipeline before and after.
Out of scope:
- result rows on Beam and Spark, which the engines don't collect at all (an
engine issue)
- *Get rows from result* emitting every row once per copy
- the general cost of cloning value metadata in core
### Issue Priority
Priority: 2
### Issue Component
Component: Transforms
--
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]
