chore(query): add system history enterprise feature (#17961)

feat(query): add system history enterprise feature

Author: zhang2014

src/common/base/src/runtime/runtime_tracker.rs

@@ -267,7 +267,10 @@ impl ThreadTracker {
     }
 
     pub fn should_log() -> bool {
-        TRACKER.with(|tracker| tracker.borrow().payload.should_log)
+        // To prevent crashes, logging will be skipped if thread local storage is inaccessible.
+        TRACKER
+            .try_with(|tracker| tracker.borrow().payload.should_log)
+            .unwrap_or(false)
     }
 }
 

src/common/license/src/license.rs

@@ -79,6 +79,8 @@ pub enum Feature {
     NgramIndex,
     #[serde(alias = "workload_group", alias = "WORKLOAD_GROUP")]
     WorkloadGroup,
+    #[serde(alias = "system_history", alias = "SYSTEM_HISTORY")]
+    SystemHistory,
     #[serde(other)]
     Unknown,
 }
@@ -128,6 +130,7 @@ impl fmt::Display for Feature {
             Feature::HilbertClustering => write!(f, "hilbert_clustering"),
             Feature::NgramIndex => write!(f, "ngram_index"),
             Feature::WorkloadGroup => write!(f, "workload_group"),
+            Feature::SystemHistory => write!(f, "system_history"),
             Feature::Unknown => write!(f, "unknown"),
         }
     }
@@ -350,6 +353,16 @@ mod tests {
             serde_json::from_str::<Feature>("\"NgramIndex\"").unwrap()
         );
 
+        assert_eq!(
+            Feature::WorkloadGroup,
+            serde_json::from_str::<Feature>("\"workload_group\"").unwrap()
+        );
+
+        assert_eq!(
+            Feature::SystemHistory,
+            serde_json::from_str::<Feature>("\"system_history\"").unwrap()
+        );
+
         assert_eq!(
             Feature::Unknown,
             serde_json::from_str::<Feature>("\"ssss\"").unwrap()
@@ -384,11 +397,13 @@ mod tests {
                 Feature::AmendTable,
                 Feature::HilbertClustering,
                 Feature::NgramIndex,
+                Feature::WorkloadGroup,
+                Feature::SystemHistory,
             ]),
         };
 
         assert_eq!(
-            "LicenseInfo{ type: enterprise, org: databend, tenants: [databend_tenant,foo], features: [aggregate_index,amend_table,attach_table,compute_quota(threads_num: 1, memory_usage: 1),computed_column,data_mask,hilbert_clustering,inverted_index,license_info,ngram_index,storage_encryption,storage_quota(storage_usage: 1),stream,vacuum,virtual_column] }",
+            "LicenseInfo{ type: enterprise, org: databend, tenants: [databend_tenant,foo], features: [aggregate_index,amend_table,attach_table,compute_quota(threads_num: 1, memory_usage: 1),computed_column,data_mask,hilbert_clustering,inverted_index,license_info,ngram_index,storage_encryption,storage_quota(storage_usage: 1),stream,system_history,vacuum,virtual_column,workload_group] }",
             license_info.to_string()
         );
     }

src/query/service/src/sessions/query_ctx.rs

@@ -40,6 +40,7 @@ use databend_common_base::runtime::profile::Profile;
 use databend_common_base::runtime::profile::ProfileStatisticsName;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_base::runtime::MemStat;
+use databend_common_base::runtime::ThreadTracker;
 use databend_common_base::runtime::TrySpawn;
 use databend_common_base::JoinHandle;
 use databend_common_catalog::catalog::CATALOG_DEFAULT;
@@ -73,6 +74,8 @@ use databend_common_expression::TableDataType;
 use databend_common_expression::TableField;
 use databend_common_expression::TableSchema;
 use databend_common_io::prelude::FormatSettings;
+use databend_common_license::license::Feature;
+use databend_common_license::license_manager::LicenseManagerSwitch;
 use databend_common_meta_app::principal::FileFormatParams;
 use databend_common_meta_app::principal::GrantObject;
 use databend_common_meta_app::principal::OnErrorMode;
@@ -1160,6 +1163,18 @@ impl TableContext for QueryContext {
         database: &str,
         table: &str,
     ) -> Result<Arc<dyn Table>> {
+        // Queries to non-internal system_history databases require license checks to be enabled.
+        if database.eq_ignore_ascii_case("system_history") && ThreadTracker::should_log() {
+            LicenseManagerSwitch::instance().check_enterprise_enabled(
+                unsafe {
+                    self.get_settings()
+                        .get_enterprise_license()
+                        .unwrap_or_default()
+                },
+                Feature::SystemHistory,
+            )?;
+        }
+
         let batch_size = self.get_settings().get_stream_consume_batch_size_hint()?;
         self.get_table_from_shared(catalog, database, table, batch_size)
             .await