feat: [HUDI-9775] Support for show_metadata_table_history with appropriate matching data table info#14303
Open
PavithranRick wants to merge 4 commits intoapache:masterfrom
Open
feat: [HUDI-9775] Support for show_metadata_table_history with appropriate matching data table info#14303PavithranRick wants to merge 4 commits intoapache:masterfrom
PavithranRick wants to merge 4 commits intoapache:masterfrom
Conversation
…data table info matched
…data table info matched
Collaborator
Contributor
|
@CTTY could you help review this PR? |
CTTY
reviewed
Dec 14, 2025
Contributor
CTTY
left a comment
CTTY
left a comment
There was a problem hiding this comment.
Hi @PavithranRick , thanks for the contribution! I've left some comments. Haven't got a chance to check the unit test yet but will do later
|
|
||
| private val PARAMETERS = Array[ProcedureParameter]( | ||
| ProcedureParameter.optional(0, "table", DataTypes.StringType), | ||
| ProcedureParameter.optional(1, "path", DataTypes.StringType, ""), |
Contributor
There was a problem hiding this comment.
Suggested change
| ProcedureParameter.optional(1, "path", DataTypes.StringType, ""), | |
| ProcedureParameter.optional(1, "path", DataTypes.StringType), |
path should not have default value
Contributor
There was a problem hiding this comment.
should the table and path be required rather than optional?
Comment on lines
+39
to
+41
| ProcedureParameter.optional(4, "filter", DataTypes.StringType, ""), | ||
| ProcedureParameter.optional(5, "startTime", DataTypes.StringType, ""), | ||
| ProcedureParameter.optional(6, "endTime", DataTypes.StringType, "") |
Contributor
There was a problem hiding this comment.
nit: We do not have to assign default values for these
|
|
||
| val filteredEntries = applyFilter(entries, filter, outputType) | ||
|
|
||
| val finalEntries = if (startTime.nonEmpty && endTime.nonEmpty) { |
Contributor
There was a problem hiding this comment.
Could it be possible that only startTime is empty, and endTime is not?
Comment on lines
+127
to
+148
| val filteredInstantTimes = if (startTime.nonEmpty && endTime.nonEmpty) { | ||
| allInstantTimes.filter { instantTime => | ||
| val timeMatches = (instantTime >= startTime && instantTime <= endTime) | ||
| timeMatches | ||
| } | ||
| } else { | ||
| allInstantTimes | ||
| } | ||
|
|
||
| val sortedInstantTimes = filteredInstantTimes.toSeq.sorted(Ordering[String].reverse) | ||
|
|
||
| val entries = sortedInstantTimes.map { instantTime => | ||
| createTimelineEntry( | ||
| instantTime, | ||
| dataActiveTimeline, | ||
| dataArchivedTimeline, | ||
| metadataActiveTimeline, | ||
| metadataArchivedTimeline, | ||
| dataInstantInfoMap, | ||
| metadataInstantInfoMap | ||
| ) | ||
| } |
Contributor
There was a problem hiding this comment.
I think this is kinda verbose, maybe
Suggested change
| val filteredInstantTimes = if (startTime.nonEmpty && endTime.nonEmpty) { | |
| allInstantTimes.filter { instantTime => | |
| val timeMatches = (instantTime >= startTime && instantTime <= endTime) | |
| timeMatches | |
| } | |
| } else { | |
| allInstantTimes | |
| } | |
| val sortedInstantTimes = filteredInstantTimes.toSeq.sorted(Ordering[String].reverse) | |
| val entries = sortedInstantTimes.map { instantTime => | |
| createTimelineEntry( | |
| instantTime, | |
| dataActiveTimeline, | |
| dataArchivedTimeline, | |
| metadataActiveTimeline, | |
| metadataArchivedTimeline, | |
| dataInstantInfoMap, | |
| metadataInstantInfoMap | |
| ) | |
| } | |
| val entries = | |
| allInstantTimes | |
| .filter(t => startTime.nonEmpty && endTime.nonEmpty || (t >= startTime && t <= endTime)) | |
| .toSeq | |
| .sorted(Ordering.String.reverse) | |
| .map { instantTime => | |
| createTimelineEntry( | |
| instantTime, | |
| dataActiveTimeline, | |
| dataArchivedTimeline, | |
| metadataActiveTimeline, | |
| metadataArchivedTimeline, | |
| dataInstantInfoMap, | |
| metadataInstantInfoMap | |
| ) | |
| } |
| completionTime: String, | ||
| modificationTimeMs: Long | ||
| ) { | ||
| def getModificationTime: Long = modificationTimeMs |
Contributor
There was a problem hiding this comment.
iirc, scala case class variables can be directly accessed like instantWithModTime.modificationTime, and we don't need an extra getter
| instantMap.getOrElseUpdate(instant.requestedTime(), scala.collection.mutable.Map[HoodieInstant.State, HoodieInstantWithModTime]()) | ||
| .put(instant.getState, instantWithModTime) | ||
| } catch { | ||
| case _: Exception => // Skip invalid files |
| timestamp -> stateMap.toMap | ||
| }.toMap | ||
| } catch { | ||
| case _: Exception => Map.empty[String, Map[HoodieInstant.State, HoodieInstantWithModTime]] |
Comment on lines
+266
to
+284
| val stateMap = instantInfoMap.get(instantTimestamp) | ||
| if (stateMap.isDefined) { | ||
| val stateInfo = stateMap.get.get(state) | ||
| if (stateInfo.isDefined) { | ||
| val modificationTime = stateInfo.get.getModificationTime | ||
| if (modificationTime > 0) { | ||
| val date = new java.util.Date(modificationTime) | ||
| val formatter = new java.text.SimpleDateFormat("MM-dd HH:mm:ss") | ||
| formatter.format(date) | ||
| } else { | ||
| instantTimestamp | ||
| } | ||
| } else { | ||
| null | ||
| } | ||
| } else { | ||
| null | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
We can simplify this to
Suggested change
| val stateMap = instantInfoMap.get(instantTimestamp) | |
| if (stateMap.isDefined) { | |
| val stateInfo = stateMap.get.get(state) | |
| if (stateInfo.isDefined) { | |
| val modificationTime = stateInfo.get.getModificationTime | |
| if (modificationTime > 0) { | |
| val date = new java.util.Date(modificationTime) | |
| val formatter = new java.text.SimpleDateFormat("MM-dd HH:mm:ss") | |
| formatter.format(date) | |
| } else { | |
| instantTimestamp | |
| } | |
| } else { | |
| null | |
| } | |
| } else { | |
| null | |
| } | |
| } | |
| val formatter = new java.text.SimpleDateFormat("MM-dd HH:mm:ss") | |
| instantInfoMap.get(instantTimestamp) | |
| .flatMap(_.get(state)) | |
| .map(_.modificationTimeMs) | |
| .filter(_ > 0) | |
| .map(time => formatter.format(new java.util.Date(time))) | |
| .getOrElse(instantTimestamp) | |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe the issue this Pull Request addresses
This PR introduces the
ShowMetadataTableHistoryProcedure, a new Spark SQL procedure that displays timeline information for both the data table and metadata table side-by-side, enabling analysis of metadata table synchronization and evolution.Summary and Changelog
MM-dd HH:mm:ssformatting for requested/inflight/completed timesshowArchivedparameterstartTime/endTime)Impact
show_metadata_table_history(table, path, limit, showArchived, filter, startTime, endTime)Risk Level
Low
Verification performed:
Documentation Update
Update Hudi Spark SQL procedures documentation to include
show_metadata_table_historyusage examples and parameter descriptions.Contributor's checklist