[spark] Add spark tiering service

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

## 1. Motivation

- **Current limitation**: The tiering service currently only supports Flink as the compute backend.
- **Why Spark**: Many users run Spark-based data platforms and don't have Flink in their infrastructure. Requiring Flink solely for tiering creates operational overhead.
- **Use case alignment**: Tiering is inherently a periodic batch operation (heartbeat -> poll -> read -> write -> commit), which maps naturally to Spark's RDD-based batch parallelism.
- **Cloud/serverless integration**: Spark is well-supported in cloud platforms (Databricks, EMR, Dataproc). Shared Spark clusters can run tiering alongside other workloads without a dedicated Flink cluster.
- **Ecosystem completeness**: Fluss 0.9 already supports Spark catalog/read/write; 1.0 plans Union Read for Spark. Adding Spark tiering completes the full lifecycle -- users can tier, read union views, and query all from Spark without Flink dependency.

---

## 2. Public Interfaces

- **New module**: `fluss-spark-tiering` -- fat JAR with main class `SparkLakeTiering`
- **New CLI interface**: `spark-submit --class SparkLakeTiering` with `--fluss.*`, `--datalake.format`, `--datalake.{format}.*`, `--lake.tiering.*` arguments (mirrors `FlussLakeTiering` CLI contract)
- **New configuration options** (in `SparkTieringOptions`):
- `tiering.poll.table.interval` (default 30s)
- `tiering.heartbeat.interval` (default 10s)
- `tiering.poll.timeout` (default 10s)
- `tiering.heartbeat.max.failures` (default 10)
- **No changes to existing public APIs**: Reuses `LakeTieringFactory`, `LakeWriter`, `LakeCommitter`, `WriterInitContext`, `CommitterInitContext` from `fluss-common`
- **No changes to coordinator heartbeat protocol**: Uses the same `lakeTieringHeartbeat` RPC as Flink tiering

---

## 3. Proposed Changes

### 3.1 Architecture Overview

```
DRIVER (long-running loop) EXECUTORS (parallel per bucket)
+-----------------------------------+ +----------------------------------+
| SparkTieringJobRunner | | TieringTask.process() |
| 1. heartbeat(request_table=true) | | 1. Open Connection to Fluss |
| 2. TieringSplitGenerator | | 2. Create LakeWriter |
| .generateSplits(tableInfo) | | 3. Read snapshot or log data |
| 3. sc.parallelize(splits) | ------> | 4. lakeWriter.write(record) |
| .map(process).collect() | <------ | 5. lakeWriter.complete() |
| 4. TieringCommitter.commitAll() | | 6. Serialize WriteResult |
| 5. Report finished via heartbeat | +----------------------------------+
+-----------------------------------+
|
| Background heartbeat thread during RDD job
v
Fluss Coordinator (lakeTieringHeartbeat RPC)
```

**Comparison with Flink model**:

| Component | Flink | Spark |
|-----------|-------|-------|
| **Execution** | Streaming Source -> Reader -> Commit Operator chain | Driver loop: heartbeat -> RDD job -> commit |
| **State** | Flink checkpointing | Driver in-memory state (recoverable via heartbeat) |
| **Parallelism** | Source Reader parallelism | RDD partition parallelism (one partition per bucket) |
| **Heartbeat** | Embedded in Source Enumerator | Background thread during RDD execution |
| **WriteResult Transport** | Flink record stream | Serialized bytes in RDD collect |

### 3.2 Module Structure

```
fluss-client/
└── src/main/java/org/apache/fluss/client/tiering/
├── FlussTableLakeSnapshotCommitter.java # Two-phase lake snapshot commit (shared by Flink & Spark)
├── TieringWriterInitContext.java # WriterInitContext impl (shared by Flink & Spark)
└── TieringCommitterInitContext.java # CommitterInitContext impl (shared by Flink & Spark)

fluss-spark/
├── fluss-spark-common/
│ └── src/main/scala/org/apache/fluss/spark/tiering/
│ ├── SparkTieringJobRunner.scala # Main driver loop
│ ├── TieringCoordinator.scala # Heartbeat RPC management
│ ├── TieringSplitGenerator.scala # Split generation logic
│ ├── TieringSplit.scala # Split types (LogSplit, SnapshotSplit)
│ ├── TieringTask.scala # Executor-side processing
│ ├── TieringTaskResult.scala # Serialized result type
│ ├── TieringCommitter.scala # Two-phase commit orchestration
│ ├── SparkTieringOptions.scala # Configuration options
│ └── package.scala # Utilities
│
└── fluss-spark-tiering/ # Entry point module
├── pom.xml
└── src/main/scala/org/apache/fluss/spark/tiering/
└── SparkLakeTiering.scala # Main entry point
```

### 3.3 Component Design

#### SparkLakeTiering (entry point)
- Parses CLI args: `--fluss.*` / `--datalake.format` / `--datalake.{format}.*` / `--lake.tiering.*`
- Uses `PropertiesUtils.extractAndRemovePrefix()` for fluss/datalake configs, `extractPrefix()` for lake.tiering
- Creates SparkSession, instantiates `SparkTieringJobRunner` (which loads `LakeStoragePlugin` internally)
- Calls `runner.startAsync()`, blocks via `Await.result(future, Duration.Inf)`
- *Ported from `FlussLakeTiering.java` (entry point logic)*

#### SparkTieringJobRunner (orchestrator)
- Async lifecycle: `startAsync()` returns `Future[Unit]`, `stop(timeout)` sets `AtomicBoolean` and awaits
- Internal `ExecutionContext` with single daemon thread for loop execution
- Main loop: heartbeat -> get table -> generate splits -> RDD job -> commit -> report
- Background `ScheduledExecutorService` for keepalive heartbeats during RDD execution
- DI factory params for testability (`coordinatorFactory`, `snapshotCommitterFactory`, `connectionFactory`, `splitGeneratorFactory`)
- *New component — combines roles of `TieringSourceEnumerator` and `LakeTieringJobBuilder` adapted to Spark's driver-loop model*

#### TieringCoordinator (heartbeat management)
- Manages heartbeat RPC with Fluss coordinator
- State: `tieringTableEpochs`, `finishedTables`, `failedTableEpochs`
- `open()` / `close()` lifecycle, `AutoCloseable`
- Close sends final heartbeat reporting in-progress tables as failed
- *Ported from `TieringSourceEnumerator.HeartBeatHelper` (heartbeat protocol logic)*

#### TieringSplitGenerator (split generation) — **full port**
- Log splits for incremental changes (both log and PK tables)
- Snapshot splits for first-time PK table tiering (when KV snapshot exists)
- Partitioned table support: generates splits per partition per bucket
- *Full port from `TieringSplitGenerator.java` — logic identical*

#### TieringTask (executor-side processing) — **full port**
- Object with `process()` static method, runs on Spark executors
- Re-initializes `LakeTieringFactory` per executor from config (avoids serialization)
- For log splits: `LogScanner` poll loop until `stoppingOffset`
- For snapshot splits: `BatchScanner` reads all KV snapshot data
- *Full port from `TieringSplitReader.java` — read/write logic identical*

#### TieringCommitter (commit pipeline) — **full port**
- 6-step pipeline: deserialize -> toCommittable -> check missing snapshots -> prepare -> commit lake -> commit Fluss
- Uses `FlussTableLakeSnapshotCommitter` from `fluss-client` (not a custom port)
- *Full port from `TieringCommitOperator.java` — commit logic identical*

#### TieringWriterInitContext / TieringCommitterInitContext — **shared from `fluss-client`**
- Implement `WriterInitContext` and `CommitterInitContext` interfaces respectively
- Java classes in `org.apache.fluss.client.tiering`; used directly by both Flink and Spark with no Scala port needed
- *Moved from `fluss-flink-common` to `fluss-client` as part of the shared component refactoring*

#### TieringSplit (data types)
- `sealed trait TieringSplit` + `TieringLogSplit` / `TieringSnapshotSplit` case classes
- Serializable for Spark closure passing
- *New Scala-idiomatic design (sealed trait + case classes) replacing Flink's `TieringSplit.java` class hierarchy*

### 3.4 Serialization Strategy

- `WriteResult` serialized on executors via `SimpleVersionedSerializer` to avoid Spark serialization issues with lake-specific types
- `LakeTieringFactory` re-initialized per executor from config (not serialized across driver-executor boundary)
- Configs passed via Spark closure serialization (not broadcast variables)

### 3.5 Heartbeat Lifecycle

```
IDLE --heartbeat(request=true)--> COORDINATOR
<-- response: table T1 --

START RDD --background heartbeat every 10s--> COORDINATOR
(tiering_tables=[T1])

RDD DONE --heartbeat(finished=[T1], request=true)--> COORDINATOR
<-- response: table T2 or empty --
```

- Background heartbeat: `ScheduledExecutorService` single thread during RDD job
- Table drop detection: checks `tableId` mismatch, sets `tableCancelled` flag, calls `sc.cancelJobGroup()`

### 3.6 Error Handling

| Scenario | Handling |
|----------|---------|
| Task failure | Spark auto-retry (`spark.task.maxFailures`); fresh `LakeWriter` per attempt |
| RDD job failure (all retries exhausted) | Mark table failed, report in next heartbeat |
| Commit failure | Abort lake commit, mark table failed |
| Heartbeat failure | Counter-based threshold (`MAX_HEARTBEAT_FAILURES` default 10) |
| Table dropped during tiering | Background heartbeat detects tableId mismatch, cancels RDD, marks failed |
| Missing lake snapshot | Commit snapshot to Fluss first, abort current committable, fail round |

### 3.7 Cooperative Shutdown

- `AtomicBoolean` stopped flag; thread interruption unreliable for `RDD.collect()`, `Thread.sleep`, RPC calls
- `stop(timeout)` for graceful shutdown from main thread
- Signal handler approach (SIGTERM/SIGINT) for production use

---

## 4. Code Reuse and Duplication Strategy

- **Shared interfaces**: Reuses all `fluss-common` lake interfaces (`LakeTieringFactory`, `LakeWriter`, `LakeCommitter`, etc.) -- zero changes
- **Shared protocol**: Same `lakeTieringHeartbeat` RPC -- zero changes to `fluss-rpc`
- **Shared component refactoring**: Three classes moved from `fluss-flink-common` to `fluss-client` under package `org.apache.fluss.client.tiering`, enabling both Flink and Spark to share the same logic without duplication:
- `FlussTableLakeSnapshotCommitter` — two-phase lake snapshot commit logic
- `TieringWriterInitContext` — `WriterInitContext` implementation
- `TieringCommitterInitContext` — `CommitterInitContext` implementation
- **Duplicated logic with TODO markers**: 3 components ported from Flink to Scala:
- `TieringSplitGenerator` <- `TieringSplitGenerator.java`
- `TieringCommitter` <- `TieringCommitOperator.java`
- `TieringTask` <- `TieringSplitReader.java`

---

## 5. Compatibility, Deprecation, and Migration Plan

- **No breaking changes**: Pure additive feature -- new module, new files only
- **No existing Spark behavior affected**: `fluss-spark-common` gains tiering package, but no existing classes/APIs modified
- **No Flink tiering changes**: Flink tiering continues to work unchanged
- **Coordinator compatibility**: Uses the same heartbeat protocol, no server-side changes needed
- **Lake plugin compatibility**: All 3 lake plugins (Paimon, Iceberg, Lance) work unmodified
- **Spark version support**: Spark 3.4 and 3.5 (matching existing `fluss-spark` module matrix)
- **Migration**: Users can switch from Flink to Spark tiering by changing the submission command. Same CLI argument convention.

---

## 6. Test Plan

- **Unit tests**: `SparkTieringLogTableTest` -- validates log table tiering end-to-end (create table -> produce data -> run tiering -> verify lake data)
- **Test infrastructure**: `SparkTieringTestBase` -- sets up embedded Fluss cluster + SparkSession + lake storage for integration testing
- **Planned tests**:
- PK table tiering (snapshot + incremental)
- Partitioned table tiering
- Error recovery (table drop during tiering)
- Multi-round tiering (multiple tables sequentially)
- **Build verification**: `mvn compile` + `mvn spotless:check` + Checkstyle (no Flink imports in Spark modules)
- **Manual testing**: `spark-submit` against a running Fluss cluster with Paimon/Iceberg/Lance enabled tables

---

## 7. Rejected Alternatives

| Alternative | Reason for Rejection |
|-------------|---------------------|
| Wrapping Flink tiering in Spark | Adds Flink runtime dependency, defeats the purpose of Spark-native solution |
| Spark Structured Streaming | Tiering is naturally a batch-per-round operation; streaming adds complexity (micro-batch overhead, checkpoint management) without benefit |
| Thread interruption for shutdown | `Thread.interrupt()` is unreliable for `RDD.collect()`, `Thread.sleep`, and RPC calls. Cooperative `stopped` flag is correct. |
| Broadcast variables for config passing | Replaced with closure serialization for simplicity (configs are small) |
| Custom `FlussTableSnapshotCommitter` / `TieringWriterInitContext` / `TieringCommitterInitContext` in Scala | Initially considered porting from Flink to Scala; replaced by moving these three classes to `fluss-client` (`org.apache.fluss.client.tiering`) so both Flink and Spark share the same Java implementations directly |

---

## 8. Usage

### Submitting the Spark Tiering Job

```bash
spark-submit \
--class org.apache.fluss.spark.tiering.SparkLakeTiering \
fluss-spark-tiering_2.12.jar \
--fluss.bootstrap.servers localhost:9123 \
--datalake.format paimon \
--datalake.paimon.warehouse /path/to/warehouse \
--fluss.tiering.poll.table.interval 30s
```

### Enabling Tiering on a Fluss Table

```sql
CREATE TABLE my_table (
id INT,
name STRING
) TBLPROPERTIES (
'table.datalake.enabled' = 'true',
'table.datalake.freshness' = '1h'
);
```

---

## 9. Future Work

- **Tiering code refacoring**: Extract more common tiering logic into fluss-client module (Java) shared by both Flink and Spark. This is a follow-up effort; the initial PR keeps Spark and Flink modules independent per the constraint "no Flink module modifications"
- **Metrics**: Add tiering metrics (bytes written, records tiered, tiering duration) in Spark UI
- **Tiering status tracking**: Integrate with FIP-30 (#2362)
- **Additional Spark version support**: Spark 4.x
- **Decouple from datalake format**: Currently each tiering service instance is bound to a single datalake format (e.g., Paimon). A single tiering service should be able to tier tables to different lake formats simultaneously.
- **Multi-table concurrent tiering**: Currently the driver processes one table per loop iteration. Support tiering multiple tables in parallel to improve throughput and reduce latency.
- **Refactor from RDD to Spark DataSource/operators**: Replace the low-level `sc.parallelize().map().collect()` pattern with Spark's DataSource V2 or custom operators for better integration with Spark's scheduler, metrics, and fault tolerance.

### Willingness to contribute

- [x] I'm willing to submit a PR!

Start by reading the proposed fluss-spark-common classes, SparkLakeTiering.scala, and the Flink tiering components they port or share. Run SparkTieringLogTableTest with SparkTieringTestBase to understand the existing integration setup. Done means the Spark tiering module, heartbeat flow, executor processing, commit pipeline, and the listed tiering tests work for the supported scenarios.

Written by the indexing model from the issue text.