Provide an option to run a DML query (i.e a DELETE) before committing…

… the staged data.

src/Load/DatabaseLoader/DatabaseLoader.php

@@ -49,24 +49,32 @@ class DatabaseLoader implements LoaderInterface
     protected $begun = false;
 
     protected $aborted = false;
+    /**
+     * @var PreCommitDmlInterface
+     */
+    private $preCommitDml;
 
     /**
      * DatabaseLoader constructor.
      * @param string $table
      * @param array $columnMapping Array key is the destination column and the array value is the source column.
      * @param LoaderFactory $dmlFactory
+     * @param \PDO $pdo
      * @param int $batchSize
+     * @param PreCommitDmlInterface|null $preCommitDml
      */
     public function __construct(
         string $table,
         array $columnMapping,
         LoaderFactory $dmlFactory,
-        int $batchSize = 100
+        int $batchSize = 100,
+        PreCommitDmlInterface $preCommitDml = null
     ) {
         $this->loaderFactory = $dmlFactory;
         $this->table = $table;
         $this->batchSize = $batchSize;
         $this->columnMapping = $columnMapping;
+        $this->preCommitDml = $preCommitDml;
     }
 
     /**
@@ -149,6 +157,11 @@ public function finalise(): void
         }
 
         $this->flush();
+
+        if ($this->preCommitDml !== null) {
+            $this->preCommitDml->execute();
+        }
+
         $this->stagedLoad->commit();
     }
 

src/Load/DatabaseLoader/PreCommitDmlInterface.php

@@ -0,0 +1,20 @@
+<?php
+/**
+ * Author: Courtney Miles
+ * Date: 22/09/18
+ * Time: 6:26 AM
+ */
+
+namespace MilesAsylum\Slurp\Load\DatabaseLoader;
+
+
+/**
+ * This class can been used to run a DML statement immediately prior to committing the loaded records.
+ */
+interface PreCommitDmlInterface
+{
+    /**
+     * @return int The number of affected rows.
+     */
+    public function execute(): int;
+}

src/Load/DatabaseLoader/PreCommitSimpleDelete.php

@@ -0,0 +1,67 @@
+<?php
+/**
+ * Author: Courtney Miles
+ * Date: 21/09/18
+ * Time: 7:18 PM
+ */
+
+namespace MilesAsylum\Slurp\Load\DatabaseLoader;
+
+
+class PreCommitSimpleDelete implements PreCommitDmlInterface
+{
+    /**
+     * @var \PDO
+     */
+    private $pdo;
+
+    /**
+     * @var string
+     */
+    private $table;
+
+    /**
+     * @var array
+     */
+    private $conditions;
+
+    public function __construct(\PDO $pdo, string $table, array $conditions = [])
+    {
+        $this->pdo = $pdo;
+        $this->table = $table;
+        $this->conditions = $conditions;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(): int
+    {
+        $conditionsStr = null;
+        $qryParams = [];
+
+        foreach ($this->conditions as $col => $value) {
+            $valuePlaceholder = $this->colNameToPlaceholder($col);
+            $conditions[] = sprintf(
+                '`%s` = %s',
+                $col,
+                $valuePlaceholder
+            );
+            $qryParams[$valuePlaceholder] = $value;
+        }
+
+        if (!empty($conditions)) {
+            $conditionsStr = ' WHERE ' . implode(' AND ', $conditions);
+        }
+
+        $stmt = $this->pdo->prepare("DELETE FROM `{$this->table}`$conditionsStr");
+        $stmt->execute($qryParams);
+
+        return $stmt->rowCount();
+    }
+
+    protected function colNameToPlaceholder($colName)
+    {
+        return ':' . str_replace([' '], ['_'], $colName);
+    }
+}

src/SlurpBuilder.php

@@ -11,6 +11,7 @@
 use League\Pipeline\PipelineBuilder;
 use MilesAsylum\Slurp\Load\DatabaseLoader\DatabaseLoader;
 use MilesAsylum\Slurp\Load\DatabaseLoader\LoaderFactory;
+use MilesAsylum\Slurp\Load\DatabaseLoader\PreCommitDmlInterface;
 use MilesAsylum\Slurp\Load\LoaderInterface;
 use MilesAsylum\Slurp\Stage\InvokeExtractionPipeline;
 use MilesAsylum\Slurp\Stage\LoadStage;
@@ -180,13 +181,19 @@ public function addLoader(LoaderInterface $loader): self
         return $this;
     }
 
-    public function createDatabaseLoader(\PDO $pdo, string $table, array $fieldMappings, int $batchSize): DatabaseLoader
-    {
+    public function createDatabaseLoader(
+        \PDO $pdo,
+        string $table,
+        array $fieldMappings,
+        int $batchSize,
+        PreCommitDmlInterface $preCommitDml = null
+    ): DatabaseLoader {
         return new DatabaseLoader(
             $table,
             $fieldMappings,
             new LoaderFactory($pdo),
-            $batchSize
+            $batchSize,
+            $preCommitDml
         );
     }
 

tests/Slurp/Load/DatabaseLoader/DatabaseLoaderTest.php

@@ -11,6 +11,7 @@
 use MilesAsylum\Slurp\Load\DatabaseLoader\DatabaseLoader;
 use MilesAsylum\Slurp\Load\DatabaseLoader\Exception\DatabaseLoaderException;
 use MilesAsylum\Slurp\Load\DatabaseLoader\LoaderFactory;
+use MilesAsylum\Slurp\Load\DatabaseLoader\PreCommitDmlInterface;
 use MilesAsylum\Slurp\Load\DatabaseLoader\StagedLoad;
 use Mockery\Adapter\Phpunit\MockeryPHPUnitIntegration;
 use Mockery\MockInterface;
@@ -35,6 +36,11 @@ class DatabaseLoaderTest extends TestCase
      */
     protected $mockStagedLoad;
 
+    /**
+     * @var \PDO|MockInterface
+     */
+    protected $mockPdo;
+
     protected $batchSize = 3;
 
     public function setUp()
@@ -138,6 +144,28 @@ public function testFlushRemainingOnFinalise()
         $databaseLoader->finalise();
     }
 
+    public function testCallPreCommitDmlOnFinalise()
+    {
+        $mockPreCommitDml = \Mockery::mock(PreCommitDmlInterface::class);
+
+        $mockPreCommitDml->shouldReceive('execute')
+            ->once();
+
+        $this->mockBatchStmt->shouldReceive('write')->byDefault();
+        $this->mockStagedLoad->shouldReceive('commit')->byDefault();
+
+        $databaseLoader = new DatabaseLoader(
+            'my_tbl',
+            ['col1' => 'col1', 'col2' => 'col2'],
+            $this->mockLoaderFactory,
+            2,
+            $mockPreCommitDml
+        );
+        $databaseLoader->begin();
+
+        $databaseLoader->finalise();
+    }
+
     public function testRemapColumns()
     {
         $row = ['col1' => 123, 'col2' => 234];

tests/Slurp/Load/DatabaseLoader/PreCommitSimpleDeleteTest.php

@@ -0,0 +1,85 @@
+<?php
+/**
+ * Author: Courtney Miles
+ * Date: 22/09/18
+ * Time: 7:01 AM
+ */
+
+namespace MilesAsylum\Slurp\Tests\Slurp\Load\DatabaseLoader;
+
+use MilesAsylum\Slurp\Load\DatabaseLoader\PreCommitSimpleDelete;
+use Mockery\Adapter\Phpunit\MockeryPHPUnitIntegration;
+use Mockery\MockInterface;
+use PHPUnit\Framework\TestCase;
+
+class PreCommitSimpleDeleteTest extends TestCase
+{
+    use MockeryPHPUnitIntegration;
+
+    /**
+     * @var \PDO|MockInterface
+     */
+    protected $mockPdo;
+
+    /**
+     * @var \PDOStatement|MockInterface
+     */
+    protected $mockDelStmt;
+
+    public function setUp()
+    {
+        parent::setUp();
+
+        $this->mockPdo = \Mockery::mock(\PDO::class);
+        $this->mockDelStmt = \Mockery::mock(\PDOStatement::class);
+    }
+
+    public function testExecuteWithoutConditions()
+    {
+        $affectedRows = 3;
+        $table = 'my_tbl';
+
+        $expectedQry = <<<SQL
+DELETE FROM `my_tbl`
+SQL;
+
+        $this->mockPdo->shouldReceive('prepare')
+            ->with($expectedQry)
+            ->andReturn($this->mockDelStmt)
+            ->once();
+        $this->mockDelStmt->shouldReceive('execute')
+            ->with([])
+            ->once();
+        $this->mockDelStmt->shouldReceive('rowCount')
+            ->andReturn($affectedRows);
+
+        $delete = new PreCommitSimpleDelete($this->mockPdo, $table);
+
+        $this->assertSame($affectedRows, $delete->execute());
+    }
+
+    public function testExecuteWithConditions()
+    {
+        $affectedRows = 3;
+        $table = 'my_tbl';
+        $conditions = ['col 1' => 123, 'col2' => 'abc'];
+
+        $expectedQry = <<<SQL
+DELETE FROM `my_tbl` WHERE `col 1` = :col_1 AND `col2` = :col2
+SQL;
+
+        $this->mockPdo->shouldReceive('prepare')
+            ->with($expectedQry)
+            ->andReturn($this->mockDelStmt)
+            ->once();
+        $this->mockDelStmt->shouldReceive('execute')
+            ->with([':col_1' => 123, ':col2' => 'abc'])
+            ->once();
+        $this->mockDelStmt->shouldReceive('rowCount')
+            ->andReturn($affectedRows);
+
+        $delete = new PreCommitSimpleDelete($this->mockPdo, $table, $conditions);
+
+        $this->assertSame($affectedRows, $delete->execute());
+    }
+}

tests/Slurp/Validate/SchemaValidation/SchemaValidatorTest.php

@@ -89,7 +89,7 @@ public function testValidateRecord()
         $this->assertEquals(new Violation($recordId, $badField, $badValue, $message), $violation);
     }
 
-    public function testValidatRecordWithUniqueField()
+    public function testValidateRecordWithUniqueField()
     {
         $field = 'id';
         $record = [$field => 123];
@@ -112,7 +112,7 @@ public function testValidatRecordWithUniqueField()
         $violation = array_pop($violations);
 
         $this->assertInstanceOf(ViolationInterface::class, $violation);
-        $this->assertEquals(new Violation(2, $field, $record[$field], "Field value is not unique."), $violation);
+        $this->assertEquals(new Violation(2, $field, $record[$field], "id: value is not unique."), $violation);
     }
 
     public function testUnknownFieldException()