common_meta/

error.rs

// Copyright 2023 Greptime Team
//
// 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.

use std::str::Utf8Error;
use std::sync::Arc;

use common_error::ext::{BoxedError, ErrorExt, RetryHint};
use common_error::status_code::StatusCode;
use common_macro::stack_trace_debug;
use common_procedure::ProcedureId;
use common_wal::kafka::rskafka_client_error_to_retry_hint;
use object_store::error::retry_hint_from_opendal_error;
use serde_json::error::Error as JsonError;
use snafu::{Location, Snafu};
use store_api::storage::RegionId;
use table::metadata::TableId;

use crate::DatanodeId;
use crate::peer::Peer;

mod retry_hint;

pub use retry_hint::retry_hint_from_etcd_error;
#[cfg(feature = "mysql_kvbackend")]
pub use retry_hint::retry_hint_from_sqlx_error;
#[cfg(feature = "pg_kvbackend")]
pub use retry_hint::{retry_hint_from_postgres_error, retry_hint_from_postgres_pool_error};

#[derive(Snafu)]
#[snafu(visibility(pub))]
#[stack_trace_debug]
pub enum Error {
    #[snafu(display("Empty key is not allowed"))]
    EmptyKey {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Another procedure is operating the region: {} on peer: {}",
        region_id,
        peer_id
    ))]
    RegionOperatingRace {
        #[snafu(implicit)]
        location: Location,
        peer_id: DatanodeId,
        region_id: RegionId,
    },

    #[snafu(display("Failed to connect to Etcd"))]
    ConnectEtcd {
        #[snafu(source)]
        error: etcd_client::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to execute via Etcd"))]
    EtcdFailed {
        #[snafu(source)]
        error: etcd_client::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to execute {} txn operations via Etcd", max_operations))]
    EtcdTxnFailed {
        max_operations: usize,
        #[snafu(source)]
        error: etcd_client::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to get sequence: {}", err_msg))]
    NextSequence {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Unexpected sequence value: {}", err_msg))]
    UnexpectedSequenceValue {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Table info not found: {}", table))]
    TableInfoNotFound {
        table: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to register procedure loader, type name: {}", type_name))]
    RegisterProcedureLoader {
        type_name: String,
        #[snafu(implicit)]
        location: Location,
        source: common_procedure::error::Error,
    },

    #[snafu(display("Failed to register repartition procedure loader"))]
    RegisterRepartitionProcedureLoader {
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Failed to create repartition procedure"))]
    CreateRepartitionProcedure {
        source: BoxedError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to persist the repartition GC requirement"))]
    PersistRepartitionGcRequirement {
        source: BoxedError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to submit procedure"))]
    SubmitProcedure {
        #[snafu(implicit)]
        location: Location,
        source: common_procedure::Error,
    },

    #[snafu(display("Failed to query procedure"))]
    QueryProcedure {
        #[snafu(implicit)]
        location: Location,
        source: common_procedure::Error,
    },

    #[snafu(display("Procedure not found: {pid}"))]
    ProcedureNotFound {
        #[snafu(implicit)]
        location: Location,
        pid: String,
    },

    #[snafu(display("Failed to parse procedure id: {key}"))]
    ParseProcedureId {
        #[snafu(implicit)]
        location: Location,
        key: String,
        #[snafu(source)]
        error: common_procedure::ParseIdError,
    },

    #[snafu(display("Unsupported operation {}", operation))]
    Unsupported {
        operation: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Trying to write to a read-only kv backend: {}", name))]
    ReadOnlyKvBackend {
        name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to get procedure state receiver, procedure id: {procedure_id}"))]
    ProcedureStateReceiver {
        procedure_id: ProcedureId,
        #[snafu(implicit)]
        location: Location,
        source: common_procedure::Error,
    },

    #[snafu(display("Procedure state receiver not found: {procedure_id}"))]
    ProcedureStateReceiverNotFound {
        procedure_id: ProcedureId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to wait procedure done"))]
    WaitProcedure {
        #[snafu(implicit)]
        location: Location,
        source: common_procedure::Error,
    },

    #[snafu(display("Failed to start procedure manager"))]
    StartProcedureManager {
        #[snafu(implicit)]
        location: Location,
        source: common_procedure::Error,
    },

    #[snafu(display("Failed to stop procedure manager"))]
    StopProcedureManager {
        #[snafu(implicit)]
        location: Location,
        source: common_procedure::Error,
    },

    #[snafu(display(
        "Failed to get procedure output, procedure id: {procedure_id}, error: {err_msg}"
    ))]
    ProcedureOutput {
        procedure_id: String,
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Primary key '{key}' not found when creating region request"))]
    PrimaryKeyNotFound {
        key: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to build table meta for table: {}", table_name))]
    BuildTableMeta {
        table_name: String,
        #[snafu(source)]
        error: table::metadata::TableMetaBuilderError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Table occurs error"))]
    Table {
        #[snafu(implicit)]
        location: Location,
        source: table::error::Error,
    },

    #[snafu(display("Failed to find table route for table id {}", table_id))]
    TableRouteNotFound {
        table_id: TableId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to find table repartition metadata for table id {}", table_id))]
    TableRepartNotFound {
        table_id: TableId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to decode protobuf"))]
    DecodeProto {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: prost::DecodeError,
    },

    #[snafu(display("Failed to decode packed file references"))]
    DecodePackedFileRefs {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: base64::DecodeError,
    },

    #[snafu(display("Invalid packed file references framing"))]
    InvalidPackedFileRefs {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to encode object into json"))]
    EncodeJson {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: JsonError,
    },

    #[snafu(display("Failed to decode object from json"))]
    DecodeJson {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: JsonError,
    },

    #[snafu(display("Failed to serialize to json: {}", input))]
    SerializeToJson {
        input: String,
        #[snafu(source)]
        error: serde_json::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to deserialize from json: {}", input))]
    DeserializeFromJson {
        input: String,
        #[snafu(source)]
        error: serde_json::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Payload not exist"))]
    PayloadNotExist {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to serde json"))]
    SerdeJson {
        #[snafu(source)]
        error: serde_json::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to parse value {} into key {}", value, key))]
    ParseOption {
        key: String,
        value: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Conflicting schema options: {}={} and {}={}",
        first_key,
        first_value,
        second_key,
        second_value
    ))]
    ConflictingSchemaOptions {
        first_key: String,
        first_value: String,
        second_key: String,
        second_value: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Illegal state from server, code: {}, error: {}", code, err_msg))]
    IllegalServerState {
        code: i32,
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to convert alter table request"))]
    ConvertAlterTableRequest {
        source: common_grpc_expr::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid protobuf message: {err_msg}"))]
    InvalidProtoMsg {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Unexpected: {err_msg}"))]
    Unexpected {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Metasrv election has no leader at this moment"))]
    ElectionNoLeader {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Metasrv election leader lease expired"))]
    ElectionLeaderLeaseExpired {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Metasrv election leader lease changed during election"))]
    ElectionLeaderLeaseChanged {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Table already exists, table: {}", table_name))]
    TableAlreadyExists {
        table_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Cannot drop table '{}': an older tombstone already uses the same full name",
        table_name
    ))]
    /// Raised when a live table is recreated with a name still reserved by an older tombstone.
    TableNameTombstoneConflict {
        table_name: String,
        existing_table_id: TableId,
        dropping_table_id: TableId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("View already exists, view: {}", view_name))]
    ViewAlreadyExists {
        view_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Flow already exists: {}", flow_name))]
    FlowAlreadyExists {
        flow_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Schema already exists, catalog:{}, schema: {}", catalog, schema))]
    SchemaAlreadyExists {
        catalog: String,
        schema: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to convert raw key to str"))]
    ConvertRawKey {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: Utf8Error,
    },

    #[snafu(display("Table not found: '{}'", table_name))]
    TableNotFound {
        table_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region not found: {}", region_id))]
    RegionNotFound {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("View not found: '{}'", view_name))]
    ViewNotFound {
        view_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Flow not found: '{}'", flow_name))]
    FlowNotFound {
        flow_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Flow route not found: '{}'", flow_name))]
    FlowRouteNotFound {
        flow_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Schema nod found, schema: {}", table_schema))]
    SchemaNotFound {
        table_schema: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Catalog not found, catalog: {}", catalog))]
    CatalogNotFound {
        catalog: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid metadata, err: {}", err_msg))]
    InvalidMetadata {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid view info, err: {}", err_msg))]
    InvalidViewInfo {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid flow request body: {:?}", body))]
    InvalidFlowRequestBody {
        body: Box<Option<api::v1::flow::flow_request::Body>>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to get kv cache, err: {}", err_msg))]
    GetKvCache { err_msg: String },

    #[snafu(display("Get null from cache, key: {}", key))]
    CacheNotGet {
        key: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Etcd txn error: {err_msg}"))]
    EtcdTxnOpResponse {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("External error"))]
    External {
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("The response exceeded size limit"))]
    ResponseExceededSizeLimit {
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Invalid heartbeat response"))]
    InvalidHeartbeatResponse {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to operate on datanode: {}", peer))]
    OperateDatanode {
        #[snafu(implicit)]
        location: Location,
        peer: Peer,
        source: BoxedError,
    },

    #[snafu(display("Retry later"))]
    RetryLater {
        source: BoxedError,
        clean_poisons: bool,
    },

    #[snafu(display("Abort procedure"))]
    AbortProcedure {
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
        clean_poisons: bool,
    },

    #[snafu(display("Failed to serialize WAL options for region: {region_id}"))]
    SerializeWalOptions {
        region_id: RegionId,
        #[snafu(source)]
        error: serde_json::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid number of topics {}", num_topics))]
    InvalidNumTopics {
        num_topics: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Failed to build a Kafka client, broker endpoints: {:?}",
        broker_endpoints
    ))]
    BuildKafkaClient {
        broker_endpoints: Vec<String>,
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: rskafka::client::error::Error,
    },

    #[snafu(display("Failed to create TLS Config"))]
    TlsConfig {
        #[snafu(implicit)]
        location: Location,
        source: common_wal::error::Error,
    },

    #[snafu(display("Failed to build a Kafka controller client"))]
    BuildKafkaCtrlClient {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: rskafka::client::error::Error,
    },

    #[snafu(display(
        "Failed to get a Kafka partition client, topic: {}, partition: {}",
        topic,
        partition
    ))]
    KafkaPartitionClient {
        topic: String,
        partition: i32,
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: rskafka::client::error::Error,
    },

    #[snafu(display(
        "Failed to get offset from Kafka, topic: {}, partition: {}",
        topic,
        partition
    ))]
    KafkaGetOffset {
        topic: String,
        partition: i32,
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: rskafka::client::error::Error,
    },

    #[snafu(display("Failed to produce records to Kafka, topic: {}", topic))]
    ProduceRecord {
        topic: String,
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: rskafka::client::error::Error,
    },

    #[snafu(display("Failed to create a Kafka wal topic"))]
    CreateKafkaWalTopic {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: rskafka::client::error::Error,
    },

    #[snafu(display("The topic pool is empty"))]
    EmptyTopicPool {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Unexpected table route type: {}", err_msg))]
    UnexpectedLogicalRouteTable {
        #[snafu(implicit)]
        location: Location,
        err_msg: String,
    },

    #[snafu(display("The tasks of {} cannot be empty", name))]
    EmptyDdlTasks {
        name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Metadata corruption: {}", err_msg))]
    MetadataCorruption {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Alter logical tables invalid arguments: {}", err_msg))]
    AlterLogicalTablesInvalidArguments {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Create logical tables invalid arguments: {}", err_msg))]
    CreateLogicalTablesInvalidArguments {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid node info key: {}", key))]
    InvalidNodeInfoKey {
        key: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid node stat key: {}", key))]
    InvalidStatKey {
        key: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to parse number: {}", err_msg))]
    ParseNum {
        err_msg: String,
        #[snafu(source)]
        error: std::num::ParseIntError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid role: {}", role))]
    InvalidRole {
        role: i32,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid set database option, key: {}, value: {}", key, value))]
    InvalidSetDatabaseOption {
        key: String,
        value: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid unset database option, key: {}", key))]
    InvalidUnsetDatabaseOption {
        key: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid prefix: {}, key: {}", prefix, key))]
    MismatchPrefix {
        prefix: String,
        key: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to move values: {err_msg}"))]
    MoveValues {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to restore tombstone, target key already exists: {key}"))]
    TombstoneTargetAlreadyExists {
        key: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to parse {} from utf8", name))]
    FromUtf8 {
        name: String,
        #[snafu(source)]
        error: std::string::FromUtf8Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Value not exists"))]
    ValueNotExist {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to get cache"))]
    GetCache { source: Arc<Error> },

    #[snafu(display(
        "Failed to get latest cache value after {} attempts due to concurrent invalidation",
        attempts
    ))]
    GetLatestCacheRetryExceeded {
        attempts: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "pg_kvbackend")]
    #[snafu(display("Failed to execute via Postgres, sql: {}", sql))]
    PostgresExecution {
        sql: String,
        #[snafu(source)]
        error: tokio_postgres::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "pg_kvbackend")]
    #[snafu(display("Failed to create connection pool for Postgres"))]
    CreatePostgresPool {
        #[snafu(source)]
        error: deadpool_postgres::CreatePoolError,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "pg_kvbackend")]
    #[snafu(display("Failed to get Postgres connection from pool: {}", reason))]
    GetPostgresConnection {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "pg_kvbackend")]
    #[snafu(display("Failed to get Postgres client"))]
    GetPostgresClient {
        #[snafu(source)]
        error: deadpool::managed::PoolError<tokio_postgres::Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "pg_kvbackend")]
    #[snafu(display("Failed to {} Postgres transaction", operation))]
    PostgresTransaction {
        #[snafu(source)]
        error: tokio_postgres::Error,
        #[snafu(implicit)]
        location: Location,
        operation: String,
    },

    #[cfg(feature = "pg_kvbackend")]
    #[snafu(display("Failed to setup PostgreSQL TLS configuration: {}", reason))]
    PostgresTlsConfig {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to load TLS certificate from path: {}", path))]
    LoadTlsCertificate {
        path: String,
        #[snafu(source)]
        error: std::io::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "pg_kvbackend")]
    #[snafu(display("Invalid TLS configuration: {}", reason))]
    InvalidTlsConfig {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "mysql_kvbackend")]
    #[snafu(display("Failed to execute via MySql, sql: {}", sql))]
    MySqlExecution {
        sql: String,
        #[snafu(source)]
        error: sqlx::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "mysql_kvbackend")]
    #[snafu(display("Failed to create connection pool for MySql"))]
    CreateMySqlPool {
        #[snafu(source)]
        error: sqlx::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "mysql_kvbackend")]
    #[snafu(display("Failed to decode sql value"))]
    DecodeSqlValue {
        #[snafu(source)]
        error: sqlx::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "mysql_kvbackend")]
    #[snafu(display("Failed to acquire mysql client from pool"))]
    AcquireMySqlClient {
        #[snafu(source)]
        error: sqlx::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "mysql_kvbackend")]
    #[snafu(display("Failed to {} MySql transaction", operation))]
    MySqlTransaction {
        #[snafu(source)]
        error: sqlx::Error,
        #[snafu(implicit)]
        location: Location,
        operation: String,
    },

    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
    #[snafu(display("Rds transaction retry failed"))]
    RdsTransactionRetryFailed {
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
    #[snafu(display("Sql execution timeout, sql: {}, duration: {:?}", sql, duration))]
    SqlExecutionTimeout {
        sql: String,
        duration: std::time::Duration,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Datanode table info not found, table id: {}, datanode id: {}",
        table_id,
        datanode_id
    ))]
    DatanodeTableInfoNotFound {
        datanode_id: DatanodeId,
        table_id: TableId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid topic name prefix: {}", prefix))]
    InvalidTopicNamePrefix {
        prefix: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("No leader found for table_id: {}", table_id))]
    NoLeader {
        table_id: TableId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Procedure poison key already exists with a different value, key: {}, value: {}",
        key,
        value
    ))]
    ProcedurePoisonConflict {
        key: String,
        value: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to put poison, table metadata may be corrupted"))]
    PutPoison {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        source: common_procedure::error::Error,
    },

    #[snafu(display("Invalid file path: {}", file_path))]
    InvalidFilePath {
        #[snafu(implicit)]
        location: Location,
        file_path: String,
    },

    #[snafu(display("Failed to serialize flexbuffers"))]
    SerializeFlexbuffers {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: flexbuffers::SerializationError,
    },

    #[snafu(display("Failed to deserialize flexbuffers"))]
    DeserializeFlexbuffers {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: flexbuffers::DeserializationError,
    },

    #[snafu(display("Failed to read flexbuffers"))]
    ReadFlexbuffers {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: flexbuffers::ReaderError,
    },

    #[snafu(display("Invalid file name: {}", reason))]
    InvalidFileName {
        #[snafu(implicit)]
        location: Location,
        reason: String,
    },

    #[snafu(display("Invalid file extension: {}", reason))]
    InvalidFileExtension {
        #[snafu(implicit)]
        location: Location,
        reason: String,
    },

    #[snafu(display("Failed to write object, file path: {}", file_path))]
    WriteObject {
        #[snafu(implicit)]
        location: Location,
        file_path: String,
        #[snafu(source)]
        error: object_store::Error,
    },

    #[snafu(display("Failed to read object, file path: {}", file_path))]
    ReadObject {
        #[snafu(implicit)]
        location: Location,
        file_path: String,
        #[snafu(source)]
        error: object_store::Error,
    },

    #[snafu(display("Missing column ids"))]
    MissingColumnIds {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Missing column in column metadata: {}, table: {}, table_id: {}",
        column_name,
        table_name,
        table_id,
    ))]
    MissingColumnInColumnMetadata {
        column_name: String,
        #[snafu(implicit)]
        location: Location,
        table_name: String,
        table_id: TableId,
    },

    #[snafu(display(
        "Mismatch column id: column_name: {}, column_id: {}, table: {}, table_id: {}",
        column_name,
        column_id,
        table_name,
        table_id,
    ))]
    MismatchColumnId {
        column_name: String,
        column_id: u32,
        #[snafu(implicit)]
        location: Location,
        table_name: String,
        table_id: TableId,
    },

    #[snafu(display("Failed to convert column def, column: {}", column))]
    ConvertColumnDef {
        column: String,
        #[snafu(implicit)]
        location: Location,
        source: api::error::Error,
    },

    #[snafu(display("Failed to convert time ranges"))]
    ConvertTimeRanges {
        #[snafu(implicit)]
        location: Location,
        source: api::error::Error,
    },

    #[snafu(display(
        "Column metadata inconsistencies found in table: {}, table_id: {}",
        table_name,
        table_id
    ))]
    ColumnMetadataConflicts {
        table_name: String,
        table_id: TableId,
    },

    #[snafu(display(
        "Column not found in column metadata, column_name: {}, column_id: {}",
        column_name,
        column_id
    ))]
    ColumnNotFound { column_name: String, column_id: u32 },

    #[snafu(display(
        "Column id mismatch, column_name: {}, expected column_id: {}, actual column_id: {}",
        column_name,
        expected_column_id,
        actual_column_id
    ))]
    ColumnIdMismatch {
        column_name: String,
        expected_column_id: u32,
        actual_column_id: u32,
    },

    #[snafu(display(
        "Timestamp column mismatch, expected column_name: {}, expected column_id: {}, actual column_name: {}, actual column_id: {}",
        expected_column_name,
        expected_column_id,
        actual_column_name,
        actual_column_id,
    ))]
    TimestampMismatch {
        expected_column_name: String,
        expected_column_id: u32,
        actual_column_name: String,
        actual_column_id: u32,
    },

    #[cfg(feature = "enterprise")]
    #[snafu(display("Too large duration"))]
    TooLargeDuration {
        #[snafu(source)]
        error: prost_types::DurationError,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "enterprise")]
    #[snafu(display("Negative duration"))]
    NegativeDuration {
        #[snafu(source)]
        error: prost_types::DurationError,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "enterprise")]
    #[snafu(display("Missing interval field"))]
    MissingInterval {
        #[snafu(implicit)]
        location: Location,
    },
}

pub type Result<T> = std::result::Result<T, Error>;

impl ErrorExt for Error {
    fn status_code(&self) -> StatusCode {
        use Error::*;
        match self {
            IllegalServerState { .. }
            | EtcdTxnOpResponse { .. }
            | EtcdFailed { .. }
            | EtcdTxnFailed { .. }
            | ConnectEtcd { .. }
            | MoveValues { .. }
            | TombstoneTargetAlreadyExists { .. }
            | GetCache { .. }
            | GetLatestCacheRetryExceeded { .. }
            | SerializeToJson { .. }
            | DeserializeFromJson { .. }
            | ElectionNoLeader { .. }
            | ElectionLeaderLeaseExpired { .. }
            | ElectionLeaderLeaseChanged { .. } => StatusCode::Internal,

            NoLeader { .. } => StatusCode::TableUnavailable,
            ValueNotExist { .. }
            | ProcedurePoisonConflict { .. }
            | ProcedureStateReceiverNotFound { .. }
            | MissingColumnIds { .. }
            | MissingColumnInColumnMetadata { .. }
            | MismatchColumnId { .. }
            | ColumnMetadataConflicts { .. }
            | ColumnNotFound { .. }
            | ColumnIdMismatch { .. }
            | TimestampMismatch { .. } => StatusCode::Unexpected,

            Unsupported { .. } | ReadOnlyKvBackend { .. } => StatusCode::Unsupported,
            WriteObject { .. } | ReadObject { .. } => StatusCode::StorageUnavailable,

            SerdeJson { .. }
            | ParseOption { .. }
            | InvalidProtoMsg { .. }
            | InvalidMetadata { .. }
            | Unexpected { .. }
            | TableInfoNotFound { .. }
            | NextSequence { .. }
            | UnexpectedSequenceValue { .. }
            | InvalidHeartbeatResponse { .. }
            | EncodeJson { .. }
            | DecodeJson { .. }
            | PayloadNotExist { .. }
            | ConvertRawKey { .. }
            | DecodeProto { .. }
            | DecodePackedFileRefs { .. }
            | InvalidPackedFileRefs { .. }
            | BuildTableMeta { .. }
            | TableRouteNotFound { .. }
            | TableRepartNotFound { .. }
            | RegionOperatingRace { .. }
            | SerializeWalOptions { .. }
            | BuildKafkaClient { .. }
            | BuildKafkaCtrlClient { .. }
            | KafkaPartitionClient { .. }
            | ProduceRecord { .. }
            | CreateKafkaWalTopic { .. }
            | EmptyTopicPool { .. }
            | UnexpectedLogicalRouteTable { .. }
            | ProcedureOutput { .. }
            | FromUtf8 { .. }
            | MetadataCorruption { .. }
            | KafkaGetOffset { .. }
            | ReadFlexbuffers { .. }
            | SerializeFlexbuffers { .. }
            | DeserializeFlexbuffers { .. }
            | ConvertTimeRanges { .. } => StatusCode::Unexpected,

            GetKvCache { .. } | CacheNotGet { .. } => StatusCode::Internal,

            SchemaAlreadyExists { .. } => StatusCode::DatabaseAlreadyExists,

            ProcedureNotFound { .. }
            | InvalidViewInfo { .. }
            | PrimaryKeyNotFound { .. }
            | EmptyKey { .. }
            | AlterLogicalTablesInvalidArguments { .. }
            | CreateLogicalTablesInvalidArguments { .. }
            | MismatchPrefix { .. }
            | TlsConfig { .. }
            | InvalidSetDatabaseOption { .. }
            | InvalidUnsetDatabaseOption { .. }
            | InvalidTopicNamePrefix { .. }
            | InvalidFileExtension { .. }
            | InvalidFileName { .. }
            | InvalidFlowRequestBody { .. }
            | InvalidFilePath { .. }
            | ConflictingSchemaOptions { .. } => StatusCode::InvalidArguments,

            #[cfg(feature = "enterprise")]
            MissingInterval { .. } | NegativeDuration { .. } | TooLargeDuration { .. } => {
                StatusCode::InvalidArguments
            }

            FlowNotFound { .. } => StatusCode::FlowNotFound,
            FlowRouteNotFound { .. } => StatusCode::Unexpected,
            FlowAlreadyExists { .. } => StatusCode::FlowAlreadyExists,

            ViewNotFound { .. } | TableNotFound { .. } | RegionNotFound { .. } => {
                StatusCode::TableNotFound
            }
            ViewAlreadyExists { .. }
            | TableAlreadyExists { .. }
            | TableNameTombstoneConflict { .. } => StatusCode::TableAlreadyExists,

            SubmitProcedure { source, .. }
            | QueryProcedure { source, .. }
            | WaitProcedure { source, .. }
            | StartProcedureManager { source, .. }
            | StopProcedureManager { source, .. } => source.status_code(),
            RegisterProcedureLoader { source, .. } => source.status_code(),
            External { source, .. } => source.status_code(),
            ResponseExceededSizeLimit { source, .. } => source.status_code(),
            OperateDatanode { source, .. } => source.status_code(),
            Table { source, .. } => source.status_code(),
            RetryLater { source, .. } => source.status_code(),
            AbortProcedure { source, .. } => source.status_code(),
            ConvertAlterTableRequest { source, .. } => source.status_code(),
            PutPoison { source, .. } => source.status_code(),
            ConvertColumnDef { source, .. } => source.status_code(),
            ProcedureStateReceiver { source, .. } => source.status_code(),
            RegisterRepartitionProcedureLoader { source, .. } => source.status_code(),
            CreateRepartitionProcedure { source, .. } => source.status_code(),
            PersistRepartitionGcRequirement { source, .. } => source.status_code(),

            ParseProcedureId { .. }
            | InvalidNumTopics { .. }
            | SchemaNotFound { .. }
            | CatalogNotFound { .. }
            | InvalidNodeInfoKey { .. }
            | InvalidStatKey { .. }
            | ParseNum { .. }
            | InvalidRole { .. }
            | EmptyDdlTasks { .. } => StatusCode::InvalidArguments,

            LoadTlsCertificate { .. } => StatusCode::Internal,

            #[cfg(feature = "pg_kvbackend")]
            PostgresExecution { .. }
            | CreatePostgresPool { .. }
            | GetPostgresConnection { .. }
            | GetPostgresClient { .. }
            | PostgresTransaction { .. }
            | PostgresTlsConfig { .. }
            | InvalidTlsConfig { .. } => StatusCode::Internal,
            #[cfg(feature = "mysql_kvbackend")]
            MySqlExecution { .. }
            | CreateMySqlPool { .. }
            | DecodeSqlValue { .. }
            | AcquireMySqlClient { .. }
            | MySqlTransaction { .. } => StatusCode::Internal,
            #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
            RdsTransactionRetryFailed { .. } | SqlExecutionTimeout { .. } => StatusCode::Internal,
            DatanodeTableInfoNotFound { .. } => StatusCode::Internal,
        }
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    fn retry_hint(&self) -> RetryHint {
        use Error::*;

        match self {
            RetryLater { .. }
            | GetLatestCacheRetryExceeded { .. }
            | NoLeader { .. }
            | ElectionNoLeader { .. }
            | ElectionLeaderLeaseExpired { .. }
            | ElectionLeaderLeaseChanged { .. } => RetryHint::Retryable,
            ConnectEtcd { error, .. } | EtcdFailed { error, .. } | EtcdTxnFailed { error, .. } => {
                retry_hint_from_etcd_error(error)
            }
            WriteObject { error, .. } | ReadObject { error, .. } => {
                retry_hint_from_opendal_error(error)
            }
            BuildKafkaClient { error, .. }
            | BuildKafkaCtrlClient { error, .. }
            | KafkaPartitionClient { error, .. }
            | KafkaGetOffset { error, .. }
            | ProduceRecord { error, .. }
            | CreateKafkaWalTopic { error, .. } => rskafka_client_error_to_retry_hint(error),
            SubmitProcedure { source, .. }
            | QueryProcedure { source, .. }
            | WaitProcedure { source, .. }
            | StartProcedureManager { source, .. }
            | StopProcedureManager { source, .. }
            | RegisterProcedureLoader { source, .. }
            | PutPoison { source, .. }
            | ProcedureStateReceiver { source, .. } => source.retry_hint(),
            External { source, .. }
            | ResponseExceededSizeLimit { source, .. }
            | OperateDatanode { source, .. }
            | AbortProcedure { source, .. }
            | RegisterRepartitionProcedureLoader { source, .. }
            | CreateRepartitionProcedure { source, .. }
            | PersistRepartitionGcRequirement { source, .. } => source.retry_hint(),
            Table { source, .. } => source.retry_hint(),
            ConvertAlterTableRequest { source, .. } => source.retry_hint(),
            ConvertColumnDef { source, .. } => source.retry_hint(),
            GetCache { source, .. } => source.retry_hint(),
            #[cfg(feature = "pg_kvbackend")]
            PostgresExecution { error, .. } | PostgresTransaction { error, .. } => {
                retry_hint_from_postgres_error(error)
            }
            #[cfg(feature = "pg_kvbackend")]
            GetPostgresClient { error, .. } => retry_hint_from_postgres_pool_error(error),
            #[cfg(feature = "mysql_kvbackend")]
            MySqlExecution { error, .. }
            | CreateMySqlPool { error, .. }
            | AcquireMySqlClient { error, .. }
            | MySqlTransaction { error, .. } => retry_hint_from_sqlx_error(error),
            #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
            RdsTransactionRetryFailed { .. } | SqlExecutionTimeout { .. } => RetryHint::Retryable,
            _ => RetryHint::NonRetryable,
        }
    }
}

impl Error {
    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
    /// Check if the error is a serialization error.
    pub fn is_serialization_error(&self) -> bool {
        match self {
            #[cfg(feature = "pg_kvbackend")]
            Error::PostgresExecution { error, .. } | Error::PostgresTransaction { error, .. } => {
                retry_hint::is_postgres_serialization_error(error)
            }
            #[cfg(feature = "mysql_kvbackend")]
            Error::MySqlExecution { error, .. } | Error::MySqlTransaction { error, .. } => {
                retry_hint::is_mysql_serialization_error(error)
            }
            _ => false,
        }
    }

    /// Creates a new [Error::RetryLater] error from source `err`.
    pub fn retry_later<E: ErrorExt + Send + Sync + 'static>(err: E) -> Error {
        Error::RetryLater {
            source: BoxedError::new(err),
            clean_poisons: false,
        }
    }

    /// Determine whether it is a retry later type through [StatusCode]
    pub fn is_retry_later(&self) -> bool {
        matches!(
            self,
            Error::RetryLater { .. } | Error::GetLatestCacheRetryExceeded { .. }
        )
    }

    /// Determine whether it needs to clean poisons.
    pub fn need_clean_poisons(&self) -> bool {
        matches!(
            self,
            Error::AbortProcedure { clean_poisons, .. } if *clean_poisons
        ) || matches!(
            self,
            Error::RetryLater { clean_poisons, .. } if *clean_poisons
        )
    }

    /// Returns true if the response exceeds the size limit.
    pub fn is_exceeded_size_limit(&self) -> bool {
        match self {
            Error::EtcdFailed {
                error: etcd_client::Error::GRpcStatus(status),
                ..
            } => status.code() == tonic::Code::OutOfRange,
            Error::ResponseExceededSizeLimit { .. } => true,
            _ => false,
        }
    }
}

#[cfg(test)]
mod retry_hint_tests {
    use std::sync::Arc;

    use common_error::mock::MockError;

    use super::*;

    #[test]
    fn test_retry_later_hint_is_retryable() {
        let err = Error::retry_later(MockError::new(StatusCode::Internal));

        assert_eq!(err.retry_hint(), RetryHint::Retryable);
    }

    #[test]
    fn test_latest_cache_retry_exceeded_hint_is_retryable() {
        let err = GetLatestCacheRetryExceededSnafu { attempts: 3_usize }.build();

        assert_eq!(err.retry_hint(), RetryHint::Retryable);
    }

    #[test]
    fn test_get_cache_forwards_retry_hint() {
        let source = Arc::new(Error::retry_later(MockError::new(StatusCode::Internal)));
        let err = Error::GetCache { source };

        assert_eq!(err.retry_hint(), RetryHint::Retryable);
    }

    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
    #[test]
    fn test_sql_execution_timeout_hint_is_retryable() {
        let err = SqlExecutionTimeoutSnafu {
            sql: "SELECT 1".to_string(),
            duration: std::time::Duration::from_secs(1),
        }
        .build();

        assert_eq!(err.retry_hint(), RetryHint::Retryable);
    }

    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
    #[test]
    fn test_rds_transaction_retry_failed_hint_is_retryable() {
        let err = RdsTransactionRetryFailedSnafu.build();

        assert_eq!(err.retry_hint(), RetryHint::Retryable);
    }

    #[test]
    fn test_default_hint_is_non_retryable() {
        let err = UnexpectedSnafu {
            err_msg: "mock error",
        }
        .build();

        assert_eq!(err.retry_hint(), RetryHint::NonRetryable);
    }
}