feat: allow for partial index conditions for on conflict statments in…

… postgres (#8971)

* feat: allow for partial index conditions for on conflict statments in postgres

* fix: change variable to match postgres documentation and convert from array to string type

* fix: generalize use of index predicate. add documentation and expand test cases

* lint fix

Co-authored-by: Umed Khudoiberdiev <pleerock.me@gmail.com>
Co-authored-by: Alex Messer <dmzt08@gmail.com>

docs/insert-query-builder.md

@@ -35,3 +35,69 @@ await dataSource
 ```
 
 This syntax doesn't escape your values, you need to handle escape on your own.
+
+### Update values ON CONFLICT
+
+If the values you are trying to insert conflict due to existing data the `orUpdate` function can be used to update specific values on the conflicted target.
+
+```typescript
+await datasource
+    .createQueryBuilder()
+    .insert()
+    .into(User)
+    .values({
+        firstName: "Timber",
+        lastName: "Saw",
+        externalId: "abc123",
+    })
+    .orUpdate(
+        ["firstName", "lastName"],
+        ["externalId"],
+    )
+    .execute()
+```
+
+### Skip data update if values have not changed (Postgres)
+
+```typescript
+await datasource
+    .createQueryBuilder()
+    .insert()
+    .into(User)
+    .values({
+        firstName: "Timber",
+        lastName: "Saw",
+        externalId: "abc123",
+    })
+    .orUpdate(
+        ["firstName", "lastName"],
+        ["externalId"],
+        {
+            skipUpdateIfNoValuesChanged: true,
+        }
+    )
+    .execute()
+```
+
+### Use partial index (Postgres)
+
+```typescript
+await datasource
+    .createQueryBuilder()
+    .insert()
+    .into(User)
+    .values({
+        firstName: "Timber",
+        lastName: "Saw",
+        externalId: "abc123",
+    })
+    .orUpdate(
+        ["firstName", "lastName"],
+        ["externalId"],
+        {
+            skipUpdateIfNoValuesChanged: true,
+            indexPredicate: "date > 2020-01-01"
+        }
+    )
+    .execute()
+```

docs/repository-api.md

@@ -191,6 +191,30 @@ await repository.upsert(
  **/
 ```
 
+```typescript
+await repository.upsert(
+    [
+        { externalId: "abc123", firstName: "Rizzrak", dateAdded: "2020-01-01" },
+        { externalId: "bca321", firstName: "Karzzir", dateAdded: "2022-01-01" },
+    ],
+    {
+        conflictPaths: ["externalId"],
+        skipUpdateIfNoValuesChanged: true, // supported by postgres, skips update if it would not change row values
+        indexPredicate: "dateAdded > 2020-01-01", // supported by postgres, allows for partial indexes
+    },
+)
+/** executes
+ *  INSERT INTO user
+ *  VALUES
+ *      (externalId = abc123, firstName = Rizzrak, dateAdded = 2020-01-01),
+ *      (externalId = cba321, firstName = Karzzir, dateAdded = 2022-01-01),
+ *  ON CONFLICT (externalId) WHERE ( dateAdded > 2021-01-01 ) DO UPDATE
+ *  SET firstName = EXCLUDED.firstName,
+ *  SET dateAdded = EXCLUDED.dateAdded,
+ *  WHERE user.firstName IS DISTINCT FROM EXCLUDED.firstName OR user.dateAdded IS DISTINCT FROM EXCLUDED.dateAdded 
+ **/
+```
+
 -   `delete` - Deletes entities by entity id, ids or given conditions:
 
 ```typescript

src/entity-manager/EntityManager.ts

@@ -726,6 +726,7 @@ export class EntityManager {
                 {
                     skipUpdateIfNoValuesChanged:
                         options.skipUpdateIfNoValuesChanged,
+                    indexPredicate: options.indexPredicate,
                 },
             )
             .execute()

src/query-builder/InsertOrUpdateOptions.ts

@@ -1,3 +1,10 @@
 export type InsertOrUpdateOptions = {
+    /**
+     * If true, postgres will skip the update if no values would be changed (reduces writes)
+     */
     skipUpdateIfNoValuesChanged?: boolean
+    /**
+     * If included, postgres will apply the index predicate to a conflict target (partial index)
+     */
+    indexPredicate?: string
 }

src/query-builder/InsertQueryBuilder.ts

@@ -390,6 +390,7 @@ export class InsertQueryBuilder<
             conflict: conflictTarget,
             skipUpdateIfNoValuesChanged:
                 orUpdateOptions?.skipUpdateIfNoValuesChanged,
+            indexPredicate: orUpdateOptions?.indexPredicate,
         }
         return this
     }
@@ -484,6 +485,7 @@ export class InsertQueryBuilder<
                     columns,
                     conflict,
                     skipUpdateIfNoValuesChanged,
+                    indexPredicate,
                 } = this.expressionMap.onUpdate
 
                 let conflictTarget = "ON CONFLICT"
@@ -492,6 +494,22 @@ export class InsertQueryBuilder<
                     conflictTarget += ` ( ${conflict
                         .map((column) => this.escape(column))
                         .join(", ")} )`
+                    if (
+                        indexPredicate &&
+                        !DriverUtils.isPostgresFamily(this.connection.driver)
+                    ) {
+                        throw new TypeORMError(
+                            `indexPredicate option is not supported by the current database driver`,
+                        )
+                    }
+                    if (
+                        indexPredicate &&
+                        DriverUtils.isPostgresFamily(this.connection.driver)
+                    ) {
+                        conflictTarget += ` WHERE ( ${this.escape(
+                            indexPredicate,
+                        )} )`
+                    }
                 } else if (conflict) {
                     conflictTarget += ` ON CONSTRAINT ${this.escape(conflict)}`
                 }

src/query-builder/QueryExpressionMap.ts

@@ -114,6 +114,7 @@ export class QueryExpressionMap {
         columns?: string[]
         overwrite?: string[]
         skipUpdateIfNoValuesChanged?: boolean
+        indexPredicate?: string
     }
 
     /**

src/repository/UpsertOptions.ts

@@ -1,9 +1,12 @@
+import { InsertOrUpdateOptions } from "../query-builder/InsertOrUpdateOptions"
+
 /**
  * Special options passed to Repository#upsert
  */
 // eslint-disable-next-line @typescript-eslint/no-unused-vars
-export interface UpsertOptions<Entity> {
+export interface UpsertOptions<Entity> extends InsertOrUpdateOptions {
     conflictPaths: string[] | { [P in keyof Entity]?: true }
+
     /**
      * If true, postgres will skip the update if no values would be changed (reduces writes)
      */

test/functional/query-builder/insert-on-conflict/query-builder-insert-on-conflict.ts

@@ -6,6 +6,7 @@ import {
 } from "../../../utils/test-utils"
 import { DataSource } from "../../../../src/data-source/DataSource"
 import { Post } from "./entity/Post"
+import { expect } from "chai"
 
 describe("query builder > insertion > on conflict", () => {
     let connections: DataSource[]
@@ -244,4 +245,91 @@ describe("query builder > insertion > on conflict", () => {
                     })
             }),
         ))
+    it("should perform insertion on partial index using orUpdate", () =>
+        Promise.all(
+            connections.map(async (connection) => {
+                if (connection.driver.options.type !== "postgres") return
+                const post1 = new Post()
+                post1.id = "post#1"
+                post1.title = "About post"
+                post1.date = new Date("06 Aug 2020 00:12:00 GMT")
+
+                const sql = connection.manager
+                    .createQueryBuilder()
+                    .insert()
+                    .into(Post)
+                    .values(post1)
+                    .orUpdate(["title"], ["date"], {
+                        indexPredicate: "date > 2020-01-01",
+                    })
+                    .setParameter("title", post1.title)
+                    .disableEscaping()
+                    .getSql()
+
+                expect(sql).to.equal(
+                    `INSERT INTO post(id, title, date) ` +
+                        `VALUES ($1, $2, $3) ON CONFLICT ( date ) ` +
+                        `WHERE ( date > 2020-01-01 ) DO UPDATE SET title = EXCLUDED.title`,
+                )
+            }),
+        ))
+    it("should perform insertion using partial index and skipping update on no change", () =>
+        Promise.all(
+            connections.map(async (connection) => {
+                if (connection.driver.options.type !== "postgres") return
+                const post1 = new Post()
+                post1.id = "post#1"
+                post1.title = "About post"
+                post1.date = new Date("06 Aug 2020 00:12:00 GMT")
+
+                const sql = connection.manager
+                    .createQueryBuilder()
+                    .insert()
+                    .into(Post)
+                    .values(post1)
+                    .orUpdate(["title"], ["date"], {
+                        skipUpdateIfNoValuesChanged: true,
+                        indexPredicate: "date > 2020-01-01",
+                    })
+                    .setParameter("title", post1.title)
+                    .disableEscaping()
+                    .getSql()
+
+                expect(sql).to.equal(
+                    `INSERT INTO post(id, title, date) ` +
+                        `VALUES ($1, $2, $3) ON CONFLICT ( date ) ` +
+                        `WHERE ( date > 2020-01-01 ) DO UPDATE SET title = EXCLUDED.title  ` +
+                        `WHERE (post.title IS DISTINCT FROM EXCLUDED.title)`,
+                )
+            }),
+        ))
+    it("should throw error if using indexPredicate amd an unsupported driver", () => {
+        Promise.all(
+            connections.map(async (connection) => {
+                if (
+                    connection.driver.supportedUpsertType !==
+                    "on-duplicate-key-update"
+                )
+                    return
+                const post1 = new Post()
+                post1.id = "post#1"
+                post1.title = "About post"
+                post1.date = new Date("06 Aug 2020 00:12:00 GMT")
+
+                const sql = connection.manager
+                    .createQueryBuilder()
+                    .insert()
+                    .into(Post)
+                    .values(post1)
+                    .orUpdate(["title"], ["date"], {
+                        indexPredicate: "date > 2020-01-01",
+                    })
+                    .setParameter("title", post1.title)
+                    .disableEscaping()
+                    .getSql()
+
+                expect(sql).to.throw(Error)
+            }),
+        )
+    })
 })

test/functional/repository/basic-methods/repository-basic-methods.ts

@@ -848,6 +848,28 @@ describe("repository > basic methods", () => {
                         .should.be.rejectedWith(TypeORMError)
                 }),
             ))
+        it("should throw if using indexPredicate with an unsupported driver", () =>
+            Promise.all(
+                connections.map(async (connection) => {
+                    if (
+                        connection.driver.supportedUpsertType !==
+                        "on-conflict-do-update"
+                    )
+                        return
+
+                    const postRepository = connection.getRepository(Post)
+                    const externalId = "external-2"
+                    await postRepository
+                        .upsert(
+                            { externalId, title: "Post title initial" },
+                            {
+                                conflictPaths: ["externalId"],
+                                indexPredicate: "dateAdded > 2020-01-01",
+                            },
+                        )
+                        .should.be.rejectedWith(TypeORMError)
+                }),
+            ))
     })
 
     describe("preload also should also implement merge functionality", function () {