src/Snowflake/ImportBase.php
private function insertOrUpdateTargetTable(string $stagingTableName, string $targetTableName, array $columns, bool $useTimestamp = true, array $convertEmptyValuesToNull = []): void
    {
        $this->connection->query('BEGIN TRANSACTION');
        $nowFormatted = $this->getNowFormatted();

        $targetTableNameWithSchema = $this->nameWithSchemaEscaped($targetTableName);
        $stagingTableNameWithSchema = $this->nameWithSchemaEscaped($stagingTableName);

        $primaryKey = $this->connection->getTablePrimaryKey($this->schemaName, $targetTableName);

        if (!empty($primaryKey)) {
            // Update target table
            $sql = "UPDATE " . $targetTableNameWithSchema . " AS \"dest\" SET ";

            $columnsSet = [];
            foreach ($columns as $columnName) {
                if (in_array($columnName, $convertEmptyValuesToNull)) {
                    $columnsSet[] = sprintf(
                        '%s = IFF("src".%s = \'\', NULL, "src".%s)',
                        $this->quoteIdentifier($columnName),
                        $this->quoteIdentifier($columnName),
                        $this->quoteIdentifier($columnName)
                    );
                } else {
                    $columnsSet[] = sprintf(
                        '%s = COALESCE("src".%s, \'\')',
                        $this->quoteIdentifier($columnName),
                        $this->quoteIdentifier($columnName)
                    );
                }
            }

            $sql .= implode(', ', $columnsSet);
            if ($useTimestamp) {
                $sql .= ", " . $this->quoteIdentifier(self::TIMESTAMP_COLUMN_NAME) . " = '{$nowFormatted}' ";
            }
            $sql .= " FROM " . $stagingTableNameWithSchema . ' AS "src" ';
            $sql .= " WHERE ";

            $pkWhereSql = [];
            foreach ($primaryKey as $pkColumn) {
                $pkWhereSql[] = sprintf(
                    '"dest".%s = COALESCE("src".%s, \'\')',
                    $this->quoteIdentifier($pkColumn),
                    $this->quoteIdentifier($pkColumn)
                );
            }

            $sql .= implode(' AND ', $pkWhereSql) . " ";

            // update only changed rows - mysql TIMESTAMP ON UPDATE behaviour simulation
            $columnsComparsionSql = array_map(function ($columnName) {
                return sprintf(
                    'COALESCE(TO_VARCHAR("dest".%s), \'\') != COALESCE("src".%s, \'\')',
                    $this->quoteIdentifier($columnName),
                    $this->quoteIdentifier($columnName)
                );
            }, $columns);
            $sql .= " AND (" . implode(' OR ', $columnsComparsionSql) . ") ";

            Debugger::timer('updateTargetTable');
            $this->connection->query($sql);
            $this->addTimer('updateTargetTable', Debugger::timer('updateTargetTable'));

            // Delete updated rows from staging table
            $sql = "DELETE FROM " . $stagingTableNameWithSchema . ' "src" ';
            $sql .= "USING " . $targetTableNameWithSchema . ' AS "dest" ';
            $sql .= "WHERE " . implode(' AND ', $pkWhereSql);

            Debugger::timer('deleteUpdatedRowsFromStaging');
            $this->connection->query($sql);
            $this->addTimer('deleteUpdatedRowsFromStaging', Debugger::timer('deleteUpdatedRowsFromStaging'));

            // Dedup staging table
            Debugger::timer('dedupStaging');
            $this->dedupe($stagingTableName, $columns, $primaryKey);
            $this->addTimer('dedupStaging', Debugger::timer('dedupStaging'));
        }

        // Insert from staging to target table
        $insColumns = ($useTimestamp) ? array_merge($columns, [self::TIMESTAMP_COLUMN_NAME]) : $columns;
        $sql = "INSERT INTO " . $targetTableNameWithSchema . ' (' . implode(', ', array_map(function ($column) {
            return $this->quoteIdentifier($column);
        }, $insColumns)) . ")";

        $columnsSetSql = [];

        foreach ($columns as $columnName) {
            if (in_array($columnName, $convertEmptyValuesToNull)) {
                $columnsSetSql[] = sprintf(
                    'IFF("src".%s = \'\', NULL, %s)',
                    $this->quoteIdentifier($columnName),
                    $this->quoteIdentifier($columnName)
                );
            } else {
                $columnsSetSql[] = sprintf(
                    'COALESCE("src".%s, \'\')',
                    $this->quoteIdentifier($columnName)
                );
            }
        }

        $sql .= " SELECT " . implode(',', $columnsSetSql);
        if ($useTimestamp) {
            $sql .= ", '{$nowFormatted}' ";
        }
        $sql .= " FROM " . $stagingTableNameWithSchema . ' AS "src"';
        Debugger::timer('insertIntoTargetFromStaging');

        $this->connection->query($sql);
        $this->addTimer('insertIntoTargetFromStaging', Debugger::timer('insertIntoTargetFromStaging'));

        $this->connection->query('COMMIT');
    }
private function insertOrUpdateTargetTable(string $stagingTableName, string $targetTableName, array $columns, bool $useTimestamp = true, array $convertEmptyValuesToNull = []): void
    {
        $this->connection->query('BEGIN TRANSACTION');
        $nowFormatted = $this->getNowFormatted();

        $targetTableNameWithSchema = $this->nameWithSchemaEscaped($targetTableName);
        $stagingTableNameWithSchema = $this->nameWithSchemaEscaped($stagingTableName);

        $primaryKey = $this->connection->getTablePrimaryKey($this->schemaName, $targetTableName);

        if (!empty($primaryKey)) {
            // Update target table
            $sql = "UPDATE " . $targetTableNameWithSchema . " AS \"dest\" SET ";

            $columnsSet = [];
            foreach ($columns as $columnName) {
                if (in_array($columnName, $convertEmptyValuesToNull)) {
                    $columnsSet[] = sprintf(
                        '%s = IFF("src".%s = \'\', NULL, "src".%s)',
                        $this->quoteIdentifier($columnName),
                        $this->quoteIdentifier($columnName),
                        $this->quoteIdentifier($columnName)
                    );
                } else {
                    $columnsSet[] = sprintf(
                        '%s = COALESCE("src".%s, \'\')',
                        $this->quoteIdentifier($columnName),
                        $this->quoteIdentifier($columnName)
                    );
                }
            }

            $sql .= implode(', ', $columnsSet);
            if ($useTimestamp) {
                $sql .= ", " . $this->quoteIdentifier(self::TIMESTAMP_COLUMN_NAME) . " = '{$nowFormatted}' ";
            }
            $sql .= " FROM " . $stagingTableNameWithSchema . ' AS "src" ';
            $sql .= " WHERE ";

            $pkWhereSql = [];
            foreach ($primaryKey as $pkColumn) {
                $pkWhereSql[] = sprintf(
                    '"dest".%s = COALESCE("src".%s, \'\')',
                    $this->quoteIdentifier($pkColumn),
                    $this->quoteIdentifier($pkColumn)
                );
            }

            $sql .= implode(' AND ', $pkWhereSql) . " ";

            // update only changed rows - mysql TIMESTAMP ON UPDATE behaviour simulation
            $columnsComparsionSql = array_map(function ($columnName) {
                return sprintf(
                    'COALESCE(TO_VARCHAR("dest".%s), \'\') != COALESCE("src".%s, \'\')',
                    $this->quoteIdentifier($columnName),
                    $this->quoteIdentifier($columnName)
                );
            }, $columns);
            $sql .= " AND (" . implode(' OR ', $columnsComparsionSql) . ") ";

            Debugger::timer('updateTargetTable');
            $this->connection->query($sql);
            $this->addTimer('updateTargetTable', Debugger::timer('updateTargetTable'));

            // Delete updated rows from staging table
            $sql = "DELETE FROM " . $stagingTableNameWithSchema . ' "src" ';
            $sql .= "USING " . $targetTableNameWithSchema . ' AS "dest" ';
            $sql .= "WHERE " . implode(' AND ', $pkWhereSql);

            Debugger::timer('deleteUpdatedRowsFromStaging');
            $this->connection->query($sql);
            $this->addTimer('deleteUpdatedRowsFromStaging', Debugger::timer('deleteUpdatedRowsFromStaging'));

            // Dedup staging table
            Debugger::timer('dedupStaging');
            $this->dedupe($stagingTableName, $columns, $primaryKey);
            $this->addTimer('dedupStaging', Debugger::timer('dedupStaging'));
        }

        // Insert from staging to target table
        $insColumns = ($useTimestamp) ? array_merge($columns, [self::TIMESTAMP_COLUMN_NAME]) : $columns;
        $sql = "INSERT INTO " . $targetTableNameWithSchema . ' (' . implode(', ', array_map(function ($column) {
            return $this->quoteIdentifier($column);
        }, $insColumns)) . ")";

        $columnsSetSql = [];

        foreach ($columns as $columnName) {
            if (in_array($columnName, $convertEmptyValuesToNull)) {
                $columnsSetSql[] = sprintf(
                    'IFF("src".%s = \'\', NULL, %s)',
                    $this->quoteIdentifier($columnName),
                    $this->quoteIdentifier($columnName)
                );
            } else {
                $columnsSetSql[] = sprintf(
                    'COALESCE("src".%s, \'\')',
                    $this->quoteIdentifier($columnName)
                );
            }
        }

        $sql .= " SELECT " . implode(',', $columnsSetSql);
        if ($useTimestamp) {
            $sql .= ", '{$nowFormatted}' ";
        }
        $sql .= " FROM " . $stagingTableNameWithSchema . ' AS "src"';
        Debugger::timer('insertIntoTargetFromStaging');

        $this->connection->query($sql);
        $this->addTimer('insertIntoTargetFromStaging', Debugger::timer('insertIntoTargetFromStaging'));

        $this->connection->query('COMMIT');
    }
Performs merge operation according to http://docs.aws.amazon.com/redshift/latest/dg/merge-specify-a-column-list.html