proto/catalog.proto

@@ -524,6 +524,7 @@ message Database {
   uint32 id = 1;
   string name = 2;
   uint32 owner = 3;
+  string resource_group = 4;
 }
 
 message Comment {

proto/common.proto

@@ -56,11 +56,14 @@ message WorkerNode {
     bool is_unschedulable = 3;
     // This is used for frontend node to register its rpc address
     string internal_rpc_host_addr = 4;
-    // Meta may assign labels to worker nodes to partition workload by label.
-    // This is used for serverless backfilling of materialized views.
-    optional string node_label = 5;
+
+    reserved 5;
+    reserved "node_label";
 
     uint32 parallelism = 6;
+
+    // resource group for scheduling
+    optional string resource_group = 7;
   }
   message Resource {
     string rw_version = 1;

proto/ddl_service.proto

@@ -144,6 +144,9 @@ message CreateMaterializedViewRequest {
 
   // The list of object IDs that this materialized view depends on.
   repeated uint32 dependencies = 4;
+
+  // The specific resource group to use for the materialized view. If not set, the database resource group is used.
+  optional string specific_resource_group = 5;
 }
 
 message CreateMaterializedViewResponse {
@@ -271,6 +274,14 @@ message AlterParallelismRequest {
 
 message AlterParallelismResponse {}
 
+message AlterResourceGroupRequest {
+  uint32 table_id = 1;
+  optional string resource_group = 2;
+  bool deferred = 3;
+}
+
+message AlterResourceGroupResponse {}
+
 message AlterOwnerResponse {
   common.Status status = 1;
   WaitVersion version = 2;
@@ -550,6 +561,7 @@ service DdlService {
   rpc AlterOwner(AlterOwnerRequest) returns (AlterOwnerResponse);
   rpc AlterSetSchema(AlterSetSchemaRequest) returns (AlterSetSchemaResponse);
   rpc AlterParallelism(AlterParallelismRequest) returns (AlterParallelismResponse);
+  rpc AlterResourceGroup(AlterResourceGroupRequest) returns (AlterResourceGroupResponse);
   rpc DropTable(DropTableRequest) returns (DropTableResponse);
   rpc RisectlListStateTables(RisectlListStateTablesRequest) returns (RisectlListStateTablesResponse);
   rpc CreateView(CreateViewRequest) returns (CreateViewResponse);

src/common/src/util/worker_util.rs

@@ -14,4 +14,4 @@
 
 pub type WorkerNodeId = u32;
 
-pub const DEFAULT_COMPUTE_NODE_LABEL: &str = "default";
+pub const DEFAULT_RESOURCE_GROUP: &str = "default";

src/compute/src/lib.rs

@@ -37,7 +37,7 @@ use risingwave_common::util::meta_addr::MetaAddressStrategy;
 use risingwave_common::util::resource_util::cpu::total_cpu_available;
 use risingwave_common::util::resource_util::memory::system_memory_available_bytes;
 use risingwave_common::util::tokio_util::sync::CancellationToken;
-use risingwave_common::util::worker_util::DEFAULT_COMPUTE_NODE_LABEL;
+use risingwave_common::util::worker_util::DEFAULT_RESOURCE_GROUP;
 use serde::{Deserialize, Serialize};
 
 /// If `total_memory_bytes` is not specified, the default memory limit will be set to
@@ -113,9 +113,9 @@ pub struct ComputeNodeOpts {
     #[override_opts(if_absent, path = streaming.actor_runtime_worker_threads_num)]
     pub parallelism: usize,
 
-    /// The parallelism that the compute node will register to the scheduler of the meta service.
-    #[clap(long, env = "RW_NODE_LABEL", default_value_t = default_node_label())]
-    pub node_label: String,
+    /// Resource group for scheduling, default value is "default"
+    #[clap(long, env = "RW_RESOURCE_GROUP", default_value_t = default_resource_group())]
+    pub resource_group: String,
 
     /// Decides whether the compute node can be used for streaming and serving.
     #[clap(long, env = "RW_COMPUTE_NODE_ROLE", value_enum, default_value_t = default_role())]
@@ -262,8 +262,8 @@ pub fn default_parallelism() -> usize {
     total_cpu_available().ceil() as usize
 }
 
-pub fn default_node_label() -> String {
-    DEFAULT_COMPUTE_NODE_LABEL.to_owned()
+pub fn default_resource_group() -> String {
+    DEFAULT_RESOURCE_GROUP.to_owned()
 }
 
 pub fn default_role() -> Role {

src/compute/src/server.rs

@@ -129,7 +129,7 @@ pub async fn compute_node_serve(
             is_serving: opts.role.for_serving(),
             is_unschedulable: false,
             internal_rpc_host_addr: "".to_owned(),
-            node_label: Some(opts.node_label.clone()),
+            resource_group: Some(opts.resource_group.clone()),
         },
         &config.meta,
     )

src/frontend/planner_test/tests/testdata/output/agg.yaml

@@ -312,8 +312,7 @@
   batch_plan: |-
     BatchExchange { order: [], dist: Single }
     └─BatchSortAgg { group_key: [mv.v1], aggs: [max(mv.v2)] }
-      └─BatchExchange { order: [mv.v1 DESC], dist: HashShard(mv.v1) }
-        └─BatchScan { table: mv, columns: [mv.v1, mv.v2], distribution: SomeShard }
+      └─BatchScan { table: mv, columns: [mv.v1, mv.v2], distribution: UpstreamHashShard(mv.v1) }
 - sql: |
     create table t(v1 int, v2 int);
     select v1, max(v2) from t group by v1 order by v1 desc;
@@ -367,8 +366,7 @@
   batch_plan: |-
     BatchExchange { order: [], dist: Single }
     └─BatchHashAgg { group_key: [mv.v1], aggs: [max(mv.v2)] }
-      └─BatchExchange { order: [], dist: HashShard(mv.v1) }
-        └─BatchScan { table: mv, columns: [mv.v1, mv.v2], distribution: SomeShard }
+      └─BatchScan { table: mv, columns: [mv.v1, mv.v2], distribution: UpstreamHashShard(mv.v1) }
   with_config_map:
     RW_BATCH_ENABLE_SORT_AGG: 'false'
 - name: Not use BatchSortAgg, when output requires order

src/frontend/planner_test/tests/testdata/output/append_only.yaml

@@ -25,9 +25,10 @@
     select v1 from t1 order by v1 limit 3 offset 3;
   stream_plan: |-
     StreamMaterialize { columns: [v1, t1._row_id(hidden)], stream_key: [t1._row_id], pk_columns: [v1, t1._row_id], pk_conflict: NoCheck }
-    └─StreamTopN [append_only] { order: [t1.v1 ASC], limit: 3, offset: 3 }
-      └─StreamExchange { dist: Single }
-        └─StreamTableScan { table: t1, columns: [t1.v1, t1._row_id], stream_scan_type: ArrangementBackfill, stream_key: [t1._row_id], pk: [_row_id], dist: UpstreamHashShard(t1._row_id) }
+    └─StreamExchange { dist: HashShard(t1.v1) }
+      └─StreamTopN [append_only] { order: [t1.v1 ASC], limit: 3, offset: 3 }
+        └─StreamExchange { dist: Single }
+          └─StreamTableScan { table: t1, columns: [t1.v1, t1._row_id], stream_scan_type: ArrangementBackfill, stream_key: [t1._row_id], pk: [_row_id], dist: UpstreamHashShard(t1._row_id) }
 - sql: |
     create table t1 (v1 int, v2 int) append only;
     select max(v1) as max_v1 from t1;

src/frontend/planner_test/tests/testdata/output/batch_dist_agg.yaml

@@ -11,9 +11,10 @@
   sql: |
     select max(v) as a1 from S;
   batch_plan: |-
-    BatchExchange { order: [], dist: Single }
-    └─BatchSimpleAgg { aggs: [max(s.v)] }
-      └─BatchScan { table: s, columns: [s.v], distribution: Single }
+    BatchSimpleAgg { aggs: [max(max(s.v))] }
+    └─BatchExchange { order: [], dist: Single }
+      └─BatchSimpleAgg { aggs: [max(s.v)] }
+        └─BatchScan { table: s, columns: [s.v], distribution: SomeShard }
   batch_local_plan: |-
     BatchSimpleAgg { aggs: [max(s.v)] }
     └─BatchExchange { order: [], dist: Single }
@@ -160,7 +161,7 @@
     └─BatchProject { exprs: [max(s.v)] }
       └─BatchHashAgg { group_key: [s.k], aggs: [max(s.v)] }
         └─BatchExchange { order: [], dist: HashShard(s.k) }
-          └─BatchScan { table: s, columns: [s.k, s.v], distribution: Single }
+          └─BatchScan { table: s, columns: [s.k, s.v], distribution: SomeShard }
   batch_local_plan: |-
     BatchProject { exprs: [max(s.v)] }
     └─BatchHashAgg { group_key: [s.k], aggs: [max(s.v)] }