Merge remote-tracking branch 'up/main' into spill-disk

Signed-off-by: coldWater <forsaken628@gmail.com>

Cargo.toml

@@ -289,7 +289,7 @@ iceberg = { version = "0.3.0" }
 iceberg-catalog-hms = { version = "0.3.0" }
 iceberg-catalog-rest = { version = "0.3.0" }
 poem = { version = "3.0", features = ["openssl-tls", "multipart", "compression"] }
-proj4rs = { version = "0.1.3", features = ["geo-types", "crs-definitions"] }
+proj4rs = { version = "0.1.4", features = ["geo-types", "crs-definitions"] }
 prometheus-client = "0.22"
 prost = { version = "0.12.1" }
 prost-build = { version = "0.12.1" }

src/common/license/src/license_manager.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::ops::Deref;
 use std::sync::Arc;
 
 use databend_common_base::base::GlobalInstance;
@@ -58,21 +59,34 @@ pub trait LicenseManager: Sync + Send {
     fn get_storage_quota(&self, license_key: String) -> Result<StorageQuota>;
 }
 
-pub struct LicenseManagerWrapper {
-    pub manager: Box<dyn LicenseManager>,
+pub struct LicenseManagerSwitch {
+    manager: Box<dyn LicenseManager>,
+}
+
+impl LicenseManagerSwitch {
+    pub fn create(manager: Box<dyn LicenseManager>) -> LicenseManagerSwitch {
+        LicenseManagerSwitch { manager }
+    }
+
+    pub fn instance() -> Arc<LicenseManagerSwitch> {
+        GlobalInstance::get()
+    }
+}
+
+impl Deref for LicenseManagerSwitch {
+    type Target = dyn LicenseManager;
+
+    fn deref(&self) -> &Self::Target {
+        self.manager.as_ref()
+    }
 }
-unsafe impl Send for LicenseManagerWrapper {}
-unsafe impl Sync for LicenseManagerWrapper {}
 
 pub struct OssLicenseManager {}
 
 impl LicenseManager for OssLicenseManager {
     fn init(_tenant: String) -> Result<()> {
         let rm = OssLicenseManager {};
-        let wrapper = LicenseManagerWrapper {
-            manager: Box::new(rm),
-        };
-        GlobalInstance::set(Arc::new(wrapper));
+        GlobalInstance::set(Arc::new(LicenseManagerSwitch::create(Box::new(rm))));
         Ok(())
     }
 
@@ -97,7 +111,3 @@ impl LicenseManager for OssLicenseManager {
         Ok(StorageQuota::default())
     }
 }
-
-pub fn get_license_manager() -> Arc<LicenseManagerWrapper> {
-    GlobalInstance::get()
-}

src/meta/api/src/kv_pb_api/mod.rs

@@ -188,6 +188,25 @@ pub trait KVPbApi: KVApi {
         }
     }
 
+    /// Same as [`get_pb_values`](Self::get_pb_values) but collect the result in a `Vec` instead of a stream.
+    fn get_pb_values_vec<K, I>(
+        &self,
+        keys: I,
+    ) -> impl Future<Output = Result<Vec<Option<SeqV<K::ValueType>>>, Self::Error>> + Send
+    where
+        K: kvapi::Key + 'static,
+        K::ValueType: FromToProto + Send + 'static,
+        I: IntoIterator<Item = K> + Send,
+        Self::Error: From<PbApiReadError<Self::Error>>,
+    {
+        async move {
+            self.get_pb_values(keys)
+                .await?
+                .try_collect::<Vec<_>>()
+                .await
+        }
+    }
+
     /// Same as `get_pb_stream` but does not return keys, only values.
     ///
     /// It guaranteed to return the same number of results as the input keys.
@@ -337,6 +356,26 @@ pub trait KVPbApi: KVApi {
         }
     }
 
+    /// Same as [`list_pb`](Self::list_pb)` but collect the result in a `Vec` instead of a stream.
+    fn list_pb_vec<K>(
+        &self,
+        prefix: &DirName<K>,
+    ) -> impl Future<Output = Result<Vec<(K, SeqV<K::ValueType>)>, Self::Error>> + Send
+    where
+        K: kvapi::Key + Send + Sync + 'static,
+        K::ValueType: FromToProto + Send,
+        Self::Error: From<PbApiReadError<Self::Error>>,
+    {
+        async move {
+            let strm = self.list_pb(prefix).await?;
+            let kvs = strm
+                .map_ok(|itm| (itm.key, itm.seqv))
+                .try_collect::<Vec<_>>()
+                .await?;
+            Ok(kvs)
+        }
+    }
+
     /// Same as `list_pb` but does not return values, only keys.
     fn list_pb_keys<K>(
         &self,

src/meta/api/src/schema_api.rs

@@ -131,9 +131,15 @@ pub trait SchemaApi: Send + Sync {
         req: RenameDatabaseReq,
     ) -> Result<RenameDatabaseReply, KVAppError>;
 
-    async fn get_database_history(
+    /// Retrieves all databases for a specific tenant, including those marked as dropped.
+    ///
+    /// * `include_non_retainable` -
+    /// If true, includes databases that are beyond the retention period.
+    /// If false, excludes such databases from the result.
+    async fn get_tenant_history_databases(
         &self,
         req: ListDatabaseReq,
+        include_non_retainable: bool,
     ) -> Result<Vec<Arc<DatabaseInfo>>, KVAppError>;
 
     // index