[FLINK-40169][table] Add target option to the EARLY_FIRE hint - #2
Draft
weiqingy wants to merge 51 commits into
Draft
[FLINK-40169][table] Add target option to the EARLY_FIRE hint#2weiqingy wants to merge 51 commits into
weiqingy wants to merge 51 commits into
Conversation
weiqingy
force-pushed
the
FLINK-36953-impl
branch
from
July 18, 2026 22:12
3131f4e to
4215571
Compare
weiqingy
force-pushed
the
FLINK-36953-pr2-target
branch
from
July 18, 2026 22:12
0b76a4b to
dc1e210
Compare
weiqingy
changed the base branch from
FLINK-36953-impl
to
FLINK-36953-pr1b-thread
July 18, 2026 22:14
weiqingy
changed the title
weiqingy
force-pushed
the
FLINK-36953-pr1b-thread
branch
from
July 21, 2026 02:17
e846e4e to
7dc4d49
Compare
weiqingy
force-pushed
the
FLINK-36953-pr2-target
branch
2 times, most recently
from
July 21, 2026 04:43
2d8840b to
8000ad1
Compare
weiqingy
force-pushed
the
FLINK-36953-pr1b-thread
branch
5 times, most recently
from
July 25, 2026 02:10
647116f to
a856401
Compare
as it requires minimum jdk17 This reverts commit 4e99512.
GHA pipelines use some of the assets from tools/azure-pipelines This commit moves them into tools/ci to reflect that they are not unique to the AZP builds, and are common across both. Signed-off-by: Dale Lane <dale.lane@uk.ibm.com>
weiqingy
force-pushed
the
FLINK-36953-pr2-target
branch
from
July 27, 2026 02:41
8000ad1 to
0ef163a
Compare
…rchiveStateTransition`
…ght generate runtime code with useless array copy
Refresh user-facing doc links that broke after source relocations and
external site restructuring. All targets verified reachable (HTTP 200
or file present on master).
gh_link source references:
- SourceFunction, SinkFunction, Trigger -> relocated legacy paths in
flink-runtime.
- WithPre/WithPostCommitTopology -> renamed Supports{Pre,Post}CommitTopology
in flink-runtime.
External links:
- Avro schema-resolution spec, Curator error handling, FLIP-187 title,
PyFlink TableEnvironment API doc, Apache Superset Hive guide.
- Gradle Shadow issue johnrengelman -> GradleUp.
- Debezium reference 1.2/1.3 -> stable (mysql events anchor updated).
…aFrame API Generated-by: OpenAI Codex (GPT-5.6 Sol) This closes apache#28797.
…_OP and a barrier-inserting in-memory implementation New RecoveryCheckpointTrigger (checkpoint.channel): void snapshotAndInsertBarriers(long checkpointId) -- narrow signature, widened to return a snapshot reader when the spilling backend lands; NO_OP singleton (no-op); NOT_READY singleton (throws CheckpointException(CHECKPOINT_DECLINED_TASK_NOT_READY) -- transient, the coordinator retries). New transitional in-memory implementation InMemoryRecoveryCheckpointTrigger: holds the task's List<RecoverableInputChannel>; snapshotAndInsertBarriers = forEach(ch -> ch.insertRecoveryCheckpointBarrierIfInRecovery(cpId)). The javadoc states why no snapshot exists (one-shot push means no undrained residue) and that the disk drainer replaces it. Deleted when the spilling backend lands. Unit tests for all three implementations. Co-authored-by: Roman Khachatryan <khachatryan.roman@gmail.com>
…ough the recovery trigger ChannelState gains a RecoveryCheckpointTrigger (the legacy 1-arg ctor defaults to NO_OP); new onCheckpointStartedForAllInputs(CheckpointBarrier): (1) trigger.snapshotAndInsertBarriers(cpId); (2) for (input : inputs) input.checkpointStarted(barrier). CheckpointException is rethrown as-is (routes to checkpoint abort, not task failure); other IOException via rethrowIOException. (Step 3 -- spilled-slice replay through the channel-state writer -- is added when the spilling backend lands.) AlternatingCollectingBarriers and AlternatingWaitingForFirstBarrierUnaligned: replace the inline per-input checkpointStarted loop with state.onCheckpointStartedForAllInputs(...) (behaviorally inert with NO_OP). Tests: ChannelStateTest (2-step scope), AlternatingCollectingBarriersDispatchHookTest, AlternatingWaitingForFirstBarrierUnalignedDispatchHookTest.
… and the StreamTask lifecycle InputProcessorUtil.createCheckpointBarrierHandler: overload taking a RecoveryCheckpointTrigger (the old signature is preserved as a NO_OP delegator); SingleCheckpointBarrierHandler.unaligned/alternating gain the trigger param (aligned keeps the NO_OP default through ChannelState's 1-arg ctor); OneInputStreamTask/TwoInputStreamTask/MultipleInputStreamTask pass StreamTask.getRecoveryCheckpointTrigger(). (The channel-state writer is threaded through the same seams when the spilling backend lands.) StreamTask: recoveryCheckpointTrigger field (starts NOT_READY) + mailbox-thread-asserting getter lambda; all asynchronous mutations via setRecoveryCheckpointTrigger (a mailbox mail). Lifecycle inside recoverChannelsWithCheckpointing: NOT_READY before/during filtering (checkpoints declined as task-not-ready; transient, the coordinator retries) -> install the in-memory barrier-inserting trigger at conversion, composing on requestPartitions' returned List<RecoverableInputChannel> -> swap to NO_OP when completeAll(gates' stateConsumedFutures) completes (gate on the futures, not on push completion; the mailbox-mail gap is safe because a barrier-inserting trigger with no in-recovery channels behaves as NO_OP). recoverChannelsWithoutCheckpointing and the empty-input-gates short-circuit go straight to NO_OP. Also fix the transitional conversion push's sentinel timing: the EndOfFetchedChannelStateEvent sentinel is no longer appended inside toInputChannelInRecovery() (where it becomes consumable before the upstream connection exists, letting a spurious post-recovery poll hit a LocalInputChannel without a subpartition view: "Queried for a buffer before requesting the subpartition"). Instead the recovery chain appends it via finishRecoveredBufferDelivery() on the channelIOExecutor after partitions are requested -- the method waits for upstream readiness, restoring the invariant the disk drainer also relies on, and mirroring where the drainer will run. Tests: TestBarrierHandlerFactory adaptation; StreamTask trigger-lifecycle source-level invariants (transitional, replaced with the disk drainer wiring when the spilling backend lands); RecoveredInputChannelTest push-conversion case adapted to the sentinel timing.
…able filtering ITCase TestStreamEnvironment: restore randomize(conf, CHECKPOINTING_DURING_RECOVERY_ENABLED, true, false), closing the transitional randomization window opened when the v1 conversion-migration path was retired. Re-enable RecoveredStateFilteringLargeRecordITCase. From here on the randomized ITCase fleet runs flag-on ~50% against the in-memory backend (same memory profile as base v1, which ran the same randomization with the same heap fallback). The spilling PRs must keep the fleet green -- this is the acceptance bar for "lightweight backend replacement".
…s used as keys This closes apache#28834
…ry twice testSavepointDeepCopy listed savepointPath1 for both stateFiles1 and stateFiles2, so the copied-files assertion compared savepoint1's directory with itself and savepoint2's directory was never examined at all. Point stateFiles2 at savepointPath2, as the variable name, the assertion messages, and the test's documented steps always intended. The mix-up was introduced when the getFileNamesInDirectory helper was extracted (f125067); before that, both checks listed savepointPath2. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…from native savepoint The first incremental checkpoint after a CLAIM-mode restore from a native savepoint reuses the savepoint's SST files as RelativeFileStateHandles. The metadata serializer stored only their relative path, so on restore they were resolved against the new checkpoint's exclusive directory and could no longer be found. The serializer now knows the exclusive directory of the checkpoint being written (CheckpointMetadataOutputStream#getExclusiveCheckpointDir, passed down as a non-null SerializationContext) and keeps the compact relative encoding only for handles whose file actually lives in that directory; a foreign handle (a reused savepoint SST) is persisted with its absolute path using the pre-existing FileStateHandle encoding. The wire format and the metadata version are unchanged. The state processor API keeps the unconditional relative encoding via the explicit Checkpoints#storeCheckpointMetadataWithoutExclusiveDir, because its deep copy places every referenced file next to the new metadata. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…availability plumbing
…patible servers that do not support AWS checksum-validation
…ITCase.testRestoreAfterSomeTasksFinished (apache#28800) There was traced the flakiness to task finished after the checkpoint plan was calculated but before triggerCheckpoint. The task is already removed from TaskExecutor#taskSlotTable So TaskExecutor#triggerCheckpoint fails the checkpoint with error 'TaskManager received a checkpoint request for unknown task <>'. Production systems are expected to tolerate an occasional checkpoint failure So we can address the flakiness without changes this known limitation for tasks finished mid checkpoint. the test provisions 2 sources passA that finishes quickly and passB that blocks forever and non-deterministically waits for at least one of passA subtasks to finish, which introduces the race. The fix here makes CheckpointAfterAllTasksFinishedITCase.testRestoreAfterSomeTasksFinished wait for all subtasks of passA to finish before triggering savepoint, to avoid the race but still achieve the test goal with passB subtasks still locking.
Introduce JobMdcRegistry, a process-wide registry mapping JobID to an enriched MDC context built from job configuration. Add MdcOptions with the mdc.job-configuration-to-mdc-keys config option (@PublicEvolving). Wire in mdc enrichment on the job/task submission paths. Generated-by: Claude Code
Generated-by: Claude Code
…ment" This reverts commit b198cdd.
…C enrichment"" This reverts commit d0ce9ea.
…riven MDC enrichment""" This reverts commit 3d23bdf.
This reverts commit ef6633f.
…estore from a NATIVE savepoint Since FLINK-39964, the first incremental checkpoint(s) taken after a CLAIM-mode restore from a NATIVE savepoint reference the reused savepoint files by their absolute location. Extend the CLAIM-mode note on the savepoints page (EN and ZH): such checkpoints are neither self-contained nor relocatable (consistent with incremental checkpoints in general), and the savepoint directory remains referenced, with its files' lifecycle managed by Flink, until every checkpoint reusing its files has been subsumed. Generated-by: Claude Fable 5
…in `VariantSerializer.deserialize`
Introduce JobMdcRegistry, a process-wide registry mapping JobID to an enriched MDC context built from job configuration. Add MdcOptions with the mdc.job-configuration-to-mdc-keys config option (@PublicEvolving). Wire in mdc enrichment on the job/task submission paths. Generated-by: Claude Code
Generated-by: Claude Code (cherry picked from commit ef6633f)
…of ForSt transfer threads (apache#28373) --------- Co-authored-by: Purushottam Sinha <sinhapurushottam911@gmail.com>
… number of ForSt transfer threads (apache#28373)" This reverts commit 01a5a7f.
…E`/`JSON_QUERY` calls when the owning call is skipped because of `NULL` or not taken branch
…ERIALIZED TABLE Persisted columns of a materialized table have to be produced by its definition query, and CREATE OR ALTER enforces that through validatePersistedColumnsUsedByQuery while a plain CREATE only checked physical columns. A CREATE declaring a non-virtual metadata column that the query does not project was therefore accepted, and the mismatch only surfaced when the refresh pipeline compiled the insert, whose consumed row type contains the persisted metadata column. Both statements share the validation now, so the DDL is rejected up front, and the error message names the statement it originates from. The test fixture for materialized tables with metadata columns projects its persisted column, which makes dropping that column report that it is used in the query rather than that persisted columns cannot be dropped; the latter stays covered by the query-evolution cases. Generated-by: Claude Opus 5
Offload the dump to ioExecutor in both TaskExecutor#requestThreadDump and Dispatcher#requestThreadDump via CompletableFuture.supplyAsync, matching the pattern already used by other heavy TE handlers.
…nections Co-authored-by: Roman Khachatryan <khachatryan.roman@gmail.com>
Add an optional `target` option to the EARLY_FIRE join hint. Only `interval_join` is accepted today; any other value fails planning. An omitted target is equivalent to `interval_join`, so existing hints keep their meaning. `target` scopes the hint to a single operator kind. The interval-join rule consumes the hint only when it targets the interval join and leaves a hint aimed at another operator kind untouched, so an untargeted hint never silently expands its scope.
weiqingy
force-pushed
the
FLINK-36953-pr2-target
branch
from
August 2, 2026 22:18
0ef163a to
3b14d56
Compare
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
20 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.
Part of the FLIP-497 stack under umbrella FLINK-36953. Stacked on PR-1b. Landing order: PR-1a → PR-1b → PR-2 (this) → PR-3 → PR-4 → PR-5 → PR-6 → PR-7.
What is the purpose of the change
Adds an optional
targetoption to theEARLY_FIREhint so a hint can be explicitly scoped to a single operator kind. This is a forward-compatibility guard: it keeps a bareEARLY_FIREhint from silently expanding to other operators if any of them honor the hint in the future.Brief change log
EarlyFireJoinHintOptions.TARGET(optionalstringType) and theINTERVAL_JOINconstant.EARLY_FIREKV option checker validatestarget ∈ {interval_join}; any other value fails planning.StreamPhysicalIntervalJoinRuleapplies the hint only when it targets the interval join; an omittedtargetis equivalent tointerval_join, so existing hints keep their meaning.Verifying this change
This change added tests and can be verified as follows:
IntervalJoinTestcases: an explicittarget='interval_join'produces the same plan as the default form, and an unsupportedtargetvalue fails planning.Does this pull request potentially affect one of the following parts:
@Public(Evolving): yes — new option on@PublicEvolvingEarlyFireJoinHintOptionsDocumentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Anthropic)