Merge remote-tracking branch 'origin/main' into bz/potential-stuck

Signed-off-by: Bugen Zhao <[email protected]>

Author: BugenZhao

Cargo.lock

@@ -1454,6 +1454,16 @@ def section_streaming_errors(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_count(
+                    "Source Reader Errors by Type",
+                    "",
+                    [
+                        panels.target(
+                            f"sum({metric('user_source_reader_error_count')}) by (error_type, error_msg, actor_id, source_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: actor_id={{actor_id}}, source_id={{source_id}})",
+                        ),
+                    ],
+                ),
             ],
         ),
     ]
@@ -2256,13 +2266,53 @@ def section_hummock_tiered_cache(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_count(
+                    "Refill Queue Length",
+                    "",
+                    [
+                        panels.target(
+                            f"sum(refill_queue_length) by (instance)",
+                            "refill queue length @ {{instance}}",
+                        ),
+                    ],
+                ),
                 panels.timeseries_ops(
-                    "Refill",
+                    "Refill Ops",
                     "",
                     [
                         panels.target(
-                            f"sum(rate({metric('compute_refill_data_file_cache_count')}[$__rate_interval])) by (extra, instance)",
-                            "refill data file cache - {{extra}} @ {{instance}}",
+                            f"sum(rate({metric('data_refill_duration_count')}[$__rate_interval])) by (op, instance)",
+                            "data file cache refill - {{op}} @ {{instance}}",
+                        ),
+                        panels.target(
+                            f"sum(rate({metric('data_refill_filtered_total')}[$__rate_interval])) by (instance)",
+                            "data file cache refill - filtered @ {{instance}}",
+                        ),
+                        panels.target(
+                            f"sum(rate({metric('meta_refill_duration_count')}[$__rate_interval])) by (op, instance)",
+                            "meta file cache refill - {{op}} @ {{instance}}",
+                        ),
+                    ],
+                ),
+                panels.timeseries_latency(
+                    "Refill Latency",
+                    "",
+                    [
+                        *quantile(
+                            lambda quantile, legend: panels.target(
+                                f"histogram_quantile({quantile}, sum(rate({metric('data_refill_duration_bucket')}[$__rate_interval])) by (le, op, instance))",
+                                f"p{legend} - " +
+                                "data file cache refill - {{op}} @ {{instance}}",
+                            ),
+                            [50, 90, 99, "max"],
+                        ),
+                        *quantile(
+                            lambda quantile, legend: panels.target(
+                                f"histogram_quantile({quantile}, sum(rate({metric('meta_refill_duration_bucket')}[$__rate_interval])) by (le, instance))",
+                                f"p{legend} - " +
+                                "meta cache refill @ {{instance}}",
+                            ),
+                            [50, 90, 99, "max"],
                         ),
                     ],
                 ),

docker/dashboards/risingwave-dev-dashboard.json

@@ -431,6 +431,7 @@ message GetReschedulePlanRequest {
     repeated uint32 include_worker_ids = 1;
     repeated uint32 exclude_worker_ids = 2;
     optional uint32 target_parallelism = 3;
+    optional uint32 target_parallelism_per_worker = 4;
   }
 
   message StableResizePolicy {

grafana/risingwave-dev-dashboard.dashboard.py

@@ -518,6 +518,9 @@ pub struct StorageConfig {
     #[serde(default)]
     pub meta_file_cache: FileCacheConfig,
 
+    #[serde(default)]
+    pub cache_refill: CacheRefillConfig,
+
     /// Whether to enable streaming upload for sstable.
     #[serde(default = "default::storage::min_sst_size_for_streaming_upload")]
     pub min_sst_size_for_streaming_upload: u64,
@@ -570,6 +573,21 @@ pub struct StorageConfig {
     pub unrecognized: Unrecognized<Self>,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
+pub struct CacheRefillConfig {
+    #[serde(default = "default::cache_refill::data_refill_levels")]
+    pub data_refill_levels: Vec<u32>,
+
+    #[serde(default = "default::cache_refill::timeout_ms")]
+    pub timeout_ms: u64,
+
+    #[serde(default = "default::cache_refill::concurrency")]
+    pub concurrency: usize,
+
+    #[serde(default, flatten)]
+    pub unrecognized: Unrecognized<Self>,
+}
+
 /// The subsection `[storage.data_file_cache]` and `[storage.meta_file_cache]` in `risingwave.toml`.
 ///
 /// It's put at [`StorageConfig::data_file_cache`] and  [`StorageConfig::meta_file_cache`].
@@ -617,9 +635,6 @@ pub struct FileCacheConfig {
     #[serde(default = "default::file_cache::reclaim_rate_limit_mb")]
     pub reclaim_rate_limit_mb: usize,
 
-    #[serde(default = "default::file_cache::refill_levels")]
-    pub refill_levels: Vec<u32>,
-
     #[serde(default, flatten)]
     pub unrecognized: Unrecognized<Self>,
 }
@@ -1110,10 +1125,20 @@ pub mod default {
         pub fn reclaim_rate_limit_mb() -> usize {
             0
         }
+    }
 
-        pub fn refill_levels() -> Vec<u32> {
+    pub mod cache_refill {
+        pub fn data_refill_levels() -> Vec<u32> {
             vec![]
         }
+
+        pub fn timeout_ms() -> u64 {
+            6000
+        }
+
+        pub fn concurrency() -> usize {
+            100
+        }
     }
 
     pub mod auto_dump_heap_profile {

grafana/risingwave-dev-dashboard.json

@@ -122,7 +122,6 @@ lfu_tiny_lru_capacity_ratio = 0.01
 rated_random_rate_mb = 0
 flush_rate_limit_mb = 0
 reclaim_rate_limit_mb = 0
-refill_levels = []
 
 [storage.meta_file_cache]
 dir = ""
@@ -138,7 +137,11 @@ lfu_tiny_lru_capacity_ratio = 0.01
 rated_random_rate_mb = 0
 flush_rate_limit_mb = 0
 reclaim_rate_limit_mb = 0
-refill_levels = []
+
+[storage.cache_refill]
+data_refill_levels = []
+timeout_ms = 6000
+concurrency = 100
 
 [system]
 barrier_interval_ms = 1000

proto/meta.proto

@@ -517,8 +517,8 @@ async fn into_chunk_stream<P: ByteStreamSourceParser>(mut parser: P, data_stream
 
                 Err(error) => {
                     tracing::warn!(%error, "message parsing failed, skipping");
-                    // This will throw an error for batch
-                    parser.source_ctx().report_user_source_error(error)?;
+                    // Skip for batch
+                    parser.source_ctx().report_user_source_error(error);
                     continue;
                 }
             }

src/common/src/config.rs

@@ -25,7 +25,7 @@ use itertools::Itertools;
 use parking_lot::Mutex;
 use risingwave_common::array::StreamChunk;
 use risingwave_common::catalog::TableId;
-use risingwave_common::error::{ErrorCode, ErrorSuppressor, Result as RwResult, RwError};
+use risingwave_common::error::{ErrorCode, ErrorSuppressor, RwError};
 use risingwave_common::types::{JsonbVal, Scalar};
 use risingwave_pb::connector_service::PbTableSchema;
 use risingwave_pb::source::ConnectorSplit;
@@ -172,10 +172,9 @@ impl SourceContext {
         ctx
     }
 
-    pub(crate) fn report_user_source_error(&self, e: RwError) -> RwResult<()> {
-        // Repropagate the error if batch
+    pub(crate) fn report_user_source_error(&self, e: RwError) {
         if self.source_info.fragment_id == u32::MAX {
-            return Err(e);
+            return;
         }
         let mut err_str = e.inner().to_string();
         if let Some(suppressor) = &self.error_suppressor
@@ -198,7 +197,6 @@ impl SourceContext {
                 &self.source_info.source_id.table_id.to_string(),
             ])
             .inc();
-        Ok(())
     }
 }
 

src/config/example.toml

@@ -120,6 +120,7 @@ pub async fn resize(context: &CtlContext, resize: ScaleResizeCommands) -> anyhow
         exclude_workers,
         include_workers,
         target_parallelism,
+        target_parallelism_per_worker,
         generate,
         output,
         yes,
@@ -132,8 +133,15 @@ pub async fn resize(context: &CtlContext, resize: ScaleResizeCommands) -> anyhow
         let include_worker_ids =
             worker_input_to_worker_ids(include_workers.unwrap_or_default(), true);
 
-        if let Some(target) = target_parallelism && target == 0 {
-            fail!("Target parallelism must be greater than 0");
+        match (target_parallelism, target_parallelism_per_worker) {
+            (Some(_), Some(_)) => {
+                fail!("Cannot specify both target parallelism and target parallelism per worker")
+            }
+            (_, Some(_)) if include_worker_ids.is_empty() => {
+                fail!("Cannot specify target parallelism per worker without including any worker")
+            }
+            (Some(target), _) if target == 0 => fail!("Target parallelism must be greater than 0"),
+            _ => {}
         }
 
         for worker_id in exclude_worker_ids.iter().chain(include_worker_ids.iter()) {
@@ -161,6 +169,7 @@ pub async fn resize(context: &CtlContext, resize: ScaleResizeCommands) -> anyhow
             include_worker_ids,
             exclude_worker_ids,
             target_parallelism,
+            target_parallelism_per_worker,
         }
     };
 

src/connector/src/parser/mod.rs

@@ -288,6 +288,15 @@ pub struct ScaleResizeCommands {
     #[clap(long)]
     target_parallelism: Option<u32>,
 
+    /// The target parallelism per worker, conflicts with `target_parallelism`, requires
+    /// `include_workers` to be set.
+    #[clap(
+        long,
+        requires = "include_workers",
+        conflicts_with = "target_parallelism"
+    )]
+    target_parallelism_per_worker: Option<u32>,
+
     /// Will generate a plan supported by the `reschedule` command and save it to the provided path
     /// by the `--output`.
     #[clap(long, default_value_t = false)]

src/connector/src/source/base.rs

@@ -14,6 +14,8 @@
 
 pub mod columns;
 pub mod tables;
+pub mod views;
 
 pub use columns::*;
 pub use tables::*;
+pub use views::*;

src/ctl/src/cmd_impl/scale/resize.rs

@@ -0,0 +1,45 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::LazyLock;
+
+use risingwave_common::catalog::INFORMATION_SCHEMA_SCHEMA_NAME;
+use risingwave_common::types::DataType;
+
+use crate::catalog::system_catalog::BuiltinView;
+
+/// The view `views` contains all views defined in the current database. Only those views
+/// are shown that the current user has access to (by way of being the owner or having
+/// some privilege).
+/// Ref: [`https://www.postgresql.org/docs/current/infoschema-views.html`]
+///
+/// In RisingWave, `views` contains information about defined views.
+pub static INFORMATION_SCHEMA_VIEWS: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
+    name: "views",
+    schema: INFORMATION_SCHEMA_SCHEMA_NAME,
+    columns: &[
+        (DataType::Varchar, "table_catalog"),
+        (DataType::Varchar, "table_schema"),
+        (DataType::Varchar, "table_name"),
+        (DataType::Varchar, "view_definition"),
+    ],
+    sql: "SELECT CURRENT_DATABASE() AS table_catalog, \
+                s.name AS table_schema, \
+                v.name AS table_name, \
+                v.definition AS view_definition \
+            FROM rw_catalog.rw_views v \
+            JOIN rw_catalog.rw_schemas s ON v.schema_id = s.id \
+        ORDER BY table_schema, table_name"
+        .to_string(),
+});

src/ctl/src/lib.rs

@@ -377,6 +377,7 @@ prepare_sys_catalog! {
     { BuiltinCatalog::View(&PG_DEPEND) },
     { BuiltinCatalog::View(&INFORMATION_SCHEMA_COLUMNS) },
     { BuiltinCatalog::View(&INFORMATION_SCHEMA_TABLES) },
+    { BuiltinCatalog::View(&INFORMATION_SCHEMA_VIEWS) },
     { BuiltinCatalog::Table(&RW_DATABASES), read_rw_database_info },
     { BuiltinCatalog::Table(&RW_SCHEMAS), read_rw_schema_info },
     { BuiltinCatalog::Table(&RW_USERS), read_rw_user_info },