thinkphp下mysql迁移至瀚高（postgreSQL）数据库

国产化需要将thinkphp系统由mysql迁移至瀚高数据库。

基本流程：

1. 备份log相关表；清理log无用日志；
2. 停用系统；
3. 导出最新的sql文件（log等大表可考虑单独导出）；
4. 导入本地mysql数据库（如不支持远程连接）；
5. 处理相关代码与数据表（如处理user、desc、year、group字段等；处理group by、order by field、find_in_set、ifnull、convert、date_format方法等）；
6. 创建远程hg数据库
7. 使用迁移工具迁移

问题处理

字段处理

• 调整char为varchar
• 调整year为int
• 调整datetime为timestamp without time zone
• 调整smallint为int（pg无unsigned）

修改数据库字符存储格式

show nls_length_semantics;
alter system set nls_length_semantics=char;
select pg_reload_conf();

导入pgsql.sql

CREATE OR REPLACE FUNCTION pgsql_type(a_type varchar) RETURNS varchar AS
$BODY$
DECLARE
     v_type varchar;
BEGIN
     IF a_type='int8' THEN
          v_type:='bigint';
     ELSIF a_type='int4' THEN
          v_type:='integer';
     ELSIF a_type='int2' THEN
          v_type:='smallint';
     ELSIF a_type='bpchar' THEN
          v_type:='char';
     ELSE
          v_type:=a_type;
     END IF;
     RETURN v_type;
END;
$BODY$
LANGUAGE PLPGSQL;

CREATE TYPE "public"."tablestruct" AS (
  "fields_key_name" varchar(100),
  "fields_name" VARCHAR(200),
  "fields_type" VARCHAR(20),
  "fields_length" BIGINT,
  "fields_not_null" VARCHAR(10),
  "fields_default" VARCHAR(500),
  "fields_comment" VARCHAR(1000)
);

CREATE OR REPLACE FUNCTION "public"."table_msg" (a_schema_name varchar, a_table_name varchar) RETURNS SETOF "public"."tablestruct" AS
$body$
DECLARE
     v_ret tablestruct;
     v_oid oid;
     v_sql varchar;
     v_rec RECORD;
     v_key varchar;
BEGIN
     SELECT
           pg_class.oid  INTO v_oid
     FROM
           pg_class
           INNER JOIN pg_namespace ON (pg_class.relnamespace = pg_namespace.oid AND lower(pg_namespace.nspname) = a_schema_name)
     WHERE
           pg_class.relname=a_table_name;
     IF NOT FOUND THEN
         RETURN;
     END IF;

     v_sql='
     SELECT
           pg_attribute.attname AS fields_name,
           pg_attribute.attnum AS fields_index,
           pgsql_type(pg_type.typname::varchar) AS fields_type,
           pg_attribute.atttypmod-4 as fields_length,
           CASE WHEN pg_attribute.attnotnull  THEN ''not null''
           ELSE ''''
           END AS fields_not_null,
           pg_get_expr(pg_attrdef.adbin, pg_attribute.attrelid) AS fields_default,
           pg_description.description AS fields_comment
     FROM
           pg_attribute
           INNER JOIN pg_class  ON pg_attribute.attrelid = pg_class.oid
           INNER JOIN pg_type   ON pg_attribute.atttypid = pg_type.oid
           LEFT OUTER JOIN pg_attrdef ON pg_attrdef.adrelid = pg_class.oid AND pg_attrdef.adnum = pg_attribute.attnum
           LEFT OUTER JOIN pg_description ON pg_description.objoid = pg_class.oid AND pg_description.objsubid = pg_attribute.attnum
     WHERE
           pg_attribute.attnum > 0
           AND attisdropped <> ''t''
           AND pg_class.oid = ' || v_oid || '
     ORDER BY pg_attribute.attnum' ;

     FOR v_rec IN EXECUTE v_sql LOOP
         v_ret.fields_name=v_rec.fields_name;
         v_ret.fields_type=v_rec.fields_type;
         IF v_rec.fields_length > 0 THEN
            v_ret.fields_length:=v_rec.fields_length;
         ELSE
            v_ret.fields_length:=NULL;
         END IF;
         v_ret.fields_not_null=v_rec.fields_not_null;
         v_ret.fields_default=v_rec.fields_default;
         v_ret.fields_comment=v_rec.fields_comment;
         SELECT constraint_name INTO v_key FROM information_schema.key_column_usage WHERE table_schema=a_schema_name AND table_name=a_table_name AND column_name=v_rec.fields_name;
         IF FOUND THEN
            v_ret.fields_key_name=v_key;
         ELSE
            v_ret.fields_key_name='';
         END IF;
         RETURN NEXT v_ret;
     END LOOP;
     RETURN ;
END;
$body$
LANGUAGE 'plpgsql' VOLATILE CALLED ON NULL INPUT SECURITY INVOKER;

COMMENT ON FUNCTION "public"."table_msg"(a_schema_name varchar, a_table_name varchar)
IS '获得表信息';

---重载一个函数
CREATE OR REPLACE FUNCTION "public"."table_msg" (a_table_name varchar) RETURNS SETOF "public"."tablestruct" AS
$body$
DECLARE
    v_ret tablestruct;
BEGIN
    FOR v_ret IN SELECT * FROM table_msg('public',a_table_name) LOOP
        RETURN NEXT v_ret;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql' VOLATILE CALLED ON NULL INPUT SECURITY INVOKER;

COMMENT ON FUNCTION "public"."table_msg"(a_table_name varchar)
IS '获得表信息';


###导入find_in_set
---创建函数
CREATE OR replace function generate_subscripts(anyarray,int)

RETURNS SETOF int AS $$

begin

SELECT generate_series(array_lower($1,$2), array_upper($1,$2));

end

$$ LANGUAGE plpgsql;


CREATE OR REPLACE FUNCTION find_in_set(str text, strlist text)

RETURNS bool AS $$

  SELECT i>0

  FROM generate_subscripts(string_to_array($2,','),1) g(i)

  WHERE (string_to_array($2, ','))[i] = $1

  UNION ALL

  SELECT false

  LIMIT 1

$$ LANGUAGE sql STRICT;


### field函数
-- FUNCTION: public.field(text, text[])

-- DROP FUNCTION IF EXISTS public.field(text, text[]);

CREATE OR REPLACE FUNCTION public.field(
    text,
    VARIADIC text[])
    RETURNS bigint
    LANGUAGE 'plpgsql'
    COST 100
    VOLATILE PARALLEL UNSAFE
AS $BODY$
  DECLARE
    value bigint;
  BEGIN
    value = field_func($1,$2);
    IF value IS NOT NULL THEN
      RETURN value;
    ELSE
      RETURN 0;
    END IF;
END; 
$BODY$;

ALTER FUNCTION public.field(text, text[])
    OWNER TO sysdba;

thinkphp新增函数支持

vendor/topthink/think-orm/src/db/builder/Pgsql.php

<?php
// +----------------------------------------------------------------------
// | ThinkPHP [ WE CAN DO IT JUST THINK ]
// +----------------------------------------------------------------------
// | Copyright (c) 2006~2019 http://thinkphp.cn All rights reserved.
// +----------------------------------------------------------------------
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
// +----------------------------------------------------------------------
// | Author: liu21st <liu21st@gmail.com>
// +----------------------------------------------------------------------
declare (strict_types = 1);

namespace think\db\builder;

use think\db\Builder;
use think\db\Query;
use think\db\Raw;

/**
 * Pgsql数据库驱动
 */
class Pgsql extends Builder
{
    /**
     * 查询表达式解析
     * @var array
     */
    protected $parser = [
        'parseCompare'     => ['=', '<>', '>', '>=', '<', '<='],
        'parseLike'        => ['LIKE', 'NOT LIKE'],
        'parseBetween'     => ['NOT BETWEEN', 'BETWEEN'],
        'parseIn'          => ['NOT IN', 'IN'],
        'parseExp'         => ['EXP'],
        'parseNull'        => ['NOT NULL', 'NULL'],
        'parseBetweenTime' => ['BETWEEN TIME', 'NOT BETWEEN TIME'],
        'parseTime'        => ['< TIME', '> TIME', '<= TIME', '>= TIME'],
        'parseExists'      => ['NOT EXISTS', 'EXISTS'],
        'parseColumn'      => ['COLUMN'],
        'parseFindInSet'   => ['FIND IN SET'],
    ];

    /**
     * INSERT SQL表达式
     * @var string
     */
    protected $insertSql = 'INSERT INTO %TABLE% (%FIELD%) VALUES (%DATA%) %COMMENT%';

    /**
     * INSERT ALL SQL表达式
     * @var string
     */
    protected $insertAllSql = 'INSERT INTO %TABLE% (%FIELD%) VALUES %DATA% %COMMENT%';

    /**
     * limit分析
     * @access protected
     * @param  Query     $query        查询对象
     * @param  mixed     $limit
     * @return string
     */
    public function parseLimit(Query $query, string $limit): string
    {
        $limitStr = '';

        if (!empty($limit)) {
            $limit = explode(',', $limit);
            if (count($limit) > 1) {
                $limitStr .= ' LIMIT ' . $limit[1] . ' OFFSET ' . $limit[0] . ' ';
            } else {
                $limitStr .= ' LIMIT ' . $limit[0] . ' ';
            }
        }

        return $limitStr;
    }

    /**
     * 字段和表名处理
     * @access public
     * @param  Query     $query     查询对象
     * @param  mixed     $key       字段名
     * @param  bool      $strict   严格检测
     * @return string
     */
    public function parseKey(Query $query, $key, bool $strict = false): string
    {
        if (is_int($key)) {
            return (string) $key;
        } elseif ($key instanceof Raw) {
            return $this->parseRaw($query, $key);
        }

        $key = trim($key);

        if (strpos($key, '->') && false === strpos($key, '(')) {
            // JSON字段支持
            [$field, $name] = explode('->', $key);
            $key            = '"' . $field . '"' . '->>\'' . $name . '\'';
        } elseif (strpos($key, '.')) {
            [$table, $key] = explode('.', $key, 2);

            $alias = $query->getOptions('alias');

            if ('__TABLE__' == $table) {
                $table = $query->getOptions('table');
                $table = is_array($table) ? array_shift($table) : $table;
            }

            if (isset($alias[$table])) {
                $table = $alias[$table];
            }

            if ('*' != $key && !preg_match('/[,\"\*\(\).\s]/', $key)) {
                $key = '"' . $key . '"';
            }
        }

        if (isset($table)) {
            $key = $table . '.' . $key;
        }

        return $key;
    }

    /**
     * 随机排序
     * @access protected
     * @param  Query     $query        查询对象
     * @return string
     */
    protected function parseRand(Query $query): string
    {
        return 'RANDOM()';
    }

    /**
     * FIND_IN_SET 查询
     * @access protected
     * @param  Query        $query        查询对象
     * @param  string       $key
     * @param  string       $exp
     * @param  mixed        $value
     * @param  string       $field
     * @return string
     */
    protected function parseFindInSet(Query $query, string $key, string $exp, $value, string $field): string
    {
        if ($value instanceof Raw) {
            $value = $this->parseRaw($query, $value);
        }
        return 'FIND_IN_SET(' . $value . ', ' . $key . ')';
    }

    /**
     * 生成insertall SQL
     * @access public
     * @param  Query     $query   查询对象
     * @param  array     $dataSet 数据集
     * @param  bool      $replace 是否replace
     * @return string
     */
    public function insertAll(Query $query, array $dataSet, bool $replace = false): string
    {
        $options = $query->getOptions();

        // 获取绑定信息
        $bind = $query->getFieldsBindType();

        // 获取合法的字段
        if (empty($options['field']) || '*' == $options['field']) {
            $allowFields = array_keys($bind);
        } else {
            $allowFields = $options['field'];
        }

        $fields = [];
        $values = [];

        foreach ($dataSet as $data) {
            $data = $this->parseData($query, $data, $allowFields, $bind);

            $values[] = '( ' . implode(',', array_values($data)) . ' )';

            if (!isset($insertFields)) {
                $insertFields = array_keys($data);
            }
        }

        foreach ($insertFields as $field) {
            $fields[] = $this->parseKey($query, $field);
        }

        return str_replace(
            ['%INSERT%', '%TABLE%', '%FIELD%', '%DATA%', '%COMMENT%'],
            [
                $replace ? 'REPLACE' : 'INSERT',
                $this->parseTable($query, $options['table']),
                implode(' , ', $fields),
                implode(' , ', $values),
                $this->parseComment($query, $options['comment']),
            ],
            $this->insertAllSql);
    }

}

修改vendor\topthink\think-orm\src\db\connector\Pgsql.php支持varchar字段默认值为NULL（建议统一调整为空字符串）

...
'autoinc' => $val['extra']?(0 === strpos($val['extra'], 'nextval(')):false,
...

解决默认查询排序问题，默认以 id asc

如order('sort_num desc')调整为order('sort_num desc, id asc')

vendor\topthink\think-orm\src\db\BaseQuery.php对应的find、select方法添加：

...select
if(empty($this->options['order'])){
    $this->order('id', 'asc');
}
$resultSet = $this->connection->select($this);

...find
if (empty($this->options['where']) && empty($this->options['order'])) {
    $result = [];
} else {
    //如不存在排序则默认以id正序
    if(empty($this->options['order'])){
        $this->order('id', 'asc');
    }
    $result = $this->connection->find($this);
}

处理getColumnComment方法

/**
 * 获取字段注释
 * @param string $column_name 字段名
 * @param string $table_name 表名
 * @param string $table_schema 数据库名
 */
public function getColumnComment($column_name='', $table_name='', $table_schema='')
{
    $table_schema = $table_schema?:env('DATABASE.DATABASE');
    $data = Db::query("select '{$table_schema}' AS table_schema,
                        c.relname AS table_name,
                        a.attname AS column_name,
                        d.description AS column_comment
                        from pg_class c, pg_attribute a , pg_type t, pg_description d 
                        where c.relname = '{$table_name}' 
                        and a.attname = '{$column_name}' 
                        and a.attrelid = c.oid 
                        and a.atttypid = t.oid 
                        and d.objoid = a.attrelid
                        and d.objsubid = a.attnum;");
    if(!$data){
        $data[] = [
            'table_schema'=>$table_schema,
            'table_name'=>$table_name,
            'column_name'=>$column_name,
            'column_comment'=>''
        ];
    }
    return $data;
}