JNSimba opened a new pull request, #68876:
URL: https://github.com/apache/doris/pull/68876
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
S3 streaming INSERT jobs currently discover new files by ordered listing. A
file uploaded with a key smaller than the committed cursor is therefore not
discovered. Task planning also lists the batch again even though the source
offset provider already selected its exact files.
This change adds AWS S3 `ObjectCreated` notification ingestion through a
dedicated SQS queue. Notifications are filtered by the job's bucket and glob,
selected messages stay together in a batch, and their receipts are deleted
after the INSERT transaction commits. The latest committed batch is restored
from the existing transaction attachment to acknowledge matching notifications
after FE recovery.
Both ordered listing and notification ingestion pass exact file paths and
sizes directly to the S3 TVF. Normal TVF schema inference failures are
preserved. Audit SQL keeps the existing brace-list URI display and credential
masking; QueryId connects it to the job/task label in FE logs. Notification
jobs also expose approximate visible message backlog and the latest committed
source event timestamp.
Scope and operational requirements:
- Modes are `ORDERED_LIST` (default), `NOTIFICATION`, and `ONE_TIME`. The
former explicit `LEXICAL` and `ONCE` values are replaced without aliases.
- Notification ingestion currently supports AWS S3 and direct S3 event
messages in a dedicated SQS queue. Objects must remain unchanged until
ingestion completes. This does not provide durable deduplication of every
historical notification or version-aware overwrite ingestion.
- SQS access introduces outbound FE requests using a user-supplied queue
URL, an SSRF surface reachable by users with `LOAD` privilege on the target
table through streaming `CREATE JOB`. IAM permissions for the queue and, when
used, `sts:AssumeRole` are also required.
#### User guide
Configure the bucket to send `ObjectCreated` notifications to an SQS queue
dedicated to one streaming job. The notification prefix should cover the S3 TVF
URI. Credentials need permission to read S3 objects and receive, change
visibility, delete messages, and read attributes on the SQS queue.
Create a notification job with AKSK credentials:
```sql
CREATE JOB load_s3_events
PROPERTIES (
"s3.ingestion_mode" = "NOTIFICATION",
"s3.event.source" = "SQS",
"s3.sqs.queue_url" =
"https://sqs.<region>.amazonaws.com/<account>/<queue>",
"s3.max_batch_files" = "256",
"s3.max_batch_bytes" = "10737418240",
"max_interval" = "10"
)
ON STREAMING DO INSERT INTO target_table
SELECT * FROM S3(
"uri" = "s3://bucket/incoming/*.csv",
"format" = "csv",
"provider" = "S3",
"s3.endpoint" = "s3.<region>.amazonaws.com",
"s3.region" = "<region>",
"s3.access_key" = "<access-key>",
"s3.secret_key" = "<secret-key>"
);
```
For IAM Role authentication, replace the access and secret keys with:
```sql
"s3.role_arn" = "arn:aws:iam::<account>:role/<role>"
-- Optional: "s3.external_id" = "<external-id>"
```
The FE uses its configured AWS credentials to assume the role.
`s3.max_batch_files` defaults to 256. `s3.max_batch_bytes` defaults to 10 GB
and accepts 100 MB through 10 GB. These are soft batch thresholds: all messages
already received stay in the same batch, and reaching a threshold stops the
next receive. `max_interval` defaults to 10 seconds and must be less than 7200
seconds in `NOTIFICATION` mode.
The TVF URI filters notification records by bucket and glob. Records outside
the URI, unsupported records, malformed notifications, and S3 test events are
skipped; messages without accepted records are acknowledged without an INSERT.
Accepted messages are acknowledged only after the INSERT commits, with
visibility renewed while the task runs. Recovery uses the latest committed
batch's persisted keys to acknowledge matching redeliveries without importing
them again.
`NOTIFICATION` consumes queued notifications and does not list existing
objects for backfill. Use `ONE_TIME` to import existing objects once, or
`ORDERED_LIST` for incremental scanning in key order.
Inspect progress with:
```sql
SELECT Name, Status, CurrentOffset, EndOffset, LagBytes,
LastSourceEventTimestamp, LastTaskSuccessTime
FROM jobs("type" = "insert")
WHERE Name = "load_s3_events";
```
- `CurrentOffset`: the last file key in the committed batch, such as
`{"fileName":"incoming/a.csv"}`.
- `EndOffset`: the latest observed batch's last file key and `lagMessages`,
such as `{"fileName":"incoming/b.csv","lagMessages":12}`. The count is SQS's
approximate visible-message count, excluding in-flight and delayed messages.
- `LagBytes`: `-1` because SQS does not expose queued object bytes.
- `LastSourceEventTimestamp`: the latest committed batch's maximum S3 event
timestamp, in Unix seconds.
Per-job FE metrics also expose:
```text
doris_fe_streaming_job_per_job_lag_messages
doris_fe_streaming_job_per_job_last_source_event_timestamp_seconds
```
### Release note
Add SQS notification-based S3 streaming ingestion, reuse exact selected
files during task execution, and expose notification backlog metrics. Rename S3
ingestion modes to `ORDERED_LIST` and `ONE_TIME`; add `NOTIFICATION`.
### Check List (For Author)
- Test
- [x] Regression test
- Passed: `test_streaming_insert_job` (default ordered listing).
- Passed: `test_streaming_insert_job_s3_event_aksk` (out-of-order
keys, pause/resume, glob exclusions, literal comma keys, metrics, and empty
messages).
- Passed: `test_streaming_insert_job_s3_event_iam_role` (AssumeRole
and subsequent notification batches).
- Added but not run: `test_streaming_insert_job_s3_event_restart_fe`.
- [x] Unit Test
- 104 targeted FE unit tests passed across the focused run and the
corrected audit-test rerun: event offsets/provider, ordered provider,
exact-file TVF, task audit, job properties/lifecycle/persistence/lag, and
metrics.
- [x] Manual test
- FE build and Checkstyle passed. Reviewed successful task audit/FE
logs: exact displayed filenames, masked credentials, QueryId/task label
linkage, and COMMITTED/VISIBLE transactions.
- Existing BE binary was reused for the above regressions; Web UI
build was skipped because of the development machine's Node version.
- Behavior changed:
- [x] Yes. Add notification ingestion and replace explicit ingestion
mode names; default ordered behavior remains.
- Does this need documentation?
- [x] Yes. The new mode, SQS setup, IAM permissions, and object
immutability requirement need documentation; documentation PR not included here.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
--
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]
