mito2/

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::any::Any;
use std::sync::Arc;

use common_datasource::compression::CompressionType;
use common_error::ext::{BoxedError, ErrorExt, RetryHint};
use common_error::status_code::StatusCode;
use common_macro::stack_trace_debug;
use common_memory_manager;
use common_runtime::JoinError;
use common_time::Timestamp;
use common_time::timestamp::TimeUnit;
use datatypes::arrow::error::ArrowError;
use datatypes::prelude::ConcreteDataType;
use object_store::ErrorKind;
use object_store::error::retry_hint_from_opendal_error;
use partition::error::Error as PartitionError;
use prost::DecodeError;
use snafu::{Location, Snafu};
use store_api::ManifestVersion;
use store_api::logstore::provider::Provider;
use store_api::storage::{FileId, RegionId};
use tokio::time::error::Elapsed;

use crate::cache::file_cache::FileType;
use crate::region::RegionRoleState;
use crate::schedule::remote_job_scheduler::JobId;
use crate::worker::WorkerId;

#[derive(Snafu)]
#[snafu(visibility(pub))]
#[stack_trace_debug]
pub enum Error {
    #[cfg(feature = "enterprise")]
    #[snafu(display("Failed to route multi-region compaction input"))]
    RouteMultiRegion {
        #[snafu(source)]
        source: PartitionError,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "enterprise")]
    #[snafu(display("Invalid multi-region compaction: {reason}"))]
    InvalidMultiRegionCompaction {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[cfg(feature = "enterprise")]
    #[snafu(display("Multi-region compaction resource limit: {reason}"))]
    MultiRegionCompactionResource {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("SST writer {kind} needs {bytes} bytes, budget is {limit}"))]
    SstWriteMemoryExceeded {
        kind: &'static str,
        bytes: usize,
        limit: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Unexpected data type"))]
    DataTypeMismatch {
        source: datatypes::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

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

    #[snafu(display("OpenDAL operator failed"))]
    OpenDal {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: object_store::Error,
    },

    #[snafu(display(
        "Manifest delta {} disappeared after it was listed, path: {}",
        version,
        path
    ))]
    ManifestDeltaNotFound {
        version: ManifestVersion,
        path: String,
        #[snafu(source)]
        error: object_store::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Fail to compress object by {}, path: {}", compress_type, path))]
    CompressObject {
        compress_type: CompressionType,
        path: String,
        #[snafu(source)]
        error: std::io::Error,
    },

    #[snafu(display("Fail to decompress object by {}, path: {}", compress_type, path))]
    DecompressObject {
        compress_type: CompressionType,
        path: String,
        #[snafu(source)]
        error: std::io::Error,
    },

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

    #[snafu(display("Failed to serialize column metadata"))]
    SerializeColumnMetadata {
        #[snafu(source)]
        error: serde_json::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Manifest version {version} conflicts in {manifest_dir}: {reason}"))]
    ManifestVersionConflict {
        version: ManifestVersion,
        manifest_dir: String,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Manifest checkpoint {version} in {manifest_dir} was removed after it was listed"
    ))]
    ManifestCheckpointVanished {
        version: ManifestVersion,
        manifest_dir: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to serialize manifest, region_id: {}", region_id))]
    SerializeManifest {
        region_id: RegionId,
        #[snafu(source)]
        error: serde_json::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid scan index, start: {}, end: {}", start, end))]
    InvalidScanIndex {
        start: ManifestVersion,
        end: ManifestVersion,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid UTF-8 content"))]
    Utf8 {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: std::str::Utf8Error,
    },

    #[snafu(display("Cannot find RegionMetadata"))]
    RegionMetadataNotFound {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to join handle"))]
    Join {
        #[snafu(source)]
        error: common_runtime::JoinError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Worker {} is stopped", id))]
    WorkerStopped {
        id: WorkerId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Engine is stopped"))]
    EngineStopped {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to recv result"))]
    Recv {
        #[snafu(source)]
        error: tokio::sync::oneshot::error::RecvError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid metadata, {}", reason))]
    InvalidMeta {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid region metadata"))]
    InvalidMetadata {
        source: store_api::metadata::MetadataError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to create RecordBatch from vectors"))]
    NewRecordBatch {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: ArrowError,
    },

    #[snafu(display("Failed to read parquet file, path: {}", path))]
    ReadParquet {
        path: String,
        #[snafu(source)]
        error: parquet::errors::ParquetError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to load range result"))]
    LoadRangeResult {
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to load SST metadata"))]
    LoadSstMeta {
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to load Parquet page"))]
    LoadParquetPage {
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to write parquet file"))]
    WriteParquet {
        #[snafu(source)]
        error: parquet::errors::ParquetError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Cannot assign a stable field id to native histogram sub-field '{}' of column id {} (unknown sub-field name or derived id overflows i32)",
        field_name,
        column_id
    ))]
    InvalidNativeHistogramSubfield {
        column_id: i32,
        field_name: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Native histogram column '{}' has no usable PARQUET:field_id to namespace its sub-field ids (missing, malformed, or exceeds i32::MAX)",
        field_name
    ))]
    InvalidNativeHistogramFieldId {
        field_name: String,
        #[snafu(implicit)]
        location: Location,
    },

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

    #[snafu(display("Object store not found: {}", object_store))]
    ObjectStoreNotFound {
        object_store: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} is corrupted, reason: {}", region_id, reason))]
    RegionCorrupted {
        region_id: RegionId,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid request to region {}, reason: {}", region_id, reason))]
    InvalidRequest {
        region_id: RegionId,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "STALE_CURSOR: incremental query stale, region: {}, given_seq: {}, min_readable_seq: {}, retry_hint: FALLBACK_FULL_RECOMPUTE",
        region_id,
        given_seq,
        min_readable_seq
    ))]
    IncrementalQueryStale {
        region_id: RegionId,
        given_seq: u64,
        min_readable_seq: u64,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "STALE_SNAPSHOT_FENCE: snapshot upper bound stale, region: {}, given_seq: {}, min_enforceable_seq: {}, retry_hint: REBIND_SNAPSHOT_FENCE",
        region_id,
        given_seq,
        min_enforceable_seq
    ))]
    SnapshotFenceStale {
        region_id: RegionId,
        given_seq: u64,
        min_enforceable_seq: u64,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "SEQUENCE_RANGE_UNSUPPORTED: exact sequence-range read unsupported, region: {}, min_seq: {}, max_seq: {}, reason: {}, retry_hint: FALLBACK_MEMTABLE_ONLY_OR_FULL_RECOMPUTE",
        region_id,
        min_seq,
        max_seq,
        reason
    ))]
    SequenceRangeUnsupported {
        region_id: RegionId,
        min_seq: u64,
        max_seq: u64,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "region {} is unusable for sequence reads: file {} declares region {}",
        region_id,
        file_id,
        file_region_id
    ))]
    RegionSequenceDomainBroken {
        region_id: RegionId,
        file_region_id: RegionId,
        file_id: FileId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Old manifest missing for region {}", region_id))]
    MissingOldManifest {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("New manifest missing for region {}", region_id))]
    MissingNewManifest {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Manifest missing for region {}", region_id))]
    MissingManifest {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("File consistency check failed for file {}: {}", file_id, reason))]
    InconsistentFile {
        file_id: FileId,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Files lost during remapping: old={}, new={}", old_count, new_count))]
    FilesLost {
        old_count: usize,
        new_count: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("No old manifests provided (need at least one for template)"))]
    NoOldManifests {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to fetch manifests"))]
    FetchManifests {
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Partition expression missing for region {}", region_id))]
    MissingPartitionExpr {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to serialize partition expression: {}", source))]
    SerializePartitionExpr {
        #[snafu(source)]
        source: PartitionError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Failed to convert ConcreteDataType to ColumnDataType, reason: {}",
        reason
    ))]
    ConvertColumnDataType {
        reason: String,
        source: api::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    /// An error type to indicate that schema is changed and we need
    /// to fill default values again.
    #[snafu(display("Need to fill default value for region {}", region_id))]
    FillDefault {
        region_id: RegionId,
        // The error is for internal use so we don't need a location.
    },

    #[snafu(display(
        "Failed to create default value for column {} of region {}",
        column,
        region_id
    ))]
    CreateDefault {
        region_id: RegionId,
        column: String,
        source: datatypes::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to build entry, region_id: {}", region_id))]
    BuildEntry {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Failed to write WAL"))]
    WriteWal {
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Failed to read WAL, provider: {}", provider))]
    ReadWal {
        provider: Provider,
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Failed to decode WAL entry, region_id: {}", region_id))]
    DecodeWal {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: DecodeError,
    },

    #[snafu(display("Failed to delete WAL, region_id: {}", region_id))]
    DeleteWal {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Failed to wait for the WAL to be durable, region_id: {}", region_id))]
    WaitWalDurable {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    #[snafu(display("Failed to check the WAL history, region_id: {}", region_id))]
    CheckWalHistory {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
        source: BoxedError,
    },

    // Shared error for each writer in the write group.
    #[snafu(display("Failed to write region"))]
    WriteGroup { source: Arc<Error> },

    #[snafu(display("Invalid parquet SST file {}, reason: {}", file, reason))]
    InvalidParquet {
        file: String,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid batch, {}", reason))]
    InvalidBatch {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid arrow record batch, {}", reason))]
    InvalidRecordBatch {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid wal read request, {}", reason))]
    InvalidWalReadRequest {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to convert array to vector"))]
    ConvertVector {
        #[snafu(implicit)]
        location: Location,
        source: datatypes::error::Error,
    },

    #[snafu(display("Failed to compute arrow arrays"))]
    ComputeArrow {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: datatypes::arrow::error::ArrowError,
    },

    #[snafu(display("Failed to evaluate partition filter"))]
    EvalPartitionFilter {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: datafusion::error::DataFusionError,
    },

    #[snafu(display("Failed to merge candidate series"))]
    MergeCandidateSeries {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: datafusion::error::DataFusionError,
    },

    #[snafu(display("Failed to compute vector"))]
    ComputeVector {
        #[snafu(implicit)]
        location: Location,
        source: datatypes::error::Error,
    },

    #[snafu(display("Primary key length mismatch, expect: {}, actual: {}", expect, actual))]
    PrimaryKeyLengthMismatch {
        expect: usize,
        actual: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid sender",))]
    InvalidSender {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid scheduler state"))]
    InvalidSchedulerState {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to stop scheduler"))]
    StopScheduler {
        #[snafu(source)]
        error: JoinError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Failed to batch delete SST files, region id: {}, file ids: {:?}",
        region_id,
        file_ids
    ))]
    DeleteSsts {
        region_id: RegionId,
        file_ids: Vec<FileId>,
        #[snafu(source)]
        error: object_store::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to delete index file, file id: {}", file_id))]
    DeleteIndex {
        file_id: FileId,
        #[snafu(source)]
        error: object_store::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to batch delete index files, file ids: {:?}", file_ids))]
    DeleteIndexes {
        file_ids: Vec<FileId>,
        #[snafu(source)]
        error: object_store::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to flush region {}", region_id))]
    FlushRegion {
        region_id: RegionId,
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} is dropped", region_id))]
    RegionDropped {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} is closed", region_id))]
    RegionClosed {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Stale compaction execution for region {}, the region may have been reopened, truncated or the compaction was superseded",
        region_id
    ))]
    StaleCompactionExecution {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} is truncated", region_id))]
    RegionTruncated {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Engine write buffer is full, rejecting write requests of region {}",
        region_id,
    ))]
    RejectWrite {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} rejects writes while in staging", region_id))]
    RejectWriteInStaging {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} still has WAL tails to recover", region_id))]
    RegionPendingTails {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} does not hold a valid leader lease", region_id))]
    RegionLeaseExpired {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Region {} lost its leader lease while writing to the WAL, the write may have been applied",
        region_id
    ))]
    UnacknowledgedWrite {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to compact region {}", region_id))]
    CompactRegion {
        region_id: RegionId,
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to edit region {}", region_id))]
    EditRegion {
        region_id: RegionId,
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Failed to compat readers for region {}, reason: {}",
        region_id,
        reason,
    ))]
    CompatReader {
        region_id: RegionId,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalue region req"))]
    InvalidRegionRequest {
        source: store_api::metadata::MetadataError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Region {} is in {:?} state, which does not permit manifest updates.",
        region_id,
        state
    ))]
    UpdateManifest {
        region_id: RegionId,
        state: RegionRoleState,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} is in {:?} state, expect: {:?}", region_id, state, expect))]
    RegionState {
        region_id: RegionId,
        state: RegionRoleState,
        expect: RegionRoleState,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Partition expr version mismatch for region {}: request {}, expected {}",
        region_id,
        request_version,
        expected_version
    ))]
    PartitionExprVersionMismatch {
        region_id: RegionId,
        request_version: u64,
        expected_version: u64,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid options"))]
    JsonOptions {
        #[snafu(source)]
        error: serde_json::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Empty region directory, region_id: {}, region_dir: {}",
        region_id,
        region_dir,
    ))]
    EmptyRegionDir {
        region_id: RegionId,
        region_dir: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Empty manifest directory, manifest_dir: {}", manifest_dir,))]
    EmptyManifestDir {
        manifest_dir: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Column not found, column: {column}"))]
    ColumnNotFound {
        column: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to build index applier"))]
    BuildIndexApplier {
        source: index::inverted_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to build index asynchronously in region {}", region_id))]
    BuildIndexAsync {
        region_id: RegionId,
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Index build did not complete for files {:?} in region {}",
        file_ids,
        region_id
    ))]
    IndexBuildIncomplete {
        region_id: RegionId,
        file_ids: Vec<FileId>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to convert value"))]
    ConvertValue {
        source: datatypes::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to apply inverted index"))]
    ApplyInvertedIndex {
        source: index::inverted_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to apply bloom filter index"))]
    ApplyBloomFilterIndex {
        source: index::bloom_filter::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to push index value"))]
    PushIndexValue {
        source: index::inverted_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to write index completely"))]
    IndexFinish {
        source: index::inverted_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Operate on aborted index"))]
    OperateAbortedIndex {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to read puffin blob"))]
    PuffinReadBlob {
        source: puffin::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to add blob to puffin file"))]
    PuffinAddBlob {
        source: puffin::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to clean dir {dir}"))]
    CleanDir {
        dir: String,
        #[snafu(source)]
        error: std::io::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid config, {reason}"))]
    InvalidConfig {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Stale log entry found during replay, region: {}, flushed: {}, replayed: {}",
        region_id,
        flushed_entry_id,
        unexpected_entry_id
    ))]
    StaleLogEntry {
        region_id: RegionId,
        flushed_entry_id: u64,
        unexpected_entry_id: u64,
    },

    #[snafu(display(
        "Failed to download file, region_id: {}, file_id: {}, file_type: {:?}",
        region_id,
        file_id,
        file_type,
    ))]
    Download {
        region_id: RegionId,
        file_id: FileId,
        file_type: FileType,
        #[snafu(source)]
        error: std::io::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Failed to upload file, region_id: {}, file_id: {}, file_type: {:?}",
        region_id,
        file_id,
        file_type,
    ))]
    Upload {
        region_id: RegionId,
        file_id: FileId,
        file_type: FileType,
        #[snafu(source)]
        error: std::io::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to create directory {}", dir))]
    CreateDir {
        dir: String,
        #[snafu(source)]
        error: std::io::Error,
    },

    #[snafu(display("Record batch error"))]
    RecordBatch {
        source: common_recordbatch::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("BiErrors, first: {first}, second: {second}"))]
    BiErrors {
        first: Box<Error>,
        second: Box<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Encode null value"))]
    IndexEncodeNull {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to encode memtable to Parquet bytes"))]
    EncodeMemtable {
        #[snafu(source)]
        error: parquet::errors::ParquetError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Partition {} out of range, {} in total", given, all))]
    PartitionOutOfRange {
        given: usize,
        all: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to iter data part"))]
    ReadDataPart {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: parquet::errors::ParquetError,
    },

    #[snafu(display("Failed to read row group in memtable"))]
    DecodeArrowRowGroup {
        #[snafu(source)]
        error: ArrowError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid region options, {}", reason))]
    InvalidRegionOptions {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("checksum mismatch (actual: {}, expected: {})", actual, expected))]
    ChecksumMismatch { actual: u32, expected: u32 },

    #[snafu(display(
        "No checkpoint found, region: {}, last_version: {}",
        region_id,
        last_version
    ))]
    NoCheckpoint {
        region_id: RegionId,
        last_version: ManifestVersion,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "No manifests found in range: [{}..{}), region: {}, last_version: {}",
        start_version,
        end_version,
        region_id,
        last_version
    ))]
    NoManifests {
        region_id: RegionId,
        start_version: ManifestVersion,
        end_version: ManifestVersion,
        last_version: ManifestVersion,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Failed to install manifest to {}, region: {}, available manifest version: {}, last version: {}",
        target_version,
        region_id,
        available_version,
        last_version
    ))]
    InstallManifestTo {
        region_id: RegionId,
        target_version: ManifestVersion,
        available_version: ManifestVersion,
        #[snafu(implicit)]
        location: Location,
        last_version: ManifestVersion,
    },

    #[snafu(display("Region {} is stopped", region_id))]
    RegionStopped {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Time range predicate overflows, timestamp: {:?}, target unit: {}",
        timestamp,
        unit
    ))]
    TimeRangePredicateOverflow {
        timestamp: Timestamp,
        unit: TimeUnit,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to open region"))]
    OpenRegion {
        #[snafu(implicit)]
        location: Location,
        source: Arc<Error>,
    },

    #[snafu(display(
        "Region {} does not satisfy requirement '{}': {}",
        region_id,
        requirement,
        reason
    ))]
    RegionRequirement {
        region_id: RegionId,
        requirement: &'static str,
        reason: &'static str,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to parse job id"))]
    ParseJobId {
        #[snafu(implicit)]
        location: Location,
        #[snafu(source)]
        error: uuid::Error,
    },

    #[snafu(display("Operation is not supported: {}", err_msg))]
    UnsupportedOperation {
        err_msg: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Failed to remotely compact region {} by job {:?} due to {}",
        region_id,
        job_id,
        reason
    ))]
    RemoteCompaction {
        region_id: RegionId,
        job_id: Option<JobId>,
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to initialize puffin stager"))]
    PuffinInitStager {
        source: puffin::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to purge puffin stager"))]
    PuffinPurgeStager {
        source: puffin::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to build puffin reader"))]
    PuffinBuildReader {
        source: puffin::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to retrieve index options from column metadata"))]
    IndexOptions {
        #[snafu(implicit)]
        location: Location,
        source: datatypes::error::Error,
        column_name: String,
    },

    #[snafu(display("Failed to retrieve compression options of column {column_name}"))]
    CompressionOptions {
        #[snafu(implicit)]
        location: Location,
        source: datatypes::error::Error,
        column_name: String,
    },

    #[snafu(display("Failed to create fulltext index creator"))]
    CreateFulltextCreator {
        source: index::fulltext_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to cast vector of {from} to {to}"))]
    CastVector {
        #[snafu(implicit)]
        location: Location,
        from: ConcreteDataType,
        to: ConcreteDataType,
        source: datatypes::error::Error,
    },

    #[snafu(display("Failed to push text to fulltext index"))]
    FulltextPushText {
        source: index::fulltext_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to finalize fulltext index creator"))]
    FulltextFinish {
        source: index::fulltext_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to apply fulltext index"))]
    ApplyFulltextIndex {
        source: index::fulltext_index::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("SST file {} does not contain valid stats info", file_path))]
    StatsNotPresent {
        file_path: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to decode stats of file {}", file_path))]
    DecodeStats {
        file_path: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid SST primary key range: {reason}"))]
    InvalidPrimaryKeyRange {
        reason: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to decode SST primary key range {endpoint} endpoint"))]
    DecodePrimaryKeyRange {
        endpoint: &'static str,
        source: mito_codec::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Region {} is busy", region_id))]
    RegionBusy {
        region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to get schema metadata"))]
    GetSchemaMetadata {
        source: common_meta::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

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

    #[snafu(display("Failed to read file metadata"))]
    Metadata {
        #[snafu(source)]
        error: std::io::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to push value to bloom filter"))]
    PushBloomFilterValue {
        source: index::bloom_filter::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to finish bloom filter"))]
    BloomFilterFinish {
        source: index::bloom_filter::error::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Manual compaction is override by following operations."))]
    ManualCompactionOverride {},

    #[snafu(display("Manual compaction is already running for region {region_id}."))]
    ManualCompactionAlreadyRunning { region_id: RegionId },

    #[snafu(display("Compaction is cancelled."))]
    CompactionCancelled {},

    #[snafu(display("Flush is cancelled."))]
    FlushCancelled {},

    #[snafu(display("Compaction memory exhausted for region {region_id} (policy: {policy})",))]
    CompactionMemoryExhausted {
        region_id: RegionId,
        policy: String,
        #[snafu(source)]
        source: common_memory_manager::Error,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Incompatible WAL provider change. This is typically caused by changing WAL provider in database config file without completely cleaning existing files. Global provider: {}, region provider: {}",
        global,
        region
    ))]
    IncompatibleWalProviderChange { global: String, region: String },

    #[snafu(display("Expected mito manifest info"))]
    MitoManifestInfo {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to scan series"))]
    ScanSeries {
        #[snafu(implicit)]
        location: Location,
        source: Arc<Error>,
    },

    #[snafu(display("Partition {} scan multiple times", partition))]
    ScanMultiTimes {
        partition: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Invalid partition expression: {}", expr))]
    InvalidPartitionExpr {
        expr: String,
        #[snafu(implicit)]
        location: Location,
        source: partition::error::Error,
    },

    #[snafu(display("Failed to decode bulk wal entry"))]
    ConvertBulkWalEntry {
        #[snafu(implicit)]
        location: Location,
        source: common_grpc::Error,
    },

    #[snafu(display("Failed to encode"))]
    Encode {
        #[snafu(implicit)]
        location: Location,
        source: mito_codec::error::Error,
    },

    #[snafu(display("Failed to decode"))]
    Decode {
        #[snafu(implicit)]
        location: Location,
        source: mito_codec::error::Error,
    },

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

    #[cfg(feature = "enterprise")]
    #[snafu(display("Failed to scan external range"))]
    ScanExternalRange {
        source: BoxedError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Inconsistent timestamp column length, expect: {}, actual: {}",
        expected,
        actual
    ))]
    InconsistentTimestampLength {
        expected: usize,
        actual: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Too many files to read concurrently: {}, max allowed: {}",
        actual,
        max
    ))]
    TooManyFilesToRead {
        actual: usize,
        max: usize,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Duration out of range: {input:?}"))]
    DurationOutOfRange {
        input: std::time::Duration,
        #[snafu(source)]
        error: chrono::OutOfRangeError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("GC job permit exhausted"))]
    TooManyGcJobs {
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Staging partition expr mismatch, manifest: {:?}, request: {}",
        manifest_expr,
        request_expr
    ))]
    StagingPartitionExprMismatch {
        manifest_expr: Option<String>,
        request_expr: String,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Invalid source and target region, source: {}, target: {}",
        source_region_id,
        target_region_id
    ))]
    InvalidSourceAndTargetRegion {
        source_region_id: RegionId,
        target_region_id: RegionId,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to prune file"))]
    PruneFile {
        source: Arc<Error>,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to cast column"))]
    CastColumn {
        #[snafu(source)]
        error: datafusion::error::DataFusionError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display("Failed to generate Arrow schema from Parquet file: {}", file))]
    ParquetToArrowSchema {
        file: String,
        #[snafu(source)]
        error: parquet::errors::ParquetError,
        #[snafu(implicit)]
        location: Location,
    },

    #[snafu(display(
        "Region {} is in {:?} state, expect: Writable, Staging or Downgrading",
        region_id,
        state
    ))]
    FlushableRegionState {
        region_id: RegionId,
        state: RegionRoleState,
        #[snafu(implicit)]
        location: Location,
    },
}

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

impl Error {
    /// Returns true if we need to fill default value for a region.
    pub(crate) fn is_fill_default(&self) -> bool {
        matches!(self, Error::FillDefault { .. })
    }

    /// Returns true if the file is not found on the object store.
    pub(crate) fn is_object_not_found(&self) -> bool {
        match self {
            Error::OpenDal { error, .. } => error.kind() == ErrorKind::NotFound,
            Error::ManifestDeltaNotFound { .. } => true,
            Error::LoadSstMeta { source, .. }
            | Error::LoadParquetPage { source, .. }
            | Error::LoadRangeResult { source, .. } => source.is_object_not_found(),
            _ => false,
        }
    }

    /// Returns whether a failed manifest update may have been persisted.
    ///
    /// Unknown errors are treated conservatively because deleting output SSTs after an ambiguous
    /// manifest write could leave the manifest referencing missing files. The retry layer returns
    /// the last attempt's error, so an error that can follow a stored attempt (throttling, or a 403
    /// from a revoked key, a changed bucket policy or clock skew) never proves nothing was written.
    pub(crate) fn may_have_persisted_manifest_update(&self) -> bool {
        match self {
            // Applicability and capability checks reject the edit before storage I/O.
            Error::InvalidRequest { .. }
            | Error::UpdateManifest { .. }
            | Error::RegionState { .. }
            | Error::RegionTruncated { .. }
            | Error::RegionStopped { .. }
            | Error::RegionLeaseExpired { .. }
            | Error::SerdeJson { .. }
            | Error::CompressObject { .. } => false,
            Error::OpenDal { error, .. } => !matches!(
                error.kind(),
                ErrorKind::Unsupported
                    | ErrorKind::ConfigInvalid
                    | ErrorKind::NotFound
                    | ErrorKind::IsADirectory
                    | ErrorKind::NotADirectory
                    | ErrorKind::AlreadyExists
                    | ErrorKind::IsSameFile
                    | ErrorKind::ConditionNotMatch
                    | ErrorKind::RangeNotSatisfied
            ),
            _ => true,
        }
    }
}

impl ErrorExt for Error {
    fn status_code(&self) -> StatusCode {
        use Error::*;

        match self {
            #[cfg(feature = "enterprise")]
            RouteMultiRegion { source, .. } => source.status_code(),
            #[cfg(feature = "enterprise")]
            InvalidMultiRegionCompaction { .. } => StatusCode::InvalidArguments,
            #[cfg(feature = "enterprise")]
            MultiRegionCompactionResource { .. } => StatusCode::RuntimeResourcesExhausted,
            SstWriteMemoryExceeded { .. } => StatusCode::RuntimeResourcesExhausted,
            DataTypeMismatch { source, .. } => source.status_code(),
            OpenDal { .. }
            | ManifestDeltaNotFound { .. }
            | ManifestVersionConflict { .. }
            | ManifestCheckpointVanished { .. }
            | ReadParquet { .. } => StatusCode::StorageUnavailable,
            WriteWal { source, .. }
            | ReadWal { source, .. }
            | DeleteWal { source, .. }
            | WaitWalDurable { source, .. }
            | CheckWalHistory { source, .. } => source.status_code(),
            CompressObject { .. }
            | DecompressObject { .. }
            | SerdeJson { .. }
            | Utf8 { .. }
            | NewRecordBatch { .. }
            | RegionCorrupted { .. }
            | InconsistentFile { .. }
            | CreateDefault { .. }
            | InvalidParquet { .. }
            | OperateAbortedIndex { .. }
            | IndexEncodeNull { .. }
            | NoCheckpoint { .. }
            | NoManifests { .. }
            | FilesLost { .. }
            | InstallManifestTo { .. }
            | Unexpected { .. }
            | SerializeColumnMetadata { .. }
            | SerializeManifest { .. }
            | StagingPartitionExprMismatch { .. } => StatusCode::Unexpected,

            RegionNotFound { .. } => StatusCode::RegionNotFound,
            ObjectStoreNotFound { .. }
            | InvalidScanIndex { .. }
            | InvalidMeta { .. }
            | InvalidRequest { .. }
            | PartitionExprVersionMismatch { .. }
            | FillDefault { .. }
            | ConvertColumnDataType { .. }
            | ColumnNotFound { .. }
            | InvalidMetadata { .. }
            | InvalidRegionOptions { .. }
            | InvalidWalReadRequest { .. }
            | PartitionOutOfRange { .. }
            | ParseJobId { .. }
            | DurationOutOfRange { .. }
            | MissingOldManifest { .. }
            | MissingNewManifest { .. }
            | MissingManifest { .. }
            | NoOldManifests { .. }
            | MissingPartitionExpr { .. }
            | SerializePartitionExpr { .. }
            | InvalidSourceAndTargetRegion { .. } => StatusCode::InvalidArguments,

            IncrementalQueryStale { .. } | SnapshotFenceStale { .. } => StatusCode::RequestOutdated,

            SequenceRangeUnsupported { .. } => StatusCode::Unsupported,

            RegionSequenceDomainBroken { .. }
            | RegionMetadataNotFound { .. }
            | Join { .. }
            | WorkerStopped { .. }
            | EngineStopped { .. }
            | Recv { .. }
            | DecodeWal { .. }
            | ComputeArrow { .. }
            | EvalPartitionFilter { .. }
            | MergeCandidateSeries { .. }
            | BiErrors { .. }
            | StopScheduler { .. }
            | ComputeVector { .. }
            | EncodeMemtable { .. }
            | CreateDir { .. }
            | ReadDataPart { .. }
            | BuildEntry { .. }
            | Metadata { .. }
            | CastColumn { .. }
            | MitoManifestInfo { .. }
            | ParquetToArrowSchema { .. } => StatusCode::Internal,

            FetchManifests { source, .. } => source.status_code(),

            OpenRegion { source, .. } => source.status_code(),

            WriteParquet { .. } => StatusCode::StorageUnavailable,
            WriteGroup { source, .. }
            | LoadSstMeta { source, .. }
            | LoadParquetPage { source, .. }
            | LoadRangeResult { source, .. } => source.status_code(),
            InvalidBatch { .. } => StatusCode::InvalidArguments,
            InvalidRecordBatch { .. } => StatusCode::InvalidArguments,
            ConvertVector { source, .. } => source.status_code(),

            PrimaryKeyLengthMismatch { .. } => StatusCode::InvalidArguments,
            InvalidSender { .. } => StatusCode::InvalidArguments,
            InvalidSchedulerState { .. } => StatusCode::InvalidArguments,
            RegionRequirement { .. } => StatusCode::InvalidArguments,
            DeleteSsts { .. } | DeleteIndex { .. } | DeleteIndexes { .. } => {
                StatusCode::StorageUnavailable
            }
            FlushRegion { source, .. } | BuildIndexAsync { source, .. } => source.status_code(),
            IndexBuildIncomplete { .. } => StatusCode::IllegalState,
            RegionDropped { .. } => StatusCode::Cancelled,
            RegionClosed { .. } => StatusCode::Cancelled,
            StaleCompactionExecution { .. } => StatusCode::Cancelled,
            RegionTruncated { .. } => StatusCode::Cancelled,
            RejectWrite { .. } | RejectWriteInStaging { .. } => StatusCode::StorageUnavailable,
            CompactRegion { source, .. } => source.status_code(),
            EditRegion { source, .. } => source.status_code(),
            CompatReader { .. } => StatusCode::Unexpected,
            InvalidRegionRequest { source, .. } => source.status_code(),
            RegionState { .. } | UpdateManifest { .. } => StatusCode::RegionNotReady,
            RegionLeaseExpired { .. } | UnacknowledgedWrite { .. } => StatusCode::RegionNotReady,
            RegionPendingTails { .. } => StatusCode::RegionNotReady,
            JsonOptions { .. } => StatusCode::InvalidArguments,
            EmptyRegionDir { .. } | EmptyManifestDir { .. } => StatusCode::RegionNotFound,
            ConvertValue { source, .. } => source.status_code(),
            ApplyBloomFilterIndex { source, .. } => source.status_code(),
            InvalidPartitionExpr { source, .. } => source.status_code(),
            BuildIndexApplier { source, .. }
            | PushIndexValue { source, .. }
            | ApplyInvertedIndex { source, .. }
            | IndexFinish { source, .. } => source.status_code(),
            PuffinReadBlob { source, .. }
            | PuffinAddBlob { source, .. }
            | PuffinInitStager { source, .. }
            | PuffinBuildReader { source, .. }
            | PuffinPurgeStager { source, .. } => source.status_code(),
            CleanDir { .. } => StatusCode::Unexpected,
            InvalidConfig { .. } => StatusCode::InvalidArguments,
            StaleLogEntry { .. }
            | InvalidNativeHistogramSubfield { .. }
            | InvalidNativeHistogramFieldId { .. } => StatusCode::Unexpected,

            External { source, .. } => source.status_code(),

            RecordBatch { source, .. } => source.status_code(),

            Download { .. } | Upload { .. } => StatusCode::StorageUnavailable,
            ChecksumMismatch { .. } => StatusCode::Unexpected,
            RegionStopped { .. } => StatusCode::RegionNotReady,
            TimeRangePredicateOverflow { .. } => StatusCode::InvalidArguments,
            UnsupportedOperation { .. } => StatusCode::Unsupported,
            RemoteCompaction { .. } => StatusCode::Unexpected,

            IndexOptions { source, .. } => source.status_code(),
            CompressionOptions { source, .. } => source.status_code(),
            CreateFulltextCreator { source, .. } => source.status_code(),
            CastVector { source, .. } => source.status_code(),
            FulltextPushText { source, .. }
            | FulltextFinish { source, .. }
            | ApplyFulltextIndex { source, .. } => source.status_code(),
            DecodeStats { .. }
            | StatsNotPresent { .. }
            | InvalidPrimaryKeyRange { .. }
            | DecodePrimaryKeyRange { .. } => StatusCode::Internal,
            RegionBusy { .. } => StatusCode::RegionBusy,
            GetSchemaMetadata { source, .. } => source.status_code(),
            Timeout { .. } => StatusCode::Cancelled,

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

            PushBloomFilterValue { source, .. } | BloomFilterFinish { source, .. } => {
                source.status_code()
            }

            ManualCompactionOverride {} | CompactionCancelled {} | FlushCancelled {} => {
                StatusCode::Cancelled
            }

            // A concurrent manual compaction fails fast instead of being queued;
            // the conflict is reported to the caller, which decides whether to
            // issue a new request after the running one finishes.
            ManualCompactionAlreadyRunning { .. } => StatusCode::RegionBusy,

            CompactionMemoryExhausted { source, .. } => source.status_code(),

            IncompatibleWalProviderChange { .. } => StatusCode::InvalidArguments,

            ScanSeries { source, .. } => source.status_code(),

            ScanMultiTimes { .. } => StatusCode::InvalidArguments,
            ConvertBulkWalEntry { source, .. } => source.status_code(),

            Encode { source, .. } | Decode { source, .. } => source.status_code(),

            #[cfg(feature = "enterprise")]
            ScanExternalRange { source, .. } => source.status_code(),

            InconsistentTimestampLength { .. } => StatusCode::InvalidArguments,

            TooManyFilesToRead { .. } | TooManyGcJobs { .. } => StatusCode::RateLimited,

            PruneFile { source, .. } => source.status_code(),

            FlushableRegionState { .. } => StatusCode::RegionNotReady,
        }
    }

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

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

        match self {
            PartitionExprVersionMismatch { .. }
            | RegionNotFound { .. }
            | RegionClosed { .. }
            | RejectWriteInStaging { .. }
            | RegionLeaseExpired { .. }
            | UnacknowledgedWrite { .. }
            | RegionState {
                state: RegionRoleState::Follower,
                ..
            } => RetryHint::RetryAfterRouteRefresh,
            ReadParquet { .. }
            | WriteParquet { .. }
            | RejectWrite { .. }
            | Download { .. }
            | Upload { .. }
            | RegionState { .. }
            | UpdateManifest { .. }
            | RegionStopped { .. }
            | RegionBusy { .. }
            | RegionPendingTails { .. }
            | ManualCompactionAlreadyRunning { .. }
            | FlushableRegionState { .. }
            | ManifestDeltaNotFound { .. }
            | ManifestVersionConflict { .. }
            | ManifestCheckpointVanished { .. } => RetryHint::Retryable,

            OpenDal { error, .. }
            | DeleteSsts { error, .. }
            | DeleteIndex { error, .. }
            | DeleteIndexes { error, .. } => retry_hint_from_opendal_error(error),

            WriteWal { source, .. }
            | ReadWal { source, .. }
            | DeleteWal { source, .. }
            | WaitWalDurable { source, .. }
            | CheckWalHistory { source, .. }
            | FetchManifests { source, .. }
            | External { source, .. } => source.retry_hint(),

            OpenRegion { source, .. }
            | LoadSstMeta { source, .. }
            | LoadParquetPage { source, .. }
            | LoadRangeResult { source, .. }
            | WriteGroup { source, .. }
            | FlushRegion { source, .. }
            | BuildIndexAsync { source, .. }
            | CompactRegion { source, .. }
            | EditRegion { source, .. }
            | ScanSeries { source, .. }
            | PruneFile { source, .. } => source.retry_hint(),

            DataTypeMismatch { source, .. }
            | ConvertVector { source, .. }
            | ConvertValue { source, .. }
            | IndexOptions { source, .. }
            | CompressionOptions { source, .. }
            | CastVector { source, .. } => source.retry_hint(),

            BuildIndexApplier { source, .. }
            | PushIndexValue { source, .. }
            | ApplyInvertedIndex { source, .. }
            | IndexFinish { source, .. } => source.retry_hint(),

            ApplyBloomFilterIndex { source, .. }
            | PushBloomFilterValue { source, .. }
            | BloomFilterFinish { source, .. } => source.retry_hint(),

            PuffinReadBlob { source, .. }
            | PuffinAddBlob { source, .. }
            | PuffinInitStager { source, .. }
            | PuffinBuildReader { source, .. }
            | PuffinPurgeStager { source, .. } => source.retry_hint(),

            CreateFulltextCreator { source, .. }
            | FulltextPushText { source, .. }
            | FulltextFinish { source, .. }
            | ApplyFulltextIndex { source, .. } => source.retry_hint(),

            InvalidRegionRequest { source, .. } => source.retry_hint(),
            InvalidPartitionExpr { source, .. } => source.retry_hint(),
            RecordBatch { source, .. } => source.retry_hint(),
            GetSchemaMetadata { source, .. } => source.retry_hint(),
            CompactionMemoryExhausted { source, .. } => source.retry_hint(),
            ConvertBulkWalEntry { source, .. } => source.retry_hint(),
            Encode { source, .. } | Decode { source, .. } => source.retry_hint(),

            #[cfg(feature = "enterprise")]
            ScanExternalRange { source, .. } => source.retry_hint(),

            _ => RetryHint::NonRetryable,
        }
    }
}

#[cfg(test)]
mod tests {
    use snafu::IntoError;

    use super::*;
    use crate::region::RegionLeaderState;

    #[test]
    fn test_partition_version_mismatch_retry_hint() {
        let error = PartitionExprVersionMismatchSnafu {
            region_id: RegionId::new(1, 1),
            request_version: 1u64,
            expected_version: 2u64,
        }
        .build();
        assert_eq!(error.status_code(), StatusCode::InvalidArguments);
        assert_eq!(error.retry_hint(), RetryHint::RetryAfterRouteRefresh);
    }

    #[test]
    fn test_region_state_retry_hint() {
        let region_id = RegionId::new(1, 1);
        let writable = RegionRoleState::Leader(RegionLeaderState::Writable);
        let follower_error = RegionStateSnafu {
            region_id,
            state: RegionRoleState::Follower,
            expect: writable,
        }
        .build();
        assert_eq!(
            follower_error.retry_hint(),
            RetryHint::RetryAfterRouteRefresh
        );
        assert_eq!(follower_error.status_code(), StatusCode::RegionNotReady);
        let leader_error = RegionStateSnafu {
            region_id,
            state: writable,
            expect: RegionRoleState::Follower,
        }
        .build();
        assert_eq!(leader_error.retry_hint(), RetryHint::Retryable);
        for (error, status, hint) in [
            (
                RejectWriteSnafu { region_id }.build(),
                StatusCode::StorageUnavailable,
                RetryHint::Retryable,
            ),
            (
                RejectWriteInStagingSnafu { region_id }.build(),
                StatusCode::StorageUnavailable,
                RetryHint::RetryAfterRouteRefresh,
            ),
            (
                RegionNotFoundSnafu { region_id }.build(),
                StatusCode::RegionNotFound,
                RetryHint::RetryAfterRouteRefresh,
            ),
            (
                RegionClosedSnafu { region_id }.build(),
                StatusCode::Cancelled,
                RetryHint::RetryAfterRouteRefresh,
            ),
            (
                RegionDroppedSnafu { region_id }.build(),
                StatusCode::Cancelled,
                RetryHint::NonRetryable,
            ),
            (
                RegionTruncatedSnafu { region_id }.build(),
                StatusCode::Cancelled,
                RetryHint::NonRetryable,
            ),
        ] {
            assert_eq!(error.status_code(), status);
            assert_eq!(error.retry_hint(), hint);
        }
    }

    #[test]
    fn test_shared_sst_metadata_preserves_error_semantics() {
        for source in [
            OpenDalSnafu {}
                .into_error(object_store::Error::new(ErrorKind::NotFound, "missing SST")),
            OpenDalSnafu {}.into_error(object_store::Error::new(
                ErrorKind::Unexpected,
                "read failed",
            )),
            RejectWriteSnafu {
                region_id: RegionId::new(1, 1),
            }
            .build(),
        ] {
            let expected_not_found = source.is_object_not_found();
            let expected_status = source.status_code();
            let expected_retry = source.retry_hint();
            let source = Arc::new(source);
            for wrapped in [
                LoadSstMetaSnafu.into_error(source.clone()),
                LoadRangeResultSnafu.into_error(source),
            ] {
                assert_eq!(expected_not_found, wrapped.is_object_not_found());
                assert_eq!(expected_status, wrapped.status_code());
                assert_eq!(expected_retry, wrapped.retry_hint());
            }
        }
    }

    #[test]
    fn test_manifest_update_persistence() {
        let rejected_kinds = [
            ErrorKind::Unsupported,
            ErrorKind::ConfigInvalid,
            ErrorKind::NotFound,
            ErrorKind::IsADirectory,
            ErrorKind::NotADirectory,
            ErrorKind::AlreadyExists,
            ErrorKind::IsSameFile,
            ErrorKind::ConditionNotMatch,
            ErrorKind::RangeNotSatisfied,
        ];
        for kind in rejected_kinds {
            let error = OpenDalSnafu {}.into_error(object_store::Error::new(kind, "test"));
            assert!(
                !error.may_have_persisted_manifest_update(),
                "error kind {kind:?} should prove the manifest was not persisted"
            );
        }

        // The last retry can fail with these kinds even though an earlier attempt was stored.
        for kind in [
            ErrorKind::Unexpected,
            ErrorKind::RateLimited,
            ErrorKind::PermissionDenied,
        ] {
            let error = OpenDalSnafu {}.into_error(object_store::Error::new(kind, "test"));
            assert!(
                error.may_have_persisted_manifest_update(),
                "error kind {kind:?} should not prove the manifest was not persisted"
            );
        }

        let region_id = RegionId::new(1, 1);
        let error = RegionTruncatedSnafu { region_id }.build();
        assert!(!error.may_have_persisted_manifest_update());
        let error = InvalidRequestSnafu {
            region_id,
            reason: "manifest applicability check rejected the edit",
        }
        .build();
        assert!(!error.may_have_persisted_manifest_update());

        // A commit refused before its PUT because the Leader deadline has passed.
        let error = RegionLeaseExpiredSnafu { region_id }.build();
        assert!(!error.may_have_persisted_manifest_update());

        // This can be raised while applying an edit after its manifest file was saved.
        let error = RegionMetadataNotFoundSnafu {}.build();
        assert!(error.may_have_persisted_manifest_update());

        let error = ManifestVersionConflictSnafu {
            version: 1u64,
            manifest_dir: "manifest/",
            reason: "test",
        }
        .build();
        assert!(error.may_have_persisted_manifest_update());
        assert!(error.retry_hint().is_retryable());
    }
}