[C++][Dataset] Support reading date/time-partitioned datasets accounting for URL encoding (Spark)

[asfimport]

I'm using Spark (3.1.1) to write a dataframe to a partitioned parquet dataset (using delta.io) which is partitioned by a timestamp field.

The relevant Spark code:

// code placeholder
(
 df.withColumn(
 "Date",
 sf.date_trunc(
 "DAY",
 sf.from_unixtime(
 (sf.col("MyEpochField")),
 ),
 ),
 )
 .write.format("delta")
 .mode("append")
 .partitionBy("Date")
 .save("...")

This gives a structure like following:

// code placeholder
/tip
/tip/Date=2021-05-04 00%3A00%3A00
/tip/Date=2021-05-04 00%3A00%3A00/Time=2021-05-04 07%3A27%3A00
/tip/Date=2021-05-04 00%3A00%3A00/Time=2021-05-04 07%3A27%3A00/part-00000-8846eb80-a369-43f6-a715-fec9cf1adf95.c000.snappy.parquet

Notice the : character is (url?) encoded because of fs protocol violation.

When i try to open this dataset using delta-rs (https://github.com/delta-io/delta-rs) which uses Arrow below the hood, then an error is raised trying to parse the Date (folder) value.

// code placeholder
pyarrow.lib.ArrowInvalid: error parsing '2021-05-03 00%3A00%3A00' as scalar of type timestamp[ns]

It seems this error is raised in ScalarParseImpl => ParseValue => StringConverter::Convert => ParseTimestampISO8601

The mentioned parse method does support for format:

// code placeholder
static inline bool ParseTimestampISO8601(const char* s, size_t length,
 TimeUnit::type unit,
 TimestampType::c_type* out) {
 using seconds_type = std::chrono::duration<TimestampType::c_type>; // We allow the following formats for all units:
 // - "YYYY-MM-DD"
 // - "YYYY-MM-DD[ T]hhZ?"
 // - "YYYY-MM-DD[ T]hh:mmZ?"
 // - "YYYY-MM-DD[ T]hh:mm:ssZ?"
<...>

But may not support (url?) decoding the value upfront?

Questions we have:

• Should Arrow support timestamp fields when used as partitioned field?

• Where to decode?

   

  Some more information from the writing side.

  The writing is initiated using FileFormatWriter.write that eventually uses a DynamicPartitionDataWriter (passing in the partitionColumns through the job description).

  Here the actual "value" is rendered and concatennated.

  // code placeholder
   /** Expression that given partition columns builds a path string like: col1=val/col2=val/... */
   private lazy val partitionPathExpression: Expression = Concat(
   description.partitionColumns.zipWithIndex.flatMap { case (c, i) =>
   val partitionName = ScalaUDF(
   ExternalCatalogUtils.getPartitionPathString _,
   StringType,
   Seq(Literal(c.name), Cast(c, StringType, Option(description.timeZoneId))))
   if (i == 0) Seq(partitionName) else Seq(Literal(Path.SEPARATOR), partitionName)
   })

  Where the encoding is done in:

  https://github.com/apache/spark/blob/v3.0.0/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala#L66

  If i understand correct, then Arrow should provide the equivalent of unescapePathName for fields used as partitioned columns.

   

Reporter: Paul Bormans
Assignee: David Li / @lidavidm

Related issues:

• [C++][Dataset] Writing partitions with timestamp type give mis-formatted (integers) directory names (relates to)
• [C++][Dataset] Support URL encoding of partition field values for the file path (relates to)
• [C++][Dataset] Support URI-decoding partition keys (is related to)

PRs and other links:

• GitHub Pull Request #10264

_{Note: This issue was originally created as ARROW-12644. Please see the migration documentation for further details.}

[asfimport]

David Li / @lidavidm:
Yes, it looks like we may need/want to URL-decode paths at some point, though we need to figure out how to handle that. (For instance: presumably it usually doesn't make sense to URL-decode local file system paths, but it looks like Spark may URL-encode paths so we need some way to enable that optionally?)

This also looks somewhat related to ARROW-11378, in that partitioning by a timestamp is rather unusual (the cardinality would normally be too high to make for a worthwhile partition, no?).

[asfimport]

Paul Bormans:
@lidavidm It looks a bit wierd at first sight maybe but the benifit of doing it like this is:

• In Spark it's a (string-typed) timestamp field, thus allowing >= / <= operators for running a query, without any risk of misinterpreting date/time as a string.

• Secondly this example is a truncated timestamp value, so with 
  sf.date_trunc("DAY",
  you do see hours/minutes, etc... but it's all trunc'ed values.

   

[asfimport]

Weston Pace / @westonpace:
I'll add that we do also have a date32 data type (32 bit days since the epoch) which might be more intuitive for truncating to the day and will certainly be more compact.

That being said, even if we decide (ala ARROW-11378) to be strict on what formats we allow for output-partitioning, we could probably be considerably more lenient on what we allow Arrow to read in.

Digging around a bit in Jira it appears that Spark is using java.sql.Timestamp to write out timestamps which states "Formats a timestamp in JDBC timestamp escape format" which at least explains why it is in this format.

Seems like there are potentially 3 issues:

• Should Arrow allow timestamps to be used as partition when writing datasets and, if so, what format (ARROW-11378)

• Add the ability for Arrow to url decode filenames (this issue?)

• Arrow should be able to infer columns written using JDBC escape format (may not be an issue after this issues is resolved but, if it is, it is probably a new issue)

   

  presumably it usually doesn't make sense to URL-decode local file system paths, but it looks like Spark may URL-encode paths so we need some way to enable that optionally?

  I don't think it's too unusual and I think URL decoding has a pretty low false positive rate (i.e. it's not likely to decode something that wasn't meant to be decoded since it's syntax is so obtuse). I'd probably vote towards making it configurable but the default.

[asfimport]

David Li / @lidavidm:
Sounds reasonable to me, thanks for digging up the format & chiming in. Let's use this issue for URL-decoding paths in datasets (and even if we don't infer the type here, it should at least be readable by setting the partition schema). And I don't know if we've 'officially' said this, but presumably we want to maintain some level of Spark compatibility.

[asfimport]

Joris Van den Bossche / @jorisvandenbossche:
Agreed that we should try to be able to read what Spark produces, and that URL-decoding the paths in datasets before parsing the values with a typed converted sounds as a good default.

[asfimport]

Antoine Pitrou / @pitrou:
Issue resolved by pull request 10264
#10264