voonhous commented on code in PR #20526:
URL: https://github.com/apache/hudi/pull/20526#discussion_r4489879035
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/DateTimeFormatRule.scala:
##########
@@ -51,12 +52,18 @@ import scala.util.Try
object DateTimeFormatRule extends RangeRule {
override def steps(term: Expression): Seq[Step] = term match {
- case DateFormatClass(timestamp, format, _)
+ case DateFormatClass(timestamp, format, timeZoneId)
Review Comment:
**major:** This conflicts with #20521, which rewrote these two arms and the
`DateTimeOrdering` scaladoc. Taking either side alone drops a guard: this side
calls the 1-arg `formatPreservesOrder(format)` and loses #20521's fixed-offset
check on time-of-day patterns. Could the resolution keep
`formatPreservesOrder(format, timeZoneId)` in the case guard, wrap the step in
`localDateReversals(timeZoneId)` as here, and delete both stale NOTEs (master's
"date-only format is assumed monotone ... see #20525" and this one's "guarded
elsewhere, by #20521")?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/DateTimeOrdering.scala:
##########
@@ -124,4 +133,59 @@ object DateTimeOrdering {
// not fixed
private def isFixedOffsetZone(timeZoneId: String): Boolean =
Try(DateTimeUtils.getZoneId(timeZoneId).getRules.isFixedOffset).getOrElse(false)
+
+ private val localDateReversalsByZoneId = new ConcurrentHashMap[String,
Option[Seq[Instant]]]()
+
+ /**
+ * Instants at which the local date runs backwards in the zone, resolved the
way Spark evaluates it and
+ * cached per zone id (see the overload over `ZoneRules`). None, so that the
rendering takes no step, if
+ * there is no zone (not analyzed), the zone is unknown, or its current
rules keep doing it.
+ */
+ def localDateReversals(timeZoneId: Option[String]): Option[Seq[Instant]] =
+ timeZoneId.flatMap { id =>
+ localDateReversalsByZoneId.computeIfAbsent(id,
+ _ => Try(DateTimeUtils.getZoneId(id).getRules).toOption.flatMap(rules
=> localDateReversals(rules)))
+ }
+
+ /**
+ * Instants at which the local date runs backwards in a zone with the given
rules, that is where its clocks fell back
+ * across midnight, or None if its current rules keep doing it. Falling back
at midnight itself (for ex, Asia/Beirut)
+ * repeats the previous date without running backwards, since the local date
before the transition is that of the
+ * instant just before it
+ */
+ def localDateReversals(rules: ZoneRules): Option[Seq[Instant]] = {
+ def isReversal(transition: ZoneOffsetTransition): Boolean =
transition.isOverlap &&
+
transition.getDateTimeAfter.toLocalDate.isBefore(transition.getDateTimeBefore.minusNanos(1).toLocalDate)
+ // NOTE: A rule falls back at the same time of day every year, so a single
year shows whether it reverses the date
+ if (rules.getTransitionRules.asScala.exists(rule =>
isReversal(rule.createTransition(2000)))) {
+ None
+ } else {
+
Some(rules.getTransitions.asScala.filter(isReversal).map(_.getInstant).toList)
+ }
+ }
+
+ /**
+ * File guard of rendering a timestamp's local date: the file's range spans
none of `instants`, since only
+ * a file spanning one can hold rows that render outside of [f(lo), f(hi)].
None, i.e. unguarded, without
+ * instants.
+ *
+ * @param toLiteral an instant as a value of the rendered input:
[[asTimestamp]] or [[asEpochSeconds]]
+ */
+ def noLocalDateReversal(instants: Seq[Instant], toLiteral: Instant =>
Literal): Option[Guard] =
+ if (instants.isEmpty) {
+ None
+ } else {
+ // NOTE: A one-sided verdict (for ex, `<`) reads one bound, while the
guard reads both, so a bound that cannot be
+ // evaluated (for ex, from_utc_timestamp of the largest timestamp)
is NULL here, which keeps the file
+ Some { r =>
+ val (lo, hi) = (TryEval(r.lo), TryEval(r.hi))
+ instants.map(toLiteral).map(t => Not(And(LessThan(lo, t),
GreaterThanOrEqual(hi, t)))).reduce[Expression](And(_, _))
Review Comment:
**nit:** In St_Johns this is 24 conjuncts; `Step.guarded` copies it into
both bounds and `Verdict` repeats both bounds per `IN` value, so `CAST(ts AS
DATE) IN (100 dates)` compiles to about 9600 comparisons per stats row. Only
the 20 reversal zones pay it, so feel free to ignore, but could the guard lead
with `hi < first OR lo >= last` (the instants are sorted), so post-2010 files
settle in two comparisons?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/DateTimeOrdering.scala:
##########
@@ -124,4 +133,59 @@ object DateTimeOrdering {
// not fixed
private def isFixedOffsetZone(timeZoneId: String): Boolean =
Try(DateTimeUtils.getZoneId(timeZoneId).getRules.isFixedOffset).getOrElse(false)
+
+ private val localDateReversalsByZoneId = new ConcurrentHashMap[String,
Option[Seq[Instant]]]()
+
+ /**
+ * Instants at which the local date runs backwards in the zone, resolved the
way Spark evaluates it and
+ * cached per zone id (see the overload over `ZoneRules`). None, so that the
rendering takes no step, if
+ * there is no zone (not analyzed), the zone is unknown, or its current
rules keep doing it.
+ */
+ def localDateReversals(timeZoneId: Option[String]): Option[Seq[Instant]] =
+ timeZoneId.flatMap { id =>
+ localDateReversalsByZoneId.computeIfAbsent(id,
+ _ => Try(DateTimeUtils.getZoneId(id).getRules).toOption.flatMap(rules
=> localDateReversals(rules)))
+ }
+
+ /**
+ * Instants at which the local date runs backwards in a zone with the given
rules, that is where its clocks fell back
+ * across midnight, or None if its current rules keep doing it. Falling back
at midnight itself (for ex, Asia/Beirut)
+ * repeats the previous date without running backwards, since the local date
before the transition is that of the
+ * instant just before it
+ */
+ def localDateReversals(rules: ZoneRules): Option[Seq[Instant]] = {
+ def isReversal(transition: ZoneOffsetTransition): Boolean =
transition.isOverlap &&
+
transition.getDateTimeAfter.toLocalDate.isBefore(transition.getDateTimeBefore.minusNanos(1).toLocalDate)
+ // NOTE: A rule falls back at the same time of day every year, so a single
year shows whether it reverses the date
+ if (rules.getTransitionRules.asScala.exists(rule =>
isReversal(rule.createTransition(2000)))) {
+ None
+ } else {
+
Some(rules.getTransitions.asScala.filter(isReversal).map(_.getInstant).toList)
+ }
+ }
+
+ /**
+ * File guard of rendering a timestamp's local date: the file's range spans
none of `instants`, since only
+ * a file spanning one can hold rows that render outside of [f(lo), f(hi)].
None, i.e. unguarded, without
+ * instants.
+ *
+ * @param toLiteral an instant as a value of the rendered input:
[[asTimestamp]] or [[asEpochSeconds]]
+ */
+ def noLocalDateReversal(instants: Seq[Instant], toLiteral: Instant =>
Literal): Option[Guard] =
+ if (instants.isEmpty) {
+ None
+ } else {
+ // NOTE: A one-sided verdict (for ex, `<`) reads one bound, while the
guard reads both, so a bound that cannot be
+ // evaluated (for ex, from_utc_timestamp of the largest timestamp)
is NULL here, which keeps the file
+ Some { r =>
+ val (lo, hi) = (TryEval(r.lo), TryEval(r.hi))
Review Comment:
**minor:** Only this guard reads its bounds through `TryEval`.
`formattableInstants` / `formattableSeconds` read `r.lo` / `r.hi` raw and
`TimeZoneShiftRule` has no overflow guard, so I'd expect
`date_format(from_utc_timestamp(C, '+08:00'), 'yyyy-MM-dd') < '2000-01-01'`
over the Phoenix test's max to still throw, in any zone. Not blocking, but
could `Step.guarded` wrap the bounds once, `guard(r.copy(lo = TryEval(r.lo), hi
= TryEval(r.hi)))`, and drop the local `TryEval` here?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala:
##########
@@ -165,6 +165,68 @@ class TestDataSkippingUtils extends
HoodieSparkClientTestBase with SparkAdapterS
}
}
+ @Test
+ def testLocalDateAcrossMidnightFallBack(): Unit = {
+ withSqlConf(SESSION_LOCAL_TIMEZONE.key -> "America/St_Johns") {
+ // At 02:31Z on 2009-11-01 the clocks went from 00:01 NDT back to 23:01
NST of the previous day, so 02:30:30Z
+ // renders as 2009-11-01 but 02:31Z as 2009-10-31. Re-applying a date
rendering over min/max stats puts the max
+ // below the min and would wrongly prune file_1 for either date; must
keep the file
+ val acrossFallBack = IndexRow("file_1", valueCount = 2,
+ A_minValue = 1257042630L, A_maxValue = 1257042660L, A_nullCount = 0,
// epoch seconds of C's bounds
+ C_minValue = new Timestamp(1257042630000L), // 2009-11-01 02:30:30Z
+ C_maxValue = new Timestamp(1257042660000L), // 2009-11-01 02:31:00Z,
the first instant after the fall-back
+ C_nullCount = 0)
+ // Files that do not span the fall-back are still pruned
+ val afterFallBack = IndexRow("file_2", valueCount = 2,
+ A_minValue = 1257042660L, A_maxValue = 1257043200L, A_nullCount = 0,
+ C_minValue = new Timestamp(1257042660000L), // 2009-11-01 02:31:00Z,
renders as 2009-10-31
+ C_maxValue = new Timestamp(1257043200000L), // 2009-11-01 02:40:00Z,
renders as 2009-10-31
+ C_nullCount = 0)
+ val earlierDay = IndexRow("file_3", valueCount = 2,
+ A_minValue = 1256904000L, A_maxValue = 1256907600L, A_nullCount = 0,
+ C_minValue = new Timestamp(1256904000000L), // 2009-10-30 12:00Z
+ C_maxValue = new Timestamp(1256907600000L), // 2009-10-30 13:00Z
+ C_nullCount = 0)
+ val files = Seq(acrossFallBack, afterFallBack, earlierDay)
+ Seq(
+ "date_format(C, 'yyyy-MM-dd') = '2009-10-31'" -> Seq("file_1",
"file_2"),
+ "date_format(C, 'yyyy-MM-dd') = '2009-11-01'" -> Seq("file_1"),
+ "from_unixtime(A, 'yyyy-MM-dd') = '2009-10-31'" -> Seq("file_1",
"file_2"),
+ "CAST(C AS DATE) IN (DATE'2009-10-29', DATE'2009-10-31')" ->
Seq("file_1", "file_2"),
+ "date_add(CAST(C AS DATE), 1) = DATE'2009-11-01'" -> Seq("file_1",
"file_2")
+ ).foreach { case (filter, expected) =>
+ assertEquals(expected, applyFilterExpr(resolveExpr(spark, filter,
sourceTableSchema), files), filter)
+ }
+ // to_date reaches data skipping as the cast it stands for, once the
optimizer replaced it. Two dates: with one,
+ // the optimizer turns IN into "=", which Spark 3.5+ rewrites into a
range on C itself
+ assertEquals(Seq("file_1", "file_2"),
+ lookupThroughFullOptimizer("to_date(C) IN (DATE'2009-10-29',
DATE'2009-10-31')", files))
+
+ // The same date at both ends does not rule out a row of the previous
date in between
+ val aroundFallBack = IndexRow("file_4", valueCount = 3,
+ C_minValue = new Timestamp(1257042630000L), // 2009-11-01 02:30:30Z,
renders as 2009-11-01
+ C_maxValue = new Timestamp(1257046800000L), // 2009-11-01 03:40:00Z,
renders as 2009-11-01
+ C_nullCount = 0)
+ assertEquals(Seq("file_4"), applyFilterExpr(
+ resolveExpr(spark, "NOT (date_format(C, 'yyyy-MM-dd') =
'2009-11-01')", sourceTableSchema), Seq(aroundFallBack)))
+ }
+ }
+
+ @Test
+ def testLocalDateGuardToleratesFailingBounds(): Unit = {
Review Comment:
**minor:** This also passes without the guard: `<` reads only the min, so
the base keeps file_1 too and the test only pins "does not throw". Not
blocking, but could we add a second file with min 2001-01-01 (past both 1944
instants) and the same max, expected pruned? That checks that a NULL hi does
not keep every file.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/rules/DateTimeOrdering.scala:
##########
@@ -124,4 +133,59 @@ object DateTimeOrdering {
// not fixed
private def isFixedOffsetZone(timeZoneId: String): Boolean =
Try(DateTimeUtils.getZoneId(timeZoneId).getRules.isFixedOffset).getOrElse(false)
+
+ private val localDateReversalsByZoneId = new ConcurrentHashMap[String,
Option[Seq[Instant]]]()
Review Comment:
**minor:** After #20521, zone resolution lives in
`dataskipping/TimeZones.scala` (`isFixedOffsetZone`), so this adds a second
resolver, and the private `isFixedOffsetZone` at L134 becomes dead after the
rebase. Not blocking, but could `localDateReversals` (both overloads and the
cache) move to `TimeZones`, with `CastRule` and `DateTimeFormatRule` calling it
there?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCatalystExpressionUtils.scala:
##########
@@ -269,9 +272,13 @@ class TestHoodieCatalystExpressionUtils extends
SparkAdapterSupport {
assertEquals(None, matched(Cast(dateAttr, StringType)))
assertEquals(None, matched(Cast(dateAttr, IntegerType)))
assertEquals(None, matched(Cast(boolAttr, DecimalType(1, 1))))
- // A date maps to the start of its day, and truncating a timestamp to its
date is non-decreasing.
+ // A date maps to the start of its day.
assertEquals(Some(dateAttr), matched(Cast(dateAttr, TimestampType)))
- assertEquals(Some(tsAttr), matched(Cast(tsAttr, DateType)))
+ // A timestamp's local date runs backwards where the clocks fell back
across midnight, so it is
+ // ranged with a per-file guard there (America/St_Johns), and not at all
without a time zone.
+ assertEquals(Some(tsAttr), matched(Cast(tsAttr, DateType, Some("UTC"))))
+ assertEquals(Some(tsAttr), matched(Cast(tsAttr, DateType,
Some("America/St_Johns"))))
Review Comment:
**minor:** These two `Some(tsAttr)` assertions also pass on 9c75d0b, where
the cast was ranged unguarded, so they do not show the guard. Not blocking, but
could they assert `TermRange.derive(Cast(tsAttr, DateType,
Some("America/St_Johns")),
ColumnStatsSource(Seq("ts"))).get.nullBoundIsUnknown` is true, and false for
UTC?
--
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]
