feat(frontend): allow users to confirm if background_ddl is enabled for a sink

e2e_test/background_ddl/background.slt

@@ -0,0 +1,138 @@
+# Test whether sinks, materialized views, and indexes run in the background
+
+# Test index part
+statement ok
+create table t(v1 int);
+
+statement ok
+insert into t select * from generate_series(1, 500000);
+
+statement ok
+set background_ddl=false;
+
+statement ok
+CREATE INDEX index1 ON t (v1);
+
+query I
+select background_ddl from rw_catalog.rw_indexes where name='index1';
+----
+f
+
+statement ok
+set background_ddl=true;
+
+statement ok
+CREATE INDEX index2 ON t (v1);
+
+query I
+select background_ddl from rw_catalog.rw_indexes where name='index2';
+----
+t
+
+statement ok
+drop index index1;
+
+statement ok
+drop index index2;
+
+statement ok
+drop table t;
+
+statement ok
+set background_ddl=false;
+
+# Test sink part
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+SET sink_decouple TO true;
+
+statement ok
+create table t(v1 int);
+
+statement ok
+set backfill_rate_limit=5000;
+
+statement ok
+insert into t select * from generate_series(1, 10000);
+
+statement ok
+CREATE SINK sink1 AS select * from t WITH (
+    connector = 'blackhole'
+);
+
+query I
+select background_ddl from rw_catalog.rw_sinks where name='sink1';
+----
+f
+
+statement ok
+set background_ddl=true;
+
+statement ok
+CREATE SINK sink2 AS select * from t WITH (
+    connector = 'blackhole'
+);
+
+query I
+select background_ddl from rw_catalog.rw_sinks where name='sink2';
+----
+t
+
+statement ok
+drop sink sink1;
+
+statement ok
+drop sink sink2;
+
+statement ok
+drop table t;
+
+statement ok
+set background_ddl=false;
+
+# Test materialized view part
+statement ok
+create table t(v1 int);
+
+statement ok
+insert into t select * from generate_series(1, 10000);
+
+statement ok
+set backfill_rate_limit=5000;
+
+statement ok
+create materialized view m1 as select * from t;
+
+query I
+select background_ddl from rw_catalog.rw_materialized_views where name='m1';
+----
+f
+
+statement ok
+set background_ddl=true;
+
+statement ok
+create materialized view m2 as select * from t;
+
+query I
+select background_ddl from rw_catalog.rw_materialized_views where name='m2';
+----
+t
+
+statement ok
+drop materialized view m1;
+
+statement ok
+drop materialized view m2;
+
+statement ok
+drop table t;
+
+statement ok
+set background_ddl=false;
+
+statement ok
+set backfill_rate_limit=default;
+

src/frontend/src/catalog/system_catalog/rw_catalog/rw_sinks.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_common::catalog::CreateType;
 use risingwave_common::id::{ConnectionId, SchemaId, SinkId, UserId};
 use risingwave_common::types::{Fields, JsonbVal, Timestamptz};
 use risingwave_connector::WithOptionsSecResolved;
@@ -38,6 +39,7 @@ struct RwSink {
     created_at: Option<Timestamptz>,
     initialized_at_cluster_version: Option<String>,
     created_at_cluster_version: Option<String>,
+    background_ddl: bool,
 
     // connector properties in json format
     connector_props: JsonbVal,
@@ -101,6 +103,7 @@ fn read_rw_sinks_info(reader: &SysCatalogReaderImpl) -> Result<Vec<RwSink>> {
                     created_at_cluster_version: sink.created_at_cluster_version.clone(),
                     connector_props,
                     format_encode_options,
+                    background_ddl: sink.create_type == CreateType::Background,
                 }
             })
         })