Omega359 commented on PR #25299:
URL: https://github.com/apache/datafusion/pull/25299#issuecomment-6372445520
I had claude do a review of this PR and it has this to say (among other
things):
Before polishing this, please reply to @comphead explaining why
make_array/lit don't fit (positional pairing vs. the same value on every row),
and address row-order determinism. As written, any multi-partition input —
including read_batches(..).filter(..) on a multi-threaded runtime — pairs rows
with the wrong values.
## 2 Blocker: rows get paired with the wrong values
`AddColumnExec::execute` (`datafusion/core/src/dataframe/mod.rs:3119`)
silently wraps any multi-partition input in a `CoalescePartitionsExec`:
```rust
// A single ArrayRef is one global vector, so the input must be one stream.
let input = if self.input.output_partitioning().partition_count() == 1 {
Arc::clone(&self.input)
} else {
Arc::new(CoalescePartitionsExec::new(Arc::clone(&self.input)))
};
```
That has two consequences:
- **Row order is effectively random.** `CoalescePartitionsExec` merges
partitions in whatever order batches arrive, so row *i* of the output is not
row *i* of the input.
- **Most inputs have multiple partitions.** The inner plan is built with
`state.create_physical_plan`, and the optimizer routinely splits even a single
in-memory source into many partitions.
### Evidence
I checked out the PR locally and ran probe tests. Each one:
- built 32,000 rows with `id` = 0..32000,
- added `tag` = 0..32000 via `array_col`,
- counted rows where `id != tag`.
| Input | Runtime | Rows mismatched (of 32,000) |
| --- | --- | --- |
| `MemTable` with 8 partitions | single-threaded | 19,000, in 10 of 10 runs |
| `ctx.read_batches(...).filter(id >= 0)` | single-threaded | 0 |
| `ctx.read_batches(...).filter(id >= 0)` | multi-threaded (8 workers) |
20,000 to 32,000, in 10 of 10 runs |
EXPLAIN for the `read_batches` case shows the optimizer split the source
into 8 partitions underneath `AddColumnExec`:
```text
AddColumnExec: tag
FilterExec: id@0 >= 0
DataSourceExec: partitions=8, partition_sizes=[4, 4, 4, 4, 4, 4, 4, 4]
```
The hidden coalesce doesn't appear in EXPLAIN at all, because it's created
inside `execute()` rather than in the plan.
### Why the PR's tests pass
They use 3-row `dataframe!` inputs on the default single-threaded
`#[tokio::test]` runtime, which is the one setup where this doesn't show up.
### What a fix needs
Declaring `required_input_distribution = SinglePartition` isn't enough on
its own. It would make the merge visible in the plan, but the merge still
wouldn't preserve order. Pairing by position needs a defined row order, and
DataFusion plans don't guarantee one.
## 3 The marker UDF is fragile
array_col is an Expr that only works in one exact position: the top level of
with_column, optionally wrapped in an alias. Anywhere else it passes planning
and then fails at execution. I confirmed both of these:
- df.select([array_col(a).alias("x")]) plans fine, then fails at collect
with "array_col() can only be used with DataFrame::with_column".
- with_column("x", array_col(a) + lit(1)) behaves the same way.
The type system says "this is an expression", but the real rule is "this is
a with_column argument". That's the kind of hidden assumption that turns into a
stream of bug reports. If the feature survives, a separately named method
states the contract honestly. That was the PR's original shape, and the issue's
with_array_columns sketch matches it. Hiding the method made the API look
smaller while making it trickier to use.
## 4 Inconsistent behaviour inside with_column
- Replacing a column: with_column is documented as "Add or replace a
column". with_column("id", lit(5)) replaces id, but with_column("id",
array_col(..)) fails with Column 'id' already exists (mod.rs:2953). The same
method now behaves differently depending on the argument.
- Nullability: the new field is always marked nullable (mod.rs:2963),
whatever the array actually contains.
- Multiple columns: each array_col call adds another table-scan wrapper and
another AddColumnExec, each planned separately. The PR's own type test chains
13 of them.
## 5 Structure and conciseness
- Wrong file: about 390 lines of UDF, TableProvider and ExecutionPlan code
went into dataframe/mod.rs, which grows from 3,055 to 3,320 lines. This is
execution-layer code. If it's kept, it belongs in its own module (for example
dataframe/add_column.rs), next to the existing ViewTable pattern
(datafusion/catalog/src/view.rs) that wraps a logical plan as a table.
- Unrelated import churn: about 60 lines of the diff reorder imports that
have nothing to do with the change, which hides the real diff. It should be
reverted.
- Duplicated example: the same doc example appears three times — on
with_column, on the private with_array_column, and on array_col. A private
helper doesn't need a full example, and array_col's copy is identical to the
one on with_array_column.
- Hidden coalesce: related to item 2, execute() decides partitioning itself
instead of declaring what it needs. That's an anti-pattern whatever happens to
the design.
- Pass-through code: replace_children / with_new_children add bulk without
doing anything useful.
Realistic options:
1. **Restrict to inputs with a guaranteed order.** For example, build the
inner plan with `target_partitions = 1` and document the limitation clearly.
2. **Use the relational approach.** Join on a key, or on `row_number()` over
an explicit `ORDER BY`.
--
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]
