typeorm / typeorm / 22160751351

import { ObjectLiteral } from "../../common/ObjectLiteral"

import { TypeORMError } from "../../error"

import { QueryFailedError } from "../../error/QueryFailedError"

import { QueryRunnerAlreadyReleasedError } from "../../error/QueryRunnerAlreadyReleasedError"

import { TransactionNotStartedError } from "../../error/TransactionNotStartedError"

import { ReadStream } from "../../platform/PlatformTools"

import { BaseQueryRunner } from "../../query-runner/BaseQueryRunner"

import { QueryResult } from "../../query-runner/QueryResult"

import { QueryRunner } from "../../query-runner/QueryRunner"

import { TableIndexOptions } from "../../schema-builder/options/TableIndexOptions"

import { Table } from "../../schema-builder/table/Table"

import { TableCheck } from "../../schema-builder/table/TableCheck"

import { TableColumn } from "../../schema-builder/table/TableColumn"

import { TableExclusion } from "../../schema-builder/table/TableExclusion"

import { TableForeignKey } from "../../schema-builder/table/TableForeignKey"

import { TableIndex } from "../../schema-builder/table/TableIndex"

import { TableUnique } from "../../schema-builder/table/TableUnique"

import { View } from "../../schema-builder/view/View"

import { Broadcaster } from "../../subscriber/Broadcaster"

import { BroadcasterResult } from "../../subscriber/BroadcasterResult"

import { InstanceChecker } from "../../util/InstanceChecker"

import { OrmUtils } from "../../util/OrmUtils"

import { VersionUtils } from "../../util/VersionUtils"

import { DriverUtils } from "../DriverUtils"

import { Query } from "../Query"

import { ColumnType } from "../types/ColumnTypes"

import { IsolationLevel } from "../types/IsolationLevel"

import { MetadataTableType } from "../types/MetadataTableType"

import { ReplicationMode } from "../types/ReplicationMode"

import { PostgresDriver } from "./PostgresDriver"

/**

 * Runs queries on a single postgres database connection.

 */

export class PostgresQueryRunner

    extends BaseQueryRunner

    implements QueryRunner

{

    // -------------------------------------------------------------------------

    // Public Implemented Properties

    // -------------------------------------------------------------------------

    /**

     * Database driver used by connection.

     */

    driver: PostgresDriver

    // -------------------------------------------------------------------------

    // Protected Properties

    // -------------------------------------------------------------------------

    /**

     * Promise used to obtain a database connection for a first time.

     */

    protected databaseConnectionPromise: Promise<any>

    /**

     * Special callback provided by a driver used to release a created connection.

     */

    protected releaseCallback?: (err: any) => void

    // -------------------------------------------------------------------------

    // Constructor

    // -------------------------------------------------------------------------

    constructor(driver: PostgresDriver, mode: ReplicationMode) {

        super()

        this.driver = driver

        this.connection = driver.connection

        this.mode = mode

        this.broadcaster = new Broadcaster(this)

    }

    // -------------------------------------------------------------------------

    // Public Methods

    // -------------------------------------------------------------------------

    /**

     * Creates/uses database connection from the connection pool to perform further operations.

     * Returns obtained database connection.

     */

    connect(): Promise<any> {

        if (this.databaseConnection)

            return Promise.resolve(this.databaseConnection)

        if (this.databaseConnectionPromise)

            return this.databaseConnectionPromise

        if (this.mode === "slave" && this.driver.isReplicated) {

            this.databaseConnectionPromise = this.driver

                .obtainSlaveConnection()

                .then(([connection, release]: any[]) => {

                    this.driver.connectedQueryRunners.push(this)

                    this.databaseConnection = connection

                    const onErrorCallback = (err: Error) =>

                    this.releaseCallback = (err?: Error) => {

                        this.databaseConnection.removeListener(

                            "error",

                            onErrorCallback,

                        )

                        release(err)

                    }

                    this.databaseConnection.on("error", onErrorCallback)

                    return this.databaseConnection

                })

        } else {

            // master

            this.databaseConnectionPromise = this.driver

                .obtainMasterConnection()

                .then(([connection, release]: any[]) => {

                    this.driver.connectedQueryRunners.push(this)

                    this.databaseConnection = connection

                    const onErrorCallback = (err: Error) =>

                    this.releaseCallback = (err?: Error) => {

                        this.databaseConnection.removeListener(

                            "error",

                            onErrorCallback,

                        )

                        release(err)

                    }

                    this.databaseConnection.on("error", onErrorCallback)

                    return this.databaseConnection

                })

        }

        return this.databaseConnectionPromise

    }

    /**

     * Release a connection back to the pool, optionally specifying an Error to release with.

     * Per pg-pool documentation this will prevent the pool from re-using the broken connection.

     * @param err

     */

    private async releasePostgresConnection(err?: Error) {

        this.isReleased = true

        if (this.releaseCallback) {

            this.releaseCallback(err)

            this.releaseCallback = undefined

        }

        const index = this.driver.connectedQueryRunners.indexOf(this)

        if (index !== -1) {

            this.driver.connectedQueryRunners.splice(index, 1)

        }

    }

    /**

     * Releases used database connection.

     * You cannot use query runner methods once its released.

     */

    release(): Promise<void> {

        return this.releasePostgresConnection()

    }

    /**

     * Starts transaction.

     * @param isolationLevel

     */

    async startTransaction(isolationLevel?: IsolationLevel): Promise<void> {

        this.isTransactionActive = true

        try {

            await this.broadcaster.broadcast("BeforeTransactionStart")

        if (this.transactionDepth === 0) {

            await this.query("START TRANSACTION")

            if (isolationLevel) {

                await this.query(

                    "SET TRANSACTION ISOLATION LEVEL " + isolationLevel,

                )

            }

        } else {

            await this.query(`SAVEPOINT typeorm_${this.transactionDepth}`)

        }

        this.transactionDepth += 1

        await this.broadcaster.broadcast("AfterTransactionStart")

    }

    /**

     * Commits transaction.

     * Error will be thrown if transaction was not started.

     */

    async commitTransaction(): Promise<void> {

        await this.broadcaster.broadcast("BeforeTransactionCommit")

        if (this.transactionDepth > 1) {

            await this.query(

                `RELEASE SAVEPOINT typeorm_${this.transactionDepth - 1}`,

            )

        } else {

            await this.query("COMMIT")

            this.isTransactionActive = false

        }

        this.transactionDepth -= 1

        await this.broadcaster.broadcast("AfterTransactionCommit")

    }

    /**

     * Rollbacks transaction.

     * Error will be thrown if transaction was not started.

     */

    async rollbackTransaction(): Promise<void> {

        await this.broadcaster.broadcast("BeforeTransactionRollback")

        if (this.transactionDepth > 1) {

            await this.query(

                `ROLLBACK TO SAVEPOINT typeorm_${this.transactionDepth - 1}`,

            )

        } else {

            await this.query("ROLLBACK")

            this.isTransactionActive = false

        }

        this.transactionDepth -= 1

        await this.broadcaster.broadcast("AfterTransactionRollback")

    }

    /**

     * Executes a given SQL query.

     * @param query

     * @param parameters

     * @param useStructuredResult

     */

    async query(

        query: string,

        parameters?: any[],

        useStructuredResult: boolean = false,

    ): Promise<any> {

        const databaseConnection = await this.connect()

        this.driver.connection.logger.logQuery(query, parameters, this)

        await this.broadcaster.broadcast("BeforeQuery", query, parameters)

        const broadcasterResult = new BroadcasterResult()

        try {

            const queryStartTime = Date.now()

            const raw = await databaseConnection.query(query, parameters)

            // log slow queries if maxQueryExecution time is set

            const maxQueryExecutionTime =

                this.driver.options.maxQueryExecutionTime

            const queryEndTime = Date.now()

            const queryExecutionTime = queryEndTime - queryStartTime

            this.broadcaster.broadcastAfterQueryEvent(

                broadcasterResult,

                query,

                parameters,

                true,

                queryExecutionTime,

                raw,

                undefined,

            )

            if (

            )

            const result = new QueryResult()

            if (raw) {

                if (raw.hasOwnProperty("rows")) {

                    result.records = raw.rows

                }

                if (raw.hasOwnProperty("rowCount")) {

                    result.affected = raw.rowCount

                }

                switch (raw.command) {

                    case "DELETE":

                    case "UPDATE":

                        // for UPDATE and DELETE query additionally return number of affected rows

                        result.raw = [raw.rows, raw.rowCount]

                        break

                    default:

                        result.raw = raw.rows

                }

                if (!useStructuredResult) {

                    return result.raw

                }

            }

            return result

        } catch (err) {

            this.driver.connection.logger.logQueryError(

                err,

                query,

                parameters,

                this,

            )

            this.broadcaster.broadcastAfterQueryEvent(

                broadcasterResult,

                query,

                parameters,

                false,

                undefined,

                undefined,

                err,

            )

            throw new QueryFailedError(query, parameters, err)

            await broadcasterResult.wait()

        }

    }

    /**

     * Returns raw data stream.

     * @param query

     * @param parameters

     * @param onEnd

     * @param onError

     */

    async stream(

        query: string,

        parameters?: any[],

        onEnd?: Function,

        onError?: Function,

    ): Promise<ReadStream> {

        const QueryStream = this.driver.loadStreamDependency()

        const databaseConnection = await this.connect()

        this.driver.connection.logger.logQuery(query, parameters, this)

        const stream = databaseConnection.query(

            new QueryStream(query, parameters),

        )

        if (onEnd) stream.on("end", onEnd)

        if (onError) stream.on("error", onError)

        return stream

    }

    /**

     * Returns all available database names including system databases.

     */

    async getDatabases(): Promise<string[]> {

    /**

     * Returns all available schema names including system schemas.

     * If database parameter specified, returns schemas of that database.

     * @param database

     */

    async getSchemas(database?: string): Promise<string[]> {

    /**

     * Checks if database with the given name exist.

     * @param database

     */

    async hasDatabase(database: string): Promise<boolean> {

        const result = await this.query(

            `SELECT * FROM pg_database WHERE datname='${database}';`,

        )

        return result.length ? true : false

    }

    /**

     * Loads currently using database

     */

    async getCurrentDatabase(): Promise<string> {

        const query = await this.query(`SELECT * FROM current_database()`)

        return query[0]["current_database"]

    }

    /**

     * Checks if schema with the given name exist.

     * @param schema

     */

    async hasSchema(schema: string): Promise<boolean> {

        const result = await this.query(

            `SELECT * FROM "information_schema"."schemata" WHERE "schema_name" = '${schema}'`,

        )

        return result.length ? true : false

    }

    /**

     * Loads currently using database schema

     */

    async getCurrentSchema(): Promise<string> {

        const query = await this.query(`SELECT * FROM current_schema()`)

        return query[0]["current_schema"]

    }

    /**

     * Checks if table with the given name exist in the database.

     * @param tableOrName

     */

    async hasTable(tableOrName: Table | string): Promise<boolean> {

        const parsedTableName = this.driver.parseTableName(tableOrName)

        const sql = `SELECT * FROM "information_schema"."tables" WHERE "table_schema" = '${parsedTableName.schema}' AND "table_name" = '${parsedTableName.tableName}'`

        const result = await this.query(sql)

        return result.length ? true : false

    }

    /**

     * Checks if column with the given name exist in the given table.

     * @param tableOrName

     * @param columnName

     */

    async hasColumn(

        tableOrName: Table | string,

        columnName: string,

    ): Promise<boolean> {

        const parsedTableName = this.driver.parseTableName(tableOrName)

        const sql = `SELECT * FROM "information_schema"."columns" WHERE "table_schema" = '${parsedTableName.schema}' AND "table_name" = '${parsedTableName.tableName}' AND "column_name" = '${columnName}'`

        const result = await this.query(sql)

        return result.length ? true : false

    }

    /**

     * Creates a new database.

     * Note: Postgres does not support database creation inside a transaction block.

     * @param database

     * @param ifNotExist

     */

    async createDatabase(

        database: string,

        ifNotExist?: boolean,

    ): Promise<void> {

        if (ifNotExist) {

            const databaseAlreadyExists = await this.hasDatabase(database)

        }

        const up = `CREATE DATABASE "${database}"`

        const down = `DROP DATABASE "${database}"`

        await this.executeQueries(new Query(up), new Query(down))

    }

    /**

     * Drops database.

     * Note: Postgres does not support database dropping inside a transaction block.

     * @param database

     * @param ifExist

     */

    async dropDatabase(database: string, ifExist?: boolean): Promise<void> {

        const up = ifExist

            ? `DROP DATABASE IF EXISTS "${database}"`

            : `DROP DATABASE "${database}"`

        const down = `CREATE DATABASE "${database}"`

        await this.executeQueries(new Query(up), new Query(down))

    }

    /**

     * Creates a new table schema.

     * @param schemaPath

     * @param ifNotExist

     */

    async createSchema(

        schemaPath: string,

        ifNotExist?: boolean,

    ): Promise<void> {

        const schema =

            schemaPath.indexOf(".") === -1

                ? schemaPath

        const up = ifNotExist

            ? `CREATE SCHEMA IF NOT EXISTS "${schema}"`

            : `CREATE SCHEMA "${schema}"`

        const down = `DROP SCHEMA "${schema}" CASCADE`

        await this.executeQueries(new Query(up), new Query(down))

    }

    /**

     * Drops table schema.

     * @param schemaPath

     * @param ifExist

     * @param isCascade

     */

    async dropSchema(

        schemaPath: string,

        ifExist?: boolean,

        isCascade?: boolean,

    ): Promise<void> {

        const schema =

            schemaPath.indexOf(".") === -1

                ? schemaPath

        const up = ifExist

        const down = `CREATE SCHEMA "${schema}"`

        await this.executeQueries(new Query(up), new Query(down))

    }

    /**

     * Creates a new table.

     * @param table

     * @param ifNotExist

     * @param createForeignKeys

     * @param createIndices

     */

    async createTable(

        table: Table,

        ifNotExist: boolean = false,

        createForeignKeys: boolean = true,

        createIndices: boolean = true,

    ): Promise<void> {

        if (ifNotExist) {

            const isTableExist = await this.hasTable(table)

            if (isTableExist) return Promise.resolve()

        }

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        // if table have column with ENUM type, we must create this type in postgres.

        const enumColumns = table.columns.filter(

            (column) => column.type === "enum" || column.type === "simple-enum",

        )

        const createdEnumTypes: string[] = []

        for (const column of enumColumns) {

            // TODO: Should also check if values of existing type matches expected ones

            const hasEnum = await this.hasEnumType(table, column)

            const enumName = this.buildEnumName(table, column)

            // if enum with the same "enumName" is defined more then once, me must prevent double creation

            if (!hasEnum && createdEnumTypes.indexOf(enumName) === -1) {

                createdEnumTypes.push(enumName)

                upQueries.push(this.createEnumTypeSql(table, column, enumName))

                downQueries.push(this.dropEnumTypeSql(table, column, enumName))

            }

        }

        // if table have column with generated type, we must add the expression to the metadata table

        const generatedColumns = table.columns.filter(

            (column) =>

                column.generatedType === "STORED" && column.asExpression,

        )

        for (const column of generatedColumns) {

            const tableNameWithSchema = (

                await this.getTableNameWithSchema(table.name)

            ).split(".")

            const tableName = tableNameWithSchema[1]

            const schema = tableNameWithSchema[0]

            const insertQuery = this.insertTypeormMetadataSql({

                database: this.driver.database,

                schema,

                table: tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

                value: column.asExpression,

            })

            const deleteQuery = this.deleteTypeormMetadataSql({

                database: this.driver.database,

                schema,

                table: tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

            })

            upQueries.push(insertQuery)

            downQueries.push(deleteQuery)

        }

        upQueries.push(this.createTableSql(table, createForeignKeys))

        downQueries.push(this.dropTableSql(table))

        // if createForeignKeys is true, we must drop created foreign keys in down query.

        // createTable does not need separate method to create foreign keys, because it create fk's in the same query with table creation.

        if (createForeignKeys)

            table.foreignKeys.forEach((foreignKey) =>

                downQueries.push(this.dropForeignKeySql(table, foreignKey)),

            )

        if (createIndices) {

            table.indices.forEach((index) => {

                // new index may be passed without name. In this case we generate index name manually.

                if (!index.name)

                    index.name = this.connection.namingStrategy.indexName(

                        table,

                        index.columnNames,

                        index.where,

                    )

                upQueries.push(this.createIndexSql(table, index))

                downQueries.push(this.dropIndexSql(table, index))

            })

        }

        if (table.comment) {

            upQueries.push(

                new Query(

                    `COMMENT ON TABLE ${this.escapePath(table)}` +

                        ` IS ${this.escapeComment(table.comment)}`,

                ),

            )

            downQueries.push(

                new Query(

                    `COMMENT ON TABLE ${this.escapePath(table)}` + ` IS NULL`,

                ),

            )

        }

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Drops the table.

     * @param target

     * @param ifExist

     * @param dropForeignKeys

     * @param dropIndices

     */

    async dropTable(

        target: Table | string,

        ifExist?: boolean,

        dropForeignKeys: boolean = true,

        dropIndices: boolean = true,

    ): Promise<void> {

        // It needs because if table does not exist and dropForeignKeys or dropIndices is true, we don't need

        // to perform drop queries for foreign keys and indices.

        if (ifExist) {

            const isTableExist = await this.hasTable(target)

        }

        // if dropTable called with dropForeignKeys = true, we must create foreign keys in down query.

        const createForeignKeys: boolean = dropForeignKeys

        const tablePath = this.getTablePath(target)

        const table = await this.getCachedTable(tablePath)

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        if (dropIndices) {

            table.indices.forEach((index) => {

                upQueries.push(this.dropIndexSql(table, index))

                downQueries.push(this.createIndexSql(table, index))

            })

        }

        if (dropForeignKeys)

            table.foreignKeys.forEach((foreignKey) =>

                upQueries.push(this.dropForeignKeySql(table, foreignKey)),

            )

        upQueries.push(this.dropTableSql(table))

        downQueries.push(this.createTableSql(table, createForeignKeys))

        // if table had columns with generated type, we must remove the expression from the metadata table

        const generatedColumns = table.columns.filter(

            (column) => column.generatedType && column.asExpression,

        )

        for (const column of generatedColumns) {

            const tableNameWithSchema = (

                await this.getTableNameWithSchema(table.name)

            ).split(".")

            const tableName = tableNameWithSchema[1]

            const schema = tableNameWithSchema[0]

            const deleteQuery = this.deleteTypeormMetadataSql({

                database: this.driver.database,

                schema,

                table: tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

            })

            const insertQuery = this.insertTypeormMetadataSql({

                database: this.driver.database,

                schema,

                table: tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

                value: column.asExpression,

            })

            upQueries.push(deleteQuery)

            downQueries.push(insertQuery)

        }

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Creates a new view.

     * @param view

     * @param syncWithMetadata

     */

    async createView(

        view: View,

        syncWithMetadata: boolean = false,

    ): Promise<void> {

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        upQueries.push(this.createViewSql(view))

        if (syncWithMetadata)

            upQueries.push(await this.insertViewDefinitionSql(view))

        downQueries.push(this.dropViewSql(view))

        if (syncWithMetadata)

            downQueries.push(await this.deleteViewDefinitionSql(view))

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Drops the view.

     * @param target

     */

    async dropView(target: View | string): Promise<void> {

        const view = await this.getCachedView(viewName)

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        upQueries.push(await this.deleteViewDefinitionSql(view))

        upQueries.push(this.dropViewSql(view))

        downQueries.push(await this.insertViewDefinitionSql(view))

        downQueries.push(this.createViewSql(view))

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Renames the given table.

     * @param oldTableOrName

     * @param newTableName

     */

    async renameTable(

        oldTableOrName: Table | string,

        newTableName: string,

    ): Promise<void> {

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        const oldTable = InstanceChecker.isTable(oldTableOrName)

            ? oldTableOrName

            : await this.getCachedTable(oldTableOrName)

        const newTable = oldTable.clone()

        const { schema: schemaName, tableName: oldTableName } =

            this.driver.parseTableName(oldTable)

        newTable.name = schemaName

            ? `${schemaName}.${newTableName}`

        upQueries.push(

            new Query(

                `ALTER TABLE ${this.escapePath(

                    oldTable,

                )} RENAME TO "${newTableName}"`,

            ),

        )

        downQueries.push(

            new Query(

                `ALTER TABLE ${this.escapePath(

                    newTable,

                )} RENAME TO "${oldTableName}"`,

            ),

        )

        // rename column primary key constraint if it has default constraint name

        if (

            newTable.primaryColumns.length > 0 &&

            !newTable.primaryColumns[0].primaryKeyConstraintName

        ) {

            const columnNames = newTable.primaryColumns.map(

                (column) => column.name,

            )

            const oldPkName = this.connection.namingStrategy.primaryKeyName(

                oldTable,

                columnNames,

            )

            const newPkName = this.connection.namingStrategy.primaryKeyName(

                newTable,

                columnNames,

            )

            upQueries.push(

                new Query(

                    `ALTER TABLE ${this.escapePath(

                        newTable,

                    )} RENAME CONSTRAINT "${oldPkName}" TO "${newPkName}"`,

                ),

            )

            downQueries.push(

                new Query(

                    `ALTER TABLE ${this.escapePath(

                        newTable,

                    )} RENAME CONSTRAINT "${newPkName}" TO "${oldPkName}"`,

                ),

            )

        }

        // rename sequences

        newTable.columns.map((col) => {

            if (col.isGenerated && col.generationStrategy === "increment") {

                const sequencePath = this.buildSequencePath(oldTable, col.name)

                const sequenceName = this.buildSequenceName(oldTable, col.name)

                const newSequencePath = this.buildSequencePath(

                    newTable,

                    col.name,

                )

                const newSequenceName = this.buildSequenceName(

                    newTable,

                    col.name,

                )

                const up = `ALTER SEQUENCE ${this.escapePath(

                    sequencePath,

                )} RENAME TO "${newSequenceName}"`

                const down = `ALTER SEQUENCE ${this.escapePath(

                    newSequencePath,

                )} RENAME TO "${sequenceName}"`

                upQueries.push(new Query(up))

                downQueries.push(new Query(down))