Unverified Commit 2e8705a4 authored by Alex Pott's avatar Alex Pott
Browse files

feat: #2547493 Add support for unique / primary key constraints composed of... 

feat: #2547493 Add support for unique / primary key constraints composed of multiple fields for Upsert queries

By: amateescu
By: daffie
By: andypost
By: ghost of drupal past
By: mondrake
By: chi
By: longwave
(cherry picked from commit 3ae3915e)
parent db002cb1

core/lib/Drupal/Core/Database/Query/Upsert.php

+10 −11
Original line number Diff line number Diff line
@@ -7,19 +7,18 @@ 
/**

 * General class for an abstracted "Upsert" (UPDATE or INSERT) query operation.

 *

 * This class can only be used with a table with a single unique index.

 * Often, this will be the primary key. On such a table this class works like

 * Insert except the rows will be set to the desired values even if the key

 * existed before.

 * This class works like Insert except the rows will be set to the desired

 * values even if the key existed before. It supports both single-field and

 * composite (multi-field) unique or primary key constraints.

 */

abstract class Upsert extends Query implements \Countable {



  use InsertTrait;



  /**

   * The unique or primary key of the table.

   * The unique or primary key column(s) of the table.

   *

   * @var string

   * @var string[]

   */

  protected $key;
@@ -39,15 +38,15 @@ public function __construct(Connection $connection, $table, array $options = []) 
  }



  /**

   * Sets the unique / primary key field to be used as condition for this query.

   * Sets the unique / primary key field(s) to be used as condition.

   *

   * @param string $field

   *   The name of the field to set.

   * @param string|string[] $field

   *   The name of the field, or an array of field names for a composite key.

   *

   * @return $this

   */

  public function key($field) {

    $this->key = $field;

  public function key(string|array $field) {

    $this->key = (array) $field;



    return $this;

  }

core/modules/mysql/src/Driver/Database/mysql/Upsert.php

+5 −2
Original line number Diff line number Diff line
@@ -18,6 +18,7 @@ public function __toString() { 


    // Default fields are always placed first for consistency.

    $insert_fields = array_merge($this->defaultFields, $this->insertFields);

    $insert_fields = array_combine($insert_fields, $insert_fields);

    $insert_fields = array_map(function ($field) {

      return $this->connection->escapeField($field);

    }, $insert_fields);
@@ -27,8 +28,10 @@ public function __toString() { 
    $values = $this->getInsertPlaceholderFragment($this->insertValues, $this->defaultFields);

    $query .= implode(', ', $values);



    // Updating the unique / primary key is not necessary.

    unset($insert_fields[$this->key]);

    // Updating the unique / primary key fields is not necessary.

    foreach ($this->key as $key) {

      unset($insert_fields[$key]);

    }



    $update = [];

    foreach ($insert_fields as $field) {

core/modules/pgsql/src/Driver/Database/pgsql/Upsert.php

+11 −3
Original line number Diff line number Diff line
@@ -28,6 +28,7 @@ public function execute() { 
    $blobs = [];

    $blob_count = 0;

    foreach ($this->insertValues as $insert_values) {

      $insert_values = array_values($insert_values);

      foreach ($this->insertFields as $idx => $field) {

        if (isset($table_information->blob_fields[$field]) && $insert_values[$idx] !== NULL) {

          $blobs[$blob_count] = fopen('php://memory', 'a');
@@ -103,8 +104,13 @@ public function __toString() { 
    // Create a sanitized comment string to prepend to the query.

    $comments = $this->connection->makeComment($this->comments);



    $keys = array_map(function ($key) {

      return $this->connection->escapeField($key);

    }, $this->key);



    // Default fields are always placed first for consistency.

    $insert_fields = array_merge($this->defaultFields, $this->insertFields);

    $insert_fields = array_combine($insert_fields, $insert_fields);

    $insert_fields = array_map(function ($field) {

      return $this->connection->escapeField($field);

    }, $insert_fields);
@@ -114,8 +120,10 @@ public function __toString() { 
    $values = $this->getInsertPlaceholderFragment($this->insertValues, $this->defaultFields);

    $query .= implode(', ', $values);



    // Updating the unique / primary key is not necessary.

    unset($insert_fields[$this->key]);

    // Updating the unique / primary key fields is not necessary.

    foreach ($this->key as $key) {

      unset($insert_fields[$key]);

    }



    $update = [];

    foreach ($insert_fields as $field) {
@@ -124,7 +132,7 @@ public function __toString() { 
      $update[] = "$field = EXCLUDED.$field";

    }



    $query .= ' ON CONFLICT (' . $this->connection->escapeField($this->key) . ') DO UPDATE SET ' . implode(', ', $update);

    $query .= ' ON CONFLICT (' . implode(', ', $keys) . ') DO UPDATE SET ' . implode(', ', $update);



    return $query;

  }

core/modules/sqlite/src/Driver/Database/sqlite/Upsert.php

+10 −3
Original line number Diff line number Diff line
@@ -18,8 +18,13 @@ public function __toString() { 
    // Create a sanitized comment string to prepend to the query.

    $comments = $this->connection->makeComment($this->comments);



    $keys = array_map(function ($key) {

      return $this->connection->escapeField($key);

    }, $this->key);



    // Default fields are always placed first for consistency.

    $insert_fields = array_merge($this->defaultFields, $this->insertFields);

    $insert_fields = array_combine($insert_fields, $insert_fields);

    $insert_fields = array_map(function ($field) {

      return $this->connection->escapeField($field);

    }, $insert_fields);
@@ -29,8 +34,10 @@ public function __toString() { 
    $values = $this->getInsertPlaceholderFragment($this->insertValues, $this->defaultFields);

    $query .= implode(', ', $values);



    // Updating the unique / primary key is not necessary.

    unset($insert_fields[$this->key]);

    // Updating the unique / primary key fields is not necessary.

    foreach ($this->key as $key) {

      unset($insert_fields[$key]);

    }



    $update = [];

    foreach ($insert_fields as $field) {
@@ -39,7 +46,7 @@ public function __toString() { 
      $update[] = "$field = EXCLUDED.$field";

    }



    $query .= ' ON CONFLICT (' . $this->connection->escapeField($this->key) . ') DO UPDATE SET ' . implode(', ', $update);

    $query .= ' ON CONFLICT (' . implode(', ', $keys) . ') DO UPDATE SET ' . implode(', ', $update);



    return $query;

  }

core/tests/Drupal/KernelTests/Core/Database/UpsertTest.php

+82 −0
Original line number Diff line number Diff line
@@ -126,6 +126,88 @@ public function testUpsertNonExistingTable(): void { 
    $upsert->execute();

  }



  /**

   * Confirms that we can upsert records with composite keys successfully.

   */

  public function testCompositeKeyUpsert(): void {

    $connection = Database::getConnection();

    $this->installSchema('database_test', ['test_composite_primary']);



    // Add some initial test data.

    $connection->insert('test_composite_primary')

      ->fields(['name', 'age', 'job'])

      ->values([

        'name' => 'Tiffany',

        'age' => 31,

        'job' => 'Presenter',

      ])

      ->values([

        'name' => 'Meredith',

        'age' => 30,

        'job' => 'Speaker',

      ])

      ->execute();



    $num_records_before = $connection->query('SELECT COUNT(*) FROM {test_composite_primary}')->fetchField();



    $upsert = $connection->upsert('test_composite_primary')

      ->key(['name', 'age'])

      // Add a new row directly from ::fields().

      ->fields([

        'name' => 'Kate',

        'age' => 25,

        'job' => 'Volunteer',

      ]);



    // Add a new row.

    $upsert->values([

      'name' => 'Karen',

      'age' => 35,

      'job' => 'Manager',

    ]);



    // Update an existing row.

    $upsert->values([

      'name' => 'Meredith',

      'age' => 30,

      // The initial job was 'Speaker'.

      'job' => 'Organizer',

    ]);



    // Add a new row by reusing a name but with a different age. This won't

    // match the composite primary key constraint.

    $upsert->values([

      'name' => 'Meredith',

      'age' => 40,

      'job' => 'Supervisor',

    ]);



    $upsert->execute();



    $num_records_after = $connection->query('SELECT COUNT(*) FROM {test_composite_primary}')->fetchField();

    $this->assertEquals($num_records_before + 3, $num_records_after, 'Rows were inserted and updated properly.');



    $person = $connection->query('SELECT * FROM {test_composite_primary} WHERE [job] = :job', [':job' => 'Volunteer'])->fetch();

    $this->assertEquals('Volunteer', $person->job, 'Job set correctly.');

    $this->assertEquals(25, $person->age, 'Age set correctly.');

    $this->assertEquals('Kate', $person->name, 'Name set correctly.');



    $person = $connection->query('SELECT * FROM {test_composite_primary} WHERE [job] = :job', [':job' => 'Manager'])->fetch();

    $this->assertEquals('Manager', $person->job, 'Job set correctly.');

    $this->assertEquals(35, $person->age, 'Age set correctly.');

    $this->assertEquals('Karen', $person->name, 'Name set correctly.');



    $person = $connection->query('SELECT * FROM {test_composite_primary} WHERE [job] = :job', [':job' => 'Organizer'])->fetch();

    $this->assertEquals('Organizer', $person->job, 'Job set correctly.');

    $this->assertEquals(30, $person->age, 'Age set correctly.');

    $this->assertEquals('Meredith', $person->name, 'Name set correctly.');



    $person = $connection->query('SELECT * FROM {test_composite_primary} WHERE [job] = :job', [':job' => 'Supervisor'])->fetch();

    $this->assertEquals('Supervisor', $person->job, 'Job set correctly.');

    $this->assertEquals(40, $person->age, 'Age set correctly.');

    $this->assertEquals('Meredith', $person->name, 'Name set correctly.');

  }



  /**

   * Tests that we can upsert a null into blob field.

   */