y2ksayed commented on code in PR #20334:
URL: https://github.com/apache/hudi/pull/20334#discussion_r4286963528
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSparkBaseAnalysis.scala:
##########
@@ -437,6 +446,56 @@ 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.
+ *
+ * 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.
+ * Instants are ordered by completion time rather than requested time:
completion order is
+ * append-only and therefore stable under multi-writer, whereas a
late-completing
+ * early-requested commit would otherwise renumber already-issued versions.
+ *
+ * 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.
+ */
+ 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 activeCommits = metaClient.getActiveTimeline.getCommitsTimeline
+
.filterCompletedInstants.getInstantsOrderedByCompletionTime.iterator().asScala.toSeq
+ val archivedCommits = metaClient.getArchivedTimeline.getCommitsTimeline
+
.filterCompletedInstants.getInstantsOrderedByCompletionTime.iterator().asScala.toSeq
+ // Archived instants always completed before active ones (archival only
moves old instants),
+ // so concatenation yields the global completion-time order.
+ val allCommits = archivedCommits ++ activeCommits
+ if (version >= allCommits.size) {
+ throw new HoodieAnalysisException(
+ s"Version $version is out of range: table has ${allCommits.size}
completed commit(s), " +
+ s"valid versions are 0 to ${allCommits.size - 1}")
+ }
+ allCommits(version.toInt).requestedTime
Review Comment:
Fixed in the latest commit. Versions are now ordered by requested time —
which is what the time-travel read path filters on — so every version maps to a
valid historical snapshot. Additionally, the resolver now detects out-of-order
multi-writer completion among data commits (commit/delta-commit; replace
commits are excluded as logically transparent) and fails with a clear error
directing to TIMESTAMP AS OF, instead of returning a misleading snapshot.
--
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]
