Skip to content

[SPARK-29904][SQL][2.4] Parse timestamps in microsecond precision by JSON/CSV datasources #26507

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MaxGekk wants to merge 22 commits into from
Closed

[SPARK-29904][SQL][2.4] Parse timestamps in microsecond precision by JSON/CSV datasources #26507

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
897bf33
Add fastParseToMicros and tests
MaxGekk Nov 13, 2019
b18c61a
Use fastParseToMicros in JacksonParser
MaxGekk Nov 13, 2019
915a755
Add a test for from_json()
MaxGekk Nov 13, 2019
3d7a611
Merge remote-tracking branch 'origin/branch-2.4' into fastdateformat-…
MaxGekk Nov 13, 2019
453ee5c
Accept only SSS or SSSSSS or without fraction
MaxGekk Nov 13, 2019
86ac2b2
Generic approach
MaxGekk Nov 13, 2019
9c82538
Make MicrosCalendar private
MaxGekk Nov 14, 2019
9a446aa
Optimizations: produce less garbage
MaxGekk Nov 14, 2019
ef8284c
Fix imports
MaxGekk Nov 14, 2019
638c640
Bug fixes
MaxGekk Nov 14, 2019
eb58d03
Improve error message
MaxGekk Nov 14, 2019
274f25a
dateTimeParser -> timestampParser
MaxGekk Nov 14, 2019
66d1100
Fix CSV as well
MaxGekk Nov 14, 2019
69c1bb4
Add a negative test
MaxGekk Nov 14, 2019
0bdb523
Add a CSV test
MaxGekk Nov 14, 2019
7d9a19d
Add comments
MaxGekk Nov 14, 2019
810a061
getDateTimeParser -> getTimestampParser
MaxGekk Nov 15, 2019
1a6608f
Fix a comment
MaxGekk Nov 15, 2019
31d7877
the array of parsed `fields` -> protected `fields`
MaxGekk Nov 15, 2019
8db95ca
Change expected value in a test
MaxGekk Nov 15, 2019
1852511
Put `digitsInFraction` and `cal` inside of `TimestampParser`
MaxGekk Nov 15, 2019
1cbd35b
Rename a test
MaxGekk Nov 15, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.catalyst.util.DateTimeUtils.fastParseToMicros
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.Utils
Expand Down Expand Up @@ -216,7 +217,7 @@ class JacksonParser(
// This one will lose microseconds parts.
// See https://issues.apache.org/jira/browse/SPARK-10681.
Long.box {
Try(options.timestampFormat.parse(stringValue).getTime * 1000L)
Try(fastParseToMicros(options.timestampFormat, stringValue, options.timeZone))
.getOrElse {
// If it fails to parse, then tries the way used in 2.0 and 1.x for backwards
// compatibility.
Expand Down
30 changes: 29 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,13 +20,15 @@ package org.apache.spark.sql.catalyst.util
import java.sql.{Date, Timestamp}
import java.text.{DateFormat, SimpleDateFormat}
import java.time.Instant
import java.util.{Calendar, Locale, TimeZone}
import java.util.{Calendar, GregorianCalendar, Locale, TimeZone}
import java.util.concurrent.ConcurrentHashMap
import java.util.function.{Function => JFunction}
import javax.xml.bind.DatatypeConverter

import scala.annotation.tailrec

import org.apache.commons.lang3.time.FastDateFormat

import org.apache.spark.unsafe.types.UTF8String

/**
Expand Down Expand Up @@ -1164,4 +1166,30 @@ object DateTimeUtils {
threadLocalTimestampFormat.remove()
threadLocalDateFormat.remove()
}

class MicrosCalendar(tz: TimeZone) extends GregorianCalendar(tz, Locale.US) {
def getMicros(): SQLTimestamp = {
Copy link
Contributor

@cloud-fan cloud-fan Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we have some comments to explain the behavior? Seems to me it's

  1. for .1, it's 1000 microseconds
  2. for .1234, it's 1234 microseconds
  3. for .1234567, it's 123456 microseconds

Copy link
Contributor

@cloud-fan cloud-fan Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have a simple rule? The rule for interval is pretty simple: adding 0 at the end until the second fraction has 9 digits, then parse the 9 digits to nanoseconds.

Copy link
Member Author

@MaxGekk MaxGekk Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have a simple rule?

I haven't found simpler approach for now. The difference between interval and timestamp is the former one may have time zone at the end or anything else. We cannot say to users don't use the pattern like mm:ss.SSSSSSXXX yyyy/MM/dd

Copy link
Contributor

@cloud-fan cloud-fan Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So what we get here is the milliseconds FastDateFormat extracts from the string. I believe FastDateFormat can handle the part after seconds. e.g. 12:12:12.1234Z, the milliseconds part should be 1234.

Copy link
Member Author

@MaxGekk MaxGekk Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we have some comments to explain the behavior? Seems to me it's
for .1, it's 1000 microseconds

This is 100 * 1000 microsecond but SimpleDateFormat and FastDateFormat have weird behavior. The example below on 2.4 without my changes:

scala> val df = Seq("""{"a":"2019-10-14T09:39:07.1Z"}""").toDF
df: org.apache.spark.sql.DataFrame = [value: string]

scala> val res = df.select(from_json('value, schema, Map("timestampFormat" -> "yyyy-MM-dd'T'HH:mm:ss.SXXX")))
res: org.apache.spark.sql.DataFrame = [jsontostructs(value): struct<a: timestamp>]

scala> res.show(false)
+-------------------------+
|jsontostructs(value)     |
+-------------------------+
|[2019-10-14 12:39:07.001]|
+-------------------------+
scala> val res = df.select(from_json('value, schema, Map("timestampFormat" -> "yyyy-MM-dd'T'HH:mm:ss.SSSXXX")))
res: org.apache.spark.sql.DataFrame = [jsontostructs(value): struct<a: timestamp>]

scala> res.show(false)
+-------------------------+
|jsontostructs(value)     |
+-------------------------+
|[2019-10-14 12:39:07.001]|
+-------------------------+

So .1 cannot be parsed correctly only 0.100:

scala> val df = Seq("""{"a":"2019-10-14T09:39:07.100Z"}""").toDF
df: org.apache.spark.sql.DataFrame = [value: string]

scala> val res = df.select(from_json('value, schema, Map("timestampFormat" -> "yyyy-MM-dd'T'HH:mm:ss.SSSXXX")))
res: org.apache.spark.sql.DataFrame = [jsontostructs(value): struct<a: timestamp>]

scala> res.show(false)
+-----------------------+
|jsontostructs(value)   |
+-----------------------+
|[2019-10-14 12:39:07.1]|
+-----------------------+

Copy link
Member Author

@MaxGekk MaxGekk Nov 13, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What I see in the source code of SimpleDateFormat, it just casts the fraction part to int. .001 and .01 are the same and equal to 1.

Copy link
Member Author

@MaxGekk MaxGekk Nov 13, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, this following check fails:

    check("yyyy-MM-dd'T'HH:mm:ss.SSSSSSXXX",
      "2019-10-14T09:39:07.000010Z", "2019-10-14T09:39:07.000010Z")

Expected :1571045947000010
Actual   :1571045947010000

because.000010 is parsed to 10 inside of SimpleDateFormat.

Copy link
Member Author

@MaxGekk MaxGekk Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The commit 86ac2b2 should fix that

var fraction = fields(Calendar.MILLISECOND)
if (fraction < MICROS_PER_MILLIS) {
fraction *= MICROS_PER_MILLIS.toInt
} else if (fraction >= MICROS_PER_SECOND) {
do {
fraction /= 10
} while (fraction >= MICROS_PER_SECOND)
}
fraction
}
}

def fastParseToMicros(parser: FastDateFormat, s: String, tz: TimeZone): SQLTimestamp = {
val pos = new java.text.ParsePosition(0)
val cal = new MicrosCalendar(tz)
cal.clear()
if (!parser.parse(s, pos, cal)) {
throw new IllegalArgumentException(s)
}
val micros = cal.getMicros()
cal.set(Calendar.MILLISECOND, 0)
cal.getTimeInMillis * MICROS_PER_MILLIS + micros
}
}
21 changes: 21 additions & 0 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ import java.sql.{Date, Timestamp}
import java.text.SimpleDateFormat
import java.util.{Calendar, Locale, TimeZone}

import org.apache.commons.lang3.time.FastDateFormat

import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.util.DateTimeUtils._
import org.apache.spark.unsafe.types.UTF8String
Expand Down Expand Up @@ -692,4 +694,23 @@ class DateTimeUtilsSuite extends SparkFunSuite {
}
}
}

test("fast parse to micros") {
val timeZone = TimeZoneUTC
def check(pattern: String, input: String, reference: String): Unit = {
val parser = FastDateFormat.getInstance(pattern, timeZone, Locale.US)
val expected = DateTimeUtils.stringToTimestamp(
UTF8String.fromString(reference), timeZone).get
val actual = fastParseToMicros(parser, input, timeZone)
assert(actual === expected)
}
check("yyyy-MM-dd'T'HH:mm:ss.SSSSSSSXXX",
"2019-10-14T09:39:07.3220000Z", "2019-10-14T09:39:07.322Z")
Copy link
Contributor

@cloud-fan cloud-fan Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this behavior the same with master branch?

check("yyyy-MM-dd'T'HH:mm:ss.SSSXXX",
"2019-10-14T09:39:07.322Z", "2019-10-14T09:39:07.322Z")
check("yyyy-MM-dd'T'HH:mm:ss.SSSSSSXXX",
"2019-10-14T09:39:07.123456Z", "2019-10-14T09:39:07.123456Z")
check("yyyy-MM-dd'T'HH:mm:ss.SSSXXX",
"2019-10-14T09:39:07.123Z", "2019-10-14T09:39:07.123Z")
}
}
10 changes: 10 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -518,4 +518,14 @@ class JsonFunctionsSuite extends QueryTest with SharedSQLContext {
jsonDF.select(to_json(from_json($"a", schema))),
Seq(Row(json)))
}

test("from_json - timestamp in micros") {
val df = Seq("""{"time": "1970-01-01T00:00:00.123456"}""").toDS()
val schema = new StructType().add("time", TimestampType)
val options = Map("timestampFormat" -> "yyyy-MM-dd'T'HH:mm:ss.SSSSSS")

checkAnswer(
df.select(from_json($"value", schema, options)),
Row(Row(java.sql.Timestamp.valueOf("1970-01-01 00:00:00.123456"))))
}
}