[SPARK-57228][SS] Support transformWithState in Real-Time Mode

[jerrypeng]

What changes were proposed in this pull request?

This PR enables the JVM (Scala/Java) transformWithState operator to run in Structured Streaming Real-Time Mode (RTM).

The main changes are:

• Allow TransformWithStateExec in RTM and propagate the authoritative RTM flag from IncrementalExecution, including batches where a source is temporarily replaced by LocalRelation.
• Process each input row independently without requiring the key-sorted input assumed by micro-batch execution.
• Use the live RTM clock for processing-time semantics, processing-time timers, and TTL state.
• Check expired timers after each input row. A reusable RocksDB iterator avoids creating a native iterator and task-completion listener per row. The first scan starts at the beginning,
  subsequent scans resume from the previous expiration threshold, and the final batch scan remains unbounded.
• Perform periodic TTL cleanup while an RTM batch is running and a final cleanup at batch completion. The internal configuration
  spark.sql.streaming.realTimeMode.transformWithState.ttlEvictionIntervalMs controls the interval and defaults to one second.
• Keep event-time watermarks fixed within a batch and update them between batches. Event-time timers already expired against the batch watermark fire incrementally as rows arrive.
• Load initial state in a finite batch-0 bootstrap before the first long-running RTM batch. Pipelined shuffle is disabled for this bootstrap batch.
• Record V2 operator metadata during planning, write it after the offset-log entry is durable, and validate recovery using metadata from the last committed batch rather than an
  uncommitted retry attempt.
• Support checkpoint recovery, transitions between micro-batch and RTM, output event-time columns, State Data Source reads, RocksDB changelog checkpointing, and state-store row checksums.

The public transformWithState API is unchanged. This PR is limited to the JVM implementation and does not enable PySpark transformWithState in RTM.

Why are the changes needed?

The existing transformWithState execution path assumes a finite micro-batch:

• Input is sorted and grouped by key.
• Processing time is fixed at the driver-provided batch timestamp.
• Timer expiration and TTL cleanup can wait until batch completion.

These assumptions do not hold for RTM. RTM uses long-running batches and pipelined shuffle, which does not provide the ordering required by grouped micro-batch processing. A fixed batch
timestamp and batch-end-only cleanup would also delay processing-time timers and TTL expiration for the duration of the RTM batch.

The recovery changes prevent an unsuccessful batch attempt from causing the next attempt to validate against uncommitted operator metadata.

Does this PR introduce any user-facing change?

Yes.

Previously, JVM transformWithState queries using Trigger.RealTime were unsupported and rejected. With this PR, Scala and Java transformWithState queries can run in RTM under the
existing RTM restrictions.

Processing-time timers and TTL state can expire while a long-running RTM batch remains open. Event-time watermarks continue to advance only between batches, while timers already expired
against the current batch watermark can fire incrementally as input arrives.

PySpark transformWithState remains unsupported in RTM.

How was this patch tested?

The main RTM coverage was run with both the standard RocksDB configuration and row checksums enabled:

build/sbt \
  'sql/testOnly org.apache.spark.sql.streaming.RealTimeTransformWithStateSuite org.apache.spark.sql.streaming.RealTimeTransformWithStateSuiteWithRowChecksum'

Additional review-driven validation included:

• Reusable timer iterator tests with processing-time and event-time timers under UnsafeRow and Avro encoding: 4/4 passed.
• The same iterator tests with row checksums enabled: 4/4 passed.
• RTM task-completion-listener reuse integration test: passed.
• Focused RTM recovery, initial-state bootstrap, changelog checkpointing, and incremental timer tests: 7/7 passed.
• Kafka null-latestOffset JVM transformWithState regression test: passed.
• sql/Test/scalastyle: 1,093 files checked with no errors or warnings.

Coverage includes:

• Value, map, and list state
• Incremental and final TTL cleanup
• Processing-time and event-time timers
• Empty-batch final timer scans
• Fixed within-batch watermarks and between-batch advancement
• Late-row filtering and watermark metrics
• Initial-state bootstrap, failure, and retry
• Batch-0 non-pipelined shuffle planning
• Checkpoint restart and metadata recovery
• RocksDB changelog checkpointing enabled and disabled
• Micro-batch to RTM to micro-batch recovery
• Stateful operator chaining and Union validation
• Output event-time columns
• Kafka recovery and null-offset planning
• State Data Source reads
• State-store row checksums

Was this patch authored or co-authored using generative AI tooling?

Co-authored with OpenAI Codex