[BEAM-11747] Make BeamCalcRel safe for ZetaSQL

[apilloud]

This makes the filter for BeamCalcRel a passing set of operators.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang				---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[apilloud]

R: @ibzib
cc: @amaliujia

[amaliujia]

To conclude a bit:

This PR rejects unsupported types, other stuff, etc. in BeamZetaSQL UDF, while #13934 rejects mixed usage of built-in operator and UDF.

So these two PRs do not conflict each other.

[amaliujia]

I am still not sure why we should reject Int64 type for UDF.

Assuming a UDF is defined as package com.xxx.client; class name; public Long udf(Long a), the code gen will generate code for UDF like Long n = com.xxx.client.name.udf((Long) value).

In this case, the value is from Row thus is compatible with ZetaSQL, the return value is still Java object which is also be compatible with ZetaSQL, so there isn't a corrupted case.

[apilloud]

That isn't entirely true. Literals are stored as a BigDecimal (which is what was disabled):

beam/sdks/java/extensions/sql/zetasql/src/main/java/org/apache/beam/sdk/extensions/sql/zetasql/ZetaSqlCalciteTranslationUtils.java

Line 211 in 9bbd5bd

new BigDecimal(value.getInt64Value()), toCalciteType(type, false, rexBuilder));

The generated code prints the literal to a string and looks like this:
c.output(org.apache.beam.sdk.values.Row.withSchema(outputSchema).attachValues(java.util.Arrays.asList(new org.apache.beam.sdk.extensions. sql.provider.UdfTestProvider.IncrementFn().increment(5L))));

In earlier testing I was seeing overflow errors from gRPC, but that appears to actually be caused by DOUBLE literals (which are also stored as BigDecimal). I've added DECIMAL to the list of allowed types for literals and reverted the test changes.

[apilloud]

This is now rebased past #13912.

[amaliujia]

The generated code prints the literal to a string and looks like this:
c.output(org.apache.beam.sdk.values.Row.withSchema(outputSchema).attachValues(java.util.Arrays.asList(new org.apache.beam.sdk.extensions. sql.provider.UdfTestProvider.IncrementFn().increment(5L))));

Based on this generated code, for a Java UDF, it's input won't lose any precision?

Is there an example that the code generation generates code that casts input which causes precision lose? If not there is probably no data corruption for Java UDF.

[apilloud]

@amaliujia One example is SELECT CAST (Double(179769313486231570814527423731704356798070567525844996598917476803157260780028538760589558632766878171540458953514382464234321326889464182768467546703537516986049910576551282076245490090389328944075868508455133942304583236903222948165808559332123348274797826204144723168738177180919299881250404026184124858368) AS NUMERIC) AS ColA It generates this code:

{
  final java.math.BigDecimal v = new java.math.BigDecimal(
    "179769313486231570814527423731704356798070567525844996598917476803157260780028538760589558632766878171540458953514382464234321326889464182768467546703537516986049910576551282076245490090389328944075868508455133942304583236903222948165808559332123348274797826204144723168738177180919299881250404026184124858368");
  c.output(org.apache.beam.sdk.values.Row.withSchema(outputSchema).attachValues(java.util.Arrays.asList(v)));
}

This fixes at least 9 additional data corruption bugs (mostly around timestamps) and doesn't break any tests in Beam. I can provide you the link to the internal run of this, but if you want to debate further I would suggest we roll back first.

[amaliujia]

My suggestion is:

1. if this PR's goal is to verify whether BeamCalRel is safe to run ZetaSQL operators, can you create a separate rule that allows ZetaSQL operators run in BeamCalRel(and reject those wrong cases)?
2. If this PR's goal is to help improve Java UDF, please hold this PR now. Kyle and I are discussing the next steps to improve current UDF implementation. This PR is a great reference to give us directions on where to check the implementation. We will need time to digest this PR and consider whether Calc splitting should go first. We will definitely consider this PR during our improving process.

[apilloud]

This PR is currently addressing the second point. There is currently data corruption introduced by UDFs. If you think it will take a significant time to address this I would suggest we roll back #13841 until you figure this out.

[ibzib]

This PR is currently addressing the second point. There is currently data corruption introduced by UDFs. If you think it will take a significant time to address this I would suggest we roll back #13841 until you figure this out.

Can we merge just the first commit from this PR, but remove BeamJavaUdfCalcRule from the planner by default?

[apilloud]

That works for me. I moved the first commit into it's own PR: #14009. Will one of you follow up and disable BeamJavaUdfCalcRule?

[ibzib]

That works for me. I moved the first commit into it's own PR: #14009. Will one of you follow up and disable BeamJavaUdfCalcRule?

SG, thanks.

[github-actions]

The Workflow run is cancelling this PR. It is an earlier duplicate of 2173354 run.

[github-actions]

The Workflow run is cancelling this PR. It is an earlier duplicate of 2173354 run.

[amaliujia]

Calcite does more for built-in operators than UDF. Using CAST as an example: https://github.com/apache/calcite/blob/d815dc1209c3cc0728941e9fc81fce7d9e44c79b/core/src/main/java/org/apache/calcite/adapter/enumerable/RexToLixTranslator.java#L248 .

For UDF, fortunately, Calcite does less. Basically Calcite will either generate the string representation of a value (i.e. Literal) or make a Row field access (i.e. column), and then make a reflection call on the UDF implementation. So for UDF, I think we only need to verify 1) whether Calcite does anything that make the value lose precision 2) whether Calcite generates a correct string representation for value or make correct field access to Row. The verification work will be much less compared to checking the built-in operators.

I agree this PR could be an upper bound for the limitation of Java UDF, what I want to see is whether we can do better.

[amaliujia]

Ok, with another thought, I think why not we merge this PR for now?

We can start from the most strict constraints, then gradually loose the constraints.

[apilloud]

The constraints added here do not limit BeamCalc to only UDFs, we would need significantly more constraints for that. They do ensure the only BeamCalc specific compliance failure is unimplemented. I'm happy to review any changes that loosen the constraints and hope some day there are none.