Preload data when loading and discard if aborted due a violation in a…

src/Load/DatabaseLoader/AbstractBatchStmt.php → src/Load/DatabaseLoader/BatchInsertManager.php

@@ -2,14 +2,14 @@
 /**
  * Author: Courtney Miles
  * Date: 26/08/18
- * Time: 9:07 AM
+ * Time: 8:53 AM
  */
 
 namespace MilesAsylum\Slurp\Load\DatabaseLoader;
 
 use MilesAsylum\Slurp\Load\Exception\MissingValueException;
 
-abstract class AbstractBatchStmt implements BatchStmtInterface
+class BatchInsertManager implements BatchManagerInterface
 {
     /**
      * @var \PDO
@@ -26,11 +26,50 @@ abstract class AbstractBatchStmt implements BatchStmtInterface
      */
     protected $columns;
 
-    public function __construct(\PDO $pdo, string $table, array $columns)
+    /**
+     * @var QueryFactory
+     */
+    private $queryFactory;
+
+    /**
+     * @var \PDOStatement[]
+     */
+    private $preparedBatchStmts = [];
+
+    public function __construct(\PDO $pdo, string $table, array $columns, QueryFactory $queryFactory)
     {
         $this->pdo = $pdo;
         $this->table = $table;
         $this->columns = $columns;
+        $this->queryFactory = $queryFactory;
+    }
+
+    /**
+     * @param array[] $rows
+     */
+    public function write(array $rows)
+    {
+        if (!empty($rows)) {
+            $this->getPreparedBatchStmt(count($rows))
+                ->execute(
+                    $this->convertRowCollectionToParams($rows)
+                );
+        }
+    }
+
+    protected function getPreparedBatchStmt($count): \PDOStatement
+    {
+        if (!isset($this->preparedBatchStmts[$count])) {
+            $this->preparedBatchStmts[$count] = $this->pdo->prepare(
+                $this->queryFactory->createInsertQuery(
+                    $this->table,
+                    $this->columns,
+                    $count
+                )
+            );
+        }
+
+        return $this->preparedBatchStmts[$count];
     }
 
     protected function ensureColumnMatch($rowId, array $rowValues): void

src/Load/DatabaseLoader/BatchStmtInterface.php → src/Load/DatabaseLoader/BatchManagerInterface.php

@@ -7,7 +7,7 @@
 
 namespace MilesAsylum\Slurp\Load\DatabaseLoader;
 
-interface BatchStmtInterface
+interface BatchManagerInterface
 {
     public function write(array $rows);
 }

src/Load/DatabaseLoader/DatabaseLoader.php

@@ -7,12 +7,13 @@
 
 namespace MilesAsylum\Slurp\Load\DatabaseLoader;
 
+use MilesAsylum\Slurp\Load\DatabaseLoader\Exception\DatabaseLoaderException;
 use MilesAsylum\Slurp\Load\LoaderInterface;
 
 class DatabaseLoader implements LoaderInterface
 {
     /**
-     * @var \PDOStatement
+     * @var BatchManagerInterface
      */
     protected $batchStmt;
 
@@ -30,35 +31,91 @@ class DatabaseLoader implements LoaderInterface
      * @var array
      */
     private $columnMapping;
+    /**
+     * @var string
+     */
+    private $table;
+
+    /**
+     * @var StagedLoad
+     */
+    private $stagedLoad;
+
+    /**
+     * @var LoaderFactory
+     */
+    private $loaderFactory;
 
     /**
      * DatabaseLoader constructor.
-     * @param BatchStmtInterface $batchStmt
-     * @param int $batchSize
+     * @param string $table
      * @param array $columnMapping Array key is the destination column and the array value is the source column.
+     * @param LoaderFactory $dmlFactory
+     * @param int $batchSize
      */
     public function __construct(
-        BatchStmtInterface $batchStmt,
-        int $batchSize = 100,
-        array $columnMapping = []
+        string $table,
+        array $columnMapping,
+        LoaderFactory $dmlFactory,
+        int $batchSize = 100
     ) {
+        $this->loaderFactory = $dmlFactory;
+        $this->table = $table;
         $this->batchSize = $batchSize;
-        $this->batchStmt = $batchStmt;
         $this->columnMapping = $columnMapping;
     }
 
+    /**
+     * @param array $values
+     * @throws DatabaseLoaderException
+     */
     public function loadValues(array $values): void
     {
+        if (!$this->hasBegun()) {
+            throw new DatabaseLoaderException(
+                sprintf(
+                    'Data cannot be loaded until %s has been called.',
+                    __CLASS__ . '::begin()'
+                )
+            );
+        }
+
         $this->rowCollection[] = $this->mapColumnNames($values);
 
         if (count($this->rowCollection) >= $this->batchSize) {
             $this->flush();
         }
     }
 
+    public function begin(): void
+    {
+        $this->stagedLoad = $this->loaderFactory->createStagedLoad(
+            $this->table,
+            array_keys($this->columnMapping)
+        );
+        $stagedTable = $this->stagedLoad->begin();
+        $this->batchStmt = $this->loaderFactory->createBatchInsStmt(
+            $stagedTable,
+            array_keys($this->columnMapping)
+        );
+    }
+
+    public function hasBegun(): bool
+    {
+        return isset($this->stagedLoad);
+    }
+
     public function finalise(): void
     {
         $this->flush();
+        $this->stagedLoad->commit();
+    }
+
+    public function abort(): void
+    {
+        $this->stagedLoad->discard();
+        $this->stagedLoad = null;
+        $this->batchStmt = null;
     }
 
     protected function flush(): void

src/Load/DatabaseLoader/Exception/DatabaseLoaderException.php

@@ -0,0 +1,15 @@
+<?php
+/**
+ * Author: Courtney Miles
+ * Date: 16/09/18
+ * Time: 3:54 PM
+ */
+
+namespace MilesAsylum\Slurp\Load\DatabaseLoader\Exception;
+
+use MilesAsylum\Slurp\Exception\ExceptionInterface;
+
+class DatabaseLoaderException extends \Exception implements ExceptionInterface
+{
+
+}

src/Load/DatabaseLoader/LoaderFactory.php

@@ -0,0 +1,36 @@
+<?php
+/**
+ * Author: Courtney Miles
+ * Date: 16/09/18
+ * Time: 3:34 PM
+ */
+
+namespace MilesAsylum\Slurp\Load\DatabaseLoader;
+
+class LoaderFactory
+{
+    /**
+     * @var \PDO
+     */
+    private $pdo;
+
+    public function __construct(\PDO $pdo)
+    {
+        $this->pdo = $pdo;
+    }
+
+    public function createBatchInsStmt(string $table, array $columns)
+    {
+        return new BatchInsertManager($this->pdo, $table, $columns, $this->createBatchInsQueryFactory());
+    }
+
+    public function createStagedLoad(string $table, array $columns)
+    {
+        return new StagedLoad($this->pdo, $table, $columns);
+    }
+
+    protected function createBatchInsQueryFactory()
+    {
+        return new QueryFactory();
+    }
+}

src/Load/DatabaseLoader/BatchInsUpdQueryFactory.php → src/Load/DatabaseLoader/QueryFactory.php

@@ -7,7 +7,7 @@
 
 namespace MilesAsylum\Slurp\Load\DatabaseLoader;
 
-class BatchInsUpdQueryFactory
+class QueryFactory
 {
     /**
      * @param $table
@@ -16,7 +16,7 @@ class BatchInsUpdQueryFactory
      * @return string
      * @throws \Exception
      */
-    public function createQuery($table, array $columns, $batchSize = 1): string
+    public function createInsertQuery($table, array $columns, $batchSize = 1): string
     {
         if (empty($columns)) {
             throw new \InvalidArgumentException('One or more columns must be supplied.');
@@ -29,18 +29,10 @@ public function createQuery($table, array $columns, $batchSize = 1): string
         $colsStr = '`' . implode('`, `', $columns) . '`';
         $valueStr = '(' . implode(', ', array_fill(0, count($columns), '?')) . ')';
         $batchValueStr = implode(",\n    ", array_fill(0, $batchSize, $valueStr));
-        $updateValues = [];
-
-        foreach ($columns as $column) {
-            $updateValues[] = "`{$column}` = VALUES(`{$column}`)";
-        }
-
-        $updateValuesStr = implode(', ', $updateValues);
 
         return <<<SQL
 INSERT INTO `{$table}` ({$colsStr})
   VALUES $batchValueStr
-  ON DUPLICATE KEY UPDATE {$updateValuesStr}
 SQL;
     }
 }