You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "y2ksayed (via GitHub)" <gi...@apache.org> on 2026/10/14 17:02:19 UTC
[PR] [HUDI] Add Delta Lake style VERSION AS OF time travel support [hudi]
y2ksayed opened a new pull request, #20334:
URL: https://github.com/apache/hudi/pull/20334
## Summary
Adds Delta Lake style `VERSION AS OF` time-travel support to Hudi Spark SQL, closing a long-standing parity gap: until now, any version-based time-travel query (`VERSION AS OF`, `FOR VERSION AS OF`, `SYSTEM_VERSION AS OF`, `FOR SYSTEM_VERSION AS OF`) was explicitly rejected with `HoodieAnalysisException: Version expression is not supported for time travel`. Only `TIMESTAMP AS OF` worked, which forces users to know exact commit timestamps.
With this change:
```sql
SELECT * FROM my_hudi_table VERSION AS OF 0; -- state as of the first commit
SELECT * FROM my_hudi_table VERSION AS OF 2; -- state as of the third commit
```
## Design
- Versions are **0-indexed**, matching Delta Lake semantics: version 0 is the table's first completed commit, version 1 the second, and so on.
- The version is resolved in `ResolveReferences` (`HoodieSparkBaseAnalysis`) by reading the table's active timeline via `HoodieTableMetaClient` and mapping the version to the Nth completed commit/delta-commit instant in timeline order. The resolved instant then flows through the existing `TIME_TRAVEL_AS_OF_INSTANT` read path, so all existing snapshot-query machinery is reused untouched.
- Only integer literals are accepted as the version (a clear error is raised otherwise); negative versions and out-of-range versions fail fast with actionable messages.
- Archived (cleaned) commits are not addressable by version; this is documented in the code.
## Changes
- `hudi-spark-datasource/hudi-spark-common/.../analysis/HoodieSparkBaseAnalysis.scala`: replaced the version-travel rejection with version-to-instant resolution (`resolveVersionToInstant`); guard against specifying both timestamp and version.
- `hudi-spark-datasource/hudi-spark/src/test/.../dml/others/TestTimeTravelTable.scala`: new test `Test version based time travel` covering all four version spellings, versions 0 and 1 snapshot correctness, plus out-of-range, negative, and non-literal version error cases. Removed the old assertion that version travel is rejected.
## Testing
- New test coverage in `TestTimeTravelTable` as described above.
- Existing `TIMESTAMP AS OF` tests are untouched and continue to pass through the same code path.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI] Add Delta Lake style VERSION AS OF time travel support [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI] Add Delta Lake style VERSION AS OF time travel support [hudi]
Posted by "y2ksayed (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI] Add Delta Lake style VERSION AS OF time travel support [hudi]
Posted by "y2ksayed (via GitHub)" <gi...@apache.org>.
y2ksayed commented on PR #20334:
URL: https://github.com/apache/hudi/pull/20334#issuecomment-6174285381
Thanks for the review! Addressed all three points in the latest commit:
1. **Version stability across archival**: versions are now resolved against the full commit history (archived + active timelines), so `VERSION AS OF N` keeps pointing at the same snapshot after archival runs. Archived instants always completed before active ones, so concatenating the two completion-ordered sequences yields the global order.
2. **Ordering and what counts as a version**: instants are now ordered by completion time instead of requested time — completion order is append-only, so a late-completing early-requested commit under multi-writer can no longer renumber already-issued versions. Kept `getCommitsTimeline()` (commit, delta-commit, replace-commit), documenting that clustering replace commits count as versions, mirroring Delta Lake where OPTIMIZE also bumps the table version.
3. **Nit**: dropped the redundant `.toString` (`asOfInstant.get`).
@yihua — flagging for your take on the version semantics as requested: 0-indexed, completion-time ordered, stable across archival, replace commits included.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI] Add Delta Lake style VERSION AS OF time travel support [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20334:
URL: https://github.com/apache/hudi/pull/20334#discussion_r4273902721
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSparkBaseAnalysis.scala:
##########
@@ -437,6 +446,44 @@ 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. Both commit and delta-commit instants
+ * that produced data are counted, in timeline order. Only completed instants on the active
+ * timeline are addressable; archived (cleaned) commits cannot be resolved by 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 completedCommits = metaClient.getActiveTimeline.getCommitsTimeline
+ .filterCompletedInstants.getInstants.asScala.toSeq.sortBy(_.requestedTime)
Review Comment:
🤖 Because versions are counted from the start of the active timeline, archival shifts every number, so `VERSION AS OF 0` returns a different snapshot after each archival run with no error; could version N come from a stable source instead, or could we fail when the timeline has been archived? @yihua, since this decides what Hudi's version semantics are, could you weigh in?
<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,44 @@ 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. Both commit and delta-commit instants
+ * that produced data are counted, in timeline order. Only completed instants on the active
+ * timeline are addressable; archived (cleaned) commits cannot be resolved by 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 completedCommits = metaClient.getActiveTimeline.getCommitsTimeline
+ .filterCompletedInstants.getInstants.asScala.toSeq.sortBy(_.requestedTime)
+ if (version >= completedCommits.size) {
Review Comment:
🤖 `getCommitsTimeline` also counts compaction commits and clustering replacecommits, and sorting by `requestedTime` lets an earlier-requested multi-writer commit that completes later renumber versions that were already handed out; would it make sense to order by completion time and decide whether table services should count as versions?
<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:
##########
@@ -75,7 +84,7 @@ case class ResolveReferences(spark: SparkSession) extends Rule[LogicalPlan]
bucketSpec = table.bucketSpec,
className = table.provider.get,
options = table.storage.properties ++ pathOption ++ Map(
- DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT.key -> timestamp.get.toString()),
+ DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT.key -> asOfInstant.get.toString),
Review Comment:
🤖 nit: could you drop `.get.toString` and use `asOfInstant.get`? `asOfInstant` is already an `Option[String]`, so the `toString` is redundant.
<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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI] Add Delta Lake style VERSION AS OF time travel support [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI] Add Delta Lake style VERSION AS OF time travel support [hudi]
Posted by "y2ksayed (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org