From c321c49cd0e4ad38d0136166fef0114b04ffcf9c Mon Sep 17 00:00:00 2001 From: Dylan Chen Date: Fri, 14 Aug 2026 11:47:34 +0800 Subject: [PATCH] feat: support streaming_cache_refill_policy config --- README.md | 2 +- dbt/adapters/risingwave/connections.py | 2 + dbt/include/risingwave/macros/adapters.sql | 1 + docs/configuration.md | 10 +++++ .../test_materialization_relation_lookup.py | 39 ++++++++++++++++++- 5 files changed, 52 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 508d652..f9a04d9 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,7 @@ See [docs/configuration.md](docs/configuration.md) for model-level and `dbt_proj ### RisingWave Native Model Configs -The adapter supports RisingWave session settings such as `streaming_parallelism`, `streaming_parallelism_for_backfill`, `backfill_rate_limit`, and `enable_index_selection` in profiles and native model configs. +The adapter supports RisingWave session settings such as `streaming_parallelism`, `streaming_parallelism_for_backfill`, `streaming_cache_refill_policy`, `backfill_rate_limit`, and `enable_index_selection` in profiles and native model configs. See [docs/configuration.md](docs/configuration.md) for the full configuration matrix. diff --git a/dbt/adapters/risingwave/connections.py b/dbt/adapters/risingwave/connections.py index b44872d..ca2637e 100644 --- a/dbt/adapters/risingwave/connections.py +++ b/dbt/adapters/risingwave/connections.py @@ -16,6 +16,7 @@ "streaming_parallelism", "streaming_parallelism_for_backfill", "streaming_max_parallelism", + "streaming_cache_refill_policy", "enable_serverless_backfill", "backfill_rate_limit", "source_rate_limit", @@ -39,6 +40,7 @@ class RisingWaveCredentials(PostgresCredentials): streaming_parallelism: Optional[Any] = None streaming_parallelism_for_backfill: Optional[Any] = None streaming_max_parallelism: Optional[Any] = None + streaming_cache_refill_policy: Optional[str] = None enable_serverless_backfill: Optional[bool] = None backfill_rate_limit: Optional[int] = None source_rate_limit: Optional[int] = None diff --git a/dbt/include/risingwave/macros/adapters.sql b/dbt/include/risingwave/macros/adapters.sql index 74fa694..5168e89 100644 --- a/dbt/include/risingwave/macros/adapters.sql +++ b/dbt/include/risingwave/macros/adapters.sql @@ -22,6 +22,7 @@ "streaming_parallelism", "streaming_parallelism_for_backfill", "streaming_max_parallelism", + "streaming_cache_refill_policy", "enable_serverless_backfill", "backfill_rate_limit", "source_rate_limit", diff --git a/docs/configuration.md b/docs/configuration.md index 267810d..8d32470 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -36,6 +36,7 @@ default: streaming_parallelism: 2 streaming_parallelism_for_backfill: 2 streaming_max_parallelism: 8 + streaming_cache_refill_policy: both enable_serverless_backfill: true backfill_rate_limit: 1000 streaming_parallelism_for_materialized_view: 4 @@ -50,6 +51,7 @@ Supported adapter-specific profile keys: | `streaming_parallelism` | Sets `SET streaming_parallelism = ...` for the session. | | `streaming_parallelism_for_backfill` | Sets `SET streaming_parallelism_for_backfill = ...` for the session. | | `streaming_max_parallelism` | Sets `SET streaming_max_parallelism = ...` for the session. | +| `streaming_cache_refill_policy` | Sets the initial cache refill policy for streaming jobs created in the session. | | `enable_serverless_backfill` | Sets `SET enable_serverless_backfill = true/false` for the session. | | `backfill_rate_limit` | Sets `SET backfill_rate_limit = ...` for the session. | | `source_rate_limit` | Sets `SET source_rate_limit = ...` for the session. | @@ -118,6 +120,7 @@ You can override supported RisingWave session settings for an individual model. streaming_parallelism=2, streaming_parallelism_for_backfill=2, streaming_max_parallelism=8, + streaming_cache_refill_policy='both', streaming_parallelism_for_materialized_view=4, backfill_rate_limit=1000, enable_index_selection=true @@ -134,6 +137,7 @@ Supported model configs: | `streaming_parallelism` | Sets the initial streaming parallelism for streaming jobs. | | `streaming_parallelism_for_backfill` | Sets streaming parallelism for backfill. | | `streaming_max_parallelism` | Sets the maximum future streaming parallelism. | +| `streaming_cache_refill_policy` | Sets the initial cache refill policy for streaming jobs created by the model. | | `streaming_parallelism_for_materialized_view` | Sets materialized-view-specific streaming parallelism. | | `streaming_parallelism_for_source` | Sets source-specific streaming parallelism. | | `streaming_parallelism_for_table` | Sets table-specific streaming parallelism. | @@ -146,12 +150,18 @@ Supported model configs: | `background_ddl` | Runs supported DDL in the background and waits before dbt continues. | | `enable_index_selection` | Enables or disables index selection while planning the model SQL. | +`streaming_cache_refill_policy` accepts `enabled`, `disabled`, `streaming`, +`serving`, or `both`. It is persisted when RisingWave creates a streaming job; +changing the dbt config alone does not update an existing job. Rebuild the model, +for example with `--full-refresh`, when changing the policy for an existing model. + These configs can also be set globally in `dbt_project.yml`: ```yaml models: my_project: +streaming_parallelism_for_backfill: 2 + +streaming_cache_refill_policy: both +backfill_rate_limit: 1000 +enable_index_selection: true ``` diff --git a/tests/unit/test_materialization_relation_lookup.py b/tests/unit/test_materialization_relation_lookup.py index e93fd96..894ee1f 100644 --- a/tests/unit/test_materialization_relation_lookup.py +++ b/tests/unit/test_materialization_relation_lookup.py @@ -24,6 +24,7 @@ "streaming_parallelism", "streaming_parallelism_for_backfill", "streaming_max_parallelism", + "streaming_cache_refill_policy", "enable_serverless_backfill", "backfill_rate_limit", "source_rate_limit", @@ -425,6 +426,39 @@ def test_profile_session_settings_are_allowlisted(): assert "background_ddl" not in connections.RISINGWAVE_PROFILE_SESSION_SETTINGS +def test_profile_credentials_accept_streaming_cache_refill_policy(): + connections = load_local_connections_module() + + credentials = connections.RisingWaveCredentials.from_dict( + { + "host": "127.0.0.1", + "user": "root", + "password": "", + "port": 4566, + "dbname": "dev", + "schema": "public", + "streaming_cache_refill_policy": "both", + } + ) + + assert credentials.streaming_cache_refill_policy == "both" + + +def test_streaming_cache_refill_policy_renders_in_model_sql_header(): + rendered = render_adapter_macro( + "risingwave__render_sql_header", + {"streaming_cache_refill_policy": "both"}, + extra_context={ + "risingwave__native_model_session_settings": lambda: ["streaming_cache_refill_policy"], + "risingwave__render_session_config_value": lambda value: render_adapter_macro( + "risingwave__render_session_config_value", {}, value + ), + }, + ) + + assert rendered == "set streaming_cache_refill_policy = 'both';" + + def test_profile_session_settings_render_safe_set_statements(): connections = load_local_connections_module() @@ -448,6 +482,7 @@ def cursor(self): handle = FakeHandle() credentials = SimpleNamespace( + streaming_cache_refill_policy="both", streaming_parallelism_for_materialized_view="bounded(16)", streaming_parallelism_for_source="ratio(0.5)", backfill_rate_limit=1000, @@ -460,6 +495,7 @@ def cursor(self): assert handle.cursor_obj.closed assert handle.cursor_obj.statements == [ "SET RW_IMPLICIT_FLUSH TO true", + "SET streaming_cache_refill_policy = 'both'", "SET enable_serverless_backfill = true", "SET backfill_rate_limit = 1000", "SET streaming_parallelism_for_materialized_view = 'bounded(16)'", @@ -477,7 +513,7 @@ def load_local_connections_module(): return module -def render_adapter_macro(name, config, *args): +def render_adapter_macro(name, config, *args, extra_context=None): def dbt_return(value): raise MacroReturn(value) @@ -492,6 +528,7 @@ def raise_compiler_error(message): ), "return": dbt_return, } + context.update(extra_context or {}) return CallableMacroGenerator(macro, context)(*args)