wombatu-kun commented on code in PR #20519:
URL: https://github.com/apache/hudi/pull/20519#discussion_r4478637396
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/TimeZoneShiftRule.scala:
##########
@@ -17,22 +17,39 @@
package org.apache.spark.sql.hudi.dataskipping.rules
-import org.apache.spark.sql.catalyst.expressions.{Expression,
FromUTCTimestamp, ToUTCTimestamp}
+import org.apache.spark.sql.catalyst.expressions.{Expression,
FromUTCTimestamp, Literal, ToUTCTimestamp}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.hudi.dataskipping.{Increasing, RangeRule, Step}
+import scala.util.Try
+
/**
* Time zone shifts: `FromUTCTimestamp`, `ToUTCTimestamp`.
*
- * Shared guard: none; any time zone operand is accepted today.
+ * Shared guard: [[isFixedOffsetTimeZone]], a literal time zone whose offset
never changes.
*
- * NOTE: The offset jumps at DST transitions, so the shift is not monotonic
across a fall-back (#20449)
+ * NOTE: Both shift by the zone's offset at that time, so they only keep the
ordering for a zone whose offset never
+ * changes, DST or historical. Local time runs backwards when the offset
drops (for ex, 05:30Z is 01:30 EDT but
+ * the later 06:10Z is 01:10 EST), and a local time in the gap of a
rising offset shifts past later ones (for ex,
+ * 02:30 maps to 07:30Z but the later 03:00 to 07:00Z)
*/
object TimeZoneShiftRule extends RangeRule {
override def steps(term: Expression): Seq[Step] = term match {
- case FromUTCTimestamp(timestamp, _) => Seq(Step(term, timestamp,
Increasing, fileGuard = None, partial = false))
- case ToUTCTimestamp(timestamp, _) => Seq(Step(term, timestamp, Increasing,
fileGuard = None, partial = false))
+ case FromUTCTimestamp(timestamp, timeZone) if
isFixedOffsetTimeZone(timeZone) =>
+ Seq(Step(term, timestamp, Increasing, fileGuard = None, partial = false))
+ case ToUTCTimestamp(timestamp, timeZone) if
isFixedOffsetTimeZone(timeZone) =>
+ Seq(Step(term, timestamp, Increasing, fileGuard = None, partial = false))
case _ => Nil
}
+
+ // Only a literal zone is matched, since the optimizer folds a constant zone
to a literal before data
+ // skipping runs. It is resolved the way Spark evaluates it, so for ex the
short ID EST is a fixed
+ // -05:00 while PST is America/Los_Angeles; an unknown zone is not matched
+ private def isFixedOffsetTimeZone(expr: Expression): Boolean = expr match {
+ case Literal(timeZone, _) if timeZone != null =>
+
Try(DateTimeUtils.getZoneId(timeZone.toString).getRules.isFixedOffset).getOrElse(false)
Review Comment:
Noted under #20449:
https://github.com/apache/hudi/issues/20449#issuecomment-6654220058
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCatalystExpressionUtils.scala:
##########
@@ -117,6 +117,23 @@ class TestHoodieCatalystExpressionUtils extends
SparkAdapterSupport {
assertEquals(Some(dateAttr), matched(DateDiff(dateAttr,
Literal(java.sql.Date.valueOf("2022-03-07")))))
}
+ @Test
+ def testUtcTimestampConversionsRequireFixedOffsetZone(): Unit = {
+ // A zone with DST runs local time backwards across a fall-back and shifts
the spring-forward gap.
+ assertEquals(None, matched(FromUTCTimestamp(tsAttr,
Literal("America/New_York"))))
+ assertEquals(None, matched(ToUTCTimestamp(tsAttr,
Literal("America/New_York"))))
+ // A fixed-offset zone shifts every timestamp by the same amount.
+ assertEquals(Some(tsAttr), matched(FromUTCTimestamp(tsAttr,
Literal("+08:00"))))
+ assertEquals(Some(tsAttr), matched(ToUTCTimestamp(tsAttr, Literal("UTC"))))
+ // Zones are resolved the way Spark evaluates them: the short ID EST is a
fixed -05:00, and +8:00 is +08:00.
+ assertEquals(Some(tsAttr), matched(FromUTCTimestamp(tsAttr,
Literal("EST"))))
+ assertEquals(Some(tsAttr), matched(ToUTCTimestamp(tsAttr,
Literal("+8:00"))))
+ // Non-literal, null and unknown zones are not matched.
+ assertEquals(None, matched(FromUTCTimestamp(tsAttr, strAttr)))
+ assertEquals(None, matched(FromUTCTimestamp(tsAttr, Literal(null,
StringType))))
+ assertEquals(None, matched(ToUTCTimestamp(tsAttr, Literal("Not/AZone"))))
Review Comment:
Done e86de12c33d7
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/TimeZoneShiftRule.scala:
##########
@@ -17,22 +17,39 @@
package org.apache.spark.sql.hudi.dataskipping.rules
-import org.apache.spark.sql.catalyst.expressions.{Expression,
FromUTCTimestamp, ToUTCTimestamp}
+import org.apache.spark.sql.catalyst.expressions.{Expression,
FromUTCTimestamp, Literal, ToUTCTimestamp}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.hudi.dataskipping.{Increasing, RangeRule, Step}
+import scala.util.Try
+
/**
* Time zone shifts: `FromUTCTimestamp`, `ToUTCTimestamp`.
*
- * Shared guard: none; any time zone operand is accepted today.
+ * Shared guard: [[isFixedOffsetTimeZone]], a literal time zone whose offset
never changes.
*
- * NOTE: The offset jumps at DST transitions, so the shift is not monotonic
across a fall-back (#20449)
+ * NOTE: Both shift by the zone's offset at that time, so they only keep the
ordering for a zone whose offset never
+ * changes, DST or historical. Local time runs backwards when the offset
drops (for ex, 05:30Z is 01:30 EDT but
+ * the later 06:10Z is 01:10 EST), and a local time in the gap of a
rising offset shifts past later ones (for ex,
+ * 02:30 maps to 07:30Z but the later 03:00 to 07:00Z)
*/
object TimeZoneShiftRule extends RangeRule {
override def steps(term: Expression): Seq[Step] = term match {
- case FromUTCTimestamp(timestamp, _) => Seq(Step(term, timestamp,
Increasing, fileGuard = None, partial = false))
- case ToUTCTimestamp(timestamp, _) => Seq(Step(term, timestamp, Increasing,
fileGuard = None, partial = false))
+ case FromUTCTimestamp(timestamp, timeZone) if
isFixedOffsetTimeZone(timeZone) =>
+ Seq(Step(term, timestamp, Increasing, fileGuard = None, partial = false))
+ case ToUTCTimestamp(timestamp, timeZone) if
isFixedOffsetTimeZone(timeZone) =>
+ Seq(Step(term, timestamp, Increasing, fileGuard = None, partial = false))
case _ => Nil
}
+
+ // Only a literal zone is matched, since the optimizer folds a constant zone
to a literal before data
+ // skipping runs. It is resolved the way Spark evaluates it, so for ex the
short ID EST is a fixed
+ // -05:00 while PST is America/Los_Angeles; an unknown zone is not matched
+ private def isFixedOffsetTimeZone(expr: Expression): Boolean = expr match {
Review Comment:
Will do in whichever of the two lands second; that rebase has to resolve the
conflict between them anyway.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/TimeZoneShiftRule.scala:
##########
@@ -17,22 +17,39 @@
package org.apache.spark.sql.hudi.dataskipping.rules
-import org.apache.spark.sql.catalyst.expressions.{Expression,
FromUTCTimestamp, ToUTCTimestamp}
+import org.apache.spark.sql.catalyst.expressions.{Expression,
FromUTCTimestamp, Literal, ToUTCTimestamp}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.hudi.dataskipping.{Increasing, RangeRule, Step}
+import scala.util.Try
+
/**
* Time zone shifts: `FromUTCTimestamp`, `ToUTCTimestamp`.
*
- * Shared guard: none; any time zone operand is accepted today.
+ * Shared guard: [[isFixedOffsetTimeZone]], a literal time zone whose offset
never changes.
Review Comment:
Updated the description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
