[SPARK-30858][SQL] Make IntegralDivide's dataType independent from SQL config changes

[MaxGekk]

What changes were proposed in this pull request?

In the PR, I propose to add the returnLong parameter to IntegralDivide, and pass the value of spark.sql.legacy.integralDivide.returnBigint if returnLong is not provided on creation of IntegralDivide.

Why are the changes needed?

This allows to avoid the issue when the configuration change between different phases of planning, and this can silently break a query plan which can lead to crashes or data corruption.

Does this PR introduce any user-facing change?

No

How was this patch tested?

By ArithmeticExpressionSuite.

[SparkQA]

Test build #118654 has finished for PR 27628 at commit ebeec38.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class IntegralDivide(

[cloud-fan]

can we just add a private val returnLong = SQLConf.get.integralDivideReturnLong in the class body? Then the config value is fixed when the expression is created. And it can be serialized to executors.

The spark Expression constructor is kind of exposed to end users when they call functions in SQL. BTW Cast already use a val to store config values.

[hvanhovell]

That can potentially change value every time you transform the tree.

[cloud-fan]

or how about we create 2 expressions IntegralDivide and IntegralDivideReturnLong? I'm just worried about we allow end users to specify the returnLong parameter which becomes an API.

[MaxGekk]

I'm just worried about we allow end users to specify the returnLong parameter which becomes an API.

We don't allow that:

SELECT div(3, 2, false);

fails with:

Invalid number of arguments for function div. Expected: 2; Found: 3; line 1 pos 7
org.apache.spark.sql.AnalysisException: Invalid number of arguments for function div. Expected: 2; Found: 3; line 1 pos 7
	at org.apache.spark.sql.catalyst.analysis.FunctionRegistry$.$anonfun$expression$7(FunctionRegistry.scala:618)
	at scala.Option.getOrElse(Option.scala:189)
	at org.apache.spark.sql.catalyst.analysis.FunctionRegistry$.$anonfun$expression$4(FunctionRegistry.scala:602)
	at org.apache.spark.sql.catalyst.analysis.SimpleFunctionRegistry.lookupFunction(FunctionRegistry.scala:121)
	at org.apache.spark.sql.catalyst.catalog.SessionCatalog.lookupFunction(SessionCatalog.scala:1418)

I ran the command on the PR changes.

[cloud-fan]

so the non-expression parameter doesn't count? Then I'm fine with it.

[MaxGekk]

Yes, we count only expressions, see

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

Lines 601 to 602 in 919d551

	val params = Seq.fill(expressions.size)(classOf[Expression])
	val f = constructors.find(_.getParameterTypes.toSeq == params).getOrElse {

[cloud-fan]

shall we define unapply as well? Most of the time we don't care about the returnLong paramter. When we do, we should write case e @ IntegralDivide(left, right) if e.returnLong.

[MaxGekk]

There is only one place where we unapply IntegralDivide. It is here https://github.com/apache/spark/pull/27628/files#diff-8e1575bb706d6f7e8b5ea0b175eaeafcR178 . And returnLong is not checked, it is just extracted and copy-pasted.

[MaxGekk]

If I define unapply, it will be not used. Not sure that is is needed.

[cloud-fan]

@MaxGekk not related to this PR, can you check other places that call SQLConf.get inside expressions? We should try to make them consistent. One exception is Cast, we decided to create AnsiCast expression.

[MaxGekk]

There are at least a few more places:

1. spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala

   Line 95 in ab186e3

   val legacySizeOfNull = SQLConf.get.legacySizeOfNull

2. spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala

   Line 96 in 4e50f02

   val nameOfCorruptRecord = SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)

3. spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala

   Line 562 in 8e280ce

   val nameOfCorruptRecord = SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)

4. spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/higherOrderFunctions.scala

   Lines 528 to 529 in b0db623

   	private val followThreeValuedLogic =
   	SQLConf.get.getConf(SQLConf.LEGACY_ARRAY_EXISTS_FOLLOWS_THREE_VALUED_LOGIC)

5. spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala

   Line 50 in b917a65

   private val nullOnOverflow = !SQLConf.get.ansiEnabled

6. spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypeCreator.scala

   Line 49 in 926e3a1

   if (SQLConf.get.getConf(SQLConf.LEGACY_CREATE_EMPTY_COLLECTION_USING_STRING_TYPE)) {

7. spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypeCreator.scala

   Line 149 in 926e3a1

   if (SQLConf.get.getConf(SQLConf.LEGACY_CREATE_EMPTY_COLLECTION_USING_STRING_TYPE)) {

[hvanhovell]

@MaxGekk do all of them influence the dataType or the nullability?

[MaxGekk]

@hvanhovell Some of them, others (except of Json/CSV exprs) influence defaultElementType

[cloud-fan]

@MaxGekk thanks for composing the list! I've created an umbrella JIRA for them: https://issues.apache.org/jira/browse/SPARK-30893

[cloud-fan]

thanks, merging to master/3.0!