delta_kernel 0.30.1

Core crate providing a Delta/Deltalake implementation focused on interoperability with a wide range of query engines.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
//! Definitions of errors that the delta kernel can encounter

use std::backtrace::{Backtrace, BacktraceStatus};
use std::convert::Infallible;
use std::num::ParseIntError;
use std::str::Utf8Error;

#[cfg(feature = "default-engine-base")]
use crate::arrow::error::ArrowError;
#[cfg(feature = "default-engine-base")]
use crate::object_store;
use crate::schema::{DataType, StructType};
use crate::table_properties::ParseIntervalError;
use crate::Version;

/// An error returned by a Delta Kernel operation.
#[derive(Debug, thiserror::Error)]
// TODO(#2630): Remove non_exhaustive once Delta and Engine variants are introduced.
#[non_exhaustive]
pub enum Error {
    /// A failure represented by a kernel implementation error.
    #[error(transparent)]
    Kernel(KernelError),
}

/// Details of a failed conversion from a scalar into a Rust value.
///
/// Conversion code adds path elements as an error unwinds, producing a path from the outermost
/// value to the value that failed without carrying mutable path state through successful parsing.
#[derive(Debug)]
pub struct ScalarConversionError {
    expected: String,
    actual: String,
    // Stored innermost-first because parent context is appended as conversion errors unwind.
    path: Vec<String>,
}

impl ScalarConversionError {
    pub(crate) fn new(expected: impl Into<String>, actual: impl Into<String>) -> Self {
        Self {
            expected: expected.into(),
            actual: actual.into(),
            path: Vec::new(),
        }
    }

    fn add_path_context(mut self, element: impl Into<String>) -> Self {
        self.path.push(element.into());
        self
    }

    fn path_string(&self) -> String {
        let mut path = String::new();
        for element in self.path.iter().rev() {
            if !path.is_empty() && !element.starts_with('[') {
                path.push('.');
            }
            path.push_str(element);
        }
        path
    }
}

impl std::fmt::Display for ScalarConversionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut target = self.path_string();
        if target.is_empty() {
            target.push_str("scalar");
        }
        write!(
            f,
            "Cannot convert {target}: expected {}, found {}",
            self.expected, self.actual
        )
    }
}

impl std::error::Error for ScalarConversionError {}

/// Adds an outer path element to a scalar conversion error as nested conversion unwinds.
///
/// Other error variants are returned unchanged: a field's `TryFrom<Scalar>` implementation may
/// report a failure unrelated to scalar shape, and this helper must not reclassify it.
pub(crate) fn add_scalar_path_context(
    error: KernelError,
    element: impl Into<String>,
) -> KernelError {
    match error {
        KernelError::ScalarConversion(error) => {
            KernelError::ScalarConversion(error.add_path_context(element))
        }
        other => other,
    }
}

/// A result whose error is the top-level [`Error`].
// Note: `delta_kernel::Result` is currently unused as part of the error migration tracked in:
// https://github.com/delta-io/delta-kernel-rs/issues/3421.
// First usages will come when the first `KernelError` is migrated to a `DeltaError`.
pub type Result<T> = std::result::Result<T, Error>;

/// A result whose error defaults to [`KernelError`].
pub type KernelResult<T, E = KernelError> = std::result::Result<T, E>;

/// A boxed, `Send` iterator of [`KernelResult<T>`] items.
pub type KernelResultIterator<'a, T> = Box<dyn Iterator<Item = KernelResult<T>> + Send + 'a>;

/// A [`KernelResultIterator`] that does not borrow data.
pub type KernelResultIteratorStatic<T> = KernelResultIterator<'static, T>;

/// A boxed, `Send` iterator of [`Result<T>`] items.
///
/// Convenience alias for fallible streams that return the top-level error envelope.
pub type ResultIterator<'a, T> = Box<dyn Iterator<Item = Result<T>> + Send + 'a>;

/// `'static` counterpart to [`ResultIterator`] for cases where the iterator does not
/// reference borrowed data.
pub type ResultIteratorStatic<T> = ResultIterator<'static, T>;

/// An error validating connector-provided state for snapshot construction.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum SnapshotHintError {
    /// A hint was combined with a log tail.
    #[error("Invalid snapshot hint: A snapshot hint cannot be combined with a log tail")]
    LogTail,
    /// A hint was combined with incremental CRC replay.
    #[error(
        "Invalid snapshot hint: A snapshot hint cannot be combined with incremental CRC replay"
    )]
    IncrementalReplay,
    /// The builder requested a version different from the hint's version.
    #[error(
        "Invalid snapshot hint: Requested version {requested} does not match snapshot hint version {hint}"
    )]
    VersionMismatch {
        /// The version requested from the snapshot builder.
        requested: Version,
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// The maximum catalog version differs from the hint when no time-travel version was
    /// requested.
    #[error(
        "Invalid snapshot hint: Max catalog version {max_catalog_version} does not match snapshot \
         hint version {hint}"
    )]
    MaxCatalogVersionMismatch {
        /// The maximum version ratified by the catalog.
        max_catalog_version: Version,
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// A hint marked latest conflicts with a later catalog-ratified version.
    #[error(
        "Invalid snapshot hint: version {hint} is marked latest but max catalog version is {max_catalog_version}"
    )]
    LatestVersionConflict {
        /// The version described by the snapshot hint.
        hint: Version,
        /// The latest version ratified by the catalog.
        max_catalog_version: Version,
    },
    /// The supplied log files contain log compaction files, which snapshot hints do not support.
    #[error("Invalid snapshot hint: log compaction files are not supported")]
    LogCompaction,
    /// A supplied log file is not beneath the table's `_delta_log` root.
    #[error("Invalid snapshot hint: log path '{path}' is not beneath log root '{log_root}'")]
    LogPathOutsideRoot {
        /// The supplied log file path.
        path: String,
        /// The expected table log root.
        log_root: String,
    },
    /// The supplied log files cannot form a valid log segment.
    #[error("Invalid snapshot hint: supplied log files do not form a valid log segment")]
    LogSegment {
        /// The log-segment construction error.
        #[source]
        source: Box<KernelError>,
    },
    /// The hint includes a published version after its snapshot version.
    #[error("Invalid snapshot hint: max_published_version exceeds snapshot hint version {hint}")]
    MaxPublishedVersion {
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// The hint has neither a complete checkpoint nor commit version zero.
    #[error("Invalid snapshot hint: snapshot history does not start at version 0")]
    MissingHistoryAnchor,
    /// The supplied CRC describes a different table version.
    #[error(
        "Invalid snapshot hint: CRC version {crc} does not match snapshot hint version {hint}"
    )]
    CrcVersion {
        /// The version described by the CRC.
        crc: Version,
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// The supplied CRC protocol differs from the hint protocol.
    #[error("Invalid snapshot hint: CRC protocol does not match snapshot hint protocol")]
    CrcProtocol,
    /// The supplied CRC metadata differs from the hint metadata.
    #[error("Invalid snapshot hint: CRC metadata does not match snapshot hint metadata")]
    CrcMetadata,
    /// A connector reported invalid snapshot-hint state, optionally with an underlying error.
    #[error("Invalid snapshot hint: {message}")]
    Connector {
        /// A description of the invalid connector state.
        message: String,
        /// The underlying validation error, if available.
        #[source]
        source: Option<Box<KernelError>>,
    },
}

impl From<SnapshotHintError> for KernelError {
    fn from(error: SnapshotHintError) -> Self {
        Box::new(error).into()
    }
}

/// All the types of errors that the kernel can run into
#[non_exhaustive]
#[derive(thiserror::Error, Debug)]
pub enum KernelError {
    /// This is an error that includes a backtrace. To have a particular type of error include such
    /// backtrace (when RUST_BACKTRACE=1), annotate the error with `#[error(transparent)]` and then
    /// add the error type and enum variant to the `from_with_backtrace!` macro invocation
    /// below. See IOError for an example.
    #[error("{source}\n{backtrace}")]
    Backtraced {
        source: Box<Self>,
        backtrace: Box<Backtrace>,
    },

    /// An error performing operations on arrow data
    #[cfg(feature = "default-engine-base")]
    #[error(transparent)]
    Arrow(ArrowError),

    #[error("Error writing checkpoint: {0}")]
    CheckpointWrite(String),

    /// User tried to convert engine data to the wrong type
    #[error("Invalid engine data type. Could not convert to {0}")]
    EngineDataType(String),

    /// Could not extract the specified type
    #[error("Error extracting type {0}: {1}")]
    Extract(&'static str, &'static str),

    /// A scalar could not be converted into the requested Rust value.
    #[error(transparent)]
    ScalarConversion(#[from] ScalarConversionError),

    /// A generic error with a message
    #[error("Generic delta kernel error: {0}")]
    Generic(String),

    /// A generic error wrapping another error
    #[error("Generic error: {source}")]
    GenericError {
        /// Source error
        source: Box<dyn std::error::Error + Send + Sync + 'static>,
    },

    /// An error involving the maximum catalog-ratified version when building a snapshot.
    #[error("Max catalog version error: {0}")]
    MaxCatalogVersion(String),

    /// The supplied log tail contains adjacent versions that are not contiguous.
    #[error("Log tail versions {first_version} and {second_version} are not contiguous")]
    LogTailVersionsNotContiguous {
        /// Earlier version in the invalid adjacent pair.
        first_version: Version,
        /// Later version in the invalid adjacent pair.
        second_version: Version,
    },

    /// Some kind of [`std::io::Error`]
    #[error(transparent)]
    IOError(std::io::Error),

    /// An internal error that means kernel found an unexpected situation, which is likely a bug
    #[error("Internal error {0}. This is a kernel bug, please report.")]
    InternalError(String),

    /// An error enountered while working with parquet data
    #[cfg(feature = "default-engine-base")]
    #[error("Arrow error: {0}")]
    Parquet(#[from] crate::parquet::errors::ParquetError),

    /// An error interacting with the object_store crate
    // We don't use [#from] object_store::Error here as our From impl transforms
    // object_store::Error::NotFound into Self::FileNotFound
    #[cfg(feature = "default-engine-base")]
    #[error("Error interacting with object store: {0}")]
    ObjectStore(object_store::Error),

    /// An error working with paths from the object_store crate
    #[cfg(feature = "default-engine-base")]
    #[error("Object store path error: {0}")]
    ObjectStorePath(#[from] object_store::path::Error),

    #[cfg(feature = "default-engine-base")]
    #[error("Reqwest Error: {0}")]
    Reqwest(#[from] reqwest::Error),

    /// A specified file could not be found
    #[error("File not found: {0}")]
    FileNotFound(String),

    /// A column was requested, but not found
    #[error("{0}")]
    MissingColumn(String),

    /// The connector-provided partition values are invalid (missing/extra/duplicate keys,
    /// or a value type does not match the schema column type).
    #[error("Invalid partition values: {0}")]
    InvalidPartitionValues(String),

    /// A column was specified with a specific type, but it is not of that type
    #[error("Expected column type: {0}")]
    UnexpectedColumnType(String),

    /// Data was expected, but not found
    #[error("Expected is missing: {0}")]
    MissingData(String),

    /// No table versions were found for the requested log operation.
    #[error("No table version found.")]
    EmptyLog,

    /// One or more table versions required by a log operation are unavailable.
    ///
    /// The payload is the lowest version that the operation requires but cannot obtain.
    #[error("Table version {0} is missing or unavailable for this log operation.")]
    MissingVersion(Version),

    /// The requested start version is unavailable from the queried log segment, though later
    /// versions remain.
    #[error(
        "Start version {requested} is not available; earliest available version is {earliest}."
    )]
    StartVersionNotFound {
        /// The start version the caller requested.
        requested: Version,
        /// The earliest version servable from the queried log segment (always > `requested`).
        /// This is the lowest version this producer can serve, not a promise about the lowest
        /// version readable on disk: a checkpoint may have trimmed the segment past commits a
        /// path-based read could still serve.
        earliest: Version,
    },

    /// A table version required by an operation has not been published to the Delta log.
    ///
    /// The payload is the first unpublished version.
    #[error("Table version {0} has not been published to the Delta log.")]
    UnpublishedVersion(Version),

    /// An error occurred while working with deletion vectors
    #[error("Deletion Vector error: {0}")]
    DeletionVector(String),

    /// A selection vector is larger than data length
    #[error("Selection vector is larger than data length: {0}")]
    InvalidSelectionVector(String),

    /// Transaction state is invalid for the requested operation
    #[error("Invalid transaction state: {0}")]
    InvalidTransactionState(String),

    /// A specified URL was invalid
    #[error("Invalid url: {0}")]
    InvalidUrl(#[from] url::ParseError),

    /// serde encountered malformed json
    #[error(transparent)]
    MalformedJson(serde_json::Error),

    /// There was no metadata action in the delta log
    #[error("No table metadata found in delta log.")]
    MissingMetadata,

    /// There was no protocol action in the delta log
    #[error("No protocol found in delta log.")]
    MissingProtocol,

    /// Invalid protocol action was read from the log
    #[error("Invalid protocol action in the delta log: {0}")]
    InvalidProtocol(String),

    /// Neither metadata nor protocol could be found in the delta log
    #[error("No table metadata or protocol found in delta log.")]
    MissingMetadataAndProtocol,

    /// A string failed to parse as the specified data type
    #[error("Failed to parse value '{0}' as '{1}'")]
    ParseError(String, DataType),

    /// A tokio executor failed to join a task
    #[error("Join failure: {0}")]
    JoinFailure(String),

    /// Could not convert to string from utf-8
    #[error("Could not convert to string from utf-8: {0}")]
    Utf8Error(#[from] Utf8Error),

    /// Could not parse an integer
    #[error("Could not parse int: {0}")]
    ParseIntError(#[from] ParseIntError),

    #[error("Invalid column mapping mode: {0}")]
    InvalidColumnMappingMode(String),

    /// Asked for a table at an invalid location
    #[error("Invalid table location: {0}.")]
    InvalidTableLocation(String),

    /// Precision or scale not compliant with delta specification
    #[error("Invalid decimal: {0}")]
    InvalidDecimal(String),

    /// Invalid CRS or other parameter for a Geometry / Geography type
    #[error("Invalid geo parameters: {0}")]
    InvalidGeoParams(String),

    /// Inconsistent data passed to struct scalar
    #[error("Invalid struct data: {0}")]
    InvalidStructData(String),

    /// Expressions did not parse or evaluate correctly
    #[error("Invalid expression evaluation: {0}")]
    InvalidExpressionEvaluation(String),

    /// Unable to parse the name of a log path
    #[error("Invalid log path: {0}")]
    InvalidLogPath(String),

    /// The assembled log segment is inconsistent with its declared file kinds, ordering, or
    /// version bounds. Malformed checkpoint file sets use [`KernelError::InvalidCheckpoint`].
    #[error("Invalid log segment: {0}")]
    InvalidLogSegment(String),

    /// Snapshot-hint validation failed. Log-segment errors caused by supplied hint state,
    /// including invalid paths and checkpoints, are wrapped in `SnapshotHintError::LogSegment`.
    /// Failures outside hint validation retain their existing categories.
    #[error(transparent)]
    SnapshotHint(#[from] Box<SnapshotHintError>),

    /// The file already exists at the path, prohibiting a non-overwrite write
    #[error("File already exists: {0}")]
    FileAlreadyExists(String),

    /// A Delta table was found during create-table validation.
    ///
    /// The payload is the caller-supplied table path, not the normalized table URL. Concurrent
    /// create attempts that pass validation surface as a commit conflict instead of this error.
    #[error("Table already exists at path: {0}")]
    TableAlreadyExistsAtPath(String),

    /// Some functionality is currently unsupported
    #[error("Unsupported: {0}")]
    Unsupported(String),

    /// Cannot write a version checksum (CRC) file for this snapshot
    #[error("Checksum write unsupported: {0}")]
    ChecksumWriteUnsupported(String),

    /// Parsing error when attempting to deserialize an interval
    #[error(transparent)]
    ParseIntervalError(#[from] ParseIntervalError),

    #[error("Change data feed is unsupported for the table at version {0}")]
    ChangeDataFeedUnsupported(Version),

    /// Row tracking (`delta.enableRowTracking`) must be enabled for the entire version range of a
    /// row-tracking change feed, but it is not enabled at the given version.
    #[error(
        "Row tracking (delta.enableRowTracking) must be enabled for the entire row-tracking change \
         feed range, but it is not enabled at version {0}"
    )]
    RowTrackingChangeFeedUnsupported(Version),

    #[error("Change data feed encountered incompatible schema. Expected {0}, got {1}")]
    ChangeDataFeedIncompatibleSchema(String, String),

    /// Invalid checkpoint files
    #[error("Invalid Checkpoint: {0}")]
    InvalidCheckpoint(String),

    /// Schema mismatch has occurred or invalid/not-kernel-supported schema used somewhere
    #[error("Schema error: {0}")]
    Schema(String),

    /// Validation error for file statistics (e.g., missing required clustering column stats)
    #[error("Stats validation error: {0}")]
    StatsValidation(String),

    /// Error during log history operations (timestamp queries, version lookups)
    #[error(transparent)]
    LogHistory(#[from] Box<crate::history_manager::error::LogHistoryError>),

    #[cfg(feature = "declarative-plans")]
    #[error("Declarative plan execution yielded the incorrect type: expected PlanResult::{expected}, got PlanResult::{actual}")]
    PlanResultTypeMismatch {
        expected: &'static str,
        actual: &'static str,
    },

    /// The operation was cancelled via a [`CancellationToken`](crate::CancellationToken).
    ///
    /// Surfaced by cancellation-aware operations when a cancellation is detected. See the
    /// [Engine operation cancellation contract](crate::cancellation#engine-operation-contract).
    #[error("Operation cancelled")]
    Cancelled,
}

// Convenience constructors for KernelError types that take a String argument
impl KernelError {
    pub(crate) fn scalar_conversion(
        expected: impl Into<String>,
        actual: impl Into<String>,
    ) -> Self {
        ScalarConversionError::new(expected, actual).into()
    }

    pub(crate) fn checkpoint_write(msg: impl ToString) -> Self {
        Self::CheckpointWrite(msg.to_string())
    }

    pub fn generic_err(source: impl Into<Box<dyn std::error::Error + Send + Sync>>) -> Self {
        Self::GenericError {
            source: source.into(),
        }
    }
    pub fn generic(msg: impl ToString) -> Self {
        Self::Generic(msg.to_string())
    }
    pub fn file_not_found(path: impl ToString) -> Self {
        Self::FileNotFound(path.to_string())
    }
    pub fn missing_column(name: impl ToString) -> Self {
        Self::MissingColumn(name.to_string()).with_backtrace()
    }
    pub fn unexpected_column_type(name: impl ToString) -> Self {
        Self::UnexpectedColumnType(name.to_string())
    }
    pub fn invalid_partition_values(msg: impl ToString) -> Self {
        Self::InvalidPartitionValues(msg.to_string())
    }
    pub fn missing_data(name: impl ToString) -> Self {
        Self::MissingData(name.to_string())
    }
    pub fn deletion_vector(msg: impl ToString) -> Self {
        Self::DeletionVector(msg.to_string())
    }
    pub fn engine_data_type(msg: impl ToString) -> Self {
        Self::EngineDataType(msg.to_string())
    }
    pub fn join_failure(msg: impl ToString) -> Self {
        Self::JoinFailure(msg.to_string())
    }
    pub fn invalid_table_location(location: impl ToString) -> Self {
        Self::InvalidTableLocation(location.to_string())
    }
    pub fn invalid_column_mapping_mode(mode: impl ToString) -> Self {
        Self::InvalidColumnMappingMode(mode.to_string())
    }
    pub fn invalid_decimal(msg: impl ToString) -> Self {
        Self::InvalidDecimal(msg.to_string())
    }
    #[cfg(feature = "geo-type-in-dev")]
    pub fn invalid_geo_params(msg: impl ToString) -> Self {
        Self::InvalidGeoParams(msg.to_string())
    }
    pub fn invalid_struct_data(msg: impl ToString) -> Self {
        Self::InvalidStructData(msg.to_string())
    }
    pub fn invalid_expression(msg: impl ToString) -> Self {
        Self::InvalidExpressionEvaluation(msg.to_string())
    }
    pub(crate) fn invalid_log_path(msg: impl ToString) -> Self {
        Self::InvalidLogPath(msg.to_string())
    }

    pub(crate) fn invalid_log_segment(msg: impl ToString) -> Self {
        Self::InvalidLogSegment(msg.to_string())
    }

    pub fn internal_error(msg: impl ToString) -> Self {
        Self::InternalError(msg.to_string()).with_backtrace()
    }

    pub fn invalid_protocol(msg: impl ToString) -> Self {
        Self::InvalidProtocol(msg.to_string())
    }

    pub fn invalid_transaction_state(msg: impl ToString) -> Self {
        Self::InvalidTransactionState(msg.to_string())
    }

    pub fn unsupported(msg: impl ToString) -> Self {
        Self::Unsupported(msg.to_string())
    }
    pub fn change_data_feed_unsupported(version: impl Into<Version>) -> Self {
        Self::ChangeDataFeedUnsupported(version.into())
    }
    /// Creates a [`KernelError::RowTrackingChangeFeedUnsupported`] for the given version, used
    /// when row tracking is not enabled at some point in a row-tracking change feed's version
    /// range.
    pub(crate) fn row_tracking_change_feed_unsupported(version: impl Into<Version>) -> Self {
        Self::RowTrackingChangeFeedUnsupported(version.into())
    }
    pub(crate) fn change_data_feed_incompatible_schema(
        expected: &StructType,
        actual: &StructType,
    ) -> Self {
        Self::ChangeDataFeedIncompatibleSchema(expected.to_string(), actual.to_string())
    }

    /// Creates an incompatible-schema error that identifies the version of `actual`.
    pub(crate) fn change_data_feed_incompatible_schema_at_version(
        expected: &StructType,
        actual: &StructType,
        version: Version,
    ) -> Self {
        Self::ChangeDataFeedIncompatibleSchema(
            expected.to_string(),
            format!("schema at version {version}: {actual}"),
        )
    }

    pub fn invalid_checkpoint(msg: impl ToString) -> Self {
        Self::InvalidCheckpoint(msg.to_string())
    }

    pub fn schema(msg: impl ToString) -> Self {
        Self::Schema(msg.to_string())
    }

    pub fn stats_validation(msg: impl ToString) -> Self {
        Self::StatsValidation(msg.to_string())
    }

    #[cfg(feature = "declarative-plans")]
    pub fn plan_result_type_mismatch(expected: &'static str, actual: &'static str) -> Self {
        Self::PlanResultTypeMismatch { expected, actual }
    }

    /// Returns the first error that is not wrapped by [`KernelError::Backtraced`].
    ///
    /// If this error has no backtrace wrapper, this returns `self`.
    pub fn without_backtrace(&self) -> &Self {
        match self {
            Self::Backtraced { source, .. } => source.without_backtrace(),
            error => error,
        }
    }

    // Capture a backtrace when the error is constructed.
    #[must_use]
    pub fn with_backtrace(self) -> Self {
        let backtrace = Backtrace::capture();
        match backtrace.status() {
            BacktraceStatus::Captured => Self::Backtraced {
                source: Box::new(self),
                backtrace: Box::new(backtrace),
            },
            _ => self,
        }
    }
}

macro_rules! from_with_backtrace(
    ( $(($error_type: ty, $error_variant: ident)), * ) => {
        $(
            impl From<$error_type> for KernelError {
                fn from(value: $error_type) -> Self {
                    Self::$error_variant(value).with_backtrace()
                }
            }
        )*
    };
);

from_with_backtrace!(
    (serde_json::Error, MalformedJson),
    (std::io::Error, IOError)
);

#[cfg(feature = "default-engine-base")]
impl From<ArrowError> for KernelError {
    fn from(value: ArrowError) -> Self {
        Self::Arrow(value).with_backtrace()
    }
}

#[cfg(feature = "default-engine-base")]
impl From<object_store::Error> for KernelError {
    fn from(value: object_store::Error) -> Self {
        match value {
            object_store::Error::NotFound { path, .. } => Self::file_not_found(path),
            err => Self::ObjectStore(err),
        }
    }
}

/// This impl is needed so the `?` operator can auto-convert `KernelResult<T, Infallible>` to
/// `KernelResult<T>`. For example, `TryFrom` impls for infallible conversions use `Infallible` as
/// their error type, and this allows those results to be propagated with `?` in functions
/// returning `KernelResult`. The match is unreachable since `Infallible` has no variants.
impl From<Infallible> for KernelError {
    fn from(value: Infallible) -> Self {
        match value {}
    }
}