voonhous commented on code in PR #13147:
URL: https://github.com/apache/hudi/pull/13147#discussion_r3589248071
##########
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/TimelineHandler.java:
##########
@@ -46,4 +135,185 @@ public List<InstantDTO> getLastInstant(String basePath) {
public TimelineDTO getTimeline(String basePath) {
return
TimelineDTO.fromTimeline(viewManager.getFileSystemView(basePath).getTimeline());
}
+
+ public UiTimelineDTO getUiTimeline(String basePath) {
+ // The active timeline is used, not the file-system-view write timeline:
the latter drops
+ // clean/rollback/savepoint/restore/indexing actions and all
requested/inflight states.
+ return
UiTimelineDTO.fromTimeline(createMetaClient(basePath).getActiveTimeline());
+ }
+
+ public Object getInstantDetails(String basePath, String requestedTime,
String action, String state) {
+ HoodieInstant.State parsedState;
+ try {
+ parsedState = HoodieInstant.State.valueOf(state);
+ } catch (IllegalArgumentException e) {
+ throw new BadRequestResponse("Invalid instant state: " + state);
+ }
+
+ if
(!Arrays.asList(HoodieTimeline.VALID_ACTIONS_IN_TIMELINE).contains(action)) {
+ throw new BadRequestResponse("Invalid instant action: " + action);
+ }
+
+ HoodieTableMetaClient metaClient = createMetaClient(basePath);
+ HoodieTimeline activeTimeline = metaClient.getActiveTimeline();
+ CommitMetadataSerDe serde = metaClient.getCommitMetadataSerDe();
+
+ // Resolve the instant against the timeline rather than constructing it
from request params:
+ // an attacker-controlled instant would otherwise flow into a StoragePath
whose URI.normalize
+ // collapses ".." segments, enabling path traversal.
+ HoodieInstant instant = activeTimeline.getInstantsAsStream()
+ .filter(i -> i.requestedTime().equals(requestedTime)
+ && i.getAction().equals(action)
+ && i.getState() == parsedState)
+ .findFirst()
+ .orElseThrow(() -> new NotFoundResponse(
+ "Instant not found in active timeline: " + requestedTime + " " +
action + " " + parsedState));
+
+ try {
+ Object result;
+ switch (instant.getAction()) {
+ case HoodieTimeline.COMMIT_ACTION:
+ case HoodieTimeline.DELTA_COMMIT_ACTION:
+ result = readCommitMetadata(serde, activeTimeline, instant);
+ break;
+ case HoodieTimeline.CLEAN_ACTION:
+ result = instant.isCompleted()
+ ? readAs(serde, activeTimeline, instant,
HoodieCleanMetadata.class)
+ : readAs(serde, activeTimeline, requestedTwin(metaClient,
instant), HoodieCleanerPlan.class);
+ break;
+ case HoodieTimeline.ROLLBACK_ACTION:
+ result = instant.isCompleted()
+ ? readAs(serde, activeTimeline, instant,
HoodieRollbackMetadata.class)
+ : readAs(serde, activeTimeline, requestedTwin(metaClient,
instant), HoodieRollbackPlan.class);
+ break;
+ case HoodieTimeline.RESTORE_ACTION:
+ result = instant.isCompleted()
+ ? readAs(serde, activeTimeline, instant,
HoodieRestoreMetadata.class)
+ : readAs(serde, activeTimeline, requestedTwin(metaClient,
instant), HoodieRestorePlan.class);
+ break;
+ case HoodieTimeline.SAVEPOINT_ACTION:
+ // Savepoint has no requested state (inflight then completed); its
inflight file is empty and
+ // now deserializes to an empty instance, so always read the instant
itself.
+ result = readAs(serde, activeTimeline, instant,
HoodieSavepointMetadata.class);
+ break;
+ case HoodieTimeline.COMPACTION_ACTION:
+ case HoodieTimeline.LOG_COMPACTION_ACTION:
+ result = instant.isCompleted()
+ ? readCommitMetadata(serde, activeTimeline, instant)
+ : readAs(serde, activeTimeline, requestedTwin(metaClient,
instant), HoodieCompactionPlan.class);
+ break;
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ case HoodieTimeline.CLUSTERING_ACTION:
+ // A completed replacecommit/clustering file is avro
HoodieReplaceCommitMetadata on disk;
+ // reading it as avro HoodieCommitMetadata fails avro record-name
resolution. Read the POJO
+ // HoodieReplaceCommitMetadata: the serde deserializes the avro
record and converts it to POJO.
+ result = instant.isCompleted()
+ ? readAs(serde, activeTimeline, instant,
HoodieReplaceCommitMetadata.class)
+ : readAs(serde, activeTimeline, requestedTwin(metaClient,
instant), HoodieRequestedReplaceMetadata.class);
+ break;
+ case HoodieTimeline.INDEXING_ACTION:
+ // A completed indexing instant stores avro
HoodieIndexCommitMetadata, not HoodieCommitMetadata.
+ result = instant.isCompleted()
+ ? readAs(serde, activeTimeline, instant,
HoodieIndexCommitMetadata.class)
+ : readAs(serde, activeTimeline, requestedTwin(metaClient,
instant), HoodieIndexPlan.class);
+ break;
+ default:
+ throw new BadRequestResponse("Unsupported action: " + action);
+ }
+
+ // Avro-generated objects (SpecificRecordBase) cannot be serialized by
+ // RequestHandler's ObjectMapper+AfterburnerModule due to module access
+ // restrictions on Avro's internal Schema classes. Convert to plain Maps
+ // using JsonUtils which accesses fields directly, bypassing getSchema().
+ if (result instanceof SpecificRecordBase) {
+ return JsonUtils.getObjectMapper().convertValue(result, Map.class);
+ }
+ return result;
+ } catch (BadRequestResponse | NotFoundResponse e) {
+ throw e;
+ } catch (Exception e) {
+ log.warn("Failed to read instant details for basePath={},
requestedTime={}, action={}, state={}",
+ basePath, requestedTime, action, state, e);
+ throw new HoodieException("Failed to read instant details", e);
+ }
+ }
+
+ public Map<String, Object> getTableConfig(String basePath) {
+ HoodieTableMetaClient metaClient = createMetaClient(basePath);
+ TreeMap<String, String> sorted = new TreeMap<>();
+ metaClient.getTableConfig().getProps().forEach((k, v) ->
sorted.put(k.toString(), v.toString()));
+ Map<String, Object> result = new HashMap<>();
+ result.put("properties", sorted);
+ return result;
+ }
+
+ public Map<String, Object> getSchemaHistory(String basePath, int limit) {
+ HoodieTableMetaClient metaClient = createMetaClient(basePath);
+ CommitMetadataSerDe serde = metaClient.getCommitMetadataSerDe();
+
+ Map<String, Object> result = new HashMap<>();
+
+ // Non-throwing accessor: a table with no commits yields null rather than
a 500.
+ result.put("currentSchema",
+ new TableSchemaResolver(metaClient)
+
.getTableSchemaIfPresent(metaClient.getTableConfig().populateMetaFields())
+ .map(schema -> schema.toString())
+ .orElse(null));
+
+ HoodieTimeline commitsTimeline = metaClient.getActiveTimeline()
+ .getCommitsTimeline().filterCompletedInstants();
+ List<HoodieInstant> instants = commitsTimeline.getInstants();
+
+ // Scan only the most recent N instants for performance.
+ int startIdx = Math.max(0, instants.size() - limit);
+ List<HoodieInstant> scanned = instants.subList(startIdx, instants.size());
+
+ List<Map<String, String>> history = new ArrayList<>();
+ String previousSchema = null;
+
+ for (HoodieInstant instant : scanned) {
+ try {
+ // Mirror TimelineUtils.getCommitMetadata: a completed
replacecommit/clustering file is avro
+ // HoodieReplaceCommitMetadata, not HoodieCommitMetadata. Read it as
the POJO (which extends
+ // HoodieCommitMetadata) or a schema change delivered by
insert_overwrite is silently skipped.
+ HoodieCommitMetadata commitMetadata =
+ (instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)
+ ||
instant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION))
+ ? readAs(serde, commitsTimeline, instant,
HoodieReplaceCommitMetadata.class)
+ : readCommitMetadata(serde, commitsTimeline, instant);
+ String schemaStr =
commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY);
+ if (schemaStr != null && !schemaStr.isEmpty() &&
!schemaStr.equals(previousSchema)) {
+ Map<String, String> entry = new LinkedHashMap<>();
+ entry.put("type", previousSchema == null ? "baseline" : "change");
Review Comment:
Done, but in the UI rather than the handler: RFC-94 pins the wire contract
(`baseline` = "the schema as of the oldest commit scanned", with
`window.truncated` as the separate signal), so changing the JSON `type` would
diverge from the merged spec. The badge label is now derived from
`window.truncated` -- when the window is truncated the oldest entry renders as
`oldest scanned` instead of `baseline`.
--
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]
