morningman opened a new issue, #70540: URL: https://github.com/apache/doris/issues/70540
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/doris/issues?q=is%3Aissue) and found no similar issues. ### Version 4.1.5, reported by a community user (Iceberg REST catalog served by Lakekeeper). master at 172120cbaa4 has the same code path. That is from reading the code; it was not reproduced on master. ### What's Wrong? A streaming insert job (continuous load) whose target is an Iceberg table never advances its offset. Every task reads the first batch of files again and commits it to Iceberg as a new append snapshot. No error is reported anywhere, so the job keeps duplicating data until someone pauses it. The job: ```sql CREATE JOB toy_events_v1_iceberg PROPERTIES ( "s3.max_batch_files" = "1", "max_interval" = "10" ) ON STREAMING DO INSERT INTO <iceberg_catalog>.<db>.toy_events_v1 (event_id, partition_timestamp) SELECT event_id, partition_timestamp FROM S3( "uri" = "s3://<bucket>/toy_events_v1/year=2026/month=11/day=12/hour=*/*", "s3.region" = "us-east-2", "s3.endpoint" = "https://s3.us-east-2.amazonaws.com";, "s3.credentials_provider_type" = "CONTAINER", "format" = "parquet" ); ``` Observed: - `CurrentOffset` in `jobs("type" = "insert")` stays empty. - Every task has the same `RunningOffset`: `startFileName = endFileName = toy_events-0.parquet`. - Every task ends `SUCCESS` with an empty `ErrorMsg`. A new task starts about every 1.5 s. - Each task commits a full append snapshot (a linear parent chain) with the same rows. With 5 files of 100 rows, the table had 1,400 rows and only 100 distinct `event_id` when the job was paused by hand. In a larger test the table reached 12 times the first batch of 13,667 rows, still with 13,667 distinct ids. - A task cancelled by `PAUSE JOB` still committed its Iceberg snapshot. Ruled out by the reporter: - The same job (same files, same `SELECT`, same properties) into a Doris internal table advances through 5 consecutive files and ends with 500 rows and 500 distinct `event_id`. - A plain `INSERT INTO <iceberg table> SELECT ... FROM S3(...)` works. - The job's current database does not matter: it loops after both `USE <iceberg_catalog>.<db>` and `USE internal.<db>`. - Lakekeeper logs show no commit conflicts or server errors. ### What You Expected? The job loads each batch once and advances `CurrentOffset`, as it does for an internal table. If Iceberg (or any external table) is not a supported target, `CREATE JOB ... ON STREAMING` should reject it instead of accepting a job that duplicates data. ### How to Reproduce? 1. Put 5 parquet files `toy_events-0.parquet` … `toy_events-4.parquet` under one S3 prefix, 100 rows each, with `event_id` unique across files: ``` message ToyEventsWriteModel { required binary event_id (STRING); required int64 partition_timestamp (TIMESTAMP(MICROS,true)); } ``` 2. Create the target table in an Iceberg catalog: ```sql CREATE TABLE <iceberg_catalog>.<db>.toy_events_v1 ( event_id STRING, partition_timestamp DATETIME(6) ) PARTITION BY LIST (DAY(partition_timestamp)) () PROPERTIES ("write.format.default" = "parquet", "compression_codec" = "zstd"); ``` 3. Create the job above and watch: ```sql SELECT Status, CurrentOffset, EndOffset, ErrorMsg FROM jobs("type" = "insert") WHERE Name = 'toy_events_v1_iceberg'; SELECT Status, RunningOffset FROM tasks("type" = "insert") WHERE JobId = '<job id>' ORDER BY CreateTime; SELECT count(*), count(DISTINCT event_id) FROM <iceberg_catalog>.<db>.toy_events_v1; ``` ### Anything Else? #### Root cause A TVF streaming job moves its offset only inside the Doris load-transaction callbacks, and only the internal-table insert path attaches its transaction to the job. An external-table insert commits through the catalog's own transaction manager, so the callbacks never run. The insert still returns OK, so the task counts as successful and the next task starts from the unchanged, empty offset. Links point to 4.1.5-release (917b20be838) and master (172120cbaa4). 1. `OlapInsertExecutor.setTxnCallbackId()` sets `TransactionState.callbackId` to the streaming job id ([4.1.5](https://github.com/apache/doris/blob/917b20be83878e98f94e3ebf7efccc80346e3e50/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java#L259-L271), [master](https://github.com/apache/doris/blob/172120cbaa46b0a2eff6c37d86fb438c3a307375/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java#L269-L281)). When `GlobalTransactionMgr` commits that transaction, it calls `StreamingInsertJob.beforeCommitted()` and `afterCommitted()`. 2. `beforeCommitted()` refuses the commit if the task was cancelled, and puts the task's end offset into a `StreamingTaskTxnCommitAttachment` ([4.1.5](https://github.com/apache/doris/blob/917b20be83878e98f94e3ebf7efccc80346e3e50/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java#L1331-L1376), [master](https://github.com/apache/doris/blob/172120cbaa46b0a2eff6c37d86fb438c3a307375/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java#L1478-L1515)). `afterCommitted()` calls `updateJobStatisticAndOffset()`, which calls `offsetProvider.updateOffset(endOffset)` ([4.1.5](https://github.com/apache/doris/blob/917b20be83878e98f94e3ebf7efccc80346e3e50/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java#L1383-L1388), [master](https://github.com/apache/doris/blob/172120cbaa46b0a2eff6c37d86fb438c3a307375/fe/fe-core/src/main/java/org/apache/doris/job/extensions/ insert/streaming/StreamingInsertJob.java#L1522-L1533)). The attachment is persisted with the transaction's edit log and replayed by `replayOnCommitted()`. While a TVF job runs, this is the only place its offset moves. The other `updateOffset()` calls handle the `offset` property of CREATE/ALTER JOB and its replay, cloud-mode replay, and the CDC `commitOffset` RPC. 3. An Iceberg insert runs `IcebergInsertExecutor` in 4.1 and `PluginDrivenInsertExecutor` on master. Both commit in `BaseExternalTableInsertExecutor.onComplete()` by calling `transactionManager.commit(txnId)` on the catalog's transaction manager ([4.1.5](https://github.com/apache/doris/blob/917b20be83878e98f94e3ebf7efccc80346e3e50/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/BaseExternalTableInsertExecutor.java#L92-L131), [master](https://github.com/apache/doris/blob/172120cbaa46b0a2eff6c37d86fb438c3a307375/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/BaseExternalTableInsertExecutor.java#L100-L140)). Nothing on this path touches `GlobalTransactionMgr` or looks up the streaming task, so neither callback runs. 4. The insert ends with `QueryState.OK`, so `StreamingInsertTask.run()` returns and `onSuccess()` calls `StreamingInsertJob.onStreamTaskSuccess()`, which creates the next task right away ([4.1.5](https://github.com/apache/doris/blob/917b20be83878e98f94e3ebf7efccc80346e3e50/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java#L882-L910), [master](https://github.com/apache/doris/blob/172120cbaa46b0a2eff6c37d86fb438c3a307375/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java#L1013-L1042)). 5. The next task calls `S3SourceOffsetProvider.getNextOffset()`, which lists files after `currentOffset.endFile`. `currentOffset` is still null, so the listing starts from the beginning and returns the same first file ([4.1.5](https://github.com/apache/doris/blob/917b20be83878e98f94e3ebf7efccc80346e3e50/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java#L62-L67), [master](https://github.com/apache/doris/blob/172120cbaa46b0a2eff6c37d86fb438c3a307375/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java#L78-L82)). `hasMoreDataToConsume()` keeps returning true for the same reason, and `CurrentOffset` shows nothing because it is rendered from the same null field. Two more consequences of the missing callbacks: - `PAUSE JOB` cancels the running task, but the check that turns a cancel into a refused commit lives in `beforeCommitted()`. A task whose BE fragments have already finished therefore still commits its Iceberg snapshot, which is the last observation above. - `beforeCommitted()` reads `LoadStatistic` from the `InsertLoadJob` registered in `LoadManager`. Only `OlapInsertExecutor` registers one (`needRegister = jobId != -1`); the external-table executors do not. The job's statistics never move either, and routing an external commit through the existing callback unchanged would fail with `load job not found`. `CREATE JOB ... ON STREAMING` never checks the target table type. `CreateJobInfo.analyzeAndCreateStreamingInsertJob()` only checks that the statement is an `InsertIntoTableCommand` and runs `initPlan()` ([4.1.5](https://github.com/apache/doris/blob/917b20be83878e98f94e3ebf7efccc80346e3e50/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateJobInfo.java#L325-L349), [master](https://github.com/apache/doris/blob/172120cbaa46b0a2eff6c37d86fb438c3a307375/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateJobInfo.java#L328-L351)). By the same code path, Hive, Paimon, JDBC and MaxCompute targets in 4.1, and every connector target on master, should loop the same way. Only Iceberg has been observed. The design in #56191 says the offset "is committed only after data is visible and durable in Doris", so external targets were never part of it, and no regression case covers them. #### Possible fix 1. Short term, and for 4.1: reject targets other than internal OLAP tables in `CREATE JOB ... ON STREAMING`, and in `ALTER JOB` when it changes the SQL. Check it again when a task starts, so a job created before the fix pauses with a clear error instead of looping. 2. To support Iceberg as a target, the external commit path has to do what the two callbacks do: check the cancel fence before the remote commit, then advance and persist the offset. Persisting the offset after a separate remote commit only gives at-least-once, because an FE crash between the Iceberg commit and the edit log replays the batch. For exactly-once, the job id and end offset can be written into the Iceberg snapshot summary in the same commit, and recovery reads the last committed offset back from the table's snapshots. The Flink Iceberg sink does this with its checkpoint id. #### Workaround Do not use a streaming job with an Iceberg target on 4.1.x. Pause or drop the job. The duplicate snapshots can be removed with `ALTER TABLE <iceberg_catalog>.<db>.<tbl> EXECUTE rollback_to_snapshot("snapshot_id" = "<last good snapshot id>")`. That also drops any other write made after that snapshot. ### Are you willing to submit PR? - [ ] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
