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
//! Errors owned by an append operation.
use arrow::error::ArrowError;
use parquet::errors::ParquetError;
use snafu::{Backtrace, Snafu};
use crate::{
coverage::{
EntityIdentity, IndexIntervalId,
index_interval::{IndexInterval, IndexIntervalMappingError},
io::CoverageSidecarError,
},
formats::parquet::SegmentCoverageError,
metadata::{
logical_schema::ArrowToLogicalSchemaError, protocol::TableProtocolError,
schema_compat::SchemaCompatibilityError,
},
storage::StorageError,
transaction_log::{CommitError, segments::SegmentError},
};
/// Errors owned by an append operation.
#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)))]
#[non_exhaustive]
pub enum AppendError {
/// The table protocol does not permit this client to append.
#[snafu(context(false), display("Table protocol error: {source}"))]
Protocol {
/// Complete table protocol failure.
#[snafu(source)]
source: TableProtocolError,
/// Backtrace captured at the append boundary.
backtrace: Backtrace,
},
/// An Arrow input reader or batch normalization failed.
#[snafu(display("Arrow input error: {source}"))]
ArrowInput {
/// Arrow error returned while reading or normalizing a batch.
#[snafu(source)]
source: ArrowError,
/// Backtrace captured at the append input boundary.
backtrace: Backtrace,
},
/// An Arrow schema could not be represented by the table logical schema.
#[snafu(display("Invalid append input schema: {source}"))]
ArrowToLogicalSchema {
/// Arrow-to-logical schema conversion failure.
#[snafu(source)]
source: ArrowToLogicalSchemaError,
/// Backtrace captured at the append schema boundary.
backtrace: Backtrace,
},
/// An append input contained no rows.
#[snafu(display("Cannot append an empty Arrow input"))]
EmptyInput,
/// A configured Parquet row group cannot contain zero rows.
#[snafu(display(
"Invalid maximum rows per Parquet row group: {max_rows_per_row_group}; expected a positive value"
))]
InvalidMaxRowsPerRowGroup {
/// Rejected per-append row-group limit.
max_rows_per_row_group: usize,
},
/// Validating the incoming and registered table schemas failed.
#[snafu(context(false), display("Schema validation failed: {source}"))]
SchemaValidation {
/// Complete schema compatibility failure.
#[snafu(source(from(SchemaCompatibilityError, Box::new)), backtrace)]
source: Box<SchemaCompatibilityError>,
},
/// A generated segment is incompatible with the table schema.
#[snafu(display(
"Generated segment {segment_path} is incompatible with the table schema: {source}"
))]
GeneratedSegmentSchemaCompatibility {
/// Generated table-relative segment path.
segment_path: String,
/// Complete schema compatibility failure.
#[snafu(source(from(SchemaCompatibilityError, Box::new)), backtrace)]
source: Box<SchemaCompatibilityError>,
},
/// Table state lacks the canonical schema required by append.
#[snafu(display(
"Table has no logical_schema at version {version}; cannot append without a canonical schema"
))]
MissingCanonicalTableSchema {
/// Transaction log version missing a canonical logical schema.
version: u64,
},
/// Reading metadata from the generated segment failed.
#[snafu(context(false), display("Generated segment metadata error: {source}"))]
SegmentMetadata {
/// Complete segment metadata failure.
#[snafu(source(from(SegmentError, Box::new)), backtrace)]
source: Box<SegmentError>,
},
/// Streaming the append input into Parquet failed.
#[snafu(display("Parquet write error: {source}"))]
ParquetWrite {
/// Parquet writer failure.
#[snafu(source)]
source: ParquetError,
/// Backtrace captured at the append writer boundary.
backtrace: Backtrace,
},
/// Deriving ordered-index coverage from the generated segment failed.
#[snafu(context(false), display("Segment coverage error: {source}"))]
GeneratedSegmentCoverage {
/// Complete segment coverage derivation failure.
#[snafu(source(from(SegmentCoverageError, Box::new)), backtrace)]
source: Box<SegmentCoverageError>,
},
/// An ordered-index interval could not be reconstructed for a diagnostic.
#[snafu(context(false), display("Index interval mapping failed: {source}"))]
IndexIntervalMapping {
/// Complete interval mapping failure.
#[snafu(source)]
source: IndexIntervalMappingError,
/// Backtrace captured at the append boundary.
backtrace: Backtrace,
},
/// A coverage sidecar could not be prepared or written.
#[snafu(context(false), display("Coverage sidecar error: {source}"))]
CoverageSidecar {
/// Complete coverage sidecar failure.
#[snafu(source(from(CoverageSidecarError, Box::new)), backtrace)]
source: Box<CoverageSidecarError>,
},
/// The generated segment overlaps coverage already persisted by the table.
#[snafu(display(
"Ordered-index interval overlap while appending {segment_path}: {overlap_count} overlapping identity/index interval pairs (example_identity={example_identity:?}, example_index_interval={example_index_interval})"
))]
PersistedIndexIntervalOverlap {
/// Relative path of the generated segment.
segment_path: String,
/// Number of overlapping identity/index interval pairs.
overlap_count: u128,
/// First overlapping identity, or `None` for a table-wide index.
example_identity: Option<EntityIdentity>,
/// Internal ID of the example interval.
example_index_interval_id: IndexIntervalId,
/// Logical ordered-index interval represented by the example ID.
example_index_interval: Box<IndexInterval>,
},
/// An entity-aware generated segment produced no entity coverage.
#[snafu(display("No entity coverage derived while appending segment {segment_path}"))]
EmptySegmentEntityCoverage {
/// Relative path of the generated segment.
segment_path: String,
},
/// One entity has rows but no usable ordered-index coverage.
#[snafu(display(
"Entity {identity:?} in segment {segment_path} has no non-null ordered-index values"
))]
EntityWithoutIndexCoverage {
/// Relative path of the generated segment.
segment_path: String,
/// Complete identity whose rows all have null ordered-index values.
identity: EntityIdentity,
},
/// An existing segment lacks coverage metadata required by append.
#[snafu(display(
"Cannot append because existing segment {segment_path} is missing coverage_path"
))]
ExistingSegmentMissingCoverageMetadata {
/// Canonical segment path missing coverage metadata.
segment_path: String,
},
/// Reading one existing segment's coverage sidecar during recovery failed.
#[snafu(display(
"Cannot recover append coverage: failed to read segment {segment_path} coverage at {coverage_path}: {source}"
))]
ExistingSegmentCoverageSidecarRead {
/// Canonical path of the segment whose sidecar could not be read.
segment_path: String,
/// Path of the failed coverage sidecar.
coverage_path: String,
/// Complete coverage sidecar failure.
#[snafu(source(from(CoverageSidecarError, Box::new)), backtrace)]
source: Box<CoverageSidecarError>,
},
/// Direct storage access for an append artifact failed.
#[snafu(context(false), display("Storage error: {source}"))]
Storage {
/// Complete storage failure.
#[snafu(source, backtrace)]
source: StorageError,
},
/// Publishing the append transaction failed with a definite outcome.
#[snafu(context(false), display("Commit error: {source}"))]
Commit {
/// Complete transaction-log failure.
#[snafu(source, backtrace)]
source: CommitError,
},
/// The append transaction may have committed, so its artifacts were preserved.
#[snafu(display(
"Commit outcome is ambiguous; generated Parquet path {segment_path} was preserved: {source}"
))]
CommitAmbiguous {
/// Generated table-relative Parquet path that was preserved.
segment_path: String,
/// Ambiguous transaction-log failure.
#[snafu(source(from(CommitError, Box::new)), backtrace)]
source: Box<CommitError>,
},
/// Append failed and one or more attempt-owned artifacts could not be removed.
#[snafu(display(
"{source}; artifact rollback also failed: [{}]",
cleanup_errors
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join("; ")
))]
Rollback {
/// Original append failure that triggered rollback.
#[snafu(source, backtrace)]
source: Box<AppendError>,
/// Typed cleanup failure for every artifact that could not be removed.
cleanup_errors: Vec<StorageError>,
},
}