KKcorps opened a new pull request, #19488:
URL: https://github.com/apache/pinot/pull/19488
## TL;DR
`BaseSingleSegmentConversionExecutor` gets an opt-in hook,
`isCopyToDeepStoreForMetadataPush()`,
default `false`. A task that returns `true` gets a METADATA push that is
safe for refreshing a
segment under its own name: the tar is staged under a name that can never be
a live segment path,
the push keeps the TAR path's `If-Match` and `REFRESH_ONLY` guards, the
controller copies the bytes
into the segment's existing deep store location, and the staged tar is
deleted afterwards. Nothing
changes for PurgeTask, RefreshSegmentTask or UpsertCompactionTask unless
they opt in.
## The problem
#17632 added a METADATA mode to this executor. It stages the converted tar at
`<outputDir>/<segment>.tar.gz` and registers that URI as the segment's
download URL, with
`COPY_SEGMENT_TO_DEEP_STORE=false`. That is the right shape for a new
segment. For a task that
refreshes an existing segment under the same name it has three gaps:
1. The `If-Match` (CRC) and `REFRESH_ONLY` headers that the TAR path sends
are dropped. A push
from a stale task can overwrite a segment refreshed after the task was
planned, and a push for a
segment deleted in the meantime recreates it.
2. `<table>/<segment>.tar.gz` is the live file for a segment that was
ingested by a metadata-push
ingestion task. With `overwriteOutput=false` the refresh fails. With
`true` it overwrites live
bytes before ZooKeeper is updated, and the failure cleanup then deletes
the live file.
3. Because the download URL moves to the staged path, the staged tar becomes
the live file and can
never be deleted, the previous deep store object is orphaned, and the
second refresh of the same
segment fails with "already exists".
```mermaid
flowchart LR
subgraph Before["Existing METADATA push (unchanged default)"]
A[stage at table/segment.tar.gz] --> B[register staged URI as download
URL]
B --> C[old object orphaned, next refresh collides]
end
subgraph After["Opt-in controller copy"]
D[stage at table/segment.taskId.tar.gz] --> E[push metadata + If-Match +
REFRESH_ONLY + copy flag]
E --> F[controller copies into table/segment]
F --> G[download URL unchanged, staged tar deleted]
end
```
## The approach
1. `isCopyToDeepStoreForMetadataPush()` returns `false` by default. The
existing METADATA method is
kept byte for byte and stays the default branch. The TAR path is
untouched.
2. When the hook returns `true`, the executor builds the metadata tar
(`metadata.properties` and
`creation.meta`) from the local converted segment before the segment
directories are deleted.
The old path downloaded the staged tar back from the output filesystem to
extract those two files.
3. If the conversion left the segment unchanged (same CRC) and the task
carries a download URL, the
executor skips tarring and staging and only re-registers the metadata
against that URL.
4. Otherwise the tar is staged at `<outputDir>/<segment>.<taskId>.tar.gz`.
Live segments are
stored as `<segment>` or `<segment>.tar.gz`, so the name can never
collide. It is stable across
retries of the same task, so a retry overwrites what an interrupted
attempt left, and unique
across tasks. A plain-path output dir (local deep store) is sent as a
`file` URI, because the
controller picks the filesystem by scheme.
5. The push reuses the TAR path's header list and adds `DOWNLOAD_URI`,
`UPLOAD_TYPE=METADATA` and
`COPY_SEGMENT_TO_DEEP_STORE=true`. The controller runs its existing CRC
check and parallel-push
lock, copies the staged tar server-side into
`<dataDir>/<table>/<segment>`, updates ZooKeeper
and sends the refresh message. That is the same order the TAR path
already uses.
6. The staged tar is deleted in a `finally`. Nothing references it after the
controller copied it,
so this is safe on success, failure and timeout.
## Key components
| Class / file | Role |
|---|---|
| `BaseSingleSegmentConversionExecutor` | New hook,
`uploadSegmentMetadataWithControllerCopy`, staging name, header assembly.
Existing `uploadSegmentWithMetadata` restored as the default branch. |
| `BaseTaskExecutor` | `moveSegmentToOutputPinotFS(configs, file,
outputFileName, overwrite)` overload (the 2-arg version delegates to it),
`deleteFromOutputPinotFS`, `createSegmentMetadataTarFile`. |
| `SegmentConversionUtils` | `uploadSegmentMetadata`, sharing the retry,
round-robin and `HOST` header logic with `uploadSegment` through one private
`uploadWithRetry`. |
## Flow (opt-in path)
```mermaid
sequenceDiagram
participant M as Minion
participant O as Output PinotFS
participant C as Controller
participant D as Deep store
participant Z as ZooKeeper
M->>M: convert, build metadata tar from local segment
alt segment unchanged (same CRC)
M->>C: POST metadata, DOWNLOAD_URI = current download URL, copy=false
C->>Z: update custom map / refresh time
else segment changed
M->>O: put table/segment.taskId.tar.gz (overwrite ok)
M->>C: POST metadata, If-Match, REFRESH_ONLY, DOWNLOAD_URI = staged,
copy=true
C->>C: CRC check, parallel-push lock
C->>D: server-side copy staged -> table/segment
C->>Z: update segment metadata, send refresh message
C-->>M: 200
M->>O: delete staged tar (always, in finally)
end
```
## Compatibility
- No behavior change for any existing task. The default branch is the
pre-existing method and the
test that pinned its failure cleanup is kept.
- No new config keys. The opt-in is a protected method, not a task config,
so it cannot be turned
on by accident from a table config.
- Everything the opt-in path sends already exists on the controller side:
`If-Match`,
`REFRESH_ONLY`, the custom-map header, `UPLOAD_TYPE=METADATA`,
`DOWNLOAD_URI` and
`COPY_SEGMENT_TO_DEEP_STORE`, including the deep store copy on an
existing-segment refresh. A new
minion works against an old controller.
- `uploadSegment` keeps its signature. Only additions to `BaseTaskExecutor`.
## Testing
`BaseSingleSegmentConversionExecutorTest`:
- Default path: still registers the staged URI through
`SegmentPushUtils.sendSegmentUriAndMetadata`
and never calls the new helper; still deletes the staged tar when the push
fails (existing test).
- Opt-in path: staging name and its deletion after success and after
failure; the header set
(`If-Match`, `REFRESH_ONLY`, custom-map, `DOWNLOAD_URI`, `UPLOAD_TYPE`,
copy flag); the metadata
tar contains exactly `metadata.properties` and `creation.meta` and
describes the converted segment;
a leftover staged tar from an earlier attempt is overwritten; a
scheme-less output dir is sent as a
`file` URI; an unchanged segment is not staged and is re-registered
against its download URL.
`PurgeMinionClusterIntegrationTest` passes, including
`testFirstRunMetadataPushPurge` on the
unchanged default path. The opt-in path was exercised end to end on a
local-FS cluster from the
StarTree purge task that overrides the hook.
Labels: `extension-point`, `bugfix`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01Y375AgHYsh1YqNSsfvF1a8
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
