hudi-agent commented on code in PR #20334:
URL: https://github.com/apache/hudi/pull/20334#discussion_r4286993637
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSparkBaseAnalysis.scala:
##########
@@ -437,6 +447,85 @@ case class PostAnalysisRule(sparkSession: SparkSession)
extends Rule[LogicalPlan
case _ => plan
}
}
+
+ /**
+ * Resolves a Delta Lake style `VERSION AS OF <version>` time-travel version
to the
+ * corresponding commit instant on the table's timeline.
+ *
+ * Versions are 0-indexed, matching Delta Lake semantics: version 0 is the
table's first
+ * completed commit, version 1 the second, and so on. Instants are ordered
by requested time,
+ * which is what Hudi's time-travel read path filters on, so every version
maps to a valid
+ * historical snapshot.
+ *
+ * Versions are resolved against the full commit history (archived + active
timelines), so
+ * version numbers stay stable even after archival moves old commits off the
active timeline.
+ * Nothing is silently dropped: instants without a known completion time
(old archive entries)
+ * still count toward version numbering.
+ *
+ * Data-producing commits (commit, delta-commit) as well as clustering
replace commits all
+ * count as versions, mirroring Delta Lake where OPTIMIZE also bumps the
table version.
+ *
+ * Limitation: correct snapshots require commits to have completed in
requested order (always
+ * true for single-writer). If out-of-order multi-writer completion is
detected among instants
+ * with known completion times, resolution fails with an actionable error
instead of returning
+ * a misleading snapshot; use TIMESTAMP AS OF in that case.
+ */
+ private def resolveVersionToInstant(table: CatalogTable, versionExpr:
Expression): String = {
+ val version = versionExpr match {
+ case Literal(v: Int, IntegerType) => v.toLong
+ case Literal(v: Long, LongType) => v
+ case UnaryMinus(Literal(v: Int, IntegerType)) => -v.toLong
+ case UnaryMinus(Literal(v: Long, LongType)) => -v
+ case _ =>
+ throw new HoodieAnalysisException(
+ s"Version expression for time travel must be an integer literal,
got: ${versionExpr.sql}")
+ }
+ if (version < 0) {
+ throw new HoodieAnalysisException(s"Version for time travel must be
non-negative, got: $version")
+ }
+ val basePath =
table.storage.locationUri.map(CatalogUtils.URIToString(_)).getOrElse(
+ throw new HoodieAnalysisException("Unable to resolve table location for
version-based time travel"))
+ val metaClient = HoodieTableMetaClient.builder()
+ .setBasePath(basePath)
+
.setConf(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()))
+ .build()
+
+ val activeTimeline =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ val archivedTimeline =
metaClient.getArchivedTimeline.getCommitsTimeline.filterCompletedInstants
+
+ // All completed instants, ordered by requested time. Unlike
getInstantsOrderedByCompletionTime,
+ // nothing is dropped here: instants with unknown completion time still
count.
+ val allInstants = (activeTimeline.getInstants.asScala ++
archivedTimeline.getInstants.asScala).toSeq
+ .sortBy(_.requestedTime)
+
+ // Completion rank for instants with known completion times (archived
always completed first,
+ // so concatenation yields the global completion order). Used only to
detect out-of-order
+ // multi-writer completion; instants without a known completion time are
not checked.
+ val completionRank = {
+ val activeOrdered =
activeTimeline.getInstantsOrderedByCompletionTime.iterator().asScala.map(_.requestedTime)
+ val archivedOrdered =
archivedTimeline.getInstantsOrderedByCompletionTime.iterator().asScala.map(_.requestedTime)
+ (archivedOrdered ++ activeOrdered).toSeq.zipWithIndex.toMap
+ }
+ // Fail loudly on out-of-order completion instead of returning a
misleading snapshot.
+ // Only data commits are checked: replace (clustering) commits are
logically transparent and
+ // cannot produce a wrong snapshot.
+ val dataCommitRanks = allInstants
+ .filter(i => i.getAction == HoodieTimeline.COMMIT_ACTION || i.getAction
== HoodieTimeline.DELTA_COMMIT_ACTION)
+ .flatMap(i => completionRank.get(i.requestedTime))
+ if (dataCommitRanks != dataCommitRanks.sorted) {
Review Comment:
🤖 On table version 6, active-timeline completion times come from file
modification times (`InstantGeneratorV1`). If the table is copied or restored
with distcp or an S3 copy, those times reset in arbitrary order, and this check
would then throw on a single-writer table. Could the check be limited to table
version 8+, where completion time is persisted?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSparkBaseAnalysis.scala:
##########
@@ -437,6 +447,85 @@ case class PostAnalysisRule(sparkSession: SparkSession)
extends Rule[LogicalPlan
case _ => plan
}
}
+
+ /**
+ * Resolves a Delta Lake style `VERSION AS OF <version>` time-travel version
to the
+ * corresponding commit instant on the table's timeline.
+ *
+ * Versions are 0-indexed, matching Delta Lake semantics: version 0 is the
table's first
+ * completed commit, version 1 the second, and so on. Instants are ordered
by requested time,
+ * which is what Hudi's time-travel read path filters on, so every version
maps to a valid
+ * historical snapshot.
+ *
+ * Versions are resolved against the full commit history (archived + active
timelines), so
+ * version numbers stay stable even after archival moves old commits off the
active timeline.
+ * Nothing is silently dropped: instants without a known completion time
(old archive entries)
+ * still count toward version numbering.
+ *
+ * Data-producing commits (commit, delta-commit) as well as clustering
replace commits all
+ * count as versions, mirroring Delta Lake where OPTIMIZE also bumps the
table version.
+ *
+ * Limitation: correct snapshots require commits to have completed in
requested order (always
+ * true for single-writer). If out-of-order multi-writer completion is
detected among instants
+ * with known completion times, resolution fails with an actionable error
instead of returning
+ * a misleading snapshot; use TIMESTAMP AS OF in that case.
+ */
+ private def resolveVersionToInstant(table: CatalogTable, versionExpr:
Expression): String = {
+ val version = versionExpr match {
+ case Literal(v: Int, IntegerType) => v.toLong
+ case Literal(v: Long, LongType) => v
+ case UnaryMinus(Literal(v: Int, IntegerType)) => -v.toLong
+ case UnaryMinus(Literal(v: Long, LongType)) => -v
+ case _ =>
+ throw new HoodieAnalysisException(
+ s"Version expression for time travel must be an integer literal,
got: ${versionExpr.sql}")
+ }
+ if (version < 0) {
+ throw new HoodieAnalysisException(s"Version for time travel must be
non-negative, got: $version")
+ }
+ val basePath =
table.storage.locationUri.map(CatalogUtils.URIToString(_)).getOrElse(
+ throw new HoodieAnalysisException("Unable to resolve table location for
version-based time travel"))
+ val metaClient = HoodieTableMetaClient.builder()
+ .setBasePath(basePath)
+
.setConf(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()))
+ .build()
+
+ val activeTimeline =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ val archivedTimeline =
metaClient.getArchivedTimeline.getCommitsTimeline.filterCompletedInstants
+
+ // All completed instants, ordered by requested time. Unlike
getInstantsOrderedByCompletionTime,
+ // nothing is dropped here: instants with unknown completion time still
count.
+ val allInstants = (activeTimeline.getInstants.asScala ++
archivedTimeline.getInstants.asScala).toSeq
+ .sortBy(_.requestedTime)
+
+ // Completion rank for instants with known completion times (archived
always completed first,
+ // so concatenation yields the global completion order). Used only to
detect out-of-order
+ // multi-writer completion; instants without a known completion time are
not checked.
+ val completionRank = {
+ val activeOrdered =
activeTimeline.getInstantsOrderedByCompletionTime.iterator().asScala.map(_.requestedTime)
+ val archivedOrdered =
archivedTimeline.getInstantsOrderedByCompletionTime.iterator().asScala.map(_.requestedTime)
+ (archivedOrdered ++ activeOrdered).toSeq.zipWithIndex.toMap
+ }
+ // Fail loudly on out-of-order completion instead of returning a
misleading snapshot.
+ // Only data commits are checked: replace (clustering) commits are
logically transparent and
+ // cannot produce a wrong snapshot.
+ val dataCommitRanks = allInstants
+ .filter(i => i.getAction == HoodieTimeline.COMMIT_ACTION || i.getAction
== HoodieTimeline.DELTA_COMMIT_ACTION)
+ .flatMap(i => completionRank.get(i.requestedTime))
+ if (dataCommitRanks != dataCommitRanks.sorted) {
Review Comment:
🤖 non-blocking: Because the check covers the whole history, one out-of-order
completion makes VERSION AS OF fail for every version on that table from then
on, including versions that map to real snapshots. Would it make sense to check
only whether the first `version + 1` instants by requested time match the first
`version + 1` by completion rank?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSparkBaseAnalysis.scala:
##########
@@ -437,6 +447,85 @@ case class PostAnalysisRule(sparkSession: SparkSession)
extends Rule[LogicalPlan
case _ => plan
}
}
+
+ /**
+ * Resolves a Delta Lake style `VERSION AS OF <version>` time-travel version
to the
+ * corresponding commit instant on the table's timeline.
+ *
+ * Versions are 0-indexed, matching Delta Lake semantics: version 0 is the
table's first
+ * completed commit, version 1 the second, and so on. Instants are ordered
by requested time,
+ * which is what Hudi's time-travel read path filters on, so every version
maps to a valid
+ * historical snapshot.
+ *
+ * Versions are resolved against the full commit history (archived + active
timelines), so
+ * version numbers stay stable even after archival moves old commits off the
active timeline.
+ * Nothing is silently dropped: instants without a known completion time
(old archive entries)
+ * still count toward version numbering.
+ *
+ * Data-producing commits (commit, delta-commit) as well as clustering
replace commits all
+ * count as versions, mirroring Delta Lake where OPTIMIZE also bumps the
table version.
+ *
+ * Limitation: correct snapshots require commits to have completed in
requested order (always
+ * true for single-writer). If out-of-order multi-writer completion is
detected among instants
+ * with known completion times, resolution fails with an actionable error
instead of returning
+ * a misleading snapshot; use TIMESTAMP AS OF in that case.
+ */
+ private def resolveVersionToInstant(table: CatalogTable, versionExpr:
Expression): String = {
+ val version = versionExpr match {
+ case Literal(v: Int, IntegerType) => v.toLong
+ case Literal(v: Long, LongType) => v
+ case UnaryMinus(Literal(v: Int, IntegerType)) => -v.toLong
+ case UnaryMinus(Literal(v: Long, LongType)) => -v
+ case _ =>
+ throw new HoodieAnalysisException(
+ s"Version expression for time travel must be an integer literal,
got: ${versionExpr.sql}")
+ }
+ if (version < 0) {
+ throw new HoodieAnalysisException(s"Version for time travel must be
non-negative, got: $version")
+ }
+ val basePath =
table.storage.locationUri.map(CatalogUtils.URIToString(_)).getOrElse(
+ throw new HoodieAnalysisException("Unable to resolve table location for
version-based time travel"))
+ val metaClient = HoodieTableMetaClient.builder()
+ .setBasePath(basePath)
+
.setConf(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()))
+ .build()
+
+ val activeTimeline =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ val archivedTimeline =
metaClient.getArchivedTimeline.getCommitsTimeline.filterCompletedInstants
+
+ // All completed instants, ordered by requested time. Unlike
getInstantsOrderedByCompletionTime,
+ // nothing is dropped here: instants with unknown completion time still
count.
+ val allInstants = (activeTimeline.getInstants.asScala ++
archivedTimeline.getInstants.asScala).toSeq
Review Comment:
🤖 If archival runs between `getActiveTimeline` and `getArchivedTimeline`, or
crashes after writing the archive but before deleting the active instant files,
the same instant shows up in both. It would then be counted twice, which shifts
every later version, and the completion-rank check could throw by mistake.
Could we dedupe by `requestedTime` before sorting?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
