fix(sink): require sink decoupling in the file sink impl itself

[yuhao-su]

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

What's changed and what's your intention?

The rule "a file sink must run decoupled" was expressed as a connector-name whitelist in the frontend catalog — SinkDesc::is_file_sink() listing fs / azblob / s3 / webhdfs — and checked in StreamSink::create. for_all_sinks! registers six FileSink<_> backends over the same OpenDAL writer, so gcs and snowflake were missing from that list.

The rule is load-bearing. OpenDalSinkWriter commits a file only once its batching strategy is met (10s rollover / 10240 rows) and truncates the log store only after a commit. On the non-decoupled in-memory log store, an untruncated checkpoint barrier parks BoundedInMemLogStoreReader in AwaitingTruncate (next_item() returns pending() forever), so the sink never gets another chance to commit and the checkpoint never completes. Verified locally against a FileSink<_> sink created with set sink_decouple = false: FLUSH hangs indefinitely, unrelated DDL on the same cluster stops making progress, and nothing is ever written to the target; with decoupling enabled the same sink flushes in 0.1s and lands its files after the rollover.

Rather than adding the two names to the list, this moves the rule to where the property lives: is_sink_decouple on impl<S: OpendalSinkBackend> Sink for FileSink<S>. One generic impl covers all six backends and any backend added later, so the list can't drift out of sync again. The whitelist and its frontend special case are deleted. This mirrors the turbopuffer sink, which has the same requirement and already handles it this way.

The frontend already dispatches SinkType::is_sink_decouple(...) through match_sink_name_str! a few lines above the special case, so no new machinery is needed.

Behavior change

The error is now a sink config error rather than ErrorCode::NotSupported with a HINT, so the remedy moves into the message text:

ERROR:  Failed to run the query

Caused by these errors (recent errors listed first):
  1: Sink error
  2: config error
  3: File sink can only be created with sink_decouple enabled. Please run `set sink_decouple = true` first.

Verification

• New unit test test_requires_sink_decouple in opendal_sink.rs (Default/Enable accepted, Disable rejected) over fs, gcs, s3 and snowflake.
• e2e_test/sink/append_only_sink.slt extended from the s3 case to s3 + gcs + snowflake; passes locally. The check happens at planning time, before the connector's validate(), so no cloud credentials are involved.
• Checked that a decoupled fs sink is unaffected: FLUSH returns immediately and a non-empty parquet file lands after the rollover.
• ./risedev c clean.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.

Documentation

• My PR needs documentation updates.

gcs and snowflake file sinks can no longer be created with sink_decouple = false (they previously appeared to succeed and then stalled the streaming job). The error message for all file sinks changes as shown above.

The check runs at plan time only, and the resulting log store type is persisted in the sink's plan, so this does not affect sinks that already exist: they keep the log store they were created with and are not re-validated on upgrade or recovery. A gcs or snowflake sink created earlier with sink_decouple = false therefore keeps its in-memory log store and its stalling behavior — it has to be dropped and recreated with decoupling enabled.

🤖 Generated with Claude Code