FEATURE: flushByTags for the PDO cache backend

With this change the `flushByTags` is optimised to run
in batches. The maximum batch size can and should be
configured based on the used data source.

This change relies on the interface changes in #2717

Author: Sebobo

Neos.Cache/Classes/Backend/PdoBackend.php

@@ -72,6 +72,11 @@ class PdoBackend extends IndependentAbstractBackend implements TaggableBackendIn
      */
     protected $tagsTableName = 'tags';
 
+    /**
+     * @var integer
+     */
+    protected $batchSize = 999;
+
     /**
      * @var \ArrayIterator
      */
@@ -137,6 +142,16 @@ protected function setTagsTableName(string $tagsTableName): void
         $this->tagsTableName = $tagsTableName;
     }
 
+    /**
+     * Sets the maximum number of items for batch operations
+     *
+     * @api
+     */
+    protected function setBatchSize(int $batchSize): void
+    {
+        $this->batchSize = $batchSize;
+    }
+
     /**
      * Saves data in the cache.
      *
@@ -348,6 +363,64 @@ public function flushByTag(string $tag): int
         return $flushed;
     }
 
+    /**
+     * Removes all cache entries of this cache which are tagged by any of the specified tags.
+     *
+     * @throws Exception
+     * @api
+     */
+    public function flushByTags(array $tags): int
+    {
+        $this->connect();
+        $this->databaseHandle->beginTransaction();
+        $flushed = 0;
+
+        // All queries need to be run in batches as every backend has a parameter limit.
+        try {
+            // Reduce batch size by 2 as we also pass the context and cache identifier parameters.
+            $batchSize = $this->batchSize - 2;
+
+            // Step 1: Collect all identifiers to be flushed for the given tags
+            $identifiers = [];
+            for ($i = 0, $iMax = count($tags); $i < $iMax; $i += $batchSize) {
+                $tagList = array_slice($tags, $i, $batchSize);
+                $tagPlaceholders = implode(',', array_fill(0, count($tagList), '?'));
+                $statementHandle = $this->databaseHandle->prepare('SELECT "identifier" FROM "' . $this->tagsTableName . '" WHERE "context"=? AND "cache"=? AND "tag" IN (' . $tagPlaceholders . ')');
+                $statementHandle->execute(array_merge([$this->context(), $this->cacheIdentifier], $tagList));
+                $result = $statementHandle->fetchAll();
+                $identifiers[]= array_column($result, 'identifier');
+            }
+            $identifiers = array_merge([], ...$identifiers);
+
+            // Step 2: Flush all collected identifiers
+            for ($i = 0, $iMax = count($identifiers); $i < $iMax; $i += $batchSize) {
+                $identifierList = array_slice($identifiers, $i, $batchSize);
+                $identifierPlaceholders = implode(',', array_fill(0, count($identifierList), '?'));
+                $statementHandle = $this->databaseHandle->prepare('DELETE FROM "' . $this->cacheTableName . '" WHERE "context"=? AND "cache"=? AND "identifier" IN (' . $identifierPlaceholders . ')');
+                $statementHandle->execute(array_merge([$this->context(), $this->cacheIdentifier], $identifierList));
+
+                // Update the flushed counter
+                $flushed += $statementHandle->rowCount();
+            }
+
+            // Step 3: Remove all given tags
+            for ($i = 0, $iMax = count($tags); $i < $iMax; $i += $batchSize) {
+                $tagList = array_slice($tags, $i, $batchSize);
+                $tagPlaceholders = implode(',', array_fill(0, count($tagList), '?'));
+                $statementHandle = $this->databaseHandle->prepare('DELETE FROM "' . $this->tagsTableName . '" WHERE "context"=? AND "cache"=? AND "tag" IN (' . $tagPlaceholders . ')');
+                $statementHandle->execute(array_merge([$this->context(), $this->cacheIdentifier], $tagList));
+            }
+
+            $this->databaseHandle->commit();
+        } catch (\Exception $exception) {
+            $this->databaseHandle->rollBack();
+
+            throw $exception;
+        }
+
+        return $flushed;
+    }
+
     /**
      * Finds and returns all cache entry identifiers which are tagged by the
      * specified tag.

Neos.Flow/Documentation/TheDefinitiveGuide/PartIII/Caching.rst

@@ -458,10 +458,22 @@ Options
 |                | * sqlite:/path/to/sqlite.db                        |           |        |         |
 |                | * sqlite::memory:                                  |           |        |         |
 +----------------+----------------------------------------------------+-----------+--------+---------+
-| username       | Username to use for the database connection        | No        |        |         |
+| username       | Username to use for the database connection.       | No        |        |         |
 +----------------+----------------------------------------------------+-----------+--------+---------+
 | password       | Password to use for the database connection.       | No        |        |         |
 +----------------+----------------------------------------------------+-----------+--------+---------+
+| cacheTableName | Table name to store cache entries                  | No        | string | cache   |
++----------------+----------------------------------------------------+-----------+--------+---------+
+| tagsTableName  | Table name to store cache tags                     | No        | string | tags    |
++----------------+----------------------------------------------------+-----------+--------+---------+
+| batchSize      | Maximum number of parameters per query for         | No        | int    | 999     |
+|                | for batch operations.                              |           |        |         |
+|                |                                                    |           |        |         |
+|                | This value should be adjusted based on type of     |           |        |         |
+|                | the data source to increase performance.           |           |        |         |
+|                | E.g. Postgres supports up to 65535 parameters,     |           |        |         |
+|                | but SQLite only 999 for older versions.            |           |        |         |
++----------------+----------------------------------------------------+-----------+--------+---------+
 
 Neos\\Cache\\Backend\\RedisBackend
 ----------------------------------