morningman opened a new issue #1429: Refactor the process of ALTER job. URL: https://github.com/apache/incubator-doris/issues/1429 # Alter Job 重构 Alter Job 主要包括 Schema Change 和 Rollup 两类。 ## 当前问题 1. Job 执行慢 所有 task 下发给 BE 后，会在指定的线程池中排队处理。每一个 task 执行前，都要等当前 BE 上正在运行的 txn 完成后，再开始处理。这样，每个 task 的等待时间累加起来，会延长整个 Job 的完成时间。 2. FINISHING 状态不可控 Alter Job 在最终完成前，有一个 FINISHING 状态。这个状态是在等待当前 Database 内正在运行的事务完成后，Job 才能进入 FINISHED 状态。而且这个状态下 Job 不能取消。这导致某些 corner case 中，Job 会有一直卡在 FINISHING 状态无法完成的风险。 3. BE 上 Tablet 的关系链 目前 BE 上的 Tablet 在执行 alter task 的时候，会维护一个源 tablet 到目的 tablet 的关系链（SchemaChangeStatus），主要用于在接收导入任务时，告知导入任务需要同时转换一份数据到目的 tablet。这个关系链的维护使得 BE 多了一个状态。并且这个关系链还需要通过 FINISHING 状态的 Job 发送 clear task 来清理。维护复杂度很高。 4. Schema Change 和 Rollup 的处理逻辑不同 Rollup 是生产一批新的 tablet，而 Schema Change 是为每一个 tablet 生产新的 schema hash。这导致，虽然两种作业绝大部分逻辑都是相同的，但是因为这个区别，很多流程都要分开考虑，代码维护成本很高。 我们希望通过重构解决以上问题，提供一个统一的，并且对 BE 较为松散的处理流程。这样做也可以为后续的存储计算分离设计提供方便。 ## 设计方案 我们通过 FE 来统一控制 Alter job 和当前正在执行的事务的关系。 1. 在 Alter Job 开始后，寻找一个时间点 X。以 X 为分界线，X 之前的事务可能只导入旧表，也可能同时导入旧表和新表。而 X 之后的事务，肯定会同时导入旧表和新表。Alter task 在 X 之后才下发给 BE。 2. Alter task 在 BE 执行时，只需要负责转换指定版本的历史数据，而不需关系正在执行的其他事务。因为有了第一步，task 需要转换的历史版本肯定是已经生效的并且小于当前正在执行的事务的版本的。所以两者互不干扰。 3. BE 上的 tablet 不再需要记录关系链，因为不再需要 BE 来进行导入事务中，从源 tablet 到目的 tablet 的数据转换工作。导入事务产生的数据，或者只针对一个 tablet，或者会同时产生两份数据。 4. 为 Schema Change 作业也产生新的 tablet。这样对于 BE 的 Alter task 来说，逻辑是一样的。 ## 具体流程 ### Rollup 1. 检查作业逻辑后，生成作业，状态为 PENDING 2. 对于 PENDING 的作业，先同步的创建所有的 rollup replica。 * rollup replica 在整个作业期间不能进行副本均衡和修复。 * rollup replica 在创建时，会在 BE 的 tablet meta 中记录状态为 NOT_READY。（后面介绍这个状态的作用） 3. 创建完成后，先将 Rollup Index 状态设为 SHADOW，并加入到元数据中。然后通过 GlobalTransactionMgr 获取一个新的 txnId，同时 Job 的状态改为 WAITING_TXN。 * SHADOW 状态的 rollup 对导入可见，对其他流程都不可见。当加入到元数据后，后续的导入任务会同时生成 Rollup 的数据。 4. 对于 WAITING_TXN 的作业。等待 txnId 之前的导入全部完成后，获取当前 partition 的visible version，开始下发 rollup task。之后，Job 的状态改为 RUNNING。 5. 对于 RUNNING 状态的作业，检查是否所有 rollup task 都完成。如果完成了，检查完整性（和当前逻辑类似）。OK 之后，将 SHADOW 状态的 Rollup 改为 NORMAL，作业完成。 几点说明: 1. 关于 rollup replica 的版本汇报 在之前的实现中，整个 Rollup 过程中，导入是不检查 rollup replica 是否成功的。也就是说，假设一个 Tablet 有3个副本A、B、C，以及对应的rollup replica A', B', C'。如果一个导入已经在 ABC 中多数副本上完成了，那么及时 A', B', C'都失败了，这个导入也会成功。A', B', C' 的最终检查会交由 Rollup Job 来完成。 但是在新的实现中，我们会将 Rollup 过程中的这些 rollup replica 视为 “正常的” replica。所以一个导入会同时检查 rollup replica 是否成功。如果多数 rollup replica 失败，则导入也会失败。这样做主要是为了统一流程，同时，将 Rollup Job 失败的概率分散到过程中的各个导入作业中（导入失败重试的代价通常小于 Rollup 失败重试的代价） 既然导入作业将 rollup replica 作为正常的 replica，就需要考虑汇报时，replica的版本问题。因为 rollup replica 在开始接收导入时，历史版本是缺失的。所以实际上，FE 会看到一个版本缺失的 replica，从而会认为这个replica 的导入是失败的。为了避免这种情况。我们引入如上所说的 NOT\_READY 状态的 replica。这种 replica 会记录一个历史版本，在汇报版本时，会忽略历史版本之前的空洞，从而保证，FE 看到的版本是完整的，这样可以避免 FE 再对 rollup replica 进行特殊版本处理的逻辑。 BE 上 replica 的 NOT\_READY 状态是持久化的。BE 重启也不会丢失。当rollup task 完成是，负责将这个状态清除。这里有个问题，就是BE上，rollup task 修改replica状态后，向FE汇报之前挂掉，重启后BE已经没有这个状态了，但FE会认为task没有完成而重发。所以，如果BE收到一个 rollup task 而replica状态不是 NOT\_READY，则可以检查下版本，如果OK，直接返回FE成功。如果有，则修改状态为 NOT\_READY 后重做。 2. Rollup task 转换数据 在 Job 的 WAITING\_TXN 阶段，我们使用一个 txnid 作为分界点。我们只保证该 txnid 之后的事务会同时导入两张表，但 txnid 之前的事务有可能导入一张或两张表。因此，最终的 rollup task 锁指定的历史版本，可能有部分已经存在于 rollup 的 replica 中。所以在 BE 执行 rollup task 转换历史数据时，对已经存在的数据，可以直接跳过。 3. Rollup 的可见性 在之前的设计中，事务在 commit 阶段如果发现一个处于 rollup 状态的 index ，则不去处理这个 index 的replica 导入成功与否。而在当前实现中，我们需要处理所有 index 的导入成功情况。 而在之前的设计中，事务并没有记录在导入开始时看到的 index，这导致在 commit 阶段，如果发现有一个 index 的tablet 在 commit tablets 中不存在，则无法判断，是因为 index 在 begin 阶段不可见，还是该 index 的所有 tablet 导入都失败了。 因此，我们需要在事务的开始阶段，在 transaction state 中记录该事务涉及的 index。而在 commit 阶段，只处理这些 index，而不需要关系其他新出现的 index。 ### Schema Change 1. 检查作业逻辑后，生成作业，状态为 PENDING 2. 对于 PENDING 的作业，先同步的创建所有的 new schema 的 tablet。 * new replica 在创建时，会在 BE 的 tablet meta 中记录状态为 NOT_READY。原因同 rollup。 3. 创建完成后，将new schema 的表也加入到元数据中。然后通过 GlobalTransactionMgr 获取一个新的 txnId，同时 Job 的状态改为 WAITING_TXN。 * 一个 schema change 作业可能涉及多个 rollup，因此，我们在 OlapTable 中专门设计一个结构来保存这些还未生效的 index（shadow index）。 * 导入作业会同时访问这些 shadow index 生成导入任务。 4. 对于 WAITING_TXN 的作业。等待 txnId 之前的导入全部完成后，获取当前 partition 的visible version，开始下发 schema change task。之后，Job 的状态改为 RUNNING。 5. 对于 RUNNING 状态的作业，检查是否所有 schema change task 都完成。如果完成了，检查完整性（和当前逻辑类似）。OK 之后，将 shadow index 原子替换 old schema 的 index。作业完成。 几点说明： 1. Schema change 作业本质和 Rollup 作业一样。只是在生成 shadow index 方面有些区别。
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: dev-unsubscr...@doris.apache.org For additional commands, e-mail: dev-h...@doris.apache.org
