feat(sink): support dynamic URL for HTTP sink

[wenym1]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

This PR extends the HTTP sink so its target URL can be supplied by the sink row itself.

• Keeps the existing single-column HTTP sink behavior unchanged: one varchar or jsonb payload column with WITH (url = ...).
• Adds a multi-column shape for dynamic targets:
  • required payload column of type varchar or jsonb
  • optional url column of type varchar
• Requires exactly one target URL source for multi-column sinks: either the static url option or the dynamic url column.
• Stores static URLs as parsed reqwest::Url, and parses dynamic row URLs before sending each request.
• Adds SQLLogicTest coverage for validation paths, dynamic URL sinks, and CREATE SINK AS SELECT.

Example SQL:

CREATE TABLE http_events (
    payload JSONB,
    url VARCHAR
);

CREATE SINK dynamic_http_sink FROM http_events WITH (
    connector = 'http',
    type = 'append-only',
    force_append_only = 'true'
);

INSERT INTO http_events VALUES
    ('{"event":"upsert","id":1}'::jsonb, 'https://example.com/workspace-a/events'),
    ('{"event":"upsert","id":2}'::jsonb, 'https://example.com/workspace-b/events');

For CREATE SINK AS SELECT, project the payload and URL columns with the required output names:

CREATE SINK dynamic_http_sink AS
SELECT
    event_body AS payload,
    format('https://example.com/workspaces/%s/events', workspace_id) AS url
FROM source_events
WITH (
    connector = 'http',
    type = 'append-only',
    force_append_only = 'true'
);

• Closes #N/A

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

HTTP sinks can now derive the target URL from a row column by outputting payload and url columns, enabling dynamic per-row HTTP destinations while preserving the existing static URL behavior.

[wenym1]

• feat(sink): support dynamic URL for HTTP sink #25699 👈 (View in Graphite)
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[yuhao-su]

How to tell the difference between use url as a dynamic url or a field when sinking

[wenym1]

How to tell the difference between use url as a dynamic url or a field when sinking

The http payload is in a single column, either jsonb or text. When the sink has only a single column, this column is the payload column. When the sink has multiple column, we check the name of each column, to be either payload and url.

[hzxa21]

Left a comment about log DDoS. You can merge the PR first and create follow-up issue on that you would like to provide a preview image for trial usage first.

Rest LGTM.

[hzxa21]

How about implementing some kind of a log suppressor to avoid log DDoS when there are many invalid URLs in the payload?

[wenym1]

Will integrate it to sink dead letter queue when it's implemented.

[github-actions]

✅ Cherry-pick PRs (or issues if encountered conflicts) have been created successfully to all target branches.