feat: support streaming_cache_refill_policy in config

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.
 

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

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",

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
 ```

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)