[GLUTEN-842][VL] convert expand op to expand exec in velox

[zhli1142015]

What changes were proposed in this pull request?

Here are project expressions we observed in expand operation:

agg exprs + group by exprs + gid.
agg exprs + group by exprs + gid + _gen_grouping_pos. --> The last column is for handling duplicate grouping sets.
group by exprs + gid + agg exprs. --> gid is calculated by different way from above two cases. it's assigned with the sequence number of project set.

Original ExpandExecTrandofrmer can only handle the first case. I'm adding the expand exec in velox side by the PR: https://github.com/oap-project/velox/pull/199/files. This PR is for converting spark ExpandExec to Expand OP in Velox, We don't need to do the columns' mapping in Gluten.

The original ExpandExecTrandofrmer is renamed to GroupIdExecTrandofrmer to not break ClicKHouse.

How was this patch tested?

Unit test.

[github-actions]

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/oap-project/gluten/issues

Then could you also rename commit message and pull request title in the following format?

[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}

See also:

• Other pull requests

[JkSelf]

@baibaichen

This PR is to implement ExpandTransformer similar with vanilla spark. And then we no need to distinguish the agg cols and group cols in Gluten. It seems the Expand PR to substrait community is ready to merge except the doc update. How about directly use PR in Gluten to support Expand in this PR?

[FelixYBW]

@baibaichen

This PR is to implement ExpandTransformer similar with vanilla spark. And then we no need to distinguish the agg cols and group cols in Gluten. It seems the Expand PR to substrait community is ready to merge except the doc update. How about directly use PR in Gluten to support Expand in this PR?

We should follow substrait's solution. Does the PR need change to substrait?

[JkSelf]

@baibaichen
This PR is to implement ExpandTransformer similar with vanilla spark. And then we no need to distinguish the agg cols and group cols in Gluten. It seems the Expand PR to substrait community is ready to merge except the doc update. How about directly use PR in Gluten to support Expand in this PR?

We should follow substrait's solution. Does the PR need change to substrait?

Yes. This PR add new ExpandRel message in algebra.proto, which follow up the substrait community solution except the definition in PR.

[zhli1142015]

@baibaichen
This PR is to implement ExpandTransformer similar with vanilla spark. And then we no need to distinguish the agg cols and group cols in Gluten. It seems the Expand PR to substrait community is ready to merge except the doc update. How about directly use PR in Gluten to support Expand in this PR?

We should follow substrait's solution. Does the PR need change to substrait?

Yes. This PR add new ExpandRel message in algebra.proto, which follow up the substrait community solution except the definition in PR.

Thanks, updated.

[github-actions]

#842

[zhli1142015]

@JkSelf , @zhouyuan and @lgbo-ustc, could you please help to take a look?

[JkSelf]

LGTM except small comments.

[lgbo-ustc]

You mention that ExpandExecTrandofrmer has benn kept but the the ExpandRel has been changed, is it backward compatible with ClickhHouse?

[zhli1142015]

It seems the ExpandRel has been changed, is it backward compatible with ClickhHouse?

new 'ExpandRel' is different and can't be compatible with CH. The original Rel is renamed to 'GroupIdRel' and changes are made in CH also.
Thanks.

[lgbo-ustc]

It's greate. This implementation is simple. I think it could solve some problems we have meet.

LGTM.

[zhli1142015]

It's greate. This implementation is simple. I think it could solve some problems we have meet.

LGTM.

Thanks for review @lgbo-ustc , will you work on CH side to consume new ExpandRel contract?

[lgbo-ustc]

It's greate. This implementation is simple. I think it could solve some problems we have meet.
LGTM.

Thanks for review @lgbo-ustc , will you work on CH side to consume new ExpandRel contract?

We will do it soon

[zhouyuan]

👍

[zhanglistar]

@zhli1142015

agg exprs + group by exprs + gid + _gen_grouping_pos. --> The last column is for handling duplicate grouping sets.
group by exprs + gid + agg exprs. --> gid is calculated by different way from above two cases. it's assigned with the sequence number of project set.

I have a question, the two cases, could you give some sql? We don't know when spark will generate the two cases.
Thanks.

[zhli1142015]

@zhli1142015

agg exprs + group by exprs + gid + _gen_grouping_pos. --> The last column is for handling duplicate grouping sets.
group by exprs + gid + agg exprs. --> gid is calculated by different way from above two cases. it's assigned with the sequence number of project set.

I have a question, the two cases, could you give some sql? We don't know when spark will generate the two cases. Thanks.

Hello @zhanglistar ,
Please check below sample code:

case class TestData3(a: Int, b: Option[Int])
val df = spark.sparkContext.parallelize(
      TestData3(1, None) ::
      TestData3(2, Some(2)) :: Nil).toDF()
import org.apache.spark.sql.functions._
df.agg(count($"a"), count($"b"), count(lit(1)), count_distinct($"a"), count_distinct($"b")).collect // case 3
df.createOrReplaceTempView("df")
spark.sql("select count(a) from df group by grouping sets((a), (a), (b))").collect // case 2

[zhanglistar]

@zhli1142015

agg exprs + group by exprs + gid + _gen_grouping_pos. --> The last column is for handling duplicate grouping sets.
group by exprs + gid + agg exprs. --> gid is calculated by different way from above two cases. it's assigned with the sequence number of project set.

I have a question, the two cases, could you give some sql? We don't know when spark will generate the two cases. Thanks.

Hello @zhanglistar , Please check below sample code:

case class TestData3(a: Int, b: Option[Int])
val df = spark.sparkContext.parallelize(
      TestData3(1, None) ::
      TestData3(2, Some(2)) :: Nil).toDF()
import org.apache.spark.sql.functions._
df.agg(count($"a"), count($"b"), count(lit(1)), count_distinct($"a"), count_distinct($"b")).collect // case 3
df.createOrReplaceTempView("df")
spark.sql("select count(a) from df group by grouping sets((a), (a), (b))").collect // case 2

@zhli1142015 Thanks! For the sql spark.sql("select count(a) from df group by grouping sets((a), (a), (b))").collect // case 2, just curious, why not just duplicate the two grouping sets (a) for optimization?