Allow usage of different schemas for documents

src/PostgresDocumentStore.php

@@ -119,6 +119,7 @@ public function hasCollection(string $collectionName): bool
 SELECT TABLE_NAME 
 FROM information_schema.tables
 WHERE TABLE_NAME = '{$this->tableName($collectionName)}'
+AND TABLE_SCHEMA = '{$this->schemaName($collectionName)}'
 EOT;
 
         $stmt = $this->connection->prepare($query);
@@ -147,8 +148,10 @@ public function addCollection(string $collectionName, Index ...$indices): void
             }
         }
 
+        $createSchemaCmd = "CREATE SCHEMA IF NOT EXISTS {$this->schemaName($collectionName)}";
+
         $cmd = <<<EOT
-CREATE TABLE {$this->tableName($collectionName)} (
+CREATE TABLE {$this->schemaName($collectionName)}.{$this->tableName($collectionName)} (
     id {$this->docIdSchema},
     doc JSONB NOT NULL,
     $metadataColumns
@@ -160,7 +163,8 @@ public function addCollection(string $collectionName, Index ...$indices): void
             return $this->indexToSqlCmd($index, $collectionName);
         }, $indices);
 
-        $this->transactional(function() use ($cmd, $indicesCmds) {
+        $this->transactional(function() use ($createSchemaCmd, $cmd, $indicesCmds) {
+            $this->connection->prepare($createSchemaCmd)->execute();
             $this->connection->prepare($cmd)->execute();
 
             array_walk($indicesCmds, function ($cmd) {
@@ -176,7 +180,7 @@ public function addCollection(string $collectionName, Index ...$indices): void
     public function dropCollection(string $collectionName): void
     {
         $cmd = <<<EOT
-DROP TABLE {$this->tableName($collectionName)};
+DROP TABLE {$this->schemaName($collectionName)}.{$this->tableName($collectionName)};
 EOT;
 
         $this->transactional(function () use ($cmd) {
@@ -190,6 +194,7 @@ public function hasCollectionIndex(string $collectionName, string $indexName): b
 SELECT INDEXNAME 
 FROM pg_indexes
 WHERE TABLENAME = '{$this->tableName($collectionName)}'
+AND SCHEMANAME = '{$this->schemaName($collectionName)}'
 AND INDEXNAME = '$indexName'
 EOT;
 
@@ -222,7 +227,7 @@ public function addCollectionIndex(string $collectionName, Index $index): void
             $columnsSql = substr($columnsSql, 2);
 
             $metadataColumnCmd = <<<EOT
-ALTER TABLE {$this->tableName($collectionName)}
+ALTER TABLE {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
     $columnsSql;
 EOT;
 
@@ -262,7 +267,7 @@ public function dropCollectionIndex(string $collectionName, $index): void
             $columnsSql = substr($columnsSql, 2);
 
             $metadataColumnCmd = <<<EOT
-ALTER TABLE {$this->tableName($collectionName)}
+ALTER TABLE {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
     $columnsSql;
 EOT;
             $index = $index->indexCmd();
@@ -312,7 +317,9 @@ public function addDoc(string $collectionName, string $docId, array $doc): void
         }
 
         $cmd = <<<EOT
-INSERT INTO {$this->tableName($collectionName)} (id, doc{$metadataKeysStr}) VALUES (:id, :doc{$metadataValsStr});
+INSERT INTO {$this->schemaName($collectionName)}.{$this->tableName($collectionName)} (
+    id, doc{$metadataKeysStr}) VALUES (:id, :doc{$metadataValsStr}
+);
 EOT;
 
         $this->transactional(function () use ($cmd, $docId, $doc, $metadata) {
@@ -345,7 +352,7 @@ public function updateDoc(string $collectionName, string $docId, array $docOrSub
         }
 
         $cmd = <<<EOT
-UPDATE {$this->tableName($collectionName)}
+UPDATE {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
 SET doc = (to_jsonb(doc) || :doc){$metadataStr}
 WHERE id = :id
 ;
@@ -384,7 +391,7 @@ public function updateMany(string $collectionName, Filter $filter, array $set):
         }
 
         $cmd = <<<EOT
-UPDATE {$this->tableName($collectionName)}
+UPDATE {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
 SET doc = (to_jsonb(doc) || :doc){$metadataStr}
 $where;
 EOT;
@@ -424,7 +431,7 @@ public function upsertDoc(string $collectionName, string $docId, array $docOrSub
     public function deleteDoc(string $collectionName, string $docId): void
     {
         $cmd = <<<EOT
-DELETE FROM {$this->tableName($collectionName)}
+DELETE FROM {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
 WHERE id = :id
 EOT;
 
@@ -447,7 +454,7 @@ public function deleteMany(string $collectionName, Filter $filter): void
         $where = $filterStr? "WHERE $filterStr" : '';
 
         $cmd = <<<EOT
-DELETE FROM {$this->tableName($collectionName)}
+DELETE FROM {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
 $where;
 EOT;
 
@@ -465,7 +472,7 @@ public function getDoc(string $collectionName, string $docId): ?array
     {
         $query = <<<EOT
 SELECT doc
-FROM {$this->tableName($collectionName)}
+FROM {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
 WHERE id = :id
 EOT;
         $stmt = $this->connection->prepare($query);
@@ -502,7 +509,7 @@ public function filterDocs(string $collectionName, Filter $filter, int $skip = n
 
         $query = <<<EOT
 SELECT doc 
-FROM {$this->tableName($collectionName)}
+FROM {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
 $where
 $orderBy
 $limit
@@ -701,7 +708,7 @@ private function indexToSqlCmd(Index $index, string $collectionName): string
         $name = $index->name() ?? '';
 
         $cmd = <<<EOT
-CREATE $type $name ON {$this->tableName($collectionName)}
+CREATE $type $name ON {$this->schemaName($collectionName)}.{$this->tableName($collectionName)}
 $fields;
 EOT;
 
@@ -751,6 +758,19 @@ private function extractFieldPartFromFieldIndex(DocumentStore\FieldIndex $fieldI
 
     private function tableName(string $collectionName): string
     {
+        if (false !== $dotPosition = strpos($collectionName, '.')) {
+            $collectionName = substr($collectionName, $dotPosition+1);
+        }
+
         return mb_strtolower($this->tablePrefix . $collectionName);
     }
+
+     private function schemaName(string $collectionName): string
+     {
+         $schemaName = 'public';
+         if (false !== $dotPosition = strpos($collectionName, '.')) {
+             $schemaName = substr($collectionName, 0, $dotPosition);
+         }
+         return mb_strtolower($schemaName);
+     }
 }

tests/SchemedPostgresDocumentStoreTest.php

@@ -0,0 +1,60 @@
+<?php
+/**
+ * This file is part of the event-engine/php-postgres-document-store.
+ * (c) 2019 prooph software GmbH <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace EventEngine\DocumentStoreTest\Postgres;
+
+use EventEngine\DocumentStore\Filter\AnyOfDocIdFilter;
+use EventEngine\DocumentStore\Filter\AnyOfFilter;
+use EventEngine\DocumentStore\Filter\DocIdFilter;
+use EventEngine\DocumentStore\Filter\NotFilter;
+use PHPUnit\Framework\TestCase;
+use EventEngine\DocumentStore\FieldIndex;
+use EventEngine\DocumentStore\Index;
+use EventEngine\DocumentStore\MultiFieldIndex;
+use EventEngine\DocumentStore\Postgres\PostgresDocumentStore;
+use Ramsey\Uuid\Uuid;
+
+class SchemedPostgresDocumentStoreTest extends TestCase
+{
+    private CONST TABLE_PREFIX = 'test_';
+    private CONST SCHEMA = 'test.';
+
+    /**
+     * @var PostgresDocumentStore
+     */
+    protected $documentStore;
+
+    /**
+     * @var \PDO
+     */
+    protected $connection;
+
+    protected function setUp(): void
+    {
+        $this->connection = TestUtil::getConnection();
+        $this->documentStore = new PostgresDocumentStore($this->connection, self::TABLE_PREFIX);
+    }
+
+    public function tearDown(): void
+    {
+        TestUtil::tearDownDatabase();
+    }
+
+    /**
+     * @test
+     */
+    public function it_adds_collection_with_schema(): void
+    {
+        $this->documentStore->addCollection(self::SCHEMA . 'test');
+        $this->assertFalse($this->documentStore->hasCollection('test'));
+        $this->assertTrue($this->documentStore->hasCollection(self::SCHEMA . 'test'));
+    }
+}