[FLINK-xxxxx] Add SQL geography construction/functions#20
Open
gkalashyan-akv wants to merge 50 commits into
Open
[FLINK-xxxxx] Add SQL geography construction/functions#20gkalashyan-akv wants to merge 50 commits into
gkalashyan-akv wants to merge 50 commits into
Conversation
…onfig) Replace .eslintrc.js and .eslintignore with eslint.config.js flat config. @angular-eslint 21 already supports eslint 9, so this adds the unified angular-eslint and typescript-eslint packages (which provide flat-config-ready presets and parser wiring) alongside the existing split @angular-eslint/* packages needed by the Angular CLI builder and schematics. eslint-plugin-prettier stays on v4.x since v5 requires prettier 3, which is a separate deferred upgrade. Generated-by: Claude Code (claude-opus-4-8)
… sort --------- Co-authored-by: Arvind Kandpal <arvind.kandpal@ksolves.com>
…e#28489) Close the JNI Statistics wrapper in `RocksDBNativeMetricMonitor.close()` and on partial init in `RocksDBHandle.loadDb()`.
…ion README (apache#28677) * [FLINK-40090][docs] Modernize cryptography export notice in distribution README * [FLINK-40090][tests] Update expected WordCount row count in CLI e2e test Generated-by: Claude Opus 4.8
…erminatedRescaleMergingWithNewRecoverableFailureTriggerCause After terminating a TaskManager to trigger a recoverable failover, the test waited via waitForVertexParallelismReachedAndJobRunning for PARALLELISM. That condition is already satisfied by the stale pre-failover state, so the wait returns immediately without synchronizing with the failover. The snapshot could therefore be taken before AdaptiveScheduler merged the failover into the in-progress UPDATE_REQUIREMENT rescale and re-stamped its trigger cause, letting the test observe UPDATE_REQUIREMENT instead of the expected RECOVERABLE_FAILOVER. Wait for the job to be RUNNING again at the parallelism the single remaining TaskManager can host (NUMBER_SLOTS_PER_TASK_MANAGER) instead. The merge re-stamps the trigger cause in goToRestarting, before the job runs again at that reduced parallelism, so the job coming back up synchronizes the snapshot with the merge. The assertion is unchanged. Generated-by: Claude Opus 4.8 (via Claude Code)
…nputs finished StreamMultipleInputProcessor.processInput returned NOTHING_AVAILABLE when no input was selectable, even once all inputs were finished. Since getAvailableFuture() returns AVAILABLE as soon as all inputs are finished, the mailbox loop then never blocks and would spin on processInput at 100% CPU if it were ever invoked again after the operator finished. This is a latent inconsistency with single-input processors (which keep returning END_OF_INPUT once finished) and is independent of any particular feature; it is just not reached today because a task suspends on the first END_OF_INPUT and processInput is not called again. The checkpointing-during- recovery work (later in this series) does re-invoke processInput after an operator finishes during recovery, which would otherwise turn this into a hang. Report END_OF_INPUT when there is nothing to read and all inputs are finished, consistent with single-input processors and MultipleInputSelectionHandler#calculateOverallStatus.
…m toBeConsumedBuffers
Split the single toBeConsumedBuffers queue into two queues with disjoint
responsibilities:
- recoveredBuffers (new): holds buffers migrated from RecoveredInputChannel
during construction; consumed by getNextRecoveredBuffer() which retains
the priority-event interleaving and last-buffer dynamic next-data-type
detection introduced by FLINK-39018.
- toBeConsumedBuffers (existing): reverted to its pre-FLINK-39018 role of
holding FullyFilledBuffer partial-buffer splits only. The recovery-aware
early branch in getNextBuffer() and the checkpointStarted inflight scan
no longer touch this queue.
Restores the checkState(toBeConsumedBuffers.isEmpty()) guard in
requestSubpartitions() (removed by cebc174). hasPendingPriorityEvent,
notifyPriorityEvent, and the constructor signature are unchanged.
Pure refactor: no public API change, no new tests; verified by the 9 existing
LocalInputChannelTest regression cases.
…andler with concrete no-filtering/filtering handlers Single-file class split of RecoveredChannelStateHandler.java; all classes stay package-private in this file. No behavior change. - New abstract AbstractInputChannelRecoveredStateHandler: fields inputGates, channelMapping, rescaledChannels, oldToNewMappings; methods getMappedChannels, calculateMapping, getChannel - verbatim extraction from InputChannelRecoveredStateHandler. Adds a closeInternal() template hook used by later phases. - New NoSpillingHandler: recover() is the verbatim old non-filtering branch (onRecoveredStateBuffer(descriptor) followed by onRecoveredStateBuffer(buffer.retainBuffer()), same try/finally recycle); inherits the verbatim network-pool getBuffer. - New FilteringHandler (the v1 filtering behavior, verbatim): recover() = old recoverWithFiltering (filteringHandler.filterAndRewrite(..., channel::requestBufferBlocking) delivering List<Buffer> via onRecoveredStateBuffer); getBuffer = old getPreFilterBuffer (reusable heap MemorySegment, preFilterBufferInUse invariant); closeInternal = old segment-freeing close. - New static create(...) factory selecting NoSpillingHandler vs FilteringHandler - equivalent to the old internal if-branch. SequentialChannelStateReaderImpl switches from direct construction to the factory (mechanical). - ResultSubpartitionRecoveredStateHandler: untouched. - Trace label in RecoveredInputChannel.onRecoveredStateBuffer: "InputChannelRecoveredStateHandler#recover" -> "NoSpillingHandler#recover" (final label). - Tests: mechanical adaptation of InputChannelRecoveredStateHandlerTest, RecoveredChannelStateHandlerTest.
- InputGate: new abstract InputChannel getChannel(InputChannelInfo). - SingleInputGate: channels[channelInfo.getInputChannelIdx()]. UnionInputGate: resolve via inputGatesByGateIndex.get(channelInfo.getGateIdx()) .getChannel(channelInfo) (correct global-vs-member index semantics). InputGateWithMetrics: delegate. CheckpointedInputGate: delegate helper. - AbstractStreamTaskNetworkInput: getChannel(channelInfo.getInputChannelIdx()) -> getChannel(channelInfo). - Test impls: MockInputGate, MockIndexedInputGate, AlignedCheckpointsMassiveRandomTest's inner gate.
…ources visibility - NetworkActionsLogger: additive tracePersist(String, Object, Object, long) overload; the existing Buffer overload delegates to it. - RecoveredInputChannel#releaseAllResources: package-private -> public.
…beros keytab logs before the application is killed
…ore effective sub-plan digest reuse (apache#27959)
…f a generic planning error This closes apache#28703.
…without an upsert key This closes apache#28704.
…se randomization Drop the randomize(conf, CHECKPOINTING_DURING_RECOVERY_ENABLED, true, false) line from TestStreamEnvironment.randomizeConfiguration (the flag stays at its default false unless a test sets it explicitly) and @disabled the explicitly flag-on RecoveredStateFilteringLargeRecordITCase. Rationale: a following commit of this PR retires the v1 conversion-migration path, so ~50% of randomized ITCases would otherwise run flag-on and fail until the StreamTask recovery rework and the checkpoint-coordination-during-recovery PRs land. The randomization and the ITCase are restored in the coordination PR (PR4 of the series).
… notifyInitiallyEnabled Add a notifyInitiallyEnabled constructor parameter to BufferManager and a new enableNotify() method: while notification is disabled, requested and recycled floating buffers are queued but never announced to the producer as credit; enableNotify() opens the gate and announces the queued buffers atomically with respect to concurrent recycle/floating-buffer callbacks. All existing callers (RecoveredInputChannel, RemoteInputChannel) pass true, so behavior is unchanged. The disabled mode is used by a later commit of this PR, where a remote channel created in recovery state withholds credit until its recovered state has been consumed.
…ery sentinels Introduce the push-based recovery contract and its sentinels: - RecoverableInputChannel (@internal): getChannelInfo(), onRecoveredStateBuffer(Buffer), finishRecoveredBufferDelivery(), insertRecoveryCheckpointBarrierIfInRecovery(long), requestRecoveryBufferBlocking(), onRecoveredStateConsumed(). - EndOfFetchedChannelStateEvent: singleton RuntimeEvent tail sentinel of the recovered-buffer stream; reflective write()/read() throw; deliberately distinct from EndOfInputChannelStateEvent. - RecoveryCheckpointBarrier (checkpoint.channel): per-checkpoint sentinel carried inside the recovery queue; write()/read() throw. - EventSerializer: type tags 13 (RECOVERY_CHECKPOINT_BARRIER_EVENT) and 14 (END_OF_FETCHED_CHANNEL_STATE_EVENT). Recovery completion is unified on the InputChannel base class: the new abstract getStateConsumedFuture() completes once a channel has consumed all of its recovered state (or never had any / was released). It is abstract rather than defaulted so a future subclass must state its recovery semantics explicitly. RecoveredInputChannel's package-private getter becomes a public override (field tightened to CompletableFuture<Void>); UnknownInputChannel returns a completed future (a channel with persisted state is always a RecoveredInputChannel at restore time). Local/RemoteInputChannel temporarily return a completed future until their push-based recovery state lands in the next commits. TestInputChannel implements the new contract as a spying test double.
Teach LocalInputChannel to be created directly in a recovery state and to receive recovered buffers through the RecoverableInputChannel interface, instead of migrating them via the constructor: - Constructor: drop ArrayDeque<Buffer> initialRecoveredBuffers, add networkBuffersPerChannel and needsRecovery. With needsRecovery=false the stateConsumedFuture is pre-completed, inRecovery stays false and no recovery BufferManager exists -- behavior is byte-for-byte unchanged. - recoveredBuffers becomes a Deque<Buffer> whose monitor also guards inRecovery and recoverySequenceNumber (starts at Integer.MIN_VALUE). - RecoverableInputChannel implementation: push append with released-state recycling; finishRecoveredBufferDelivery() waits for upstreamReady and appends the EndOfFetchedChannelStateEvent sentinel; insertRecoveryCheckpointBarrierIfInRecovery(); requestRecoveryBufferBlocking() lends exclusive buffers via the recovery-only BufferManager (setup() requests them) -- no production caller until the drainer PR; onRecoveredStateConsumed() flips recovery off and completes the stateConsumedFuture. - getNextBuffer() is rewritten under a single inRecovery predicate: recovered buffer first, then a pending priority event pulled from the subpartition view, otherwise empty (live data stays hidden while in recovery). wrapRecoveredBufferAsAvailability() materializes FileRegionBuffer/CompositeBuffer, assigns recovery sequence numbers and computes the next data type via peekNextDataType. - upstreamReady completes on subpartition-view creation; completeUpstreamReadyForTest() for tests. - checkpointStarted() splits into mutually exclusive in-recovery/normal branches. In recovery it only starts persisting (the persist window stays open) with the buffers collected by collectPreRecoveryBarrier, which walks the queue to the matching RecoveryCheckpointBarrier, retains pre-barrier data buffers and removes the sentinel; a missing barrier is an IOException wrapped as CheckpointException(CHECKPOINT_DECLINED). Mechanical constructor-caller adaptations ride along (UnknownInputChannel, LocalRecoveredInputChannel, InputChannelBuilder, benchmark factory), all passing needsRecovery=false at this stage. Includes a port of the FLINK-40016 double-persist regression test to the push-based recovery API.
Teach RemoteInputChannel to be created directly in a recovery state and to receive recovered buffers through the RecoverableInputChannel interface. The not-in-recovery checkpointStarted branch stays byte-identical to the previous behavior, and with needsRecovery=false the channel behaves exactly as before. - Constructor: drop ArrayDeque<Buffer> initialRecoveredBuffers, add needsRecovery (pre-completed stateConsumedFuture when false, same as LocalInputChannel). The BufferManager starts with credit notification disabled while in recovery (notifyInitiallyEnabled=!needsRecovery). - appendRecoveredBuffer() appends straight into receivedBuffers (NONE subpartition id, recovery sequence numbers from Integer.MIN_VALUE), so the consume path stays identical to the normal case. - recoveryEventStash (@GuardedBy receivedBuffers): ordinary upstream events arriving while credit is suppressed are stashed and replayed after the sentinel is consumed; live data buffers during recovery are a protocol violation asserted in onBuffer. - upstreamReady as CountDownLatch(1), counted down by the first onBuffer or by release; finishRecoveredBufferDelivery() awaits it and appends the EndOfFetchedChannelStateEvent sentinel. - onRecoveredStateConsumed(): unstash + bufferManager.enableNotify() (credit reopens) + complete stateConsumedFuture. - checkReadability() allows in-recovery reads before the partition request client is initialized. - checkpointStarted() in-recovery branch persists the buffers collected by collectPreRecoveryBarrier (walks receivedBuffers past the priority head to the matching RecoveryCheckpointBarrier, retains pre-barrier data buffers, removes the sentinel); only startPersisting is called, so the persist window stays open. A missing barrier declines the checkpoint (CHECKPOINT_DECLINED). - New public needsRecovery() getter, consumed by the netty stack in a later commit of this PR. Mechanical constructor-caller adaptations ride along (UnknownInputChannel, RemoteRecoveredInputChannel, InputChannelBuilder, benchmark factory, netty test channels), all passing needsRecovery=false. RemoteRecoveredInputChannel keeps calling remoteInputChannel.setup() until the conversion rework in the next commit moves setup() into RecoveredInputChannel#toInputChannel.
…ace; thread needsRecovery through gates RecoveredInputChannel#toInputChannel(boolean needsRecovery) replaces the no-arg variant; constructor migration of recovered buffers is retired: - needsRecovery=false: checkState(receivedBuffers.isEmpty()) + inputChannel.setup() (setup moves from RemoteRecoveredInputChannel into the conversion) + checkpointStopped(...). Flag-off conversion always happens after full consumption, so default behavior is unchanged. - needsRecovery=true (first used by the StreamTask recovery rework PR): create the physical channel in recovery state, then synchronously push every queued recovered data buffer via onRecoveredStateBuffer and append the EndOfFetchedChannelStateEvent sentinel (the legacy EndOfInputChannelStateEvent in the queue is dropped in translation). This is the simple in-memory drain -- FLINK-38544 transitional code, replaced by the disk drain when the spilling backend lands. The sentinel is appended directly rather than via finishRecoveredBufferDelivery(), which waits for upstream readiness that cannot arrive while the mailbox thread is still converting channels. - abstract toInputChannelInternal(ArrayDeque<Buffer>) becomes toInputChannelInternal(boolean). Gates thread the flag through: InputGate gains a concrete requestPartitions(boolean) (default overload ignores the flag); SingleInputGate implements requestPartitions(boolean) and convertRecoveredInputChannels(boolean) (no-arg kept as a @VisibleForTesting shim); getStateConsumedFuture() aggregation is unified on InputChannel#getStateConsumedFuture across all channels; UnionInputGate fans the flag out to member gates; InputGateWithMetrics delegates. Deliberate consequence: the base v1 flag-on StreamTask still converts at filtering-complete with unconsumed buffers, so the empty-queue checkState fires and the flag-on path is degraded until the StreamTask recovery rework lands -- covered by the randomization pin in the first commit of this PR. Default (flag-off) configuration is unaffected. Tests: RecoveredInputChannelTest is rewritten around the new conversion semantics (reject-while-unconsumed, empty-queue assert, and a transitional push-conversion case); Local/RemoteRecoveredInputChannelTest cover the empty-queue assert of the subclasses; UnionInputGateTest's helper adapts to the new toInputChannelInternal signature.
…tart view reader with zero credit Propagate the consumer channel's needsRecovery flag to the producer so it withholds credit while the consumer's exclusive buffers are on loan to recovery: - NettyMessage.PartitionRequest gains a needsRecovery boolean (ctor arg, writeBoolean/readBoolean, +Byte.BYTES in the length calculation). This is a wire-format change (+1 byte per PartitionRequest); safe under the usual single-version-cluster assumption for the TM network protocol, stated in the PR description. - CreditBasedSequenceNumberingViewReader ctor gains needsRecovery; numCreditsAvailable = needsRecovery ? 0 : initialCredit. - PartitionRequestServerHandler passes request.needsRecovery; NettyPartitionRequestClient passes inputChannel.needsRecovery(). Mechanical test adaptations: NettyMessageServerSideSerializationTest (round-trips the new field), PartitionRequestServerHandlerTest, PartitionRequestQueueTest, PartitionRequestRegistrationTest, CreditBasedSequenceNumberingViewReaderTest, CancelPartitionRequestTest, ServerTransportErrorHandlingTest.
…nnelStateEvent When the consume path polls the EndOfFetchedChannelStateEvent sentinel, assert the originating channel is a RecoverableInputChannel and call its onRecoveredStateConsumed(): the channel flips out of recovery, releases any upstream events held back during recovery and reopens the upstream so live data may flow again. The event itself is never delivered to the operator. The sentinel-consumption path is exercised end to end by the Local/RemoteInputChannel recovery tests of this PR (onRecoveredStateConsumed after polling the sentinel) and by the StreamTask recovery tests of the follow-up PRs.
…is killed FLINK-40099 moved YARNSessionFIFOSecuredITCase's keytab check into a pre-kill callback polling waitUntilCondition(cond, 500). The 500 int literal binds to the (cond, int retryAttempts) overload (500*100ms, then "Exhausted retry attempts."), and reading the JobManager log while the application is still alive is racy: the polled jobmanager.log does not reliably contain the keytab login line before teardown. Read the container logs after the application is killed (output flushed) but poll each log with a bounded wait instead of a single read, preserving the FLINK-17662 intent without the pre-kill race. On timeout, fail with the log file and expected strings so a recurrence is diagnosable from the uploaded container logs. Generated-by: Claude Opus 4.8 (1M context)
….runJob() (apache#28722) JarService.runJob() appended the programArgs value under the savepointPath query key instead of savepointPath itself. This has been harmless in practice because the backend (JarRunHandler) prefers the value already present in the POST body over the query parameter, but the query parameter itself was wrong. Generated-by: Claude Code (claude-sonnet-5)
The web-dashboard had no automated tests, leaving lint and the production build as the only guardrails against breakage during Angular/ng-zorro upgrades. Add a Vitest-based test setup with smoke-level component specs for high-traffic views, and run them in CI. - Wire Vitest via Angular's @angular/build:unit-test builder (the default test runner in Angular 21; Karma is deprecated). Tests run in jsdom, so no browser is needed in CI. Add tsconfig.spec.json and src/test-setup.ts with the jsdom stubs ng-zorro/CDK require (matchMedia, ResizeObserver, IntersectionObserver). - Migrate the build and serve targets from @angular-devkit/build-angular:browser-esbuild to @angular/build:application and :dev-server, which the unit-test builder requires and which is the Angular 21 default. Output stays flat under web/ so packaging is unchanged. Remove the now-unused @angular-devkit/build-angular. - Add smoke specs covering rendering and basic interactions for the Overview statistic cards, the Task Manager list, and the Job Exceptions view. - Add test / test:ci npm scripts and include test:ci in ci-check so the specs run as part of the existing Maven frontend build. Generated-by: Claude Code (claude-opus-4-8)
…nOperator (apache#28664) Rewrites a join over a SNAPSHOT table function in a LATERAL clause into a dedicated FlinkLogicalLateralSnapshotJoin and converts it to a StreamPhysicalLateralSnapshotJoin / StreamExecLateralSnapshotJoin backed by the LateralSnapshotJoinOperator. The nodes derive their own output row type (materializing the build-side rowtime) and carry the SNAPSHOT arguments as fields. Generated-By: Claude Opus 4.8 (1M context)
…rness and state-parameterized tests (apache#28589)
…OAD/JOIN execution (apache#28329) Stateful keyed two-input operator backing LATERAL SNAPSHOT joins. * Two phases (LOAD/JOIN) with flip triggers: build-side WM reaches loadCompletedTime, or a processing-time idle-timeout fallback. * Probe records are buffered in LOAD and joined on flip. Build-side changes are buffered per-key and applied lazily in event-time order on the next per-key access once the build-side WM advanced, preserving atomic -U/+U visibility. * Watermarks: build-side absorbed; probe-side held back in LOAD, forwarded in JOIN. NULL equi-keys filtered via JoinConditionWithNullFilters. * State TTL via keyed processing-time timers with a post-flip grace window; rearms are amortized so effective time-to-eviction is in [1.0×, 1.5×] stateTtlMs. Generated-By: Claude Opus 4.7 (1M context)
…minatedByJobFinished On a loaded machine the cooldown-driven Idling transition terminates the in-progress rescale with NO_RESOURCES_OR_PARALLELISMS_CHANGE before the job finishes, so goToFinished's later JOB_FINISHED stamp is dropped and the wait can never succeed. Widen the fixture cooldown to keep the rescale in-progress until the job finishes, mirroring FLINK-39903/FLINK-40010. Generated-by: Claude Opus 4.8 (1M context)
…n group window aggregate (apache#28682)
…rtex disagree on max parallelism During restore, each operator's recorded max parallelism is reconciled onto its shared vertex individually. When chaining has regrouped operators since the checkpoint, they can disagree, and the reconciliation silently keeps whichever value is applied last -- restoring a keyed operator under a foreign key-group count. Before the loop, verify that all operators chained into a vertex carrying keyed state agree, and fail with a clear error otherwise. The check spans all operators, since a non-keyed one's value can win the reconciliation and misroute a keyed operator's state. Adds ChainingMaxParallelismStateLossITCase covering both directions (explicit value below and above the chain head) on the HashMap and RocksDB backends.
…e#28674) Adds ForbidSnapshotOutsideLateralRule, which rejects any SNAPSHOT scan that survives the LATERAL SNAPSHOT rewrite with a clear message instead of failing later in the generic PTF translation. Generated-By: Claude Opus 4.8 (1M context)
…e TLS check testHttpsConnectionWithDefaultCerts only needs to prove that RestClient completes a TLS handshake against a public endpoint using the default JDK truststore. It additionally required the response body to be valid JSON, so it failed whenever GitHub returned a plain-text 429 Too Many Requests. This is structural on GitHub Actions, where runners share egress IPs, and will recur more often as CI moves from Azure to GHA. Any completed HTTP response proves the handshake succeeded, so assert that the failure chain contains a RestClientException (which is only produced after a full HTTP response is received). A genuine certificate/trust regression surfaces as an SSLException instead and still fails the test, as do connect-refused, DNS, and connection-reset failures where no handshake happened at all. Generated-by: Claude Fable 5
gkalashyan-akv
force-pushed
the
task/add-sql-support
branch
2 times, most recently
from
July 16, 2026 10:18
3a0d5b5 to
7440225
Compare
gkalashyan-akv
force-pushed
the
task/add-sql-support
branch
from
July 16, 2026 14:38
7440225 to
48e0255
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
17 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.
What is the purpose of the change
(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)
Brief change log
(for example:)
Verifying this change
Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.
(Please pick either of the following options)
This change is a trivial rework / code cleanup without any test coverage.
(or)
This change is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Does this pull request potentially affect one of the following parts:
@Public(Evolving): (yes / no)Documentation
Was generative AI tooling used to co-author this PR?