refactor: extract SchemaApiTestSuite to standalone crate (#19219)

* refactor: use FnOnce for TimedFuture callback

The timing callback is only invoked once when the future completes, so
`FnOnce` is more appropriate than `Fn`. Combined `start` and `callback`
into a single `on_ready` field since they are always consumed together.

Changes:
- Change callback bound from `Fn` to `FnOnce`
- Combine `start` and `callback` into `on_ready: Option<(Instant, F)>`
- Simplify `poll()` to consume both fields atomically

* refactor: extract SchemaApiTestSuite to standalone crate

Move `schema_api_test_suite.rs` from `databend-common-meta-api` to a new
standalone crate `databend-common-meta-schema-api-test-suite`. This follows
the existing pattern of `databend-common-meta-kvapi-test-suite`.

Also fix `TimedFuture` to initialize the start time on the first poll
instead of when constructing, so that idle time before polling is excluded
from total duration measurement.

Changes:
- Create `databend-common-meta-schema-api-test-suite` crate
- Move `SchemaApiTestSuite` to the new crate
- Copy helper functions `get_kv_data()` and `get_kv_u64_data()` locally
- Update consumer test imports in `databend-meta` service tests
- Fix `TimedFuture` to start timing on first poll, not construction

Author: drmingdrmer

Cargo.lock

@@ -63,6 +63,7 @@ databend-common-meta-kvapi = { path = "src/meta/kvapi" }
 databend-common-meta-kvapi-test-suite = { path = "src/meta/kvapi-test-suite" }
 databend-common-meta-process = { path = "src/meta/process" }
 databend-common-meta-raft-store = { path = "src/meta/raft-store" }
+databend-common-meta-schema-api-test-suite = { path = "src/meta/schema-api-test-suite" }
 databend-common-meta-semaphore = { path = "src/meta/semaphore" }
 databend-common-meta-sled-store = { path = "src/meta/sled-store" }
 databend-common-meta-stoerr = { path = "src/meta/stoerr" }

Cargo.toml

@@ -32,41 +32,42 @@ pin_project! {
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     pub struct TimedFuture<'a, Fu, F>
     where
-        F: Fn(&Fu::Output, Duration, Duration),
+        F: FnOnce(&Fu::Output, Duration, Duration),
         F: 'a,
         Fu: Future,
     {
         #[pin]
         inner: Fu,
 
-        start: Option<Instant>,
         busy: Duration,
-        callback: F,
-        _p : PhantomData<&'a ()>,
-
+        // Start time, initialized on first poll.
+        start: Option<Instant>,
+        // Callback, consumed when the future completes.
+        on_ready: Option<F>,
+        _p: PhantomData<&'a ()>,
     }
 }
 
 impl<'a, Fu, F> TimedFuture<'a, Fu, F>
 where
-    F: Fn(&Fu::Output, Duration, Duration),
+    F: FnOnce(&Fu::Output, Duration, Duration),
     F: 'a,
     Fu: Future,
 {
     pub fn new(inner: Fu, callback: F) -> Self {
         Self {
             inner,
-            start: None,
             busy: Duration::default(),
-            callback,
-            _p: Default::default(),
+            start: None,
+            on_ready: Some(callback),
+            _p: PhantomData,
         }
     }
 }
 
 impl<'a, Fu, F> Future for TimedFuture<'a, Fu, F>
 where
-    F: Fn(&Fu::Output, Duration, Duration),
+    F: FnOnce(&Fu::Output, Duration, Duration),
     F: 'a,
     Fu: Future,
 {
@@ -75,20 +76,21 @@ where
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
 
-        if this.start.is_none() {
+        // Initialize start time on first poll, only if callback is present.
+        if this.start.is_none() && this.on_ready.is_some() {
             *this.start = Some(Instant::now());
         }
 
         let t0 = Instant::now();
-
         let res = this.inner.poll(cx);
-
         *this.busy += t0.elapsed();
 
         match &res {
             Poll::Ready(output) => {
-                let total = this.start.unwrap().elapsed();
-                (this.callback)(output, total, *this.busy);
+                if let Some(callback) = this.on_ready.take() {
+                    let total = this.start.map(|s| s.elapsed()).unwrap_or_default();
+                    (callback)(output, total, *this.busy);
+                }
             }
             Poll::Pending => {}
         }
@@ -104,7 +106,7 @@ where Self: Future
     /// Wrap the future with a timing future.
     fn with_timing<'a, F>(self, f: F) -> TimedFuture<'a, Self, F>
     where
-        F: Fn(&Self::Output, Duration, Duration) + 'a,
+        F: FnOnce(&Self::Output, Duration, Duration) + 'a,
         Self: Future + Sized;
 
     /// Wrap the future with a timing future,
@@ -113,9 +115,9 @@ where Self: Future
         self,
         threshold: Duration,
         f: F,
-    ) -> TimedFuture<'a, Self, impl Fn(&Self::Output, Duration, Duration)>
+    ) -> TimedFuture<'a, Self, impl FnOnce(&Self::Output, Duration, Duration)>
     where
-        F: Fn(&Self::Output, Duration, Duration) + 'a,
+        F: FnOnce(&Self::Output, Duration, Duration) + 'a,
         Self: Future + Sized,
     {
         self.with_timing::<'a>(move |output, total, busy| {
@@ -130,7 +132,7 @@ where Self: Future
     fn log_elapsed_debug<'a>(
         self,
         ctx: impl fmt::Display + 'a,
-    ) -> TimedFuture<'a, Self, impl Fn(&Self::Output, Duration, Duration)>
+    ) -> TimedFuture<'a, Self, impl FnOnce(&Self::Output, Duration, Duration)>
     where
         Self: Future + Sized,
     {
@@ -158,7 +160,7 @@ where Self: Future
     fn log_elapsed_info<'a>(
         self,
         ctx: impl fmt::Display + 'a,
-    ) -> TimedFuture<'a, Self, impl Fn(&Self::Output, Duration, Duration)>
+    ) -> TimedFuture<'a, Self, impl FnOnce(&Self::Output, Duration, Duration)>
     where
         Self: Future + Sized,
     {
@@ -187,7 +189,7 @@ where T: Future + Sized
 {
     fn with_timing<'a, F>(self, f: F) -> TimedFuture<'a, Self, F>
     where
-        F: Fn(&Self::Output, Duration, Duration),
+        F: FnOnce(&Self::Output, Duration, Duration),
         F: 'a,
     {
         TimedFuture::new(self, f)

src/common/base/src/future.rs

@@ -40,13 +40,11 @@ pub mod name_id_value_api;
 pub mod name_value_api;
 pub mod reply;
 mod schema_api;
-mod schema_api_test_suite;
 pub mod security_api;
 mod sequence_api;
 pub mod serialization_util;
 pub mod table_api;
 pub mod tag_api;
-pub(crate) mod testing;
 pub mod txn_backoff;
 pub mod txn_condition_util;
 pub mod txn_core_util;
@@ -96,7 +94,6 @@ pub use kv_fetch_util::mget_pb_values;
 pub use lock_api::LockApi;
 pub use row_access_policy_api::RowAccessPolicyApi;
 pub use schema_api::SchemaApi;
-pub use schema_api_test_suite::SchemaApiTestSuite;
 pub use security_api::SecurityApi;
 pub use sequence_api::SequenceApi;
 // Re-export from new serialization_util module for backward compatibility

src/meta/api/src/lib.rs

@@ -0,0 +1,25 @@
+[package]
+name = "databend-common-meta-schema-api-test-suite"
+version = { workspace = true }
+authors = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+edition = { workspace = true }
+
+[dependencies]
+anyhow = { workspace = true }
+chrono = { workspace = true }
+databend-common-base = { workspace = true }
+databend-common-exception = { workspace = true }
+databend-common-expression = { workspace = true }
+databend-common-meta-api = { workspace = true }
+databend-common-meta-app = { workspace = true }
+databend-common-meta-kvapi = { workspace = true }
+databend-common-meta-types = { workspace = true }
+databend-common-proto-conv = { workspace = true }
+fastrace = { workspace = true }
+log = { workspace = true }
+maplit = { workspace = true }
+
+[lints]
+workspace = true

src/meta/schema-api-test-suite/Cargo.toml

@@ -0,0 +1,25 @@
+// Copyright 2021 Datafuse 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.
+
+#![allow(clippy::uninlined_format_args)]
+#![allow(clippy::diverging_sub_expression)]
+#![allow(clippy::type_complexity)]
+#![allow(clippy::collapsible_if)]
+#![allow(clippy::unnecessary_unwrap)]
+#![feature(try_blocks)]
+
+mod schema_api_test_suite;
+mod testing;
+
+pub use schema_api_test_suite::SchemaApiTestSuite;

src/meta/schema-api-test-suite/src/lib.rs

@@ -30,6 +30,18 @@ use databend_common_expression::TableDataType;
 use databend_common_expression::TableField;
 use databend_common_expression::TableSchema;
 use databend_common_expression::types::NumberDataType;
+use databend_common_meta_api::DEFAULT_MGET_SIZE;
+use databend_common_meta_api::DatamaskApi;
+use databend_common_meta_api::RowAccessPolicyApi;
+use databend_common_meta_api::SchemaApi;
+use databend_common_meta_api::SequenceApi;
+use databend_common_meta_api::TableApi;
+use databend_common_meta_api::deserialize_struct;
+use databend_common_meta_api::kv_app_error::KVAppError;
+use databend_common_meta_api::kv_pb_api::KVPbApi;
+use databend_common_meta_api::kv_pb_api::UpsertPB;
+use databend_common_meta_api::serialize_struct;
+use databend_common_meta_api::util::IdempotentKVTxnSender;
 use databend_common_meta_app::KeyWithTenant;
 use databend_common_meta_app::app_error::AppError;
 use databend_common_meta_app::data_mask::CreateDatamaskReq;
@@ -145,20 +157,8 @@ use fastrace::func_name;
 use log::debug;
 use log::info;
 
-use crate::DEFAULT_MGET_SIZE;
-use crate::DatamaskApi;
-use crate::RowAccessPolicyApi;
-use crate::SchemaApi;
-use crate::SequenceApi;
-use crate::TableApi;
-use crate::deserialize_struct;
-use crate::kv_app_error::KVAppError;
-use crate::kv_pb_api::KVPbApi;
-use crate::kv_pb_api::UpsertPB;
-use crate::serialize_struct;
 use crate::testing::get_kv_data;
 use crate::testing::get_kv_u64_data;
-use crate::util::IdempotentKVTxnSender;
 
 /// Test suite of `SchemaApi`.
 ///

src/meta/api/src/schema_api_test_suite.rs …-test-suite/src/schema_api_test_suite.rssrc/meta/api/src/schema_api_test_suite.rs renamed to src/meta/schema-api-test-suite/src/schema_api_test_suite.rs src/meta/api/src/schema_api_test_suite.rs renamed to src/meta/schema-api-test-suite/src/schema_api_test_suite.rs

@@ -14,6 +14,9 @@
 
 //! Supporting utilities for tests.
 
+use databend_common_meta_api::deserialize_struct;
+use databend_common_meta_api::kv_app_error::KVAppError;
+use databend_common_meta_api::serialization_util::deserialize_u64;
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::KvApiExt;
 use databend_common_meta_types::MetaAPIError;
@@ -23,9 +26,6 @@ use databend_common_meta_types::MetaError;
 use databend_common_meta_types::anyerror::AnyError;
 use databend_common_proto_conv::FromToProto;
 
-use crate::kv_app_error::KVAppError;
-use crate::serialization_util::deserialize_u64;
-
 /// Get existing value by key. Panic if key is absent.
 pub(crate) async fn get_kv_data<T>(
     kv_api: &(impl kvapi::KVApi<Error = MetaError> + ?Sized),
@@ -36,7 +36,7 @@ where
 {
     let res = kv_api.get_kv(&key.to_string_key()).await?;
     if let Some(res) = res {
-        let s = crate::deserialize_struct(&res.data)?;
+        let s = deserialize_struct(&res.data)?;
         return Ok(s);
     };
 

src/meta/api/src/testing.rs …eta/schema-api-test-suite/src/testing.rssrc/meta/api/src/testing.rs renamed to src/meta/schema-api-test-suite/src/testing.rs src/meta/api/src/testing.rs renamed to src/meta/schema-api-test-suite/src/testing.rs

@@ -73,6 +73,7 @@ watcher = { workspace = true }
 [dev-dependencies]
 databend-common-meta-cache = { workspace = true }
 databend-common-meta-kvapi-test-suite = { workspace = true }
+databend-common-meta-schema-api-test-suite = { workspace = true }
 databend-common-meta-semaphore = { workspace = true }
 databend-common-version = { workspace = true }
 env_logger = { workspace = true }

src/meta/service/Cargo.toml

@@ -18,7 +18,7 @@ use std::sync::Arc;
 use std::sync::Mutex;
 
 use databend_common_meta_api::AutoIncrementApiTestSuite;
-use databend_common_meta_api::SchemaApiTestSuite;
+use databend_common_meta_schema_api_test_suite::SchemaApiTestSuite;
 use test_harness::test;
 
 use crate::testing::meta_service_test_harness;