[SPARK-53915] Add RealTimeScanExec and ability to execute long running batches

[jerrypeng]

What changes were proposed in this pull request?

The following changes are added to support running queries in Real-time Mode

1. RealTimeScanExec. This is a new scan added for Real-time mode which enables queries to run batches for a fix time duration. It is also allows us to collect the offsets tasks have processed up to at the end of the batch.
2. Offset management changes needed for Real-time Mode. In Real-time Mode, offsets that represent the point that tasks has processed up to are collected at the end of a batch via RealTimeScanExec and sent to the driver. The driver will then write the offsets to the offset log to persist / checkpoint the progress of the batch. The commit log for the batch will also be written afterwards to ensure compatibility with existing streaming queries. This differs from existing streaming queries as offsets that a batch will process up to are determined and written to the offset log before executing the batch.

Why are the changes needed?

With the changes in this PR, simple single stage stateless queries can be executed in Real-time Mode. With this PR, we can run queries in Real-time Mode that read from a MemorySource (LowLatencyMemorySource), do some processing, and write the result to MemorySink (ContinuousMemorySink).

Does this PR introduce any user-facing change?

Yes, adds functionality to run queries in Real-time Mode

How was this patch tested?

Many tests are added in this PR. Additional tests will be added in a subsequent PR to limit the size of this PR.

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

No