[Improve](streaming-job) support specifying offset for StreamingInsertJob create and alter

[JNSimba]

Summary

• Support specifying offset (binlog position, LSN, named modes) when creating or altering a StreamingInsertJob via FROM MYSQL/POSTGRES path.
• FE: extend DataSourceConfigValidator to validate offset formats (initial/snapshot/latest/earliest/JSON); earliest is MySQL-only, rejected for PostgreSQL. Implement JdbcSourceOffsetProvider.deserializeOffsetProperty() for named modes and JSON offset parsing. Remove S3-only restriction in StreamingInsertJob.initInsertJob() and modifyPropertiesInternal() so CDC jobs can also use offset property. On ALTER, sync offset to sourceProperties for the FROM...TO path.
• BE: support JSON LSN offset {"lsn":"N"} in PostgresSourceReader.generatePostgresConfig(), and handle SPECIFIC_OFFSETS mode in JdbcIncrementalSourceReader.getStartOffsetFromConfig().
• Fix JdbcOffset.isValidOffset() and toSerializedJson() to return meaningful values instead of hardcoded false/null.

Test plan

• Regression test: test_streaming_mysql_job_special_offset — earliest/latest/JSON binlog offset with data sync verification, ALTER JOB offset change, invalid format rejection
• Regression test: test_streaming_postgres_job_special_offset — initial/latest with data sync, ALTER JOB with JSON LSN offset and data sync verification, earliest rejection for PG, invalid format rejection
• Regression test: test_streaming_mysql_job_special_offset_restart_fe — create job with JSON binlog offset, verify data sync, restart FE, verify job recovery and continued sync

🤖 Generated with Claude Code

[Thearas]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

[JNSimba]

run buildall

[JNSimba]

/review

[Copilot]

Pull request overview

This PR extends StreamingInsertJob CDC offset support so users can specify named offsets (initial/snapshot/latest/earliest) and JSON “specific offsets” when creating jobs via FROM MYSQL/POSTGRES, and JSON offsets when altering jobs (with validation and BE parsing support).

Changes:

• FE: extend source config validation to accept additional offset modes and JSON offsets; wire validateSource to be data-source-type aware; allow CDC jobs to use offset property and restrict ALTER to JSON offsets for CDC.
• FE: implement/adjust JDBC offset property deserialization and improve JdbcOffset behaviors (isValidOffset, toSerializedJson).
• BE: add PostgreSQL JSON LSN offset support and handle SPECIFIC_OFFSETS startup mode.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql_alter_offset.groovy	Adds TVF-path regression coverage for ALTER with JSON MySQL binlog offset.
regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_special_offset.groovy	Adds PG regression coverage for create offsets + ALTER with JSON LSN, plus invalid cases.
regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset_restart_fe.groovy	Adds FE-restart regression coverage for MySQL JSON binlog offset persistence.
regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy	Adds MySQL regression coverage for earliest/latest/JSON offsets and ALTER behavior.
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java	Adds JSON LSN startup offset handling for PostgreSQL.
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java	Handles SPECIFIC_OFFSETS startup mode by constructing offset from config.
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateJobCommand.java	Passes source type into source config validation for offset-type-specific rules.
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java	Passes data source type into source config validation during ALTER.
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java	Implements offset property deserialization for named modes and JSON.
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcOffset.java	Adjusts offset validity and serialization behavior.
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java	Enables offset property validation beyond S3 and restricts CDC ALTER to JSON offsets.
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java	Adds data-source-type-aware offset validation and JSON detection helper.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[github-actions]

Found 1 blocking issue.

1. High - now serializes JDBC offsets in a different shape from what still reads. For binlog offsets the new JSON nests fields under , but the deserialize/replay path still expects a flattened list of string maps. persists this serialized value immediately after , so cloud replay / offset restore can no longer round-trip the altered CDC offset correctly.

Critical checkpoints:

• Goal of task: Support special offsets on create/alter. Partially achieved; create/alter paths were extended and regression tests were added, but the new serialized offset format breaks persisted/replayed CDC offsets in cloud mode.
• Modification size/focus: Focused overall.
• Concurrency: No new concurrency concerns identified in the touched code paths.
• Lifecycle/static init: No special lifecycle or static-init issues identified.
• Configuration: No new configs added.
• Compatibility/storage format: Blocking issue. The PR changes serialized JDBC offset shape without updating the corresponding deserialize/replay path.
• Parallel code paths: Checked both and job paths; the persistence problem sits in shared JDBC offset serialization.
• Special conditional checks: The new ALTER restriction to JSON offsets is understandable and commented.
• Test coverage: New regression tests cover MySQL/Postgres create/alter and one FE-restart scenario, but they do not cover cloud-mode reset/replay of an altered JSON offset.
• Observability: Existing logs seem adequate for this path.
• Transaction/persistence: Problematic because offset reset/replay persists a format the current reader does not consume.
• Data write/atomicity: No direct new write-atomicity issue beyond the persisted offset corruption risk.
• FE-BE variable passing: No new FE-BE field propagation issues identified.
• Performance: No material performance regressions identified in the reviewed diff.
• Other issues: None beyond the blocking serialization mismatch.

[JNSimba]

/review

[JNSimba]

run buildall

[github-actions]

Found 2 blocking correctness issues.

1. FE accepts any JSON offset on ALTER for JDBC jobs. JdbcSourceOffsetProvider.validateAlterOffset() and deserializeOffsetProperty() only gate on isJsonOffset(), so a PostgreSQL job can accept MySQL-style JSON like {"file":"binlog.000001","pos":"154"} (and vice versa). ALTER succeeds, but the resumed job then fails in the broker because PostgresSourceReader.generatePostgresConfig() requires lsn.
2. ALTER on the FROM MYSQL/POSTGRES ... TO path does not persist the new startup offset into sourceProperties. modifyPropertiesInternal() updates only the in-memory provider, while replay/reconstruction uses sourceProperties (for example in gsonPostProcess() and when creating the next multi-table task). If FE restarts before the first post-ALTER commit, the altered offset is lost and the job resumes from the old mode/offset.

Critical checkpoint conclusions:

• Goal of current task: Partially accomplished. The PR adds special-offset support, but the ALTER path is still incorrect in the two scenarios above, so the feature is not correct end-to-end.
• Modification size/clarity: Mostly focused, but the ALTER validation/persistence logic is split in a way that misses source-specific constraints and durable state updates.
• Concurrency: No new lock-order or thread-safety issue found in the touched code.
• Lifecycle/static initialization: No special lifecycle or static-init issue found.
• Configuration items: No new config items added.
• Compatibility: No incompatible wire/storage format change identified.
• Parallel code paths: The CREATE path and downstream readers enforce source-specific semantics, but the ALTER path does not; that inconsistency is the root cause of one blocker.
• Special conditional checks: The new JSON-only check is too broad and should be source-type-aware.
• Test coverage: New regression tests cover many happy paths, but they do not cover invalid cross-source JSON on ALTER or FE restart before the first post-ALTER commit on the FROM ... TO path.
• Observability: Existing logging looks sufficient to diagnose these failures.
• Transaction/persistence: There is a persistence bug for altered offsets on the FROM ... TO path because replay reconstructs from stale sourceProperties.
• Data writes/modifications: The restart case can resume consumption from the wrong offset, so data correctness is at risk.
• FE/BE variable passing: No new propagation issue beyond the stale sourceProperties state above.
• Performance: No material performance regression found in the touched code.
• Other issues: None beyond the two blocking findings above.

[JNSimba]

run buildall

[hello-stephen]

FE Regression Coverage Report

Increment line coverage 12.50% (7/56) 🎉
Increment coverage report
Complete coverage report

[JNSimba]

/review

[github-actions]

Findings:

1. fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java: FE now accepts any JSON object as a JDBC offset, but the MySQL/PostgreSQL readers only support source-specific shapes (file+pos for MySQL, lsn for PostgreSQL). That means malformed JSON offsets can pass DDL-time validation and only fail later at runtime.
2. fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java: ALTER JOB ... FROM MYSQL/POSTGRES (... 'offset'=...) is validated as if it were supported, but the altered source offset is never applied to offsetProvider.currentOffset, so the resumed job continues from the old position.

Critical checkpoint conclusions:

• Goal of the task: Partially accomplished. The PR adds offset support for CDC streaming jobs, but the two issues above leave important alter/validation paths incorrect. Existing regression tests cover happy paths, but they do not prove malformed JSON shape rejection or the ALTER JOB ... FROM ... offset-reset behavior.
• Small / clear / focused: Mostly yes. The change set is focused on CDC offset handling, FE validation, and regression coverage.
• Concurrency: No new blocking concurrency issue found in the reviewed FE/CDC paths. The changes stay on existing job/task flows and do not introduce new lock ordering problems in the touched code.
• Lifecycle / static initialization: No new special lifecycle or static initialization issue found.
• Configuration items: No new config item added.
• Compatibility / incompatible change: No storage-format or protocol compatibility break was identified, but FE/BE offset-shape validation is now inconsistent.
• Parallel code paths: Not fully handled. The TVF PROPERTIES('offset'=...) path and the FROM MYSQL/POSTGRES alter path are not updated consistently.
• Special conditional checks: Not sufficient. JSON-offset checks currently validate only “is object JSON”, not whether the shape is valid for the selected source.
• Test coverage: Incomplete for the new behavior. Missing negative coverage for source-specific malformed JSON offsets and for ALTER JOB ... FROM MYSQL/POSTGRES (... 'offset'=...) actually taking effect.
• Modified test results: The added .out files look consistent with the checked-in suites, but they do not cover the blocking cases above.
• Observability: Existing logging looks adequate for these paths.
• Transaction / persistence: No new edit-log format issue found, but the FROM ... alter path fails to update the provider state that actually drives subsequent consumption.
• Data writes / modifications: Blocking issue present. A user can request an offset reset through the FROM MYSQL/POSTGRES alter syntax and the job can continue consuming from the previous offset instead.
• FE/BE variable passing: The logical contract for offset format is incomplete; unsupported JSON shapes are allowed through FE and then rejected later by the CDC reader implementation.
• Performance: No new blocking performance issue found.
• Other issues: None beyond the two findings above.

Requesting changes because both findings can produce incorrect user-visible behavior for the new offset feature.

[JNSimba]

run buildall

[hello-stephen]

FE Regression Coverage Report

Increment line coverage 24.19% (15/62) 🎉
Increment coverage report
Complete coverage report

[JNSimba]

run buildall

[hello-stephen]

FE UT Coverage Report

Increment line coverage 0.00% (0/62) 🎉
Increment coverage report
Complete coverage report

[hello-stephen]

FE Regression Coverage Report

Increment line coverage 14.52% (9/62) 🎉
Increment coverage report
Complete coverage report

[JNSimba]

/review

[github-actions]

Found 3 blocking correctness issues.

1. High - StreamingInsertJob.replayOnUpdated() reuses modifyPropertiesInternal() during edit-log and image replay, and the new CDC path now drives resetCloudProgress() from that helper. In cloud mode reset_streaming_job_offset overwrites the commit-attach offset stored in meta-service. Replaying an old ALTER entry can therefore rewind the cloud progress back to the ALTER offset, so the next replayOnCloudMode() may resume from stale progress and re-consume data.
2. High - JdbcSourceOffsetProvider.validateAlterOffset() and deserializeOffsetProperty() still treat any JSON object as a valid JDBC offset. For example, a MySQL job can now be altered to PROPERTIES("offset"="{"lsn":"123"}"). FE accepts it, but MySqlSourceReader.createBinlogSplit() cannot build a specific binlog position from that meta and falls back to the old startup offset or earliest, so the requested reset is silently ignored.
3. High - the new offset validation only covers PROPERTIES("offset"=...). ALTER JOB <name> DO INSERT ... SELECT ... FROM cdc_stream(...) can still change the TVF SQL offset, because checkUnmodifiableProperties() does not forbid it and alterJob() only refreshes originTvfProps. The provider state is never validated or reset from that new SQL, so the resumed job keeps consuming from the old offset while the SQL now shows a different one.

Critical checkpoint conclusions:

• Goal of the current task: Partially accomplished. The PR adds special-offset support, but the three paths above still leave ALTER and replay behavior incorrect end to end.
• Modification size and focus: Focused overall, but the ALTER logic is now split across multiple surfaces and they are not kept consistent.
• Concurrency: No new lock-order or thread-safety bug found in the touched FE and CDC code paths.
• Lifecycle and static initialization: No special lifecycle or static-init issue found.
• Configuration items: No new config item added.
• Compatibility and persistence: Blocking issue present in cloud mode because replay now performs a destructive progress-reset RPC from an edit-log replay path.
• Parallel code paths: Not fully handled. PROPERTIES("offset"=...) got new validation, but the SQL rewrite path for cdc_stream still bypasses it.
• Special conditional checks: The new JSON-only check is too broad. It validates only that the value is a JSON object, not that the shape matches the selected source.
• Test coverage: New regression tests cover happy paths, but they do not cover cloud replay after ALTER, malformed or cross-source JSON on ALTER, or SQL-based ALTER of cdc_stream offsets.
• Modified test results: The added .out files are consistent with the checked-in suites, but they do not exercise the blocking cases above.
• Observability: Existing logs are sufficient to diagnose these failures.
• Transaction and persistence: Blocking issue present in cloud replay, because progress in meta-service can be rewound during journal replay.
• Data writes and modifications: Blocking issue present. Both the replay regression and the malformed-offset acceptance can make the job resume from the wrong position.
• FE and BE variable passing: No new field-propagation issue found, but the FE and BE offset contract is still incomplete because FE accepts unsupported JSON shapes.
• Performance: No material performance regression found in the reviewed diff.
• Other issues: None beyond the three blockers above.

Requesting changes because these issues can rewind or ignore the requested CDC offset.

[github-actions]

replayOnUpdated() calls modifyPropertiesInternal() while the update-job edit log is being replayed. With this change, any CDC job that has properties["offset"] now reaches resetCloudProgress() here as well. In cloud mode that RPC is not read-only: reset_streaming_job_offset overwrites the commit-attach offset stored in meta-service. Replaying an old ALTER entry on follower startup can therefore rewind the job progress back to the ALTER offset, and the next replayOnCloudMode() will resume from stale progress. This helper needs a replay-safe path that skips the RPC, or replay should restore only persisted provider state.

[github-actions]

isJsonOffset() is too weak for JDBC ALTER validation. deserializeOffsetProperty() plus JdbcOffset.isValidOffset() treat any non-empty JSON object as valid, so a MySQL job can be altered to something like {"lsn":"123"} or {"foo":"bar"}. On resume, MySqlSourceReader.createBinlogSplit() cannot build a specific binlog offset from that meta and falls back to the old startup offset or earliest, so the requested reset is silently ignored. This validator needs to enforce source-specific keys such as file plus pos for MySQL and lsn for PostgreSQL before FE accepts the ALTER.

[github-actions]

This only validates the PROPERTIES("offset"=...) path. ALTER JOB <name> DO INSERT ... SELECT ... FROM cdc_stream(...) still goes through the SQL-rewrite branch: checkUnmodifiableProperties() allows changing offset, alterJob() refreshes originTvfProps, but it never calls validateAlterOffset(), validateOffset(), or offsetProvider.updateOffset(). After resume, JdbcTvfSourceOffsetProvider.getNextOffset() keeps using the old provider state, so the SQL-level offset change is silently ignored and it also bypasses the new JSON-only rule. Please either reject SQL-side offset changes for cdc_stream or route them through the same validated reset path.