Struct Cluster

pub struct Cluster {
    config: Configuration,
    handle: Handle,
}

A risingwave cluster.

Nodes

Name	IP
meta-x	192.168.1.x
frontend-x	192.168.2.x
compute-x	192.168.3.x
compactor-x	192.168.4.x
kafka-broker	192.168.11.1
kafka-producer	192.168.11.2
object_store_sim	192.168.12.1
client	192.168.100.1
ctl	192.168.101.1

Fields

config: Configuration

handle: Handle

Implementations

impl Cluster

pub async fn start(conf: Configuration) -> Result<Self>

Start a RisingWave cluster for testing.

This function should be called exactly once in a test.

fn per_session_queries(&self) -> Arc<Vec<String>>

pub fn start_session(&mut self) -> Session

Start a SQL session on the client node.

pub async fn run(&mut self, sql: impl Into<String>) -> Result<String>

Run a SQL query on a new session of the client node.

This is a convenience method that creates a new session and runs the query on it. If you want to run multiple queries on the same session, use start_session and Session::run.

pub async fn run_on_client<F>(&self, future: F) -> F::Output

where F: Future + Send + 'static, F::Output: Send + 'static,

Run a future on the client node.

pub async fn get_random_worker_nodes(&self, n: usize) -> Result<Vec<WorkerNode>>

pub async fn wait_until( &mut self, sql: impl Into<String> + Clone, p: impl FnMut(&str) -> bool, interval: Duration, timeout: Duration, ) -> Result<String>

Run a SQL query from the client and wait until the condition is met.

pub async fn wait_until_non_empty( &mut self, sql: &str, interval: Duration, timeout: Duration, ) -> Result<String>

Run a SQL query from the client and wait until the return result is not empty.

pub async fn kill_node(&self, opts: &KillOpts)

Generate a list of random worker nodes to kill by opts, then call kill_nodes to kill and restart them.

pub async fn kill_nodes( &self, nodes: impl IntoIterator<Item = impl AsRef<str>>, restart_delay_secs: u32, )

Kill the given nodes by their names and restart them in 2s + restart_delay_secs with a probability of 0.1.

pub async fn kill_nodes_and_restart( &self, nodes: impl IntoIterator<Item = impl AsRef<str>>, restart_delay_secs: u32, )

pub async fn simple_kill_nodes( &self, nodes: impl IntoIterator<Item = impl AsRef<str>>, )

pub async fn simple_restart_nodes( &self, nodes: impl IntoIterator<Item = impl AsRef<str>>, )

pub async fn create_kafka_producer(&self, datadir: &str)

Create a node for kafka producer and prepare data.

pub fn create_kafka_topics(&self, topics: HashMap<String, i32>)

Create a kafka topic.

pub fn config(&self) -> Configuration

pub fn handle(&self) -> &Handle

pub async fn graceful_shutdown(&self)

Graceful shutdown all RisingWave nodes.

impl Cluster

pub async fn locate_fragments( &mut self, predicates: impl IntoIterator<Item = BoxedPredicate>, ) -> Result<Vec<Fragment>>

Locate fragments that satisfy all the predicates.

pub async fn locate_one_fragment( &mut self, predicates: impl IntoIterator<Item = BoxedPredicate>, ) -> Result<Fragment>

Locate exactly one fragment that satisfies all the predicates.

pub async fn locate_random_fragment(&mut self) -> Result<Fragment>

Locate a random fragment that is reschedulable.

pub async fn locate_random_fragments(&mut self) -> Result<Vec<Fragment>>

Locate some random fragments that are reschedulable.

pub async fn locate_fragment_by_id(&mut self, id: u32) -> Result<Fragment>

Locate a fragment with the given id.

pub async fn get_cluster_info(&self) -> Result<GetClusterInfoResponse>

pub async fn list_source_splits(&self) -> Result<BTreeMap<u32, String>>

actor_id -> splits

async fn update_worker_node_schedulability( &self, worker_ids: Vec<u32>, target: Schedulability, ) -> Result<()>

pub async fn cordon_worker(&self, id: u32) -> Result<()>

pub async fn uncordon_worker(&self, id: u32) -> Result<()>

pub async fn reschedule(&mut self, plan: impl Into<String>) -> Result<()>

Reschedule with the given plan. Check the document of [risingwave_ctl::cmd_impl::meta::reschedule] for more details.

pub async fn reschedule_resolve_no_shuffle( &mut self, plan: impl Into<String>, ) -> Result<()>

Same as reschedule, but resolve the no-shuffle upstream

async fn reschedule_helper( &mut self, plan: impl Into<String>, resolve_no_shuffle_upstream: bool, ) -> Result<()>

pub async fn pause(&mut self) -> Result<()>

Pause all data sources in the cluster.

pub async fn resume(&mut self) -> Result<()>

Resume all data sources in the cluster.

pub async fn throttle_mv( &mut self, table_id: TableId, rate_limit: Option<u32>, ) -> Result<()>

Throttle a Mv in the cluster

pub async fn split_compaction_group( &mut self, compaction_group_id: CompactionGroupId, table_id: HummockSstableId, ) -> Result<()>

pub async fn trigger_manual_compaction( &mut self, compaction_group_id: CompactionGroupId, level_id: u32, ) -> Result<()>