[FLINK-10845][table] Support multiple different DISTINCT aggregates for batch

[xueyumusic]

What is the purpose of the change

This PR tried to support multiple distinct aggregates with different arguments for batch. For other cases such as single distinct or multiple with the same arguments it seemed to have supported. Actually it is the problem of IS_NOT_DISTINCT_FROM codegen support result from calcite rewrite

Brief change log

• CodeGenerator and ScalarOperator

Verifying this change

This change added tests and can be verified as follows:

• Added AggregateITCase batch test

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not documented)

[hequn8128]

@xueyumusic Hi, thanks for the PR. The change looks good from my side. I only left some minor suggestions about test improvement.
Best, Hequn

[hequn8128]

Add some null data in tests to test this logic.

[hequn8128]

Replace this with CollectionDataSets.get5TupleDataSet(env).
I think we can add a case when to generate null data. For example(just an example):

    val t = CollectionDataSets.get5TupleDataSet(env).toTable(tEnv).as('a, 'b, 'c, 'd, 'e)
    tEnv.registerTable("MyTable", t)
    val sqlAddNull = "SELECT a, b, c, e, CASE d WHEN 'Hallo' THEN null ELSE d END AS d From MyTable"
    val sqlQuery =
      "SELECT d, " +
      "  COUNT(DISTINCT d), " +
      "  COUNT(DISTINCT e) " +
      s"FROM ($sqlAddNull) " +
      "GROUP BY d " +
      "ORDER BY d"

[xueyumusic]

updated, thank you very much for review and suggestions~, @hequn8128

[hequn8128]

@xueyumusic thanks for the update. Looks good from my side.

[twalthr]

Thank you @xueyumusic. I added some comments.

[twalthr]

This implicitly assumes that the result term is of type boolean which might not always be the case. This line looks like a bug to me.

[twalthr]

Use explicit parameters for all methods in code generation instead of default values.

[twalthr]

Could you also add a IS NOT DISTINCT FROM expression test to org.apache.flink.table.expressions.ScalarOperatorsTest?

[xueyumusic]

Thanks for review, @twalthr , I updated the codes.

[twalthr]

@xueyumusic After looking at the amount of changes, I would suggest to leave generateOperatorIfNotNull untouched. This method is correct as it is. If you need some special null comparing logic. We can create a specialized method only for equals. We could create a method generateDistinctFrom that wraps generateEquals. What do you think?

[xueyumusic]

Yes, I think so, generateOperatorIfNotNull is a low level common method. I updated the code, thanks~ @twalthr

[twalthr]

@xueyumusic I just found the following rule org.apache.calcite.rel.rules.FilterRemoveIsNotDistinctFromRule. Would this solve our problem?
Planner rule that replaces {@code IS NOT DISTINCT FROM} in a {@link Filter} with logically equivalent operations.

[xueyumusic]

hi, @twalthr , I tried simply by putting FilterRemoveIsNotDistinctFromRule.INSTANCE after AggregateExpandDistinctAggregatesRule.JOIN in FlinkRuleSets, it seems that does not work.
I guess that FilterRemoveIsNotDistinctFromRule finds IS_NOT_DISTINCT_FROM in the first rel filter, but AggregateExpandDistinctAggregatesRule rewrites using IS_NOT_DISTINCT_FROM as multiple join condition.

[twalthr]

Thanks for the update @xueyumusic. Then let's at least do my other comment.

[twalthr]

@xueyumusic I think this solution is not correct. I think we should simply code generate:
CASE WHEN x IS NULL THEN y IS NULL WHEN y IS NULL THEN x IS NULL ELSE x = y END
this is also what Calcite's rule is doing.

[twalthr]

This looks not correct to me. By setting the expression to never null, you basically compare the default values of expressions. Long has 0L for example. For example, a 0 IS NOT DISTINCT FROM NULL might return true.

[xueyumusic]

@twalthr I thought the similar way ago and met a problem. If we check left is null then need to generate left code first. However in generateEqual (generateOperatorIfNotNull) it also generate left code, such as

${left.code}
${right.code}
if (${left.nullTerm}) {
  return ${right.nullTerm}
} else if (${right.nullTerm}) {
  return ${left.nullTerm}
} else {
  return generateEqual(left, right)
 (here will also generate ${left.code} and ${right.code} in generateOperatorIfNotNull)
}

I don't know how to make left and right code generate once if don't modify generateOperatorIfNotNull..

[xueyumusic]

It looks that there is one way which generateEqual first such as

generateEquals(left, right)
check left.nullTerm and right.nullTerm
reset nullTerm=false and resultTerm

Do you think it is reasonable? Thanks~ @twalthr

[twalthr]

@xueyumusic You can also do it like this (pseudo code):

${left.code}
${right.code}
resultTerm;
if (${left.nullTerm}) {
  resultTerm = ${right.nullTerm}
} else if (${right.nullTerm}) {
  resultTerm = ${left.nullTerm}
} else {
  resultTerm = generateEqual(
    false, // no null check required in equals
    ${left.copy(code = NO_CODE)},
    ${right.copy(code = NO_CODE)})
}

[xueyumusic]

Thank you very much, @twalthr , This is an elegant solution. I updated the code

[twalthr]

Thank you @xueyumusic. Looks to me now. Will merge...