[jira] [Created] (SPARK-49510) The withWatermark method in Dataset.scala is throwing a ParseException when a number is passed to the eventTime parameter

["Anitesh Minj (Jira)" <ji...@apache.org>]

Anitesh Minj created SPARK-49510:
------------------------------------

             Summary: The withWatermark method in Dataset.scala is throwing a ParseException when a number is passed to the eventTime parameter
                 Key: SPARK-49510
                 URL: https://issues.apache.org/jira/browse/SPARK-49510
             Project: Spark
          Issue Type: Bug
          Components: SQL, Structured Streaming
    Affects Versions: 3.3.2, 3.3.0
            Reporter: Anitesh Minj


def withWatermark(eventTime: String, delayThreshold: String): Dataset[T] = withTypedPlan {
val parsedDelay = IntervalUtils.fromIntervalString(delayThreshold)
require(!IntervalUtils.isNegative(parsedDelay),
s"delay threshold ($delayThreshold) should not be negative.")
EliminateEventTimeWatermark(
EventTimeWatermark(UnresolvedAttribute(eventTime), parsedDelay, logicalPlan))
}

In the above withWatermark function of Dataset.scala, if we pass number as eventTime parameter. It fails with below ParseException

{color:#de350b}Caused by: org.apache.spark.sql.catalyst.parser.ParseException: {color}
{color:#de350b}Syntax error at or near '1'(line 1, pos 0){color}

{color:#de350b}== SQL =={color}
{color:#de350b}1{color}
{color:#de350b}^^^{color}

{color:#de350b}    at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:315){color}
{color:#de350b}    at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:152){color}
{color:#de350b}    at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parseMultipartIdentifier(ParseDriver.scala:68){color}
{color:#de350b}    at org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute$.apply(unresolved.scala:161){color}
{color:#de350b}    at org.apache.spark.sql.Dataset.withWatermark(Dataset.scala:748){color}

 

In spark 3.1.1 it was able to parse number properly. It was not throwing any error.

The current issue is seen because of the checkin done for the issue https://issues.apache.org/jira/browse/SPARK-34636.

The changes made to the {{apply}} method in the {{Unresolved.scala}} file are causing the issue. I have attached the relevant code sections below for Spark 3.1 and Spark 3.2, highlighting the differences.:-

*Spark 3.1*
def apply(name: String): UnresolvedAttribute = {color:#00875a}{color:#172b4d}new UnresolvedAttribute({color:#00875a}name.split("\\."){color}){color}

{color:#172b4d}*Spark 3.2*{color}
{color}def apply(name: String): UnresolvedAttribute =
    new UnresolvedAttribute({color:#de350b}CatalystSqlParser.parseMultipartIdentifier(name){color})

Please visit this link to see the code changes:- [https://github.com/apache/spark/commit/94fd6cb0ce56684de47eb5cca925055859c968ef]



In above code for spark 3.1, we can see that in object instantiation of the apply Method only name.split is being done.
But with the above mentioned changed in spark 3.2 code and in later versions of spark  CatalystSqlParser.parseMultipartIdentifier(name) is being called.
In turn this parseMultipartIdentifier method will call singleMultipartIdentifier method of SQLParser. we can see in the below code snippet.

override def parseMultipartIdentifier(sqlText: String): Seq[String] = {
parse(sqlText) { parser =>
astBuilder.visitSingleMultipartIdentifier({color:#4c9aff}parser.singleMultipartIdentifier(){color})
}
}

Now, once {color:#4c9aff}parser.singleMultipartIdentifier(){color} method is called. It will try to validate sqlText using the singleMultipartIdentifier rule. This sqlText parameter will contain the same value which we passed to eventTime  parameter of withWaterMark function. 
Now when this sqlText is a number(For Example "1"), than it will throw {color:#de350b}ParseException{color} which i mentioned in the beginning, as singleMultipartIdentifier rule can't parse Integer Value (numbers).



Possible Solution that i can suggest: 
Introduce a new method in object of UnresolvedAttribute as shown below

object UnresolvedAttribute {

{color:#4c9aff}def numberString(name:String){color}

{color:#4c9aff}: UnresolvedAttribute = new UnresolvedAttribute(name.split("\\.")){color}

{color:#172b4d}}{color}

and in withWatermark function of Dataset.scala do a function call to UnresolvedAttribute.numberString(eventTime) inplace of calling a constructor UnresolvedAttribute(eventTime) as shown below

def withWatermark(eventTime: String, delayThreshold: String): Dataset[T] = withTypedPlan {
val parsedDelay = IntervalUtils.fromIntervalString(delayThreshold)
require(!IntervalUtils.isNegative(parsedDelay),
s"delay threshold ($delayThreshold) should not be negative.")
EliminateEventTimeWatermark(
EventTimeWatermark({color:#4c9aff}UnresolvedAttribute.numberString(eventTime){color}, parsedDelay, logicalPlan))
}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org