依靠Expand实现aggr distinct

现在的aggr distinct支持

现在IMCI依靠Join + HashGroupby实现对aggr distinct的支持，在这里分类列出各个情况下的改写策略

不含grouping clause的情况

仅含一个agg discinct

形如以下的SQL

SELECT 
  COUNT(DISTINCT col1) 
FROM t1;
复制

会被改写为

SELECT 
  COUNT(tmp.col1) 
FROM 
  (
     SELECT
       col1
     FROM
       t1
     GROUP BY col1);
复制

仅含有多个agg discinct

形如以下的SQL

SELECT 
  COUNT(DISTINCT col1), COUNT(DISTINCT col2)
FROM t1;
复制

会被改写为

CREATE VIEW v1 AS 
  SELECT 
    COUNT(tmp.col1)
  FROM 
    (
       SELECT
         col1
       FROM
         t1
       GROUP BY col1);

CREATE VIEW v2 AS 
  SELECT 
    COUNT(tmp.col2)
  FROM 
    (
       SELECT
         col2
       FROM
         t1
       GROUP BY col2);
       
SELECT * FROM v1, v2;
复制

同时含有普通agg和agg discinct

形如以下的SQL

SELECT 
  COUNT(DISTINCT col1), COUNT(DISTINCT col2), SUM(col3)
FROM t1;
复制

会被改写为

CREATE VIEW v1 AS 
  SELECT 
    COUNT(tmp.col1)
  FROM 
    (
       SELECT
         col1
       FROM
         t1
       GROUP BY col1);

CREATE VIEW v2 AS 
  SELECT 
    COUNT(tmp.col2)
  FROM 
    (
       SELECT
         col2
       FROM
         t1
       GROUP BY col2);

CREATE VIEW v3 AS 
  SELECT SUM(col3) FROM t1;
       
SELECT * FROM v1, v2, v3;
复制

含grouping clause的情况

仅含一个agg discinct

形如以下的SQL

SELECT 
  COUNT(DISTINCT col1) 
FROM t1
GROUP BY col2;
复制

会被改写为

SELECT 
  COUNT(tmp.col1) 
FROM 
  (
     SELECT
       col1, col2
     FROM
       t1
     GROUP BY col1, col2)
GROUP BY col2;
复制

仅含有多个agg discinct

形如以下的SQL

SELECT 
  COUNT(DISTINCT col1), COUNT(DISTINCT col2)
FROM t1
GROUP BY col3;
复制

会被改写为

 CREATE VIEW v1 AS 
  SELECT 
    col3, COUNT(tmp.col1)
  FROM 
    (
       SELECT
         col1, col3
       FROM
         t1
       GROUP BY col1, col3
    )
  GROUP BY col3;

CREATE VIEW v2 AS 
  SELECT 
    col3, COUNT(tmp.col2)
  FROM 
    (
       SELECT
         col2, col3
       FROM
         t1
       GROUP BY col2, col3
     )
   GROUP BY col3;
       
SELECT * FROM v1 JOIN v2 ON v1.col3 = v2.col3;
复制

同时含有普通agg和agg discinct

形如以下的SQL

SELECT 
  COUNT(DISTINCT col1), COUNT(DISTINCT col2), SUM(col3)
FROM t1
GROUP BY col4;
复制

会被改写为

CREATE VIEW v1 AS 
  SELECT 
    col4, COUNT(tmp.col1)
  FROM 
    (
       SELECT
         col1, col4
       FROM
         t1
       GROUP BY col1, col4
    )
  GROUP BY col4;

CREATE VIEW v2 AS 
  SELECT 
    col4, COUNT(tmp.col2)
  FROM 
    (
       SELECT
         col2, col4
       FROM
         t1
       GROUP BY col2, col4
     )
   GROUP BY col4;

CREATE VIEW v3 AS 
  SELECT SUM(col3) FROM t1 GROUP BY col4;
       
SELECT * FROM v1, v2, v3 WHERE v1.col4 = v2.col4 and v2.col4 = v3.col4;
复制

可以看出，大体的思路是：

1. 对于单个agg(distinct col), 使用叠加一个group by的方式使y = distinct col，然后计算agg(y)
2. 对于含有多个agg的情形，先拆分为多个group by，每个group by只含有一个agg，因此每个含有agg distinct的group by都可以依赖1来解决

但是，这样的作法在OSS外表的查询执行下有一些问题，因为这样的改写会生成大量的TableScan，在单机引擎中，我们认为TableScan算子是一个开销很小的算子，因为在单机上我们可以利用lru cache, pruner, prefetch等优化，同时读local disk和polar store相对较快，但是OSS上的TableScan和单机引擎有一些不同

• 以上的lru cache, pruner, prefetch等优化均不存在
• 目前的过滤条件使通过Scan + Filter两个算子实现，Filter没有下推到scan，scan输出的数据量很大

这直接导致了OSS上含有多个agg, distinct的query查询效果很差，因此，可能需要一个不含有join的处理agg distinct的方式

通过Expand实现不带join的agg distinct

为了解决上述方法在OSS表上的问题，我们可以借鉴上文方法的思路，即：

将数据复制出多份，每次只处理一部分agg函数，分而治之。

在上文的办法中，这个复制操作是通过Join+多个TableScan来完成的，为了减少这个操作的次数，我们可以用Expand操作+多个Groupby来完成这个操作

Expand操作

为了完成上文提到的复制操作，expand算子的操作需要满足以下两点

• 对于输入的一行，需要输出n行数据
• 这n行数据需要有标识符，使每行数据只对部分agg函数有意义

基于这两个假设，这里我们提出一个**Expand(N)**操作，对于一个M列的行(M > N)

Expand(N)，将输出N+1行，记作，每一行有M+1列，其内容为

方法概述

• 首先，求出不同的DISTINCT col的数量，例如SUM(distinct c1), SUM(distinct c2)中不同的列数就是2，如果再加一列SUM(distinct a+b)的话就是 3，这里a + b也视作源表中的一列，这个结果记作N

• 对输入表应用Expand(N)，得到一张更大的表，源表中的每一行，对应大表中的N+1行，这n+1行有序号0-n

• 之后连续应用N+1次group by，第1次group by处理所有非distinct的agg，后续每个groupby处理1个distinct agg，这里每个group by的grouping列中都要添加额外的列：

  • 尚未处理的distinct列，这实际上是在做后续distinct agg中的distinct操作，这样下一个AGG(DISTINCT col)就可以通过AGG(col)来完成
  • Expand中的序号，这个序号保证了不同份数据的复制之间不会彼此干扰

• 在这些操作完成之后，会得到一个含有序号的group by结果，其中序号0的行是所有group非distinct agg的结果，序号1的行是所有group第一个distinct agg的结果，依次类推，最后再采用一个group by来组合这些结果，得到最终的结果

一个使用Expand实现agg distinct的例子

以下这个SQL是一个例子

SELECT
  col1, SUM(col2), SUM(col3), SUM(DISTINCT col4), SUM(DISTINCT col 5)
FROM t1
GROUP BY col1;
复制

t1表的数据如以下表格所示

很显然，上文的SQL结果应该是

现在我们通过Expand+多个Groupby来完成这个SQL的计算

对源表应用Expand

因为我们这里需要的distinct列有两个，分别为COL4和COL5，因此，我们对t1表应用Expand(2)，Expand后的表记作tex1，数据如下表所示

处理非DISTINCT的agg

这一步，我们先处理SUM(col2)与SUM(col3)，对Expand后的表应用如下GROUPBY

SELECT COL1, SUM(COL2), SUM(COL3), COL4, COL5, E_FLAG
FROM tex1
GROUP BY COL1, COL4, COL5, E_FLAG;
复制

这个SQL的结果我们记作tmp1，tmp1的数据如下表所示

注意，这里的头两行就是我们所有非agg distinct的结果

依次处理DISTINCT的agg

处理SUM(DISTINCT COL4)

注意tmp1的数据中，所有E_FLAG = 1的数据中的col4已经变成了distinct的数据，我们接下来在tmp1上做第二个groupby，结果记为tmp2

SELECT COL1, SUM(COL2), SUM(COL3), SUM(COL4), COL5, E_FLAG
FROM tmp1
GROUP BY COL1, COL5, E_FLAG;
复制

tmp2的结果如下表所示

这里EFLAG = 1的数据即为SUM(DISTINCT COL4)的结果

处理SUM(DISTINCT COL5)

类似的，继续执行

SELECT COL1, SUM(COL2), SUM(COL3), SUM(COL4), SUM(COL5), E_FLAG
FROM tmp2
GROUP BY COL1, E_FLAG;
复制

得到表tmp3

tmp3中包含了SUM(DISTINCT col5)的结果

合并结果集

上文中的tmp3已经包含了所有agg函数的结果，但是行中都只有部分agg的结果，现在需要把这些结果合并起来，使用以下SQL

SELECT 
  col1, 
  MIN(IF(E_FLAG = 0, SUM(COL2), NULL)),
  MIN(IF(E_FLAG = 0, SUM(COL3), NULL)),
  MIN(IF(E_FLAG = 1, SUM(COL4), NULL)),
  MIN(IF(E_FLAG = 2, SUM(COL5), NULL))
FROM tmp3
GROUP BY COL1
复制

这样就得到了最终的结果