feat: add sequence meta api

[lichuang]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

feat: add sequence meta api and create and drop sequence ddl.

• Fixes add sequence meta API #15246

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Explain why

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

---

This change is 

[drmingdrmer]

Reviewed 2 of 50 files at r1.
Reviewable status: 2 of 50 files reviewed, 7 unresolved discussions (waiting on @lichuang)

---

src/meta/api/src/schema_api_sequence_impl.rs line 57 at r1 (raw file):

use crate::txn_op_put;

pub async fn do_create_sequence(

This operation does not need a transaction to run the CAS loop.

An upsert(SeqMatch::EQ(0)) would be just enough.
If the upsert return value contains an existing value, return it.

---

src/meta/api/src/schema_api_sequence_impl.rs line 340 at r1 (raw file):

    Ok((id_seq, id, sequence_seq, sequence_meta.unwrap()))
}

What's the necessity of using a two level map? i.e., SequenceNameIdent -> SequenceId -> SequenceMeta?

If there is no need to operate SequenceId directly in our codebase(No API accepts a SequenceId as argument), SequenceId can be removed.

Code quote:

async fn get_sequence_or_err(
    kv_api: &(impl kvapi::KVApi<Error = MetaError> + ?Sized),
    name_key: &SequenceNameIdent,
    msg: impl Display,
) -> Result<(u64, u64, u64, SequenceMeta), KVAppError> {
    let (id_seq, id) = get_u64_value(kv_api, name_key).await?;
    sequence_has_to_exist(id_seq, name_key, &msg)?;

    let id_key = SequenceId { id };

    let (sequence_seq, sequence_meta) = get_pb_value(kv_api, &id_key).await?;
    sequence_has_to_exist(sequence_seq, name_key, msg)?;

    Ok((id_seq, id, sequence_seq, sequence_meta.unwrap()))
}

---

src/meta/app/src/app_error.rs line 948 at r1 (raw file):

impl SequenceAlreadyExists {
    pub fn new(name: impl Into<String>, context: impl Into<String>) -> Self {

Using ToString to make the API more generic.

Suggestion:

    pub fn new(name: impl ToString, context: impl ToString) -> Self {

---

src/meta/app/src/app_error.rs line 964 at r1 (raw file):

impl UnknownSequence {
    pub fn new(name: impl Into<String>, context: impl Into<String>) -> Self {

ToString

---

src/meta/app/src/app_error.rs line 980 at r1 (raw file):

impl OutofSequenceRange {
    pub fn new(name: impl Into<String>, context: impl Into<String>) -> Self {

ToString

---

src/meta/app/src/app_error.rs line 995 at r1 (raw file):

impl WrongSequenceCount {
    pub fn new(name: impl Into<String>) -> Self {

ToString

---

src/meta/app/src/app_error.rs line 1184 at r1 (raw file):

    #[error(transparent)]
    WrongSequenceCount(#[from] WrongSequenceCount),
}

There have been a lot of errors in AppError. Consider encapsulate sequence related error in a sub enum:

enum AppError {
    SequenceError(#[from] SequenceError)
    //...
}
enum SequenceError {
    #[error(transparent)]
    SequenceAlreadyExists(#[from] SequenceAlreadyExists),

    #[error(transparent)]
    UnknownSequence(#[from] UnknownSequence),

    #[error(transparent)]
    OutofSequenceRange(#[from] OutofSequenceRange),

    #[error(transparent)]
    WrongSequenceCount(#[from] WrongSequenceCount),
}

Code quote:

    // sequence
    #[error(transparent)]
    SequenceAlreadyExists(#[from] SequenceAlreadyExists),

    #[error(transparent)]
    UnknownSequence(#[from] UnknownSequence),
    #[error(transparent)]
    OutofSequenceRange(#[from] OutofSequenceRange),
    #[error(transparent)]
    WrongSequenceCount(#[from] WrongSequenceCount),
}

[drmingdrmer]

Reviewed 1 of 50 files at r1, all commit messages.
Reviewable status: 2 of 51 files reviewed, 8 unresolved discussions (waiting on @lichuang)

---

src/meta/api/src/schema_api_sequence_impl.rs line 66 at r2 (raw file):

    let mut trials = txn_backoff(None, func_name!());
    let mut match_seq = MatchSeq::Exact(0);

I did not noticed that this method allows to override existing Sequence.
In such case, MatchSeq should be decided by the CreateOption: if CreateOption is CreateOrReplace, MatchSeq should be GE(0), i.e., allows any existing value of the seq.
Otherwise, MatchSeq should be Exact(0).

With this approach, the loop can be removed.

[drmingdrmer]

Reviewed 1 of 7 files at r2, all commit messages.
Reviewable status: 3 of 51 files reviewed, 11 unresolved discussions (waiting on @lichuang)

---

src/meta/api/src/schema_api_sequence_impl.rs line 56 at r3 (raw file):

use crate::txn_op_put;

pub async fn do_create_sequence(

If you'd like implement it in a separate file. It can be done with a pure-trait implementation.
For example:

pub trait SequenceApi: KVApi {
    async fn create_sequence() -> Result<_,_> {
          // move `do_create_sequence` body here.
    }
    // ...
}

Refer to: KVPbApi:
https://github.com/drmingdrmer/databend/blob/38e20916335824f70c7c553395fd380efd77603c/src/meta/api/src/kv_pb_api/mod.rs#L48

---

src/meta/api/src/schema_api_sequence_impl.rs line 69 at r3 (raw file):

    } else {
        MatchSeq::Exact(0)
    };

Suggestion:

let match_seq = MatchSeq::from(req.create_option);

---

src/meta/api/src/schema_api_sequence_impl.rs line 71 at r3 (raw file):

    };
    let reply = kv_api
        .upsert_kv(UpsertKVReq::new(

Using KVPbApi would make life easier here:

        let upsert = UpsertPB::update(...).with(match_seq);

        let res = kv_api.upsert_pb(&upsert).await?;

---

src/meta/app/src/schema/sequence.rs line 22 at r3 (raw file):

#[derive(Hash, Clone, Debug, PartialEq, Eq)]
pub struct SequenceNameIdent {

Since it is a simple (tenant, string) tuple like struct, define it with TIdent would be simler:
https://github.com/drmingdrmer/databend/blob/3aeb2d0290ac004adda4a8e13306b50e33798d4f/src/meta/app/src/principal/connection_ident.rs#L18

[drmingdrmer]

Reviewed 1 of 7 files at r2, 3 of 8 files at r4, all commit messages.
Reviewable status: 5 of 52 files reviewed, 6 unresolved discussions (waiting on @lichuang)

---

src/meta/app/src/id_generator.rs line 102 at r4 (raw file):

    }

    pub fn sequence_id() -> Self {

This function does not seem to be used.

---

src/meta/api/src/sequence_api_impl.rs line 122 at r4 (raw file):

                return Err(err);
            }
        };

Why not use ? to return the error? get_sequence_or_err(...).await?

Code quote:

        let result = get_sequence_or_err(
            self,
            name_key,
            format!("get_sequence_next_values: {:?}", name_key),
        )
        .await;

        let (_sequence_seq, sequence_meta) = match result {
            Ok((sequence_seq, meta)) => (sequence_seq, meta),
            Err(err) => {
                return Err(err);
            }
        };

---

src/meta/api/src/sequence_api_impl.rs line 213 at r4 (raw file):

    }

    async fn drop_sequence(&self, req: DropSequenceReq) -> Result<DropSequenceReply, KVAppError> {

This method can be implemented with a single upsert too.

[drmingdrmer]

Reviewed 10 of 50 files at r1, 1 of 3 files at r5, all commit messages.
Reviewable status: 16 of 53 files reviewed, 4 unresolved discussions (waiting on @lichuang)

---

src/meta/app/src/schema/sequence.rs line 123 at r6 (raw file):

    impl From<ExistError<Resource>> for ErrorCode {
        fn from(err: ExistError<Resource>) -> Self {
            ErrorCode::ConnectionAlreadyExists(err.to_string())

Wrong Error name

---

src/meta/api/src/sequence_api_impl.rs line 152 at r6 (raw file):

                    return Err(err);
                }
            };

Use .await? ?

Code quote:

            .await;

            let (sequence_seq, mut sequence_meta) = match result {
                Ok((sequence_seq, meta)) => (sequence_seq, meta),
                Err(err) => {
                    return Err(err);
                }
            };

[drmingdrmer]

Reviewed all commit messages.
Reviewable status: 15 of 53 files reviewed, 9 unresolved discussions (waiting on @lichuang)

---

src/meta/app/src/app_error.rs line 1006 at r7 (raw file):

#[derive(thiserror::Error, Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
#[error("DropSequenceError: `{name}`")]
pub struct DropSequenceError {

Similar to BTreeMap::remove(), returning the removed value(an Option<V>) would be enough for the caller to know if drop is successful.

---

src/meta/api/src/sequence_api_impl.rs line 88 at r7 (raw file):

                    SequenceError::SequenceAlreadyExists(SequenceAlreadyExists::new(
                        name_key.sequence_name.clone(),
                        format!("create sequence: {:?}", name_key),

Do not expose tenant info.

---

src/meta/api/src/sequence_api_impl.rs line 96 at r7 (raw file):

                        SequenceError::CreateSequenceError(CreateSequenceError::new(
                            name_key.sequence_name.clone(),
                            format!("replace sequence: {:?} fail", name_key),

Never expose tenant to user in a error: use name_key.name() to retrieve the sequence name.

---

src/meta/api/src/sequence_api_impl.rs line 144 at r7 (raw file):

                self,
                name_key,
                format!("get_sequence_next_values: {:?}", name_key),

Do not expose tenant info.

---

src/meta/api/src/sequence_api_impl.rs line 156 at r7 (raw file):

                        format!(
                            "{:?}: current: {}, count: {}",
                            name_key, sequence_meta.current, count

Do not expose tenant info.

---

src/meta/api/src/sequence_api_impl.rs line 207 at r7 (raw file):

        let result =
            get_sequence_or_err(self, name_key, format!("drop_sequence: {:?}", name_key)).await;

You do not need to get it before dropping. Just drop it.

---

src/meta/api/src/sequence_api_impl.rs line 257 at r7 (raw file):

            SequenceError::UnknownSequence(UnknownSequence::new(
                name_key.sequence_name.clone(),
                format!("{}: {:?}", msg, name_key),

Do not expose tenant info.

[drmingdrmer]

Reviewed 4 of 50 files at r1, 1 of 3 files at r8, all commit messages.
Reviewable status: 20 of 53 files reviewed, 6 unresolved discussions (waiting on @lichuang)

---

src/meta/proto-conv/tests/it/proto_conv.rs line 553 at r8 (raw file):

    }

    // lvt

Invalid comment

---

src/meta/api/src/sequence_api_impl.rs line 207 at r8 (raw file):

        let seq = MatchSeq::GE(0);
        let key = SequenceIdent::new(&name_key.tenant, &name_key.sequence_name);

The name_key should be of type SequenceIdent. There is no need to re use two types for the identifier of a sequence. Remove SequenceNameIdent and keep only SequenceIdent.

[drmingdrmer]

Reviewable status: 20 of 53 files reviewed, 7 unresolved discussions (waiting on @lichuang)

---

src/meta/api/src/sequence_api_impl.rs line 228 at r8 (raw file):

        };

        Ok(DropSequenceReply { prev })

In fact req.if_exists does not affect the behavior of drop_sequence at all.
It just defines how to respond to the caller.

This field if_exists could be removed and let the caller consume if_exists and decide what to do.

Code quote:

        let prev = if let Some(prev) = &reply.prev {
            if !reply.is_changed() {
                if req.if_exists { Some(prev.seq) } else { None }
            } else {
                Some(prev.seq)
            }
        } else {
            None
        };

        Ok(DropSequenceReply { prev })

[drmingdrmer]

Reviewed 1 of 50 files at r1, 1 of 8 files at r4, 6 of 13 files at r9, all commit messages.
Reviewable status: 23 of 53 files reviewed, 8 unresolved discussions (waiting on @lichuang)

---

src/meta/app/src/schema/sequence.rs line 27 at r9 (raw file):

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct SequenceMeta {

Remove serde derive if it is not required.

Code quote:

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct SequenceMeta {

---

src/meta/app/src/schema/sequence.rs line 54 at r9 (raw file):

    pub tenant: Tenant,
    pub sequence_name: String,
    pub create_on: DateTime<Utc>,

Use SequenceIdent to replace tenant, sequence_name. make it more structural.

Suggestion:

    pub sequence_ident: SequenceIdent,
    pub create_on: DateTime<Utc>,

---

src/meta/app/src/schema/sequence.rs line 64 at r9 (raw file):

pub struct GetSequenceNextValueReq {
    pub tenant: Tenant,
    pub sequence_name: String,

use SequenceIdent.

Code quote:

    pub tenant: Tenant,
    pub sequence_name: String,

---

src/meta/app/src/schema/sequence.rs line 79 at r9 (raw file):

pub struct GetSequenceReq {
    pub tenant: Tenant,
    pub sequence_name: String,

use SequenceIdent.

Code quote:

    pub tenant: Tenant,
    pub sequence_name: String,

---

src/meta/app/src/schema/sequence.rs line 91 at r9 (raw file):

    pub if_exists: bool,
    pub tenant: Tenant,
    pub sequence_name: String,

use SequenceIdent.

Code quote:

    pub tenant: Tenant,
    pub sequence_name: String,

---

src/meta/api/src/sequence_api_impl.rs line 212 at r9 (raw file):

        let seq = MatchSeq::GE(0);
        let key = SequenceIdent::new(tenant, sequence_name);
        let reply = self.upsert_pb(&UpsertPB::delete(key).with(seq)).await?;

with(seq) can be removed.

[drmingdrmer]

Reviewed 4 of 50 files at r1, 1 of 7 files at r2, 1 of 8 files at r4.
Reviewable status: 28 of 53 files reviewed, 10 unresolved discussions (waiting on @lichuang)

---

src/query/sql/src/planner/plans/ddl/sequence.rs line 22 at r9 (raw file):

    pub create_option: CreateOption,
    pub tenant: Tenant,
    pub sequence: String,

Use SequenceIdent.

Code quote:

    pub tenant: Tenant,
    pub sequence: String,

---

src/query/sql/src/planner/plans/ddl/sequence.rs line 29 at r9 (raw file):

pub struct DropSequencePlan {
    pub tenant: Tenant,
    pub sequence: String,

Use SequenceIdent.

Code quote:

    pub tenant: Tenant,
    pub sequence: String,

[drmingdrmer]

Reviewed 7 of 50 files at r1, 1 of 8 files at r4, 1 of 3 files at r6, 2 of 13 files at r9, 3 of 5 files at r10, all commit messages.
Reviewable status: 42 of 53 files reviewed, 9 unresolved discussions (waiting on @lichuang)

---

src/meta/api/src/sequence_api_impl.rs line 67 at r10 (raw file):

        let tenant = req.ident.tenant();
        let sequence_name = req.ident.name();

Needless reassignment.

Code quote:

        let tenant = req.ident.tenant();
        let sequence_name = req.ident.name();

---

src/meta/api/src/sequence_api_impl.rs line 71 at r10 (raw file):

        let seq = MatchSeq::from(req.create_option);
        let key = SequenceIdent::new(tenant, sequence_name);

Just use req.ident.

Suggestion:

        let key = req.indeng;

---

src/meta/api/src/sequence_api_impl.rs line 78 at r10 (raw file):

        debug!(
            tenant :?= (tenant),
            sequence_name :? =(sequence_name),

Suggestion:

            ident = (req.ident),

---

src/meta/api/src/sequence_api_impl.rs line 173 at r10 (raw file):

                current :? =(&sequence_meta.current),
                tenant :?= (tenant),
                sequence_name :? =(sequence_name);

Suggestion:

            ident = (req.ident),

---

src/meta/api/src/sequence_api_impl.rs line 188 at r10 (raw file):

                current :? =(&sequence_meta.current),
                tenant :?= (tenant),
                sequence_name :? =(sequence_name),

Suggestion:

            ident = (req.ident),

---

src/meta/api/src/sequence_api_impl.rs line 210 at r10 (raw file):

        debug!(
            tenant :?= (req.ident.tenant()),
            sequence_name :? =(req.ident.name()),

Suggestion:

            ident = (req.ident),

[drmingdrmer]

Reviewed 2 of 50 files at r1, 1 of 13 files at r9, 2 of 5 files at r11, all commit messages.
Reviewable status: 46 of 53 files reviewed, 4 unresolved discussions (waiting on @lichuang)

---

src/query/service/src/interpreters/interpreter_sequence_drop.rs line 54 at r11 (raw file):

        };
        let catalog = self.ctx.get_default_catalog()?;
        let _reply = catalog.drop_sequence(req).await?;

if _reply.prev.is_none() && !self.plan.if_exists {
    return Err(...)
}

---

tests/sqllogictests/suites/base/05_ddl/05_0036_sequence.test line 8 at r11 (raw file):

statement ok
DROP SEQUENCE seq

Need to test DROP IF EXISTS

[drmingdrmer]

The DROP IF EXISTS is not tested:

• Test DROP IF EXIST for SEQUENCE #15263

Reviewable status: 46 of 53 files reviewed, 2 unresolved discussions (waiting on @lichuang)