[BUG] Flink COW insert/upsert can hide existing base-file records after unaligned-checkpoint recovery

### Bug Description

**What happened:**

We can reproducibly trigger silent data loss in the Flink COW write path after restoring from an unaligned checkpoint. The problem is not limited to insert clustering: we reproduced it with `write.operation=insert` and `write.insert.cluster=true`, and independently observed the same physical overwrite pattern with `write.operation=upsert`.

An in-flight record tagged as INSERT (`instantTime = "I"`) is restored from checkpoint channel state and reaches the writer before UPDATE records for the same `partitionPath + fileId`. The writer bucket therefore starts with an INSERT record even though the file group already has a committed base file.

The write path selects `FlinkCreateHandle` for the existing fileId instead of a merge/concat handle. A new parquet file is written with the same fileId but only the small set of records from the current batch. Snapshot reads select this newer base-file version and hide the records that remain physically present in the older parquet file.

There is no exception and the commit succeeds.

Observed evidence from a deterministic reproduction:

- Existing file group at instant A contained 82,182 records.
- After recovery, the next writer bucket contained 2,028 records:
- 57 records tagged INSERT
- 1,971 records tagged UPDATE
- first record type was INSERT
- Hudi selected `FlinkCreateHandle`.
- No previous base-file path was opened by the handle.
- The next commit metadata reported `prevCommit = null` and `numWrites = numInserts`.
- The new parquet reused the existing fileId but did not carry forward the old base-file records.
- Snapshot row count decreased although `numDeletes = 0` and there were no write errors.

The same pattern was independently observed on a COW table configured with `write.operation=upsert`. Three existing file groups were replaced as follows:

- 408 previous rows -> 3 rows in the new base file
- 569 previous rows -> 2 rows in the new base file
- 448 previous rows -> 3 rows in the new base file

For all three new versions, commit metadata reported `prevCommit = null`, `numWrites = numInserts`, `numUpdates = 0`, and no deletes or write errors. Direct physical parquet reads confirmed that the old files still contained the hidden records.

Therefore, `write.insert.cluster` increases exposure in the insert path but is not a necessary condition for the underlying handle-selection failure.

**What you expected:**

Restoring records from a valid Flink checkpoint must not cause an existing Hudi file group to be treated as a new file handle in either INSERT or UPSERT mode.

For a fileId that already has a committed base file, Hudi should either:

1. select the merge/concat handle and preserve the existing base-file records; or
2. fail fast when a writer bucket contains inconsistent INSERT/UPDATE tags for the same existing file group.

A successful commit with zero deletes must not reduce the visible records of an existing file group.

**Steps to reproduce:**

Scenario A — deterministic insert-path reproduction:

1. Create a COPY_ON_WRITE table with a primary key.
2. Configure:
- `write.operation = insert`
- `write.insert.cluster = true`
- unaligned checkpoints enabled
3. Write enough data to create and commit a base file for a fileId.
4. Arrange for an INSERT-tagged record for that fileId to remain in channel state when an unaligned checkpoint completes.
5. Restart the job from that checkpoint.
6. After recovery, send UPDATE-tagged records routed to the same `partitionPath + fileId`, with the restored INSERT record arriving first.
7. Let the next Hudi instant commit.
8. Compare the old/new parquet files, commit metadata, and snapshot row count.

Scenario B — upsert path:

1. Create a COPY_ON_WRITE table with a primary key and `write.operation = upsert`.
2. Enable unaligned checkpoints; `write.insert.cluster` is not required.
3. Create and commit existing file groups.
4. Restore from a checkpoint that contains in-flight INSERT-tagged records.
5. Allow restored INSERT-tagged and current UPDATE-tagged records for the same fileId to enter the next writer bucket.
6. Inspect the next commit for an existing fileId with `prevCommit = null` and `numWrites = numInserts`.
7. Physically compare the old and new parquet files for that fileId.

The issue becomes easier to reproduce with Flink mini-batch enabled because more records are released together after recovery.

**Likely code path:**

In Hudi 0.15.x, `StreamWriteFunction` groups records by `partitionPath + fileId`. Handle selection ultimately depends on the first record passed to `FlinkWriteHandleFactory`:

```java
if (loc.getInstantTime().equals("I")) {
writeHandle = new FlinkCreateHandle<>(...);
} else {
writeHandle = createMergeHandle(...);
}
```

This makes handle selection dependent on restored record ordering instead of whether the file group already has a committed base file. The condition is shared by the COW commit write path used by both insert and upsert.

**Workaround:**

Disabling unaligned checkpoints prevents the reproduction in our tests:

```
execution.checkpointing.unaligned.enabled = false
```

This is only a workaround. Unaligned checkpoints are a supported Flink exactly-once feature, and Hudi should safely handle restored in-flight records.

### Environment

**Hudi version:** 0.15.0

**Query engine:** Flink 1.18.1 / Flink SQL

**Table type:** COPY_ON_WRITE

**Storage:** S3-compatible object storage

**Write mode:** Streaming

**Reproduced/observed write configurations:**

Insert path:

```properties
write.operation=insert
write.insert.cluster=true
write.precombine=true
execution.checkpointing.unaligned.enabled=true
table.exec.mini-batch.enabled=true
table.exec.mini-batch.size=5000
table.exec.mini-batch.allow-latency=5s
```

Upsert path:

```properties
write.operation=upsert
write.precombine=true
execution.checkpointing.unaligned.enabled=true
```

The mini-batch settings and `write.insert.cluster` increase reproduction probability in the insert scenario, but neither is required for the upsert observation.

### Logs and Stack Trace

No exception is thrown. The job and Hudi commit both complete successfully.

Deterministic insert-path diagnostic sequence:

```text
instant A:
fileId=
baseFileRows=82182

restored writer bucket for instant B:
records=2028
insertTagged=57
updateTagged=1971
firstRecordType=I

handle selection:
handleClass=FlinkCreateHandle
previousWritePath=NONE

commit metadata for instant B:
fileId=
prevCommit=null
numWrites=2028
numInserts=2028
numUpdates=0
numDeletes=0
totalWriteErrors=0
```

Independent upsert-path observations:

```text
file group A: previousWrites=408, currentWrites=3, currentPrevCommit=null
file group B: previousWrites=569, currentWrites=2, currentPrevCommit=null
file group C: previousWrites=448, currentWrites=3, currentPrevCommit=null

all current versions:
numWrites=numInserts
numUpdates=0
numDeletes=0
totalWriteErrors=0
```

Physical validation:

```text
old parquet: same fileId, hidden records are present
new parquet: same fileId, hidden records are absent
snapshot: selects new parquet, so old records are no longer visible
```

Start with StreamWriteFunction and trace how records grouped by partitionPath + fileId reach FlinkWriteHandleFactory, FlinkCreateHandle, or createMergeHandle. Reproduce the insert and upsert scenarios with unaligned checkpoints, then verify that an existing file group preserves its base-file records or fails fast instead of committing a replacement with prevCommit=null.

Written by the indexing model from the issue text.