feat(flow-php/etl): explain() renders expressions instead of class names

documentation/components/adapters/parquet.md

@@ -140,5 +140,6 @@ Partition columns list fewer files instead ([partitioning](/documentation/compon
          Extractor: ParquetExtractor
          Source: file:///orders/*.parquet
          Limit: 10
-         Filter: All
+         Filter: id > 1000
+         Filter: status = 'paid'
 ```

documentation/components/core/core.md

@@ -64,18 +64,17 @@ echo $dataFrame->explain()->toString();
 
 ```text
 Outputs
-├─ #4 Result
+├─ #3 Result
 │  │  Rows this plan hands out: to the trigger, or to the node reading it
-│  └─ #3 Select
+│  └─ #2 Select
 │     │  Keeps only: id, email
-│     └─ #2 Filter
-│        │  Condition: IsNotNull
-│        └─ #1 Read
-│              Extractor: CSVExtractor
-│              Source: file:///data/orders.csv
-└─ #5 Write
+│     └─ #1 Read
+│           Extractor: CSVExtractor
+│           Source: file:///data/orders.csv
+│           Filter: email IS NOT NULL
+└─ #4 Write
    │  Loader: JsonLoader
-   └─ #2 Filter (shared)
+   └─ #1 Read (shared)
 ```
 
 ```php
@@ -87,21 +86,20 @@ echo data_frame()
 ```
 
 ```text
-#3 Write
+#2 Write
 │  Loader: JsonLoader
-└─ #2 Filter
-   │  Condition: IsNotNull
-   └─ #1 Read
-         Extractor: CSVExtractor
-         Source: file:///data/orders.csv
+└─ #1 Read
+      Extractor: CSVExtractor
+      Source: file:///data/orders.csv
+      Filter: email IS NOT NULL
 ```
 
 `run()` takes no rows, so its plan carries only the sinks - the `Write` is the root and there is no `Result`. A
 `Result` comes back only when a verb follows the last `write()`: nothing would read that chain end, so one is put on
 it to pull the rows, and `run(analyze: ...)` still counts them.
 
 `Outputs` lists everything the frame produces: `Result` is what a rows-returning trigger reads, `Write` is the sink.
-A node several consumers read is printed once, and every other consumer points back at it by number - `#2 Filter (shared)` is that same filter, not a second one. A node keeps its number in every format.
+A node several consumers read is printed once, and every other consumer points back at it by number - `#1 Read (shared)` is that same read, not a second one. A node keeps its number in every format.
 A frame joined with `join()` / `crossJoin()` is part of the same tree: the join reads it the way it reads any other
 input, numbered with the rest of the plan, and it runs with this frame's configuration. A frame read with
 `from_data_frame()` is a `Read` of `DataFrameExtractor` and runs with its own.
@@ -126,14 +124,13 @@ echo $dataFrame->explain()->toString(format: Format::flow);
 #1 Read
 │  Extractor: CSVExtractor
 │  Source: file:///data/orders.csv
-└─ #2 Filter
-   │  Condition: IsNotNull
-   ├─ #3 Select
-   │  │  Keeps only: id, email
-   │  └─ #4 Result
-   │        Rows this plan hands out: to the trigger, or to the node reading it
-   └─ #5 Write
-         Loader: JsonLoader
+│  Filter: email IS NOT NULL
+├─ #2 Select
+│  │  Keeps only: id, email
+│  └─ #3 Result
+│        Rows this plan hands out: to the trigger, or to the node reading it
+└─ #4 Write
+      Loader: JsonLoader
 ```
 
 ```php
@@ -145,28 +142,24 @@ echo $dataFrame->explain()->toString(format: Format::boxes);
 │          Outputs          ├──────────────┐
 └─────────────┬─────────────┘              │
 ┌─────────────┴─────────────┐┌─────────────┴─────────────┐
-│         #4 Result         ││         #5 Write          │
+│         #3 Result         ││         #4 Write          │
 │   ────────────────────    ││   ────────────────────    │
 │ Rows this plan hands out: ││    Loader: JsonLoader     │
 │ to the trigger, or to the ││                           │
 │      node reading it      ││                           │
 └─────────────┬─────────────┘└─────────────┬─────────────┘
 ┌─────────────┴─────────────┐┌─────────────┴─────────────┐
-│         #3 Select         ││         #2 Filter         │
+│         #2 Select         ││          #1 Read          │
 │   ────────────────────    ││         (shared)          │
 │   Keeps only: id, email   ││                           │
 └─────────────┬─────────────┘└───────────────────────────┘
 ┌─────────────┴─────────────┐
-│         #2 Filter         │
-│   ────────────────────    │
-│   Condition: IsNotNull    │
-└─────────────┬─────────────┘
-┌─────────────┴─────────────┐
 │          #1 Read          │
 │   ────────────────────    │
 │  Extractor: CSVExtractor  │
 │          Source:          │
 │  file:///data/orders.csv  │
+│ Filter: email IS NOT NULL │
 └───────────────────────────┘
 ```
 

documentation/components/core/partitioning.md

@@ -290,6 +290,21 @@ data_frame()
     ->run();
 ```
 
+`explain()` lists what each file's path must pass under `Files:`, what each row must pass under `Filter:`:
+
+```text
+#3 Write
+│  Loader: StreamLoader
+└─ #2 Limit
+   │  Limit: 10
+   └─ #1 Read
+         Extractor: CSVExtractor
+         Source: file:///output/date=*/department=*/*.csv
+         Limit: 10
+         Files: date >= '2024-01-01'
+         Filter: date >= '2024-01-01'
+```
+
 ### Path Partitions
 
 `from_path_partitions()` reads partition metadata without opening the files.

src/core/etl/src/Flow/ETL/Filesystem/ScalarFunctionFilter.php

@@ -23,7 +23,7 @@
     private ScalarFunction $resolved;
 
     public function __construct(
-        ScalarFunction $function,
+        private ScalarFunction $function,
         private Schema $partitions,
         private FlowContext $context,
     ) {
@@ -38,6 +38,11 @@ public function __construct(
         }
     }
 
+    public function function(): ScalarFunction
+    {
+        return $this->function;
+    }
+
     public function accept(FileStatus $status): bool
     {
         $values = [];

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

@@ -6,6 +6,7 @@
 
 use Flow\ETL\Extractor\File\FileExtractor;
 use Flow\ETL\Extractor\File\FileScan;
+use Flow\ETL\Filesystem\ScalarFunctionFilter;
 use Flow\ETL\Join\Comparison\Any;
 use Flow\ETL\Plan\Materialization;
 use Flow\ETL\Plan\Node;
@@ -32,10 +33,13 @@
 use Flow\ETL\Plan\Node\WithColumn;
 use Flow\ETL\Plan\Node\Write;
 use Flow\ETL\Row\Reference;
+use Flow\Filesystem\Path\Filter as PathFilter;
+use Flow\Filesystem\Path\Filter\Filters;
 use Flow\Filesystem\Path\Filter\OnlyFiles;
 use ReflectionClass;
 
 use function array_map;
+use function array_merge;
 use function implode;
 use function sprintf;
 use function str_ends_with;
@@ -45,6 +49,7 @@
 {
     public function __construct(
         private Condition $condition = new Condition(),
+        private Expression $expression = new Expression(),
     ) {}
 
     public function name(object $object): string
@@ -72,9 +77,11 @@ public function labelled(Node $node): array
             $node instanceof Join => $this->join($node),
             $node instanceof CrossJoin => $this->crossJoin($node),
             $node instanceof Result => ['Rows this plan hands out: to the trigger, or to the node reading it'],
-            $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 Filter => ['Condition: ' . $this->expression->of($node->function)],
+            $node instanceof Until => ['Until: ' . $this->expression->of($node->function)],
+            $node instanceof WithColumn => [
+                sprintf('Column: %s = %s', $node->name(), $this->expression->of($node->function)),
+            ],
             $node instanceof ExpandColumn => $this->expandColumn($node),
             $node instanceof Write => ['Loader: ' . $this->name($node->loader)],
             $node instanceof Limit => ['Limit: ' . $node->limit],
@@ -168,7 +175,7 @@ public function columns(array $entries): string
      */
     public function expandColumn(ExpandColumn $node): array
     {
-        $lines = [sprintf('Column: %s = %s', $node->name(), $this->name($node->function))];
+        $lines = [sprintf('Column: %s = %s', $node->name(), $this->expression->of($node->function))];
 
         return match (true) {
             $node->carries->isAll() => $lines,
@@ -223,18 +230,33 @@ public function scan(FileScan $scan): array
             $lines[] = 'Limit: ' . $scan->limit;
         }
 
-        // every file source reads only files unless a filter was pushed into it
-        if (!$scan->pathFilter instanceof OnlyFiles) {
-            $lines[] = 'Files: ' . $this->name($scan->pathFilter);
+        foreach ($this->files($scan->pathFilter) as $filter) {
+            $lines[] = 'Files: ' . $filter;
         }
 
         foreach ($scan->conjuncts as $conjunct) {
-            $lines[] = 'Filter: ' . $this->name($conjunct);
+            $lines[] = 'Filter: ' . $this->expression->of($conjunct);
         }
 
         return $lines;
     }
 
+    /**
+     * Each filter a file must pass, one per line.
+     *
+     * @return list<string>
+     */
+    public function files(PathFilter $filter): array
+    {
+        return match (true) {
+            // every file source reads only files, saying so on every read would only add noise
+            $filter instanceof OnlyFiles => [],
+            $filter instanceof Filters => array_merge(...array_map($this->files(...), $filter->filters())),
+            $filter instanceof ScalarFunctionFilter => [$this->expression->of($filter->function())],
+            default => [$this->name($filter)],
+        };
+    }
+
     /**
      * The declarations below in plain words, for the ones that change how the rows flow.
      *