cj-zhukov commented on PR #25299:
URL: https://github.com/apache/datafusion/pull/25299#issuecomment-6385648280
> 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`:
>
> ```
> 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`.
@Omega359
Thanks for reviewing the PR and pointing out these issues! I’ll investigate
the row-order determinism and multi-partition handling and work on addressing
them.
--
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]
