拓展 sequelize-typescript 库支持 Post

作者: Billlllllllly | 来源:发表于2019-10-09 13:13 被阅读0次

一、背景

sequelize是一个基于promise的关系型数据库Node.js ORM框架，提供了支持建立model结构，连接并封装数据库操作函数等功能，而sequelize-typescript是为了支持typescript语法而基于sequelize封装的框架。

由于项目前期阶段方便db维护，应用服务启动时会调用sequelize-typescript提供的sync函数根据model结构自动同步到数据库，使用这种方式同步普通表是没有很大问题的，但是目前sync函数不支持同步分区表，导致出现分区表只能另外手动创建和维护的情况。

所以为了解决上述这个问题，只能基于sequelize-typescript来实现分区表自动同步的功能了，下面开始讨论拓展sequelize-typescript使其支持分区表的方案和实现。

首先通过源码了解sequelize是怎样将model结构同步到数据库的，才能去对其进行改造，先看sequelize.sync函数的实现：

// 文件路径为<project_path>/node_modules/_sequelize@4.44.2@sequelize/lib/sequelize.js
// 路径中的"4.44.2"是sequelize-typescript库引用的sequelize版本号，可能sequelize-typescript版本不同使得引用的sequelize版本有所差异
sync(options) {
  options = _.clone(options) || {};
  options.hooks = options.hooks === undefined ? true : !!options.hooks;
  options = _.defaults(options, this.options.sync, this.options);
  if (options.match) {
    if (!options.match.test(this.config.database)) {
      return Promise.reject(new Error(`Database "${this.config.database}" does not match sync match parameter "${options.match}"`));
    }
  }
  return Promise.try(() => {
    if (options.hooks) {
      return this.runHooks('beforeBulkSync', options);
    }
  }).then(() => {
    if (options.force) {
      return this.drop(options);
    }
  }).then(() => {
    const models = [];

    // Topologically sort by foreign key constraints to give us an appropriate
    // creation order
    this.modelManager.forEachModel(model => {
      if (model) {
        models.push(model);
      } else {
        // DB should throw an SQL error if referencing inexistant table
      }
    });
    return Promise.each(models, model => model.sync(options));
  }).then(() => {
    if (options.hooks) {
      return this.runHooks('afterBulkSync', options);
    }
  }).return(this);
}

从以上代码可知，sequelize.sync函数中除了做一些前后置的hook操作外就是调用每个model的sync函数同步表结构，再贴下model.sync函数代码：

// 文件路径为<project_path>/node_modules/_sequelize@4.44.2@sequelize/lib/model.js
// 路径中的"4.44.2"是sequelize-typescript库引用的sequelize版本号，可能sequelize-typescript版本不同使得引用的sequelize版本有所差异
static sync(options) {
  options = _.extend({}, this.options, options);
  options.hooks = options.hooks === undefined ? true : !!options.hooks;

  const attributes = this.tableAttributes;

  return Promise.try(() => {
    if (options.hooks) {
      return this.runHooks('beforeSync', options);
    }
  }).then(() => {
    if (options.force) {
      return this.drop(options);
    }
  })
    .then(() => this.QueryInterface.createTable(this.getTableName(options), attributes, options, this))
    .then(() => {
    if (options.alter) {
      return Promise.all([
        this.QueryInterface.describeTable(this.getTableName(options)),
        this.QueryInterface.getForeignKeyReferencesForTable(this.getTableName(options))
      ])
        .then(tableInfos => {
        const columns = tableInfos[0];
        // Use for alter foreign keys
        const foreignKeyReferences = tableInfos[1];

        const changes = []; // array of promises to run
        const removedConstraints = {};

        _.each(attributes, (columnDesc, columnName) => {
          if (!columns[columnName]) {
            changes.push(() => this.QueryInterface.addColumn(this.getTableName(options), columnName, attributes[columnName]));
          }
        });
        _.each(columns, (columnDesc, columnName) => {
          const currentAttributes = attributes[columnName];
          if (!currentAttributes) {
            changes.push(() => this.QueryInterface.removeColumn(this.getTableName(options), columnName, options));
          } else if (!currentAttributes.primaryKey) {
            // Check foreign keys. If it's a foreign key, it should remove constraint first.
            const references = currentAttributes.references;
            if (currentAttributes.references) {
              const database = this.sequelize.config.database;
              const schema = this.sequelize.config.schema;
              // Find existed foreign keys
              _.each(foreignKeyReferences, foreignKeyReference => {
                const constraintName = foreignKeyReference.constraintName;
                if (!!constraintName
                    && foreignKeyReference.tableCatalog === database
                    && (schema ? foreignKeyReference.tableSchema === schema : true)
                    && foreignKeyReference.referencedTableName === references.model
                    && foreignKeyReference.referencedColumnName === references.key
                    && (schema ? foreignKeyReference.referencedTableSchema === schema : true)
                    && !removedConstraints[constraintName]) {
                  // Remove constraint on foreign keys.
                  changes.push(() => this.QueryInterface.removeConstraint(this.getTableName(options), constraintName, options));
                  removedConstraints[constraintName] = true;
                }
              });
            }
            changes.push(() => this.QueryInterface.changeColumn(this.getTableName(options), columnName, attributes[columnName]));
          }
        });
        return changes.reduce((p, fn) => p.then(fn), Promise.resolve());
      });
    }
  })
    .then(() => this.QueryInterface.showIndex(this.getTableName(options), options))
    .then(indexes => {
    // Assign an auto-generated name to indexes which are not named by the user
    this.options.indexes = this.QueryInterface.nameIndexes(this.options.indexes, this.tableName);

    indexes = _.filter(this.options.indexes, item1 =>
                       !_.some(indexes, item2 => item1.name === item2.name)
                      ).sort((index1, index2) => {
      if (this.sequelize.options.dialect === 'postgres') {
        // move concurrent indexes to the bottom to avoid weird deadlocks
        if (index1.concurrently === true) return 1;
        if (index2.concurrently === true) return -1;
      }

      return 0;
    });

    return Promise.map(indexes, index => this.QueryInterface.addIndex(
      this.getTableName(options),
      _.assign({
        logging: options.logging,
        benchmark: options.benchmark,
        transaction: options.transaction
      }, index),
      this.tableName
    ));
  }).then(() => {
    if (options.hooks) {
      return this.runHooks('afterSync', options);
    }
  }).return(this);
}

model.sync代码稍微有点长，其中大致执行流程如下：

model.sync执行流程.png

二、方案

理解model结构同步到数据库的实现过程后，先设计一种支持定义分区表的方式，我的方案是在定义表的配置项中新增几个新属性：partition、partitionKey、partitionRule分别表示分区类型、分区字段、分区规则。

其中partition支持RANGE/LIST/HASH三种类型；partitionRule是一个object， key是分区子表名后缀，且key数量即是分区子表数量，value可以分别是RANGE区间/LIST值列表/HASH映射值；生成的分区子表名就是由主表名+partitionRule中每个key生成。

然后需要重写分区表的sync函数，其中实现主要流程如下：

model.sync改进后的执行流程.png

因为拓展 sequelize 时使用的是 Postgresql 10.x 版本，我们需要先了解 Postgresql 10.x 和 Postgresql 11.x 版本所支持的分区表有几个区别：

Postgresql 10.x 只支持 RANGE 和 LIST 分区方式，而 Postgresql 11.x 除此外还支持HASH分区方式；
Postgresql 10.x 主表不支持创建主键/索引，必须单独操作每个子表的索引，而 Postgresql 11.x 可以直接操作主表的索引并会自动同步到每个子表；
Postgresql 10.x 没有默认分区，当插入数据找不到所对应的分区时会插入失败，而Postgresql 11.x 支持默认分区；
Postgresql 10.x 只支持一级分区，即不能对子表再次做分区，而Postgresql 11.x 支持多级分区；

三、实现

接下来会详细说明几个关键过程（在流程图中用深色表示）的具体实现，值得一提的是这里将在支持Postgresql 10.x版本前提下使用typescript语法重写model.sync函数。

以下为代码会分段贴出来讲解，完整代码在 github 上可以自取。

1. 新建分区主表

这一步需要新建分区主表而不是普通表，所以需要在原来创建表sql的基础上进行修改，具体实现是调用createTableQuery生成创建普通表sql后，在该sql末尾加上PARTITION BY <partition> (<partitionKey>)即可通过直接执行生成分区主表。

// 生成创建分区主表的SQL
const attrs = this.QueryInterface.QueryGenerator.attributesToSQL(attributes, {
  context: 'createTable',
});
let createTableSql = this.QueryInterface.QueryGenerator.createTableQuery(this.getTableName(options), attrs, options);
createTableSql = createTableSql.substring(0, createTableSql.length - 1) + ` PARTITION BY ${options.partition} ("${options.partitionKey}");`;
return this.sequelize.query(createTableSql, options);

但是这个环节需要注意的是，由于Postgresql 10.x版本的分区主表不支持创建主键和索引，所以生成创建分区主表sql之前需要先移除主键，并在分区子表中使用唯一索引替代主键。

// postgresql 10.x版本中分区主表不能有主键和索引
const primaryKey: string[] = [];
for (const name in attributes) {
  const attr = attributes[name];
  if (attr.primaryKey) {
    primaryKey.push(name);
    attr.allowNull = false;
    delete attr.primaryKey;
  }
  if (name === this.options.partitionKey) {
    prititionColumn = attributes[name];
  }
}
// 分区子表使用unique索引代替主键
if (!this.options.indexes) {
  this.options.indexes = [];
}
this.options.indexes.push({
  fields: primaryKey,
  unique: true,
});

2. 新建以 partitionRule 的每个键为后缀的分区子表

这一步只需根据分区方式和分区主表名拼接出创建子表的sql，因为子表会自动与主表的字段结构保持一致，所以生成sql也比较简单。因为每次都会遍历partitionRule，所以当新增键值和分区规则后也能自动创建新的子表。

// 同步分区子表
const pmList: any = [];
for (const suffix in options.partitionRule) {
  let rule = options.partitionRule[suffix];
  if (prititionColumn.type instanceof DataType.STRING || prititionColumn.type instanceof DataType.TEXT || prititionColumn.type instanceof DataType.CHAR) {
    rule = rule.map(val => `'${val}'`);
  }
  let sql = `CREATE TABLE IF NOT EXISTS "${this.tableName + suffix}" PARTITION OF "${this.tableName}" FOR VALUES`;
  if (options.partition.toUpperCase() === 'LIST') {
    sql += ` IN (${rule.join(',')});`;
  } else if (options.partition.toUpperCase() === 'RANGE') {
    sql += ` FROM (${rule[0]}) TO (${rule[1]});`;
  }
  pmList.push(this.sequelize.query(sql, options));
}
return Promise.all(pmList);

3. 对比并增删改分区主表字段

由于增删改分区主表字段会自动将改动同步到每个分区子表，这一步只需要直接操作分区主表就可以，又因为对分区主表和普通表的逻辑操作基本没有区别，所以这一步和原生sync函数实现基本一致。

但是Postgresql分区表在创建之后就不允许修改分区字段，所以在调用changeColumn时需要将分区字段排除掉。

if (columnName !== options.partitionKey) {
  changes.push(() => this.QueryInterface.changeColumn(this.getTableName(options), columnName, attributes[columnName]));
}

4. 新增分区子表索引

Postgresql分区主表是不能创建索引的，故这里需要遍历每个分区子表，从db查询每个子表已有的索引，并创建每个子表新添加的索引，不过需要注意的是索引名须用子表名而不是主表名去生成，否则会发生创建索引冲突。

// 同步每个分区子表的索引结构
const tableNameList: string[] = [];
let indexOptions: any = this.options.indexes;
for (const suffix in options.partitionRule) {
  tableNameList.push(`${this.getTableName(options) + suffix}`);
}
return Promise.map(tableNameList, tableName => this.QueryInterface.showIndex(tableName, options))
  .then(indexesList => {
  const createIdxPrmList: any = [];
  for (let i = 0 ; i < indexesList.length; i++) {
    let indexes = indexesList[i];
    const tableName = tableNameList[i];
    for (const index of indexOptions) {
      delete index.name;
    }
    indexOptions = this.QueryInterface.nameIndexes(indexOptions, tableName);
    indexes = _.filter(indexOptions, item1 =>
                       !_.some(indexes, item2 => item1.name === item2.name),
                      ).sort((index1, index2) => {
      if (this.sequelize.options.dialect === 'postgres') {
        // move concurrent indexes to the bottom to avoid weird deadlocks
        if (index1.concurrently === true) return 1;
        if (index2.concurrently === true) return -1;
      }
      return 0;
    });
    for (const index of indexes) {
      createIdxPrmList.push(this.QueryInterface.addIndex(
        tableName,
        _.assign({
          logging: options.logging,
          benchmark: options.benchmark,
          transaction: options.transaction,
        }, index),
        tableName,
      ));
    }
  }
  return Promise.all(createIdxPrmList);
});

四、结论

文章基本上是讲如何拓展sequelize支持分区表结构同步到数据库的方案和实现，却没有说怎么对已有分区表的一些读写操作，这是因为分区表是物理存储上做分区而操作逻辑统一，对外的读写和普通表几乎没有区别，所以sequelize封装的大部分函数都可以直接适用于分区表的操作。

然后为什么sequelize不提供分区表结构同步到数据库的功能呢？我觉得有以下几个可能的原因：

使用sequelize.sync来同步表结构本身不是sequelize自动维护表结构的初衷，项目逐渐发展后会使用sequelize-cli库来单独维护数据库表结构的改动；
-sequelize.sync同步表结构执行时间会较长，若同步分区数量很大的表则可能直接影响整个数据库其他IO操作，而且在Postgresql 11.x版本当分区规则发生改动的时候还需要自动迁移现有数据，这会带来更大的性能影响；
分区表被应用的程度不是很广，主要是大部分场景下随着数据量的增加往往会性能会下降得很快，因为分区表存储上做了分区但提供对外的操作逻辑和普通表没有差别，即数据库本身要做更多的工作如定位分区、数据整合等，甚至某些情况下需要对每个子表进行扫描，由于这些场景的存在都使分区表没有使用物理分表那样灵活可控；

既然分区表性能上表现不理想，那么为什么还要使用分区表，或者哪些场景下可以考虑使用分区表呢？

我觉得是在一些项目初期还未有成熟的物理分表库可用，线上数据量会在短期内变得很大，而且对这些数据需要做多种比较复杂查询的情况下，可以先考虑使用分区表。

因为项目前期可能有根据查询offset+limit查询大表数据这种看似简单的需求，这种一般要使用物理分表结合相应的数据冗余表来实现，但是前期出于减少开发成本和考虑项目进度，先使用分区表完成类似的需求，但总之分区表是不适用较成熟的应用。

拓展 sequelize-typescript 库支持 Post

一、背景

二、方案

三、实现

1. 新建分区主表

2. 新建以 partitionRule 的每个键为后缀的分区子表

3. 对比并增删改分区主表字段

4. 新增分区子表索引

四、结论

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读