fix: disable sink decouple for trivial sink (#18302) (#18307)

Co-authored-by: Zhanxiang (Patrick) Huang <[email protected]>

src/connector/src/sink/trivial.rs

@@ -15,7 +15,9 @@
 use std::marker::PhantomData;
 
 use async_trait::async_trait;
+use risingwave_common::session_config::sink_decouple::SinkDecouple;
 
+use super::catalog::desc::SinkDesc;
 use crate::sink::log_store::{LogStoreReadItem, TruncateOffset};
 use crate::sink::{
     DummySinkCommitCoordinator, LogSinker, Result, Sink, SinkError, SinkLogReader, SinkParam,
@@ -64,6 +66,11 @@ impl<T: TrivialSinkName> Sink for TrivialSink<T> {
 
     const SINK_NAME: &'static str = T::SINK_NAME;
 
+    // Disable sink decoupling for all trivial sinks because it introduces overhead without any benefit
+    fn is_sink_decouple(_desc: &SinkDesc, _user_specified: &SinkDecouple) -> Result<bool> {
+        Ok(false)
+    }
+
     async fn new_log_sinker(&self, _writer_env: SinkWriterParam) -> Result<Self::LogSinker> {
         Ok(Self(PhantomData))
     }