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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
