common_meta/ddl/
create_database.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashMap;
16use std::sync::Arc;
17
18use async_trait::async_trait;
19use common_error::ext::{BoxedError, ErrorExt};
20use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
21use common_procedure::{
22    Context as ProcedureContext, EventContext, EventTrigger, LockKey, Procedure, ProcedureId,
23    Status,
24};
25use serde::{Deserialize, Serialize};
26use serde_with::{DefaultOnNull, serde_as};
27use snafu::{ResultExt, ensure};
28use strum::AsRefStr;
29
30use crate::ddl::DdlContext;
31use crate::ddl::event::database::{CREATE_DATABASE_EVENT_TYPE, DatabaseDdlEvent};
32use crate::ddl::utils::map_to_procedure_error;
33use crate::error::{self, Result};
34use crate::instruction::{CacheIdent, UserCacheIdent};
35use crate::key::schema_name::{SchemaNameKey, SchemaNameValue};
36use crate::lock_key::{CatalogLock, SchemaLock};
37use crate::rpc::ddl::CreatorGrantIntent;
38
39/// Describes the creator-access result of an atomic create.
40#[derive(Debug, Clone, Copy, PartialEq, Eq)]
41pub enum AtomicCreateAccess {
42    /// Added an exact `ALL` ACL entry for the new database.
43    AddedExactAll,
44    /// The creator already had effective `ALL` access, so their ACL was unchanged.
45    AlreadyEffectiveAll,
46}
47
48/// Outcome of atomically creating database metadata and ensuring creator access.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum AtomicCreateOutcome {
51    /// This attempt created the database with the given access result.
52    Created(AtomicCreateAccess),
53    /// The same create procedure committed during an earlier attempt.
54    AlreadyCommitted,
55    /// The creator row changed concurrently and the operation should be retried.
56    UserChanged,
57    /// The creator is missing or no longer identifies the same account generation.
58    UserMissing,
59    /// A different value already exists for the database metadata.
60    SchemaConflict,
61}
62
63/// Commits database metadata and any required creator access atomically.
64///
65/// Implementations classify conditional-write failures with [`AtomicCreateOutcome`] so the
66/// procedure can retry without overwriting concurrent user changes.
67#[async_trait]
68pub trait CreateDatabaseMetadataCommitter: Send + Sync {
69    async fn commit(
70        &self,
71        catalog: &str,
72        schema: &str,
73        value: &SchemaNameValue,
74        creator: &CreatorGrantIntent,
75    ) -> std::result::Result<AtomicCreateOutcome, BoxedError>;
76}
77
78/// Shared [`CreateDatabaseMetadataCommitter`].
79pub type CreateDatabaseMetadataCommitterRef = Arc<dyn CreateDatabaseMetadataCommitter>;
80
81/// Procedure that creates a database and optionally ensures its creator has access.
82pub struct CreateDatabaseProcedure {
83    pub context: DdlContext,
84    pub data: CreateDatabaseData,
85}
86
87impl CreateDatabaseProcedure {
88    pub const TYPE_NAME: &'static str = "metasrv-procedure::CreateDatabase";
89
90    pub fn new(
91        catalog: String,
92        schema: String,
93        create_if_not_exists: bool,
94        options: HashMap<String, String>,
95        creator: Option<CreatorGrantIntent>,
96        context: DdlContext,
97    ) -> Self {
98        Self {
99            context,
100            data: CreateDatabaseData {
101                state: CreateDatabaseState::Prepare,
102                catalog,
103                schema,
104                create_if_not_exists,
105                options,
106                creator,
107            },
108        }
109    }
110
111    pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
112        let data = serde_json::from_str(json).context(FromJsonSnafu)?;
113
114        Ok(Self { context, data })
115    }
116
117    pub async fn on_prepare(&mut self) -> Result<Status> {
118        let exists = self
119            .context
120            .table_metadata_manager
121            .schema_manager()
122            .exists(SchemaNameKey::new(&self.data.catalog, &self.data.schema))
123            .await?;
124
125        if exists && self.data.create_if_not_exists {
126            return Ok(Status::done());
127        }
128
129        ensure!(
130            !exists,
131            error::SchemaAlreadyExistsSnafu {
132                catalog: &self.data.catalog,
133                schema: &self.data.schema,
134            }
135        );
136
137        self.data.state = CreateDatabaseState::CreateMetadata;
138        Ok(Status::executing(true))
139    }
140
141    pub async fn on_create_metadata(&mut self, procedure_id: ProcedureId) -> Result<Status> {
142        let mut value: SchemaNameValue = (&self.data.options).try_into()?;
143
144        if let Some(creator) = self.data.creator.as_ref() {
145            // Distinguishes a replay of this procedure from a competing create.
146            value.create_procedure_id = Some(procedure_id.to_string());
147            let Some(committer) = self.context.create_database_metadata_committer.as_ref() else {
148                return error::UnexpectedSnafu {
149                    err_msg: "Create-database creator grant committer is not configured"
150                        .to_string(),
151                }
152                .fail();
153            };
154            let _timer = crate::metrics::METRIC_META_CREATE_SCHEMA.start_timer();
155            let outcome = match committer
156                .commit(&self.data.catalog, &self.data.schema, &value, creator)
157                .await
158            {
159                Ok(outcome) => outcome,
160                Err(err) if err.is_retryable() => {
161                    return Err(error::Error::RetryLater {
162                        source: err,
163                        clean_poisons: false,
164                    });
165                }
166                Err(err) => return Err(err).context(error::ExternalSnafu),
167            };
168
169            match outcome {
170                AtomicCreateOutcome::Created(_) => {
171                    crate::metrics::METRIC_META_CREATE_SCHEMA_COUNTER.inc();
172                }
173                AtomicCreateOutcome::AlreadyCommitted => {}
174                AtomicCreateOutcome::UserChanged => {
175                    let source = error::UnexpectedSnafu {
176                        err_msg: format!(
177                            "User '{}' changed while creating database '{}.{}'",
178                            creator.username, self.data.catalog, self.data.schema
179                        ),
180                    }
181                    .build();
182                    return Err(error::Error::retry_later(source));
183                }
184                AtomicCreateOutcome::UserMissing => {
185                    return error::UnexpectedSnafu {
186                        err_msg: format!(
187                            "Creator '{}' no longer exists in catalog '{}'",
188                            creator.username, self.data.catalog
189                        ),
190                    }
191                    .fail();
192                }
193                AtomicCreateOutcome::SchemaConflict if self.data.create_if_not_exists => {
194                    return Ok(Status::done());
195                }
196                AtomicCreateOutcome::SchemaConflict => {
197                    return error::SchemaAlreadyExistsSnafu {
198                        catalog: &self.data.catalog,
199                        schema: &self.data.schema,
200                    }
201                    .fail();
202                }
203            }
204
205            // The durable transaction is complete; retry cache invalidation separately.
206            self.data.state = CreateDatabaseState::InvalidateCreatorCache;
207            return Ok(Status::executing(true));
208        }
209
210        self.context
211            .table_metadata_manager
212            .schema_manager()
213            .create(
214                SchemaNameKey::new(&self.data.catalog, &self.data.schema),
215                Some(value),
216                self.data.create_if_not_exists,
217            )
218            .await?;
219
220        Ok(Status::done())
221    }
222
223    pub async fn on_invalidate_creator_cache(&mut self) -> Result<Status> {
224        let Some(creator) = self.data.creator.as_ref() else {
225            return Ok(Status::done());
226        };
227
228        let ident = CacheIdent::User(UserCacheIdent {
229            catalog: self.data.catalog.clone(),
230            username: creator.username.clone(),
231        });
232        let ctx = crate::cache_invalidator::Context {
233            subject: Some(format!(
234                "Invalidate creator cache after creating database '{}.{}' for '{}'",
235                self.data.catalog, self.data.schema, creator.username
236            )),
237        };
238        if let Err(source) = self
239            .context
240            .cache_invalidator
241            .invalidate(&ctx, &[ident])
242            .await
243        {
244            return Err(error::Error::retry_later(source));
245        }
246
247        Ok(Status::done())
248    }
249}
250
251#[async_trait]
252impl Procedure for CreateDatabaseProcedure {
253    fn type_name(&self) -> &str {
254        Self::TYPE_NAME
255    }
256
257    async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult<Status> {
258        let state = &self.data.state;
259
260        match state {
261            CreateDatabaseState::Prepare => self.on_prepare().await,
262            CreateDatabaseState::CreateMetadata => self.on_create_metadata(ctx.procedure_id).await,
263            CreateDatabaseState::InvalidateCreatorCache => self.on_invalidate_creator_cache().await,
264        }
265        .map_err(map_to_procedure_error)
266    }
267
268    fn dump(&self) -> ProcedureResult<String> {
269        serde_json::to_string(&self.data).context(ToJsonSnafu)
270    }
271
272    fn lock_key(&self) -> LockKey {
273        let lock_key = vec![
274            CatalogLock::Read(&self.data.catalog).into(),
275            SchemaLock::write(&self.data.catalog, &self.data.schema).into(),
276        ];
277
278        LockKey::new(lock_key)
279    }
280
281    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
282        if !ctx.event_type_filter.allows(CREATE_DATABASE_EVENT_TYPE) {
283            return None;
284        }
285
286        let event = if matches!(&ctx.trigger, EventTrigger::Submitted) {
287            DatabaseDdlEvent::create_submitted(
288                &self.data.catalog,
289                &self.data.schema,
290                self.data.create_if_not_exists,
291                &self.data.options,
292            )
293        } else {
294            DatabaseDdlEvent::create_lifecycle(&self.data.catalog, &self.data.schema)
295        };
296        Some(Box::new(event))
297    }
298}
299
300/// Persistent states of [`CreateDatabaseProcedure`].
301#[derive(Debug, Clone, Serialize, Deserialize, AsRefStr)]
302pub enum CreateDatabaseState {
303    /// Checks whether the database already exists.
304    Prepare,
305    /// Creates metadata and, when supplied, ensures the creator has access atomically.
306    CreateMetadata,
307    /// Invalidates the creator's cached user data after the durable commit.
308    InvalidateCreatorCache,
309}
310
311/// Persistent data of [`CreateDatabaseProcedure`].
312#[serde_as]
313#[derive(Debug, Serialize, Deserialize)]
314pub struct CreateDatabaseData {
315    pub state: CreateDatabaseState,
316    pub catalog: String,
317    pub schema: String,
318    pub create_if_not_exists: bool,
319    #[serde_as(deserialize_as = "DefaultOnNull")]
320    pub options: HashMap<String, String>,
321    /// Authenticated creator whose access is ensured, absent for legacy schema-only requests.
322    #[serde(default)]
323    pub creator: Option<CreatorGrantIntent>,
324}
325
326#[cfg(test)]
327mod tests {
328    use std::collections::VecDeque;
329    use std::sync::Mutex;
330    use std::sync::atomic::{AtomicUsize, Ordering};
331
332    use common_error::ext::ErrorExt;
333
334    use super::*;
335    use crate::cache_invalidator::{CacheInvalidator, Context};
336    use crate::test_util::{MockDatanodeManager, new_ddl_context};
337
338    const PROCEDURE_ID: &str = "4ee0ba94-11f0-4d4d-9468-5ebf732e3ab2";
339
340    fn procedure_id() -> ProcedureId {
341        ProcedureId::parse_str(PROCEDURE_ID).unwrap()
342    }
343
344    struct MockCommitter {
345        calls: AtomicUsize,
346        outcomes: Mutex<VecDeque<std::result::Result<AtomicCreateOutcome, BoxedError>>>,
347    }
348
349    impl MockCommitter {
350        fn new(
351            outcomes: impl IntoIterator<Item = std::result::Result<AtomicCreateOutcome, BoxedError>>,
352        ) -> Self {
353            Self {
354                calls: AtomicUsize::new(0),
355                outcomes: Mutex::new(outcomes.into_iter().collect()),
356            }
357        }
358    }
359
360    #[async_trait]
361    impl CreateDatabaseMetadataCommitter for MockCommitter {
362        async fn commit(
363            &self,
364            _catalog: &str,
365            _schema: &str,
366            value: &SchemaNameValue,
367            _creator: &CreatorGrantIntent,
368        ) -> std::result::Result<AtomicCreateOutcome, BoxedError> {
369            assert_eq!(value.create_procedure_id.as_deref(), Some(PROCEDURE_ID));
370            self.calls.fetch_add(1, Ordering::Relaxed);
371            self.outcomes
372                .lock()
373                .unwrap()
374                .pop_front()
375                .expect("unexpected commit call")
376        }
377    }
378
379    #[derive(Default)]
380    struct FailOnceInvalidator {
381        calls: AtomicUsize,
382    }
383
384    #[async_trait]
385    impl CacheInvalidator for FailOnceInvalidator {
386        async fn invalidate(&self, _ctx: &Context, _caches: &[CacheIdent]) -> Result<()> {
387            if self.calls.fetch_add(1, Ordering::Relaxed) == 0 {
388                return error::UnexpectedSnafu {
389                    err_msg: "injected cache invalidation failure",
390                }
391                .fail();
392            }
393            Ok(())
394        }
395
396        fn invalidate_all(&self) -> Result<()> {
397            Ok(())
398        }
399    }
400
401    fn procedure(
402        outcomes: impl IntoIterator<Item = std::result::Result<AtomicCreateOutcome, BoxedError>>,
403        cache_invalidator: Arc<dyn CacheInvalidator>,
404    ) -> (CreateDatabaseProcedure, Arc<MockCommitter>) {
405        let mut context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
406        context.cache_invalidator = cache_invalidator;
407        let committer = Arc::new(MockCommitter::new(outcomes));
408        context.create_database_metadata_committer = Some(committer.clone());
409        let procedure = CreateDatabaseProcedure::new(
410            "greptime".to_string(),
411            "metrics".to_string(),
412            false,
413            HashMap::new(),
414            Some(CreatorGrantIntent {
415                username: "alice".to_string(),
416                created_at_ns: 1,
417            }),
418            context,
419        );
420        (procedure, committer)
421    }
422
423    #[tokio::test]
424    async fn created_and_recovered_commits_retry_only_cache_invalidation() {
425        let invalidator = Arc::new(FailOnceInvalidator::default());
426        let (mut procedure, committer) = procedure(
427            [Ok(AtomicCreateOutcome::Created(
428                AtomicCreateAccess::AddedExactAll,
429            ))],
430            invalidator.clone(),
431        );
432        procedure.data.state = CreateDatabaseState::CreateMetadata;
433
434        assert!(
435            !procedure
436                .on_create_metadata(procedure_id())
437                .await
438                .unwrap()
439                .is_done()
440        );
441        assert!(matches!(
442            procedure.data.state,
443            CreateDatabaseState::InvalidateCreatorCache
444        ));
445        assert_eq!(committer.calls.load(Ordering::Relaxed), 1);
446
447        assert!(
448            procedure
449                .on_invalidate_creator_cache()
450                .await
451                .unwrap_err()
452                .is_retryable()
453        );
454        assert!(
455            procedure
456                .on_invalidate_creator_cache()
457                .await
458                .unwrap()
459                .is_done()
460        );
461        assert_eq!(committer.calls.load(Ordering::Relaxed), 1);
462        assert_eq!(invalidator.calls.load(Ordering::Relaxed), 2);
463    }
464
465    #[tokio::test]
466    async fn user_change_is_explicitly_retryable() {
467        let (mut procedure, committer) = procedure(
468            [Ok(AtomicCreateOutcome::UserChanged)],
469            Arc::new(FailOnceInvalidator::default()),
470        );
471        procedure.data.state = CreateDatabaseState::CreateMetadata;
472
473        let err = procedure
474            .on_create_metadata(procedure_id())
475            .await
476            .unwrap_err();
477        assert!(err.is_retryable());
478        assert!(matches!(
479            procedure.data.state,
480            CreateDatabaseState::CreateMetadata
481        ));
482        assert_eq!(committer.calls.load(Ordering::Relaxed), 1);
483    }
484
485    #[tokio::test]
486    async fn lost_commit_response_replays_as_already_committed() {
487        let (mut procedure, committer) = procedure(
488            [
489                Err(BoxedError::new(
490                    error::ElectionLeaderLeaseChangedSnafu.build(),
491                )),
492                Ok(AtomicCreateOutcome::AlreadyCommitted),
493            ],
494            Arc::new(FailOnceInvalidator::default()),
495        );
496        procedure.data.state = CreateDatabaseState::CreateMetadata;
497
498        let err = procedure
499            .on_create_metadata(procedure_id())
500            .await
501            .unwrap_err();
502        assert!(matches!(err, error::Error::RetryLater { .. }));
503        assert!(matches!(
504            procedure.data.state,
505            CreateDatabaseState::CreateMetadata
506        ));
507        assert!(
508            !procedure
509                .on_create_metadata(procedure_id())
510                .await
511                .unwrap()
512                .is_done()
513        );
514        assert!(matches!(
515            procedure.data.state,
516            CreateDatabaseState::InvalidateCreatorCache
517        ));
518        assert_eq!(committer.calls.load(Ordering::Relaxed), 2);
519    }
520
521    #[tokio::test]
522    async fn lost_state_transition_recovers_committed_business_state() {
523        let (mut procedure, committer) = procedure(
524            [
525                Ok(AtomicCreateOutcome::Created(
526                    AtomicCreateAccess::AddedExactAll,
527                )),
528                Ok(AtomicCreateOutcome::AlreadyCommitted),
529            ],
530            Arc::new(FailOnceInvalidator::default()),
531        );
532        let context = procedure.context.clone();
533        procedure.data.state = CreateDatabaseState::CreateMetadata;
534        let stale_json = procedure.dump().unwrap();
535
536        assert!(
537            !procedure
538                .on_create_metadata(procedure_id())
539                .await
540                .unwrap()
541                .is_done()
542        );
543        assert!(matches!(
544            procedure.data.state,
545            CreateDatabaseState::InvalidateCreatorCache
546        ));
547
548        let mut recovered = CreateDatabaseProcedure::from_json(&stale_json, context).unwrap();
549        assert_eq!(recovered.data.creator.as_ref().unwrap().username, "alice");
550        assert!(
551            !recovered
552                .on_create_metadata(procedure_id())
553                .await
554                .unwrap()
555                .is_done()
556        );
557        assert!(matches!(
558            recovered.data.state,
559            CreateDatabaseState::InvalidateCreatorCache
560        ));
561        assert_eq!(committer.calls.load(Ordering::Relaxed), 2);
562    }
563
564    #[tokio::test]
565    async fn missing_creator_uses_legacy_schema_path() {
566        let mut context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
567        let committer = Arc::new(MockCommitter::new([Ok(AtomicCreateOutcome::Created(
568            AtomicCreateAccess::AddedExactAll,
569        ))]));
570        context.create_database_metadata_committer = Some(committer.clone());
571        let mut procedure = CreateDatabaseProcedure::new(
572            "greptime".to_string(),
573            "legacy".to_string(),
574            false,
575            HashMap::new(),
576            None,
577            context,
578        );
579        procedure.data.state = CreateDatabaseState::CreateMetadata;
580
581        assert!(
582            procedure
583                .on_create_metadata(procedure_id())
584                .await
585                .unwrap()
586                .is_done()
587        );
588        assert_eq!(committer.calls.load(Ordering::Relaxed), 0);
589        assert!(
590            procedure
591                .context
592                .table_metadata_manager
593                .schema_manager()
594                .exists(SchemaNameKey::new("greptime", "legacy"))
595                .await
596                .unwrap()
597        );
598    }
599
600    #[tokio::test]
601    async fn creator_without_committer_fails_closed() {
602        let context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
603        let mut procedure = CreateDatabaseProcedure::new(
604            "greptime".to_string(),
605            "missing_committer".to_string(),
606            false,
607            HashMap::new(),
608            Some(CreatorGrantIntent {
609                username: "alice".to_string(),
610                created_at_ns: 1,
611            }),
612            context,
613        );
614        procedure.data.state = CreateDatabaseState::CreateMetadata;
615
616        assert!(procedure.on_create_metadata(procedure_id()).await.is_err());
617        assert!(
618            !procedure
619                .context
620                .table_metadata_manager
621                .schema_manager()
622                .exists(SchemaNameKey::new("greptime", "missing_committer"))
623                .await
624                .unwrap()
625        );
626    }
627
628    #[test]
629    fn legacy_procedure_json_defaults_creator() {
630        let context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
631        let procedure = CreateDatabaseProcedure::from_json(
632            r#"{
633                "state":"CreateMetadata",
634                "catalog":"greptime",
635                "schema":"metrics",
636                "create_if_not_exists":false,
637                "options":{}
638            }"#,
639            context,
640        )
641        .unwrap();
642
643        assert!(procedure.data.creator.is_none());
644    }
645}
common_meta/ddl/create_database.rs

common_meta/ddl/
create_database.rs
