wombatu-kun commented on code in PR #20526:
URL: https://github.com/apache/hudi/pull/20526#discussion_r4490645741
##########
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:
Done f817838133ed: that `date_format` filter did throw (`long overflow`),
and `testFormatGuardToleratesFailingBounds` now pins that it keeps the file.
`Step.guarded` leaves a bare stats column unwrapped, as it cannot fail, so only
the 19 golden lines whose guard reads a derived bound change.
##########
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:
Done f817838133ed.
##########
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:
Done f817838133ed. file_1 now starts 30 seconds before the first 1944
instant, with `< DATE'1944-01-01'`, so it is kept only because its guard is
unknown, not because its min matches.
--
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]
