Merge branch 'main' into xxh/add-sink-metr

Signed-off-by: Richard Chien <[email protected]>

.github/workflows/typo.yml

@@ -10,4 +10,4 @@ jobs:
       uses: actions/checkout@v4
 
     - name: Check spelling of the entire repository
-      uses: crate-ci/typos@v1.23.2
+      uses: crate-ci/typos@v1.28.2

.pre-commit-config.yaml

@@ -9,7 +9,7 @@ repos:
         exclude: 'src/frontend/planner_test/tests/testdata/.*'
     -   id: trailing-whitespace
 -   repo: https://github.com/crate-ci/typos
-    rev: v1.23.1
+    rev: v1.28.2
     hooks:
       - id: typos
 -   repo: local

Cargo.toml

@@ -80,7 +80,7 @@ license = "Apache-2.0"
 repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
-foyer = { version = "0.13.0", features = ["tracing", "nightly", "prometheus"] }
+foyer = { version = "0.13.1", features = ["tracing", "nightly", "prometheus"] }
 apache-avro = { git = "https://github.com/risingwavelabs/avro", rev = "25113ba88234a9ae23296e981d8302c290fdaa4b", features = [
     "snappy",
     "zstandard",

Makefile.toml

@@ -1158,7 +1158,7 @@ fi
 private = true
 category = "RiseDev - Check"
 description = "Run cargo typos-cli check"
-install_crate = { min_version = "1.23.2", crate_name = "typos-cli", binary = "typos", test_arg = [
+install_crate = { min_version = "1.28.2", crate_name = "typos-cli", binary = "typos", test_arg = [
     "--help",
 ], install_command = "binstall" }
 script = """

e2e_test/source_inline/kafka/avro/ref.slt

@@ -109,7 +109,11 @@ select
   (bar).b.y
 from s;
 ----
-3 4 5 6 6 5 4 3
+3 4 5 6 NULL NULL NULL NULL
+
+# Parsing of column `bar` fails even with ints because now `schema` is required.
+# This will be fully supported in the next PR
+# 3 4 5 6 6 5 4 3
 
 
 statement ok

grafana/risingwave-dev-dashboard.dashboard.py

@@ -1756,7 +1756,7 @@ def section_streaming_actors(outer_panels: Panels):
                     "it's very likely to be the performance bottleneck",
                     [
                         panels.target(
-                            # Here we use `min` but actually no much difference. Any of the sampled epoches makes sense.
+                            # Here we use `min` but actually no much difference. Any of the sampled epochs makes sense.
                             f"min({metric('stream_actor_current_epoch')} != 0) by (fragment_id)",
                             "fragment {{fragment_id}}",
                             ),

proto/compute.proto

@@ -12,6 +12,14 @@ message ShowConfigResponse {
   string stream_config = 2;
 }
 
+message ResizeCacheRequest {
+  uint64 meta_cache_capacity = 1;
+  uint64 data_cache_capacity = 2;
+}
+
+message ResizeCacheResponse {}
+
 service ConfigService {
   rpc ShowConfig(ShowConfigRequest) returns (ShowConfigResponse);
+  rpc ResizeCache(ResizeCacheRequest) returns (ResizeCacheResponse);
 }

src/common/src/config.rs

@@ -869,7 +869,7 @@ pub struct StorageConfig {
     #[serde(default = "default::storage::mem_table_spill_threshold")]
     pub mem_table_spill_threshold: usize,
 
-    /// The concurrent uploading number of `SSTables` of buidler
+    /// The concurrent uploading number of `SSTables` of builder
     #[serde(default = "default::storage::compactor_concurrent_uploading_sst_count")]
     pub compactor_concurrent_uploading_sst_count: Option<usize>,
 

src/compute/src/rpc/service/config_service.rs

@@ -13,16 +13,24 @@
 // limitations under the License.
 use std::sync::Arc;
 
+use foyer::HybridCache;
 use risingwave_batch::task::BatchManager;
 use risingwave_common::error::tonic::ToTonicStatus;
+use risingwave_hummock_sdk::HummockSstableObjectId;
 use risingwave_pb::compute::config_service_server::ConfigService;
-use risingwave_pb::compute::{ShowConfigRequest, ShowConfigResponse};
+use risingwave_pb::compute::{
+    ResizeCacheRequest, ResizeCacheResponse, ShowConfigRequest, ShowConfigResponse,
+};
+use risingwave_storage::hummock::{Block, Sstable, SstableBlockIndex};
 use risingwave_stream::task::LocalStreamManager;
+use thiserror_ext::AsReport;
 use tonic::{Code, Request, Response, Status};
 
 pub struct ConfigServiceImpl {
     batch_mgr: Arc<BatchManager>,
     stream_mgr: LocalStreamManager,
+    meta_cache: Option<HybridCache<HummockSstableObjectId, Box<Sstable>>>,
+    block_cache: Option<HybridCache<SstableBlockIndex, Box<Block>>>,
 }
 
 #[async_trait::async_trait]
@@ -42,13 +50,53 @@ impl ConfigService for ConfigServiceImpl {
         };
         Ok(Response::new(show_config_response))
     }
+
+    async fn resize_cache(
+        &self,
+        request: Request<ResizeCacheRequest>,
+    ) -> Result<Response<ResizeCacheResponse>, Status> {
+        let req = request.into_inner();
+
+        if let Some(meta_cache) = &self.meta_cache
+            && req.meta_cache_capacity > 0
+        {
+            match meta_cache.memory().resize(req.meta_cache_capacity as _) {
+                Ok(_) => tracing::info!(
+                    "resize meta cache capacity to {:?}",
+                    req.meta_cache_capacity
+                ),
+                Err(e) => return Err(Status::internal(e.to_report_string())),
+            }
+        }
+
+        if let Some(block_cache) = &self.block_cache
+            && req.data_cache_capacity > 0
+        {
+            match block_cache.memory().resize(req.data_cache_capacity as _) {
+                Ok(_) => tracing::info!(
+                    "resize data cache capacity to {:?}",
+                    req.data_cache_capacity
+                ),
+                Err(e) => return Err(Status::internal(e.to_report_string())),
+            }
+        }
+
+        Ok(Response::new(ResizeCacheResponse {}))
+    }
 }
 
 impl ConfigServiceImpl {
-    pub fn new(batch_mgr: Arc<BatchManager>, stream_mgr: LocalStreamManager) -> Self {
+    pub fn new(
+        batch_mgr: Arc<BatchManager>,
+        stream_mgr: LocalStreamManager,
+        meta_cache: Option<HybridCache<HummockSstableObjectId, Box<Sstable>>>,
+        block_cache: Option<HybridCache<SstableBlockIndex, Box<Block>>>,
+    ) -> Self {
         Self {
             batch_mgr,
             stream_mgr,
+            meta_cache,
+            block_cache,
         }
     }
 }

src/compute/src/server.rs

@@ -404,10 +404,10 @@ pub async fn compute_node_serve(
     let monitor_srv = MonitorServiceImpl::new(
         stream_mgr.clone(),
         config.server.clone(),
-        meta_cache,
-        block_cache,
+        meta_cache.clone(),
+        block_cache.clone(),
     );
-    let config_srv = ConfigServiceImpl::new(batch_mgr, stream_mgr.clone());
+    let config_srv = ConfigServiceImpl::new(batch_mgr, stream_mgr.clone(), meta_cache, block_cache);
     let health_srv = HealthServiceImpl::new();
 
     let telemetry_manager = TelemetryManager::new(

src/config/docs.md

@@ -120,7 +120,7 @@ This page is automatically generated by `./risedev generate-example-config`
 | block_cache_capacity_mb | DEPRECATED: This config will be deprecated in the future version, use `storage.cache.block_cache_capacity_mb` instead. |  |
 | check_compaction_result |  | false |
 | compact_iter_recreate_timeout_ms |  | 600000 |
-| compactor_concurrent_uploading_sst_count | The concurrent uploading number of `SSTables` of buidler |  |
+| compactor_concurrent_uploading_sst_count | The concurrent uploading number of `SSTables` of builder |  |
 | compactor_fast_max_compact_delete_ratio |  | 40 |
 | compactor_fast_max_compact_task_size |  | 2147483648 |
 | compactor_iter_max_io_retry_times |  | 8 |