feat: introduce license manager and feature tiers (#17396)

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

.typos.toml

@@ -25,6 +25,7 @@ extend-exclude = [
     "src/sqlparser/tests/testdata/",
     "src/frontend/planner_test/tests/testdata",
     "src/tests/sqlsmith/tests/freeze",
+    "src/license/src/manager.rs",
     "Cargo.lock",
     "**/Cargo.toml",
     "**/go.mod",

Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "src/frontend/planner_test",
     "src/java_binding",
     "src/jni_core",
+    "src/license",
     "src/meta",
     "src/meta/dashboard",
     "src/meta/model_v2",
@@ -217,6 +218,7 @@ risingwave_frontend = { path = "./src/frontend" }
 risingwave_hummock_sdk = { path = "./src/storage/hummock_sdk" }
 risingwave_hummock_test = { path = "./src/storage/hummock_test" }
 risingwave_hummock_trace = { path = "./src/storage/hummock_trace" }
+risingwave_license = { path = "./src/license" }
 risingwave_mem_table_spill_test = { path = "./src/stream/spill_test" }
 risingwave_meta = { path = "./src/meta" }
 risingwave_meta_dashboard = { path = "./src/meta/dashboard" }

e2e_test/batch/catalog/pg_settings.slt.part

@@ -13,6 +13,7 @@ postmaster backup_storage_url
 postmaster barrier_interval_ms
 postmaster checkpoint_frequency
 postmaster enable_tracing
+postmaster license_key
 postmaster max_concurrent_creating_streaming_jobs
 postmaster pause_on_next_bootstrap
 user application_name

e2e_test/error_ui/simple/license.slt

@@ -0,0 +1,59 @@
+# Set the license key to a free tier key.
+statement ok
+ALTER SYSTEM SET license_key TO 'eyJhbGciOiJSUzUxMiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJydy10ZXN0IiwidGllciI6ImZyZWUiLCJpc3MiOiJ0ZXN0LnJpc2luZ3dhdmUuY29tIiwiZXhwIjo5OTk5OTk5OTk5fQ.ALC3Kc9LI6u0S-jeMB1YTxg1k8Azxwvc750ihuSZgjA_e1OJC9moxMvpLrHdLZDzCXHjBYi0XJ_1lowmuO_0iPEuPqN5AFpDV1ywmzJvGmMCMtw3A2wuN7hhem9OsWbwe6lzdwrefZLipyo4GZtIkg5ZdwGuHzm33zsM-X5gl_Ns4P6axHKiorNSR6nTAyA6B32YVET_FAM2YJQrXqpwA61wn1XLfarZqpdIQyJ5cgyiC33BFBlUL3lcRXLMLeYe6TjYGeV4K63qARCjM9yeOlsRbbW5ViWeGtR2Yf18pN8ysPXdbaXm_P_IVhl3jCTDJt9ctPh6pUCbkt36FZqO9A';
+
+query error
+SELECT rw_test_paid_tier();
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `test_paid_tier()`
+  3: feature TestPaid is only available for tier Paid and above, while the current tier is Free
+
+Hint: You may want to set a license key with `ALTER SYSTEM SET license_key = '...';` command.
+
+
+# Set the license key to an invalid key.
+statement ok
+ALTER SYSTEM SET license_key TO 'invalid';
+
+query error
+SELECT rw_test_paid_tier();
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `test_paid_tier()`
+  3: feature TestPaid is not available due to license error
+  4: invalid license key
+  5: InvalidToken
+
+
+# Set the license key to empty. This demonstrates the default behavior in production, i.e., free tier.
+statement ok
+ALTER SYSTEM SET license_key TO '';
+
+query error
+SELECT rw_test_paid_tier();
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `test_paid_tier()`
+  3: feature TestPaid is only available for tier Paid and above, while the current tier is Free
+
+Hint: You may want to set a license key with `ALTER SYSTEM SET license_key = '...';` command.
+
+
+# Set the license key to default. In debug mode, this will set the license key to a paid tier key.
+statement ok
+ALTER SYSTEM SET license_key TO DEFAULT;
+
+query T
+SELECT rw_test_paid_tier();
+----
+t

proto/expr.proto

@@ -286,6 +286,7 @@ message ExprNode {
     // ------------------------
     // Internal functions
     VNODE = 1101;
+    TEST_PAID_TIER = 1102;
     // Non-deterministic functions
     PROCTIME = 2023;
     PG_SLEEP = 2024;

proto/meta.proto

@@ -603,6 +603,7 @@ message SystemParams {
   optional string wasm_storage_url = 14 [deprecated = true];
   optional bool enable_tracing = 15;
   optional bool use_new_object_prefix_strategy = 16;
+  optional string license_key = 17;
 }
 
 message GetSystemParamsRequest {}

src/common/Cargo.toml

@@ -87,6 +87,7 @@ risingwave_common_estimate_size = { workspace = true }
 risingwave_common_metrics = { path = "./metrics" }
 risingwave_common_proc_macro = { workspace = true }
 risingwave_error = { workspace = true }
+risingwave_license = { workspace = true }
 risingwave_pb = { workspace = true }
 rust_decimal = { version = "1", features = ["db-postgres", "maths"] }
 rw_iter_util = { workspace = true }

src/common/src/config.rs

@@ -2278,6 +2278,13 @@ pub struct CompactionConfig {
 mod tests {
     use super::*;
 
+    fn default_config_for_docs() -> RwConfig {
+        let mut config = RwConfig::default();
+        // Set `license_key` to empty to avoid showing the test-only license key in the docs.
+        config.system.license_key = Some("".to_owned());
+        config
+    }
+
     /// This test ensures that `config/example.toml` is up-to-date with the default values specified
     /// in this file. Developer should run `./risedev generate-example-config` to update it if this
     /// test fails.
@@ -2287,7 +2294,7 @@ mod tests {
 # Check detailed comments in src/common/src/config.rs";
 
         let actual = expect_test::expect_file!["../../config/example.toml"];
-        let default = toml::to_string(&RwConfig::default()).expect("failed to serialize");
+        let default = toml::to_string(&default_config_for_docs()).expect("failed to serialize");
 
         let expected = format!("{HEADER}\n\n{default}");
         actual.assert_eq(&expected);
@@ -2328,7 +2335,7 @@ mod tests {
             .collect();
 
         let toml_doc: BTreeMap<String, toml::Value> =
-            toml::from_str(&toml::to_string(&RwConfig::default()).unwrap()).unwrap();
+            toml::from_str(&toml::to_string(&default_config_for_docs()).unwrap()).unwrap();
         toml_doc.into_iter().for_each(|(name, value)| {
             set_default_values("".to_string(), name, value, &mut configs);
         });

src/common/src/lib.rs

@@ -75,12 +75,12 @@ pub mod field_generator;
 pub mod hash;
 pub mod log;
 pub mod memory;
-pub use risingwave_common_metrics as metrics;
 pub use risingwave_common_metrics::{
     monitor, register_guarded_gauge_vec_with_registry,
     register_guarded_histogram_vec_with_registry, register_guarded_int_counter_vec_with_registry,
     register_guarded_int_gauge_vec_with_registry,
 };
+pub use {risingwave_common_metrics as metrics, risingwave_license as license};
 pub mod lru;
 pub mod opts;
 pub mod range;

src/common/src/system_param/common.rs

@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_license::LicenseManager;
+
 use super::diff::SystemParamsDiff;
 use super::reader::SystemParamsReader;
 use crate::util::tracing::layer::toggle_otel_layer;
 
 /// Node-independent handler for system parameter changes.
-///
-/// Currently, it is only used to enable or disable the distributed tracing layer.
 #[derive(Debug)]
 pub struct CommonHandler;
 
@@ -32,8 +32,13 @@ impl CommonHandler {
 
     /// Handle the change of system parameters.
     pub fn handle_change(&self, diff: &SystemParamsDiff) {
+        // Toggle the distributed tracing layer.
         if let Some(enabled) = diff.enable_tracing {
-            toggle_otel_layer(enabled)
+            toggle_otel_layer(enabled);
+        }
+        // Refresh the license key.
+        if let Some(key) = diff.license_key.as_ref() {
+            LicenseManager::get().refresh(key);
         }
     }
 }

src/common/src/system_param/local_manager.rs

@@ -48,6 +48,7 @@ impl LocalSystemParamsManager {
         let this = Self::new_inner(initial_params.clone());
 
         // Spawn a task to run the common handler.
+        // TODO(bugen): this may be spawned multiple times under standalone deployment, though idempotent.
         tokio::spawn({
             let mut rx = this.tx.subscribe();
             async move {

src/common/src/system_param/mod.rs

@@ -30,6 +30,7 @@ use std::ops::RangeBounds;
 use std::str::FromStr;
 
 use paste::paste;
+use risingwave_license::TEST_PAID_LICENSE_KEY;
 use risingwave_pb::meta::PbSystemParams;
 
 use self::diff::SystemParamsDiff;
@@ -60,6 +61,15 @@ impl_param_value!(u64);
 impl_param_value!(f64);
 impl_param_value!(String => &'a str);
 
+/// Set the default value of `license_key` to [`TEST_PAID_LICENSE_KEY`] in debug mode.
+fn default_license_key() -> String {
+    if cfg!(debug_assertions) {
+        TEST_PAID_LICENSE_KEY.to_owned()
+    } else {
+        "".to_owned()
+    }
+}
+
 /// Define all system parameters here.
 ///
 /// To match all these information, write the match arm as follows:
@@ -87,7 +97,8 @@ macro_rules! for_all_params {
             { max_concurrent_creating_streaming_jobs,   u32,    Some(1_u32),                                true,   "Max number of concurrent creating streaming jobs.", },
             { pause_on_next_bootstrap,                  bool,   Some(false),                                true,   "Whether to pause all data sources on next bootstrap.", },
             { enable_tracing,                           bool,   Some(false),                                true,   "Whether to enable distributed tracing.", },
-            { use_new_object_prefix_strategy,           bool,   None,                                       false,   "Whether to split object prefix.", },
+            { use_new_object_prefix_strategy,           bool,   None,                                       false,  "Whether to split object prefix.", },
+            { license_key,                              String, Some(default_license_key()),                true,   "The license key to activate enterprise features.", },
         }
     };
 }
@@ -148,6 +159,8 @@ macro_rules! def_default {
 pub mod default {
     use std::sync::LazyLock;
 
+    use super::*;
+
     for_all_params!(def_default_opt);
     for_all_params!(def_default);
 }
@@ -444,6 +457,7 @@ mod tests {
             (PAUSE_ON_NEXT_BOOTSTRAP_KEY, "false"),
             (ENABLE_TRACING_KEY, "true"),
             (USE_NEW_OBJECT_PREFIX_STRATEGY_KEY, "false"),
+            (LICENSE_KEY_KEY, "foo"),
             ("a_deprecated_param", "foo"),
         ];
 

src/common/src/system_param/reader.rs

@@ -168,4 +168,8 @@ where
             .enable_tracing
             .unwrap_or_else(default::enable_tracing)
     }
+
+    fn license_key(&self) -> &str {
+        self.inner().license_key.as_deref().unwrap_or_default()
+    }
 }

src/config/docs.md

@@ -163,6 +163,7 @@ This page is automatically generated by `./risedev generate-example-config`
 | checkpoint_frequency | There will be a checkpoint for every n barriers. | 1 |
 | data_directory | Remote directory for storing data and metadata objects. |  |
 | enable_tracing | Whether to enable distributed tracing. | false |
+| license_key | The license key to activate enterprise features. | "" |
 | max_concurrent_creating_streaming_jobs | Max number of concurrent creating streaming jobs. | 1 |
 | parallel_compact_size_mb | The size of parallel task for one compact/flush job. | 512 |
 | pause_on_next_bootstrap | Whether to pause all data sources on next bootstrap. | false |

src/config/example.toml

@@ -238,3 +238,4 @@ bloom_false_positive = 0.001
 max_concurrent_creating_streaming_jobs = 1
 pause_on_next_bootstrap = false
 enable_tracing = false
+license_key = ""

src/expr/impl/src/scalar/mod.rs

@@ -84,6 +84,7 @@ pub use to_jsonb::*;
 mod encrypt;
 mod external;
 mod inet;
+mod test_license;
 mod to_timestamp;
 mod translate;
 mod trigonometric;

src/expr/impl/src/scalar/test_license.rs

@@ -0,0 +1,27 @@
+// Copyright 2024 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 risingwave_common::license::Feature;
+use risingwave_expr::{function, ExprError, Result};
+
+/// A function that checks if the `TestPaid` feature is available.
+///
+/// It's mainly for testing purposes only.
+#[function("test_paid_tier() -> boolean")]
+pub fn test_paid_tier() -> Result<bool> {
+    Feature::TestPaid
+        .check_available()
+        .map_err(|e| ExprError::Internal(anyhow::Error::from(e)))?;
+    Ok(true)
+}

src/frontend/src/binder/expr/function.rs

@@ -1426,6 +1426,7 @@ impl Binder {
                 ("pg_is_in_recovery", raw_literal(ExprImpl::literal_bool(false))),
                 // internal
                 ("rw_vnode", raw_call(ExprType::Vnode)),
+                ("rw_test_paid_tier", raw_call(ExprType::TestPaidTier)), // for testing purposes
                 // TODO: choose which pg version we should return.
                 ("version", raw_literal(ExprImpl::literal_varchar(current_cluster_version()))),
                 // non-deterministic

src/frontend/src/expr/pure.rs

@@ -258,6 +258,7 @@ impl ExprVisitor for ImpureAnalyzer {
             }
             // expression output is not deterministic
             Type::Vnode
+            | Type::TestPaidTier
             | Type::Proctime
             | Type::PgSleep
             | Type::PgSleepFor

src/frontend/src/optimizer/plan_expr_visitor/strong.rs

@@ -291,6 +291,7 @@ impl Strong {
             | ExprType::JsonbToRecord
             | ExprType::JsonbSet
             | ExprType::Vnode
+            | ExprType::TestPaidTier
             | ExprType::Proctime
             | ExprType::PgSleep
             | ExprType::PgSleepFor