fix(flow-php/etl): array_expand() no longer copies every input column per element

.github/actions/setup-php-env/action.yml

@@ -76,7 +76,7 @@ runs:
       run: sudo apt-get update && sudo apt-get install -y $APT_PACKAGES
 
     - name: "Install PHP"
-      uses: shivammathur/setup-php@f3e473d116dcccaddc5834248c87452386958240 # v2
+      uses: shivammathur/setup-php@eb7c497e18156a6bbabfef1d3a82760b9eda3962 # 2.40.0
       with:
         tools: ${{ inputs.tools }}
         php-version: ${{ inputs.php-version }}

documentation/upgrading.md

@@ -141,6 +141,28 @@ $df->withEntry('date', ref('date')->cast(type_date()))
 
 `same()` / `notSame()` stay strict.
 
+### 15) `flow-php/etl` - an expanding `withEntry()` emits batches of at most 1,000 rows
+
+| Before                                            | After                                                                            |
+|---------------------------------------------------|----------------------------------------------------------------------------------|
+| one output batch per input batch, of any size     | batches of at most 1,000 rows; an input that expands to nothing: one empty batch |
+| every input column copied into every expanded row | only the columns a later `select()` / `drop()` keeps                             |
+
+### 16) `flow-php/etl` - `ScalarFunctionTransformer` refuses `array_expand()`, `NestedExpandTransformer` removed
+
+```php
+// Before - now throws "transform() cannot contain array_expand() ..."
+$df->transform(new ScalarFunctionTransformer('item', array_expand(ref('items'))));
+
+// After
+$df->withEntry('item', array_expand(ref('items')));
+```
+
+### 17) `flow-php/etl` - a serialized `Rows` holds a list of frames
+
+A `Rows` serialized by 0.45 (e.g. a `FilesystemCache` entry) is not read by 0.46: clear such caches after upgrading. A batch
+over 2 GiB of strings in one column is now split across frames instead of throwing `OffsetOverflow`.
+
 ---
 
 ## Upgrading from 0.44.x to 0.45.x

src/core/etl/src/Flow/ETL/DataFrame.php

@@ -15,6 +15,7 @@
 use Flow\ETL\Executor\StatisticsCollector;
 use Flow\ETL\Formatter\AsciiTableFormatter;
 use Flow\ETL\Function\AggregatingFunction;
+use Flow\ETL\Function\ExpandingFunctions;
 use Flow\ETL\Function\ScalarFunction;
 use Flow\ETL\Function\WindowFunction;
 use Flow\ETL\Join\Expression;
@@ -835,9 +836,11 @@ public function withEntries(array $references): self
      */
     public function withEntry(string|Definition $entry, ScalarFunction|WindowFunction $reference): self
     {
-        $this->root = $reference instanceof WindowFunction
-            ? new Node\WindowColumn($this->root, $entry, $reference)
-            : new Node\WithColumn($this->root, $entry, $reference);
+        $this->root = match (true) {
+            $reference instanceof WindowFunction => new Node\WindowColumn($this->root, $entry, $reference),
+            (new ExpandingFunctions())->in($reference) !== [] => new Node\ExpandColumn($this->root, $entry, $reference),
+            default => new Node\WithColumn($this->root, $entry, $reference),
+        };
 
         return $this;
     }

src/core/etl/src/Flow/ETL/Optimizer.php

@@ -12,6 +12,7 @@
 use Flow\ETL\Optimizer\Rule\CountFromStatistics;
 use Flow\ETL\Optimizer\Rule\PushFilterIntoSource;
 use Flow\ETL\Optimizer\Rule\PushLimitIntoSource;
+use Flow\ETL\Optimizer\Rule\PushProjectionIntoExpand;
 use Flow\ETL\Plan\LogicalPlan;
 
 use function array_diff;
@@ -39,6 +40,7 @@ public static function default(): self
             new PushLimitIntoSource(),
             new PushFilterIntoSource(),
             new CountFromStatistics(),
+            new PushProjectionIntoExpand(),
         );
     }
 

src/core/etl/src/Flow/ETL/Optimizer/ProjectionWalk.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Optimizer;
+
+use Flow\ETL\Function\ReferencedColumns;
+use Flow\ETL\Function\ScalarFunction\UnpackResults;
+use Flow\ETL\Plan\Node;
+use Flow\ETL\Plan\RequiredColumns;
+use Flow\ETL\Row\References;
+
+final readonly class ProjectionWalk
+{
+    /**
+     * What the node's row input must hold so that $above holds what its consumers read.
+     */
+    public function below(Node $node, RequiredColumns $above): RequiredColumns
+    {
+        $refs = new ReferencedColumns();
+
+        return match (true) {
+            $node instanceof Node\Select => RequiredColumns::only(...References::init(...$node->entries)->names()),
+            $node instanceof Node\Drop => $above->without(...References::init(...$node->entries)->names()),
+            // an unpack writes "name.*", known only once bound: keep the demand, add what it reads
+            ($node instanceof Node\WithColumn || $node instanceof Node\ExpandColumn)
+                && $node->function instanceof UnpackResults
+                => $above->with(...$refs->in($node->function)->names()),
+            $node instanceof Node\WithColumn, $node instanceof Node\ExpandColumn => $above
+                ->without($node->name())
+                ->with(...$refs->in($node->function)->names()),
+            $node instanceof Node\Filter => $above->with(...$refs->in($node->function)->names()),
+            $node instanceof Node\Limit, $node instanceof Node\Offset => $above,
+            default => RequiredColumns::all(),
+        };
+    }
+}

src/core/etl/src/Flow/ETL/Optimizer/Rule/PushProjectionIntoExpand.php

@@ -0,0 +1,72 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Optimizer\Rule;
+
+use Flow\ETL\FlowContext;
+use Flow\ETL\Optimizer\ProjectionWalk;
+use Flow\ETL\Optimizer\Rule;
+use Flow\ETL\Plan\LogicalPlan;
+use Flow\ETL\Plan\Node;
+use Flow\ETL\Plan\ReplaceLeaf;
+use Flow\ETL\Plan\RequiredColumns;
+use SplObjectStorage;
+
+use function arsort;
+use function spl_object_id;
+
+/**
+ * Columns no consumer reads are not copied into the rows an expand emits (Spark: ColumnPruning setting
+ * Generate.unrequiredChildIndex).
+ */
+final readonly class PushProjectionIntoExpand implements Rule
+{
+    public function apply(LogicalPlan $plan, FlowContext $context): LogicalPlan
+    {
+        $walk = new ProjectionWalk();
+        /** @var SplObjectStorage<Node\ExpandColumn, RequiredColumns> $demands */
+        $demands = new SplObjectStorage();
+
+        foreach ($plan->consumerInputs() as $input) {
+            $required = RequiredColumns::all();
+
+            for ($node = $input; $node->children() !== []; $node = $node->children()[0]) {
+                if ($node instanceof Node\ExpandColumn) {
+                    $demands[$node] = $demands->offsetExists($node) ? $demands[$node]->union($required) : $required;
+                }
+
+                $required = $walk->below($node, $required);
+            }
+        }
+
+        $depths = [];
+        $targets = [];
+
+        foreach ($demands as $node) {
+            if ($demands[$node]->isAll()) {
+                continue;
+            }
+
+            $id = spl_object_id($node);
+            $targets[$id] = $node;
+            $depths[$id] = 0;
+
+            for ($below = $node; $below->children() !== []; $below = $below->children()[0]) {
+                $depths[$id]++;
+            }
+        }
+
+        // TransformUp hands Rewrite::of() the node rebuilt over its new children, so a node whose descendant was
+        // replaced first would no longer be the target; the one farthest from the leaf goes first
+        arsort($depths);
+
+        foreach ($depths as $id => $_) {
+            $plan = $plan->transformUp(
+                new ReplaceLeaf($targets[$id], $targets[$id]->withCarries($demands[$targets[$id]])),
+            );
+        }
+
+        return $plan;
+    }
+}

src/core/etl/src/Flow/ETL/Plan/Explain/Details.php

@@ -14,6 +14,7 @@
 use Flow\ETL\Plan\Node\CrossJoin;
 use Flow\ETL\Plan\Node\Distinct;
 use Flow\ETL\Plan\Node\Drop;
+use Flow\ETL\Plan\Node\ExpandColumn;
 use Flow\ETL\Plan\Node\Filter;
 use Flow\ETL\Plan\Node\Join;
 use Flow\ETL\Plan\Node\Limit;
@@ -73,6 +74,7 @@ public function labelled(Node $node): array
             $node instanceof Filter => ['Condition: ' . $this->name($node->function)],
             $node instanceof Until => ['Until: ' . $this->name($node->function)],
             $node instanceof WithColumn => [sprintf('Column: %s = %s', $node->name(), $this->name($node->function))],
+            $node instanceof ExpandColumn => $this->expandColumn($node),
             $node instanceof Write => ['Loader: ' . $this->name($node->loader)],
             $node instanceof Limit => ['Limit: ' . $node->limit],
             $node instanceof Offset => ['Skip: ' . $node->offset],
@@ -160,6 +162,21 @@ public function columns(array $entries): string
             : $entry, $entries));
     }
 
+    /**
+     * @return list<string>
+     */
+    public function expandColumn(ExpandColumn $node): array
+    {
+        $lines = [sprintf('Column: %s = %s', $node->name(), $this->name($node->function))];
+
+        return match (true) {
+            $node->carries->isAll() => $lines,
+            $node->carries->allBut => [...$lines, 'Keeps: every column but ' . implode(', ', $node->carries->names)],
+            $node->carries->names === [] => [...$lines, 'Keeps: no input column'],
+            default => [...$lines, 'Keeps: ' . implode(', ', $node->carries->names)],
+        };
+    }
+
     /**
      * @return list<string>
      */

src/core/etl/src/Flow/ETL/Plan/Node/ExpandColumn.php

@@ -0,0 +1,77 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Plan\Node;
+
+use Flow\ETL\Function\ScalarFunction;
+use Flow\ETL\Plan\Materialization;
+use Flow\ETL\Plan\Node;
+use Flow\ETL\Plan\Redefined;
+use Flow\ETL\Plan\RequiredColumns;
+use Flow\ETL\Plan\RowCount;
+use Flow\ETL\Plan\Transparency;
+use Flow\ETL\Schema\Definition;
+
+final readonly class ExpandColumn implements Node
+{
+    public RequiredColumns $carries;
+
+    /**
+     * @param Definition<mixed>|string $entry
+     * @param null|RequiredColumns $carries what the consumers above read; null means every column
+     */
+    public function __construct(
+        private Node $input,
+        public string|Definition $entry,
+        public ScalarFunction $function,
+        ?RequiredColumns $carries = null,
+    ) {
+        $this->carries = $carries ?? RequiredColumns::all();
+    }
+
+    /**
+     * @return list<Node>
+     */
+    public function children(): array
+    {
+        return [$this->input];
+    }
+
+    public function withChildren(array $children): self
+    {
+        return $children[0] === $this->input
+            ? $this
+            : new self($children[0], $this->entry, $this->function, $this->carries);
+    }
+
+    public function withCarries(RequiredColumns $carries): self
+    {
+        return new self($this->input, $this->entry, $this->function, $carries);
+    }
+
+    public function rowCount(): RowCount
+    {
+        return RowCount::expanding;
+    }
+
+    public function transparency(): Transparency
+    {
+        return Transparency::transparent;
+    }
+
+    public function materialization(): Materialization
+    {
+        return Materialization::streaming;
+    }
+
+    public function redefines(): Redefined
+    {
+        return Redefined::names($this->name());
+    }
+
+    public function name(): string
+    {
+        return $this->entry instanceof Definition ? $this->entry->entry()->name() : $this->entry;
+    }
+}

src/core/etl/src/Flow/ETL/Plan/Node/WithColumn.php

@@ -4,7 +4,6 @@
 
 namespace Flow\ETL\Plan\Node;
 
-use Flow\ETL\Function\ExpandingFunctions;
 use Flow\ETL\Function\ScalarFunction;
 use Flow\ETL\Plan\Materialization;
 use Flow\ETL\Plan\Node;
@@ -16,19 +15,14 @@
 
 final readonly class WithColumn implements Node
 {
-    private RowCount $rowCount;
-
     /**
      * @param Definition<mixed>|string $entry
      */
     public function __construct(
         private Node $input,
         public string|Definition $entry,
         public ScalarFunction $function,
-    ) {
-        // array_expand() can sit anywhere in the function tree, not only at its root
-        $this->rowCount = (new ExpandingFunctions())->in($function) === [] ? RowCount::preserving : RowCount::expanding;
-    }
+    ) {}
 
     /**
      * @return list<Node>
@@ -45,7 +39,7 @@ public function withChildren(array $children): self
 
     public function rowCount(): RowCount
     {
-        return $this->rowCount;
+        return RowCount::preserving;
     }
 
     public function transparency(): Transparency