hudi-agent commented on code in PR #20334:
URL: https://github.com/apache/hudi/pull/20334#discussion_r4276337338
##########
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:
🤖 Versions are now ordered by completion time, but the requested time is
what gets passed to `TIME_TRAVEL_AS_OF_INSTANT`, and time travel filters on
requested time. So if writer A requests at t1, writer B requests at t2, and B
completes before A, then version k (t2) also includes A's data, while version
k+1 (t1) is missing B. Could the snapshot be resolved by completion time as
well, or could this case at least fail or be documented?
<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 +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:
🤖 `getInstantsOrderedByCompletionTime` drops instants whose completion time
is null. On table version 6 the archived V1 timeline takes completion time from
`stateTransitionTime`, which older archive entries don't have, so those commits
silently fall out of the count and every version number shifts. Could we fall
back to requested-time order for V1 timelines, or fail when an instant has no
completion time?
<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]
