**Flink SQL scan on Fluss table with DataLake enabled reads historical Iceberg files despite `table.log.ttl`, causing `ConnectionClosedException` on old Parquet files**

### Search before asking

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

### Fluss version

0.9.0 (latest release)

### Please describe the bug 🐞

## Body

````markdown
### Fluss Version

0.9.1-incubating

### Deployment

- Kubernetes
- Flink 2.2
- Fluss Coordinator + 2 TabletServers
- Nessie Catalog
- Iceberg
- MinIO (S3)
- table.datalake.enabled=true

---

## Description

We have a Fluss pipeline that continuously writes Kafka messages into a Fluss table.

Example table:

```sql
CREATE TABLE rlc_stream_message_v2 (
kafka_partition INT,
kafka_offset BIGINT,
kafka_timestamp TIMESTAMP(3),
message_type STRING,
payload STRING
)
WITH (
'table.log.ttl'='24 h',
'table.datalake.enabled'='true',
'table.datalake.format'='iceberg'
);
````

This table has been running for several days without any visible problems.

We then create another Flink SQL job that reads from this Fluss table and writes selected records into another Fluss table.

Example:

```sql
INSERT INTO rlc_trade_flat
SELECT ...
FROM rlc_stream_message_v2
WHERE message_type='Asa.Mdp.Protobuf.Rlc.Messages.Trade';
```

The destination table may or may not have DataLake enabled.

---

## Problem

After running for some time (sometimes minutes, sometimes hours), the reader job always fails with:

```
org.apache.iceberg.exceptions.RuntimeIOException

Caused by:

org.apache.http.ConnectionClosedException:
Premature end of Content-Length delimited message body
(expected: 442,423,704;
received: 135,330,000)
```

The failure always references old Iceberg parquet files.

Example:

```
442423708 bytes
```

This file is several days old.

---

## Observations

### 1)

The source table continues ingesting Kafka normally.

The original ingestion pipeline never fails.

---

### 2)

The reader pipeline restarts.

After restart it again eventually attempts to read the same historical parquet files and fails again.

---

### 3)

Running

```sql
SELECT *
FROM rlc_stream_message_v2
/*+ OPTIONS('scan.startup.mode'='latest') */;
```

starts from current records only and returns today's data.

This appears to avoid reading historical Iceberg files.

---

### 4)

Running

```sql
EXPLAIN PLAN FOR
SELECT *
FROM rlc_stream_message_v2;
```

does not indicate that Iceberg is being used.

The execution plan only shows

```
TableSourceScan
```

---

### 5)

The table has

```
table.log.ttl='24 h'
```

We expected a normal Fluss scan to read only the Fluss log.

Instead it appears that historical Iceberg files are also scanned.

---

### 6)

Removing

```
table.datalake.enabled
```

reduces the frequency of the problem, but eventually the same exception still appears.

---

## Expected Behavior

A continuous Flink SQL streaming job reading a Fluss table should continue reading from the Fluss log.

Old Iceberg files should not be revisited during normal streaming unless explicitly requested.

A table with

```
table.log.ttl='24 h'
```

should not require reading parquet files that are several days old.

---

## Actual Behavior

The streaming reader eventually attempts to read historical Iceberg parquet files.

If one of those large parquet reads fails, the whole Flink job fails and restarts.

After restart the same behavior repeats.

---

## Questions

1. Is a normal Fluss streaming scan expected to read historical Iceberg files?

2. Under what conditions does Fluss switch from reading the log to reading Iceberg snapshots?

3. Does `table.log.ttl` affect only the Fluss log while snapshot planning still includes all Iceberg history?

4. Is this expected behavior or a bug?

````

---

I would add **one more piece of evidence** at the bottom before posting. It's quite compelling:

```markdown
Additional Observation

The source table has been ingesting continuously for multiple days without errors.

Only downstream Flink SQL jobs that read from this Fluss table experience the failure.

This suggests the issue occurs during Fluss table reading rather than Fluss ingestion.
````

This report gives the Fluss maintainers:

* your architecture,
* the exact exception,
* what you've already ruled out,
* your expectation,
* and the specific question about why a streaming read appears to fall back to historical Iceberg files.

It's focused on facts rather than assumptions, which makes it much easier for maintainers to investigate.

### Solution

_No response_

### Are you willing to submit a PR?

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

Start with the Flink SQL TableSourceScan behavior for a DataLake-enabled Fluss table, using the reported CREATE TABLE and INSERT INTO statements. Compare a normal scan with scan.startup.mode='latest' while table.log.ttl is 24 h, and trace when historical Iceberg files enter the read plan. Done means determining whether this fallback is expected and preventing unintended historical reads or documenting the required behavior.

Written by the indexing model from the issue text.