[1.0.0] Optimize the SQL in the PDO storage adapter for some searches.

src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/Pdo/PdoListQueryBuilder.php

@@ -153,8 +153,11 @@ private function buildChildQuery(
             return $query;
         }
 
+        // short-circuits the child scan instead of an IN list materialising the whole match set (O(directory)).
+        $filterSql = $filterResult->correlatedSql ?? $filterResult->sql;
+
         return $query->appending(
-            ' AND (' . $filterResult->sql . ')',
+            ' AND (' . $filterSql . ')',
             $filterResult->params,
         );
     }

src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/SqlFilter/SqlFilterResult.php

@@ -22,11 +22,17 @@
  */
 final class SqlFilterResult
 {
+    /**
+     * Correlated `EXISTS` form referencing the outer `lc_dn` so an outer LIMIT can short-circuit, or null when none.
+     */
+    public readonly ?string $correlatedSql;
+
     /**
      * @param list<string> $params
      * @param list<string> $referencedAttributes Attributes whose absence makes the filter undefined under RFC 4511
      * @param ?string $sidecarCondition Single drivable leaf's sidecar WHERE body for the streaming fast path.
      * @param list<SidecarLeaf> $drivableLeaves A composed filter's drivable child leaves, for composed-filter streaming.
+     * @param ?string $correlatedSql Explicit correlated form for composites; leaves derive it from $sidecarCondition.
      */
     public function __construct(
         public readonly string $sql,
@@ -35,5 +41,23 @@ public function __construct(
         public readonly array $referencedAttributes = [],
         public readonly ?string $sidecarCondition = null,
         public readonly array $drivableLeaves = [],
-    ) {}
+        ?string $correlatedSql = null,
+    ) {
+        $this->correlatedSql = $correlatedSql
+            ?? ($sidecarCondition !== null ? self::correlatedLeaf($sidecarCondition) : null);
+    }
+
+    /**
+     * Wraps a sidecar WHERE body as an `EXISTS` correlated to the outer entries row via the unqualified `lc_dn`.
+     */
+    public static function correlatedLeaf(string $sidecarCondition): string
+    {
+        return <<<SQL
+            EXISTS (
+                SELECT 1
+                FROM entry_attribute_values s
+                WHERE s.entry_lc_dn = lc_dn
+                  AND $sidecarCondition)
+            SQL;
+    }
 }

src/FreeDSx/Ldap/Server/Backend/Storage/Adapter/SqlFilter/SqlFilterTranslatorTrait.php

@@ -391,9 +391,11 @@ private function prepareMatchValue(string $value): string
     private function translateAnd(AndFilter $filter): ?SqlFilterResult
     {
         $parts = [];
+        $correlatedParts = [];
         $params = [];
         $drivableLeaves = [];
         $hasUntranslatable = false;
+        $allCorrelatable = true;
 
         foreach ($filter->get() as $child) {
             $result = $this->dispatch($child);
@@ -407,6 +409,12 @@ private function translateAnd(AndFilter $filter): ?SqlFilterResult
             $parts[] = '(' . $result->sql . ')';
             array_push($params, ...$result->params);
             array_push($drivableLeaves, ...$this->drivableLeavesOf($result));
+
+            if ($result->correlatedSql !== null) {
+                $correlatedParts[] = '(' . $result->correlatedSql . ')';
+            } else {
+                $allCorrelatable = false;
+            }
         }
 
         if ($parts === []) {
@@ -418,6 +426,9 @@ private function translateAnd(AndFilter $filter): ?SqlFilterResult
             $params,
             isExact: !$hasUntranslatable,
             drivableLeaves: $drivableLeaves,
+            correlatedSql: $allCorrelatable
+                ? implode(' AND ', $correlatedParts)
+                : null,
         );
     }
 
@@ -443,8 +454,10 @@ private function drivableLeavesOf(SqlFilterResult $result): array
     private function translateOr(OrFilter $filter): ?SqlFilterResult
     {
         $parts = [];
+        $correlatedParts = [];
         $params = [];
         $hasInexact = false;
+        $allCorrelatable = true;
 
         foreach ($filter->get() as $child) {
             $result = $this->dispatch($child);
@@ -456,6 +469,12 @@ private function translateOr(OrFilter $filter): ?SqlFilterResult
             }
             $parts[] = '(' . $result->sql . ')';
             array_push($params, ...$result->params);
+
+            if ($result->correlatedSql !== null) {
+                $correlatedParts[] = '(' . $result->correlatedSql . ')';
+            } else {
+                $allCorrelatable = false;
+            }
         }
 
         if ($parts === []) {
@@ -466,6 +485,9 @@ private function translateOr(OrFilter $filter): ?SqlFilterResult
             implode(' OR ', $parts),
             $params,
             isExact: !$hasInexact,
+            correlatedSql: $allCorrelatable
+                ? implode(' OR ', $correlatedParts)
+                : null,
         );
     }
 
@@ -485,6 +507,9 @@ private function translateNot(NotFilter $filter): ?SqlFilterResult
                 'NOT (' . $result->sql . ')',
                 $result->params,
                 isExact: $result->isExact,
+                correlatedSql: $result->correlatedSql !== null
+                    ? 'NOT (' . $result->correlatedSql . ')'
+                    : null,
             );
         }
 
@@ -493,15 +518,19 @@ private function translateNot(NotFilter $filter): ?SqlFilterResult
         // simple filters (those that populated referencedAttributes) we AND
         // in a presence guard so missing-attribute rows are excluded.
         if ($result->referencedAttributes !== []) {
+            $attributes = array_values(array_unique($result->referencedAttributes));
             $guards = array_map(
                 fn(string $attribute): string => $this->buildPresenceCheck($attribute),
-                array_values(array_unique($result->referencedAttributes)),
+                $attributes,
             );
 
             return new SqlFilterResult(
                 '(NOT (' . $result->sql . ') AND ' . implode(' AND ', $guards) . ')',
                 $result->params,
                 isExact: $result->isExact,
+                correlatedSql: $result->correlatedSql !== null
+                    ? '(NOT (' . $result->correlatedSql . ') AND ' . implode(' AND ', $this->correlatedGuards($attributes)) . ')'
+                    : null,
             );
         }
 
@@ -513,6 +542,28 @@ private function translateNot(NotFilter $filter): ?SqlFilterResult
             'NOT (' . $result->sql . ')',
             $result->params,
             isExact: false,
+            correlatedSql: $result->correlatedSql !== null
+                ? 'NOT (' . $result->correlatedSql . ')'
+                : null,
+        );
+    }
+
+    /**
+     * Correlated `EXISTS` presence guards matching the IN-form presence guards for a NOT(value) filter.
+     *
+     * @param list<string> $attributes
+     * @return list<string>
+     */
+    private function correlatedGuards(array $attributes): array
+    {
+        return array_map(
+            fn(string $attribute): string => SqlFilterResult::correlatedLeaf(
+                $this->sidecarCondition(
+                    $attribute,
+                    null,
+                ),
+            ),
+            $attributes,
         );
     }
 

tests/unit/Server/Backend/Storage/Adapter/Pdo/PdoListQueryBuilderTest.php

@@ -263,6 +263,57 @@ public function test_child_scope_disables_the_streaming_fast_path(): void
         );
     }
 
+    public function test_child_scope_uses_the_correlated_exists_form(): void
+    {
+        $query = (new PdoListQueryBuilder(new SqliteDialect()))->build(
+            'ou=people,dc=foo,dc=bar',
+            false,
+            $this->sidecarLeaf(),
+            5001,
+            [],
+        );
+
+        self::assertStringContainsString(
+            'lc_parent_dn = ?',
+            $query->sql,
+        );
+        self::assertStringContainsString(
+            'EXISTS (',
+            $query->sql,
+        );
+        self::assertStringContainsString(
+            's.entry_lc_dn = lc_dn',
+            $query->sql,
+        );
+        self::assertStringNotContainsString(
+            'lc_dn IN (',
+            $query->sql,
+        );
+    }
+
+    public function test_child_scope_falls_back_to_the_in_form_without_a_correlated_form(): void
+    {
+        $query = (new PdoListQueryBuilder(new SqliteDialect()))->build(
+            'ou=people,dc=foo,dc=bar',
+            false,
+            new SqlFilterResult(
+                "lc_dn IN (SELECT s.entry_lc_dn FROM entry_attribute_values s WHERE s.attr_name_lower = 'cn')",
+                [],
+            ),
+            5001,
+            [],
+        );
+
+        self::assertStringContainsString(
+            'lc_dn IN (',
+            $query->sql,
+        );
+        self::assertStringNotContainsString(
+            'EXISTS (',
+            $query->sql,
+        );
+    }
+
     private function sidecarLeaf(): SqlFilterResult
     {
         return new SqlFilterResult(

tests/unit/Server/Backend/Storage/Adapter/SqliteFilterTranslatorTest.php

@@ -86,6 +86,82 @@ public function test_equality_emits_sidecar_value_equality(): void
         );
     }
 
+    public function test_leaf_correlated_sql_is_an_exists_form(): void
+    {
+        $result = $this->subject->translate(new EqualityFilter(
+            'cn',
+            'Alice',
+        ));
+
+        self::assertNotNull($result);
+        self::assertNotNull($result->correlatedSql);
+        self::assertStringContainsString(
+            'EXISTS (',
+            $result->correlatedSql,
+        );
+        self::assertStringContainsString(
+            's.entry_lc_dn = lc_dn',
+            $result->correlatedSql,
+        );
+        self::assertStringNotContainsString(
+            'lc_dn IN (',
+            $result->correlatedSql,
+        );
+    }
+
+    public function test_and_composes_correlated_exists_with_and(): void
+    {
+        $result = $this->subject->translate(new AndFilter(
+            new EqualityFilter('cn', 'Alice'),
+            new EqualityFilter('sn', 'Smith'),
+        ));
+
+        self::assertNotNull($result);
+        self::assertNotNull($result->correlatedSql);
+        self::assertSame(
+            2,
+            substr_count($result->correlatedSql, 'EXISTS ('),
+        );
+        self::assertStringContainsString(
+            ') AND (',
+            $result->correlatedSql,
+        );
+    }
+
+    public function test_or_composes_correlated_exists_with_or(): void
+    {
+        $result = $this->subject->translate(new OrFilter(
+            new EqualityFilter('cn', 'Alice'),
+            new EqualityFilter('cn', 'Bob'),
+        ));
+
+        self::assertNotNull($result);
+        self::assertNotNull($result->correlatedSql);
+        self::assertStringContainsString(
+            ') OR (',
+            $result->correlatedSql,
+        );
+    }
+
+    public function test_not_value_correlated_sql_keeps_the_presence_guard(): void
+    {
+        $result = $this->subject->translate(new NotFilter(
+            new EqualityFilter('cn', 'Alice'),
+        ));
+
+        self::assertNotNull($result);
+        self::assertNotNull($result->correlatedSql);
+        self::assertStringContainsString(
+            'NOT (',
+            $result->correlatedSql,
+        );
+        // Presence guard EXISTS plus the negated value EXISTS.
+        self::assertSame(
+            2,
+            substr_count($result->correlatedSql, 'EXISTS ('),
+        );
+    }
+
     public function test_approximate_translates_same_as_equality(): void
     {
         $result = $this->subject->translate(new ApproximateFilter(