y2ksayed commented on code in PR #20334:
URL: https://github.com/apache/hudi/pull/20334#discussion_r4286942173
##########
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
Review Comment:
Good catch — fixed in the latest commit. The version list no longer uses
getInstantsOrderedByCompletionTime (which silently filters out nulls); it now
orders all completed instants by requested time via getInstants(), so nothing
is ever dropped from the count. Completion times are only used to build a rank
map for the out-of-order check, where instants without a known completion time
are simply skipped rather than excluded.
--
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]
