mito2/series_index/
builder.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Independently builds range and aggregate series indexes from SST readers.
16
17use std::sync::Arc;
18
19use async_stream::try_stream;
20use common_telemetry::warn;
21use futures::TryStreamExt;
22use object_store::ObjectStore;
23use snafu::{OptionExt, ensure};
24use store_api::metadata::RegionMetadataRef;
25
26use crate::error::{Result, UnexpectedSnafu};
27use crate::read::BoxedRecordBatchStream;
28use crate::read::compat::FlatCompatBatch;
29use crate::read::flat_merge::FlatMergeReader;
30use crate::read::flat_projection::FlatProjectionMapper;
31use crate::read::prune::FlatPruneReader;
32use crate::read::read_columns::ReadColumns;
33use crate::region::MitoRegionRef;
34use crate::series_index::bucket::SeriesBucket;
35use crate::series_index::catalog::{
36    RangeIndexEntry, SeriesIndexEntry, range_index_path, series_index_path, series_metadata,
37};
38use crate::series_index::purger::{IndexFilePurger, IndexFileType, file_operation};
39use crate::series_index::version::SeriesIndexFileHandle;
40use crate::series_index::{SeriesIndexWriter, SeriesIndexWriterOptions};
41use crate::sst::file::FileHandle;
42use crate::sst::parquet::reader::{FlatRowGroupReader, ReaderMetrics};
43use crate::sst::parquet::row_group::ParquetFetchMetrics;
44use crate::sst::range_index::{SstRangeIndexWriter, SstRangeIndexWriterOptions};
45
46async fn reader_input(
47    region: &MitoRegionRef,
48    file: FileHandle,
49) -> Result<
50    Option<(
51        Arc<crate::sst::parquet::file_range::FileRangeContext>,
52        crate::sst::parquet::row_selection::RowGroupSelection,
53    )>,
54> {
55    Ok(region
56        .access_layer
57        .read_sst(file)
58        .projection(Some(ReadColumns::new([])))
59        .build_reader_input(&mut ReaderMetrics::default())
60        .await?
61        .map(|(context, selection)| (Arc::new(context), selection)))
62}
63
64/// Builds one range index, or returns `None` when the SST has no readable input.
65pub(crate) async fn build_range_index(
66    store: &ObjectStore,
67    region: &MitoRegionRef,
68    metadata: &RegionMetadataRef,
69    file: FileHandle,
70) -> Result<Option<RangeIndexEntry>> {
71    let file_id = file.file_id().file_id();
72    let Some((context, mut selection)) = reader_input(region, file).await? else {
73        return Ok(None);
74    };
75    let mapper = FlatProjectionMapper::new(metadata, [])?;
76    let compat = FlatCompatBatch::try_new(&mapper, context.read_format(), false)?;
77    let path = range_index_path(region.region_id, file_id);
78    let mut writer = SstRangeIndexWriter::try_new(
79        metadata.clone(),
80        store.clone(),
81        &path,
82        SstRangeIndexWriterOptions::default(),
83    )
84    .await?;
85    let result: Result<()> = async {
86        let fetch_metrics = ParquetFetchMetrics::default();
87        while let Some((row_group_id, row_selection)) = selection.pop_first() {
88            let parquet_reader = context
89                .reader_builder()
90                .build(context.build_context(
91                    row_group_id,
92                    Some(row_selection),
93                    Some(&fetch_metrics),
94                ))
95                .await?;
96            let mut reader = FlatPruneReader::new_with_row_group_reader(
97                context.clone(),
98                FlatRowGroupReader::new(context.clone(), parquet_reader),
99                context.pre_filter_mode().skip_fields(),
100            );
101            while let Some(batch) = reader.next_batch().await? {
102                let batch = match &compat {
103                    Some(compat) => compat.compat(batch)?,
104                    None => batch,
105                };
106                writer.write(row_group_id as u32, &batch).await?;
107            }
108        }
109        Ok(())
110    }
111    .await;
112    if let Err(error) = result {
113        if let Err(cleanup_error) = writer.abort().await {
114            warn!(cleanup_error; "Failed to abort range-index build");
115        }
116        return Err(error);
117    }
118    let metrics = writer.finish().await?;
119    file_operation(IndexFileType::Range, "build", "success");
120    Ok(Some(RangeIndexEntry {
121        file_id,
122        file_size: metrics.output_bytes,
123    }))
124}
125
126/// Builds only the series index. Callers build needed range indexes separately.
127pub(crate) async fn build_series_index(
128    store: &ObjectStore,
129    region: &MitoRegionRef,
130    metadata: &RegionMetadataRef,
131    bucket: &SeriesBucket,
132    entry: &SeriesIndexEntry,
133    purger: &IndexFilePurger,
134) -> Result<SeriesIndexFileHandle> {
135    let mut sources = Vec::<BoxedRecordBatchStream>::new();
136    let mapper = FlatProjectionMapper::new(metadata, [])?;
137    let schema = mapper.input_arrow_schema(false);
138    for file in &bucket.files {
139        let Some((context, mut selection)) = reader_input(region, file.clone()).await? else {
140            continue;
141        };
142        let compat = FlatCompatBatch::try_new(&mapper, context.read_format(), false)?;
143        sources.push(Box::pin(try_stream! {
144            let fetch_metrics = ParquetFetchMetrics::default();
145            while let Some((row_group_id, row_selection)) = selection.pop_first() {
146                let parquet_reader = context.reader_builder().build(context.build_context(
147                    row_group_id,
148                    Some(row_selection),
149                    Some(&fetch_metrics),
150                )).await?;
151                let mut reader = FlatPruneReader::new_with_row_group_reader(
152                    context.clone(),
153                    FlatRowGroupReader::new(context.clone(), parquet_reader),
154                    context.pre_filter_mode().skip_fields(),
155                );
156                while let Some(batch) = reader.next_batch().await? {
157                    yield match &compat {
158                        Some(compat) => compat.compat(batch)?,
159                        None => batch,
160                    };
161                }
162            }
163        }));
164    }
165    ensure!(
166        !sources.is_empty(),
167        UnexpectedSnafu {
168            reason: "series-index bucket has no readable SST",
169        }
170    );
171    let mut visible: BoxedRecordBatchStream = if sources.len() == 1 {
172        sources.pop().context(UnexpectedSnafu {
173            reason: "series-index source disappeared",
174        })?
175    } else {
176        Box::pin(
177            FlatMergeReader::new(schema, sources, 8192, None)
178                .await?
179                .into_stream(),
180        )
181    };
182    // TODO(yingwen): Deduplicate update-mode rows before series indexes are used by queries.
183    let path = series_index_path(region.region_id, entry.index_uuid);
184    let mut writer = SeriesIndexWriter::try_new(
185        metadata.clone(),
186        store.clone(),
187        &path,
188        SeriesIndexWriterOptions::default(),
189        Some(series_metadata(entry)?),
190    )
191    .await?;
192    let result: Result<()> = async {
193        while let Some(batch) = visible.try_next().await? {
194            writer.write(&batch).await?;
195        }
196        Ok(())
197    }
198    .await;
199    if let Err(error) = result {
200        if let Err(cleanup_error) = writer.abort().await {
201            warn!(cleanup_error; "Failed to abort series-index build");
202        }
203        return Err(error);
204    }
205    let metrics = writer.finish().await?;
206    file_operation(IndexFileType::Series, "build", "success");
207    Ok(SeriesIndexFileHandle::new(
208        region.region_id,
209        SeriesIndexEntry {
210            file_size: metrics.output_bytes,
211            ..entry.clone()
212        },
213        purger.clone(),
214    ))
215}
216
217#[cfg(test)]
218mod tests {
219    use std::collections::{BTreeMap, HashMap};
220    use std::sync::Mutex;
221    use std::sync::atomic::{AtomicBool, Ordering};
222
223    use datatypes::data_type::ConcreteDataType;
224    use object_store::layers::mock::{self, MockLayerBuilder, oio};
225    use object_store::services::Memory;
226    use store_api::region_engine::RegionEngine;
227
228    use super::*;
229    use crate::region::version::VersionRef;
230    use crate::series_index::bucket::{group_files_into_series_buckets, plan_series_indexes};
231    use crate::series_index::catalog::{
232        SeriesIndexCatalog, load_version_control, series_catalog_path, store_catalog,
233    };
234    use crate::series_index::purger::series_index_channel;
235    use crate::series_index::tests::prepare_region;
236    use crate::test_util::TestEnv;
237
238    fn build_input(version: &VersionRef) -> (SeriesBucket, SeriesIndexEntry) {
239        let files = version
240            .ssts
241            .levels()
242            .iter()
243            .flat_map(|level| level.files())
244            .cloned()
245            .collect::<Vec<_>>();
246        let mut plan = plan_series_indexes(
247            group_files_into_series_buckets(&files, 100, 10),
248            BTreeMap::new(),
249            None,
250            0,
251        );
252        assert_eq!(1, plan.builds.len());
253        plan.builds.pop().unwrap()
254    }
255
256    fn metadata_with_seconds(version: &VersionRef) -> store_api::metadata::RegionMetadataRef {
257        let mut metadata = (*version.metadata).clone();
258        let time_index = metadata.time_index_column_pos();
259        metadata.column_metadatas[time_index]
260            .column_schema
261            .data_type = ConcreteDataType::timestamp_second_datatype();
262        Arc::new(metadata)
263    }
264
265    #[rstest::rstest]
266    #[case::range_then_series(true, false)]
267    #[case::series_only(false, false)]
268    #[case::series_failure(true, true)]
269    #[tokio::test]
270    async fn test_build_indexes_without_publication(
271        #[case] build_ranges: bool,
272        #[case] fail_series: bool,
273    ) {
274        let mut env = TestEnv::with_prefix("series-builder").await;
275        let (engine, region) = prepare_region(&mut env).await;
276        let version = region.version();
277        let (bucket, entry) = build_input(&version);
278        let files = &bucket.files;
279        let store = ObjectStore::new(Memory::default()).unwrap();
280        let (purger, mut receiver) = series_index_channel(store.clone());
281        let mut range_bytes = HashMap::new();
282        if build_ranges {
283            for file in files {
284                let completed = build_range_index(&store, &region, &version.metadata, file.clone())
285                    .await
286                    .unwrap()
287                    .unwrap();
288                let id = completed.file_id;
289                assert_eq!(
290                    completed.file_size,
291                    store
292                        .stat(&range_index_path(region.region_id, id))
293                        .await
294                        .unwrap()
295                        .content_length()
296                );
297                range_bytes.insert(
298                    id,
299                    store
300                        .read(&range_index_path(region.region_id, id))
301                        .await
302                        .unwrap()
303                        .to_bytes(),
304                );
305            }
306        }
307        // Both builders use the captured version, even if live metadata changes.
308        region
309            .version_control
310            .alter_metadata(metadata_with_seconds(&version));
311        let layer = writer_layer(&WriterStates::default(), move |path| {
312            assert!(path.contains("/series/"), "series stage opened {path}");
313            if fail_series {
314                WriterFailure::Finish
315            } else {
316                WriterFailure::None
317            }
318        });
319        let result = build_series_index(
320            &store.clone().layer(layer),
321            &region,
322            &version.metadata,
323            &bucket,
324            &entry,
325            &purger,
326        )
327        .await;
328        if fail_series {
329            assert!(result.is_err());
330        } else {
331            let handle = result.unwrap();
332            assert_eq!(
333                handle.entry().file_size,
334                store
335                    .stat(&series_index_path(region.region_id, entry.index_uuid))
336                    .await
337                    .unwrap()
338                    .content_length()
339            );
340            assert_eq!(
341                handle.entry(),
342                &SeriesIndexEntry {
343                    file_size: handle.entry().file_size,
344                    ..entry.clone()
345                }
346            );
347            assert!(
348                store
349                    .exists(&series_index_path(region.region_id, entry.index_uuid))
350                    .await
351                    .unwrap()
352            );
353            store_catalog(
354                &store,
355                &series_catalog_path(region.region_id),
356                &SeriesIndexCatalog {
357                    indexes: vec![handle.entry().clone()],
358                },
359            )
360            .await
361            .unwrap();
362            let recovered = load_version_control(&store, region.region_id, &purger)
363                .await
364                .current();
365            let repeated = plan_series_indexes(
366                group_files_into_series_buckets(files, 100, 10),
367                recovered.index_buckets.clone(),
368                None,
369                0,
370            );
371            assert!(repeated.builds.is_empty());
372            assert_eq!(recovered.index_buckets, repeated.index_buckets);
373        }
374        // Series success or failure leaves completed range indexes unchanged.
375        for file in files {
376            let id = file.file_id().file_id();
377            let path = range_index_path(region.region_id, id);
378            if let Some(bytes) = range_bytes.get(&id) {
379                assert_eq!(*bytes, store.read(&path).await.unwrap().to_bytes());
380            } else {
381                assert!(!store.exists(&path).await.unwrap());
382            }
383        }
384        assert!(region.series_index_version().range_indexes.is_empty());
385        assert!(region.series_index_version().series_indexes.is_empty());
386        assert!(receiver.try_recv().is_err());
387        engine.stop().await.unwrap();
388    }
389
390    #[tokio::test]
391    async fn test_build_indexes_after_time_index_widening() {
392        use api::v1::helper::row;
393        use api::v1::value::ValueData;
394        use api::v1::{ColumnDataType, Rows, SemanticType, WriteHint};
395        use datatypes::arrow::array::{TimestampMicrosecondArray, UInt64Array};
396        use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
397        use store_api::region_request::{
398            AlterKind, ModifyColumnType, RegionAlterRequest, RegionPutRequest, RegionRequest,
399        };
400        use store_api::storage::consts::PRIMARY_KEY_COLUMN_NAME;
401
402        use crate::test_util::sst_util::new_sparse_primary_key;
403        use crate::test_util::{CreateRequestBuilder, flush_region, rows_schema};
404
405        let mut env = TestEnv::with_prefix("series-builder-widen").await;
406        let (engine, region) = prepare_region(&mut env).await;
407        let metadata = region.version().metadata.clone();
408        engine
409            .handle_request(
410                region.region_id,
411                RegionRequest::Alter(RegionAlterRequest {
412                    kind: AlterKind::ModifyColumnTypes {
413                        columns: vec![ModifyColumnType {
414                            column_name: metadata.time_index_column().column_schema.name.clone(),
415                            target_type: ConcreteDataType::timestamp_microsecond_datatype(),
416                        }],
417                    },
418                }),
419            )
420            .await
421            .unwrap();
422
423        let mut request = CreateRequestBuilder::new().build();
424        request.column_metadatas = metadata.column_metadatas.clone();
425        request.primary_key = metadata.primary_key.clone();
426        let full_schema = rows_schema(&request);
427        let mut pk = full_schema[0].clone();
428        pk.column_name = PRIMARY_KEY_COLUMN_NAME.to_string();
429        pk.datatype = ColumnDataType::Binary.into();
430        pk.semantic_type = SemanticType::Tag.into();
431        let mut ts = full_schema[5].clone();
432        ts.datatype = ColumnDataType::TimestampMicrosecond.into();
433        engine
434            .handle_request(
435                region.region_id,
436                RegionRequest::Put(RegionPutRequest {
437                    skip_wal: false,
438                    rows: Rows {
439                        schema: vec![pk, ts, full_schema[4].clone()],
440                        rows: [500_500, 2_500_500, 3_500_500]
441                            .into_iter()
442                            .map(|ts| {
443                                row(vec![
444                                    ValueData::BinaryValue(new_sparse_primary_key(
445                                        &["a", "x"],
446                                        &metadata,
447                                        10,
448                                        0,
449                                    )),
450                                    ValueData::TimestampMicrosecondValue(ts),
451                                    ValueData::U64Value(1),
452                                ])
453                            })
454                            .collect(),
455                    },
456                    hint: Some(WriteHint {
457                        primary_key_encoding: api::v1::PrimaryKeyEncoding::Sparse.into(),
458                    }),
459                    partition_expr_version: None,
460                }),
461            )
462            .await
463            .unwrap();
464        flush_region(&engine, region.region_id, None).await;
465        let version = region.version();
466        let (bucket, entry) = build_input(&version);
467        let store = ObjectStore::new(Memory::default()).unwrap();
468        let (purger, _receiver) = series_index_channel(store.clone());
469        for file in &bucket.files {
470            build_range_index(&store, &region, &version.metadata, file.clone())
471                .await
472                .unwrap()
473                .unwrap();
474        }
475        let _handle =
476            build_series_index(&store, &region, &version.metadata, &bucket, &entry, &purger)
477                .await
478                .unwrap();
479        let bytes = store
480            .read(&series_index_path(region.region_id, entry.index_uuid))
481            .await
482            .unwrap()
483            .to_bytes();
484        let batches = ParquetRecordBatchReaderBuilder::try_new(bytes)
485            .unwrap()
486            .build()
487            .unwrap()
488            .collect::<std::result::Result<Vec<_>, _>>()
489            .unwrap();
490        assert_eq!(1, batches.len());
491        let batch = &batches[0];
492        assert_eq!(1, batch.num_rows());
493        for (column, expected) in [(0, 500_500), (1, 4_000_000)] {
494            assert_eq!(
495                expected,
496                batch
497                    .column(column)
498                    .as_any()
499                    .downcast_ref::<TimestampMicrosecondArray>()
500                    .unwrap()
501                    .value(0)
502            );
503        }
504        assert_eq!(
505            7,
506            batch
507                .column(2)
508                .as_any()
509                .downcast_ref::<UInt64Array>()
510                .unwrap()
511                .value(0)
512        );
513        engine.stop().await.unwrap();
514    }
515
516    #[tokio::test]
517    async fn test_range_failure_preserves_completed_indexes_and_stops_series_stage() {
518        let mut env = TestEnv::with_prefix("range-builder-stage-failure").await;
519        let (engine, region) = prepare_region(&mut env).await;
520        let version = region.version();
521        let (bucket, entry) = build_input(&version);
522        let files = &bucket.files;
523        let failed_path = range_index_path(region.region_id, files[1].file_id().file_id());
524        let states = WriterStates::default();
525        let layer = writer_layer(&states, move |path| {
526            if path == failed_path {
527                WriterFailure::Finish
528            } else {
529                WriterFailure::None
530            }
531        });
532        let store = ObjectStore::new(Memory::default()).unwrap().layer(layer);
533        let (purger, _receiver) = series_index_channel(store.clone());
534        let mut completed = Vec::new();
535        let result: Result<Option<SeriesIndexFileHandle>> = async {
536            for file in files {
537                let Some(entry) =
538                    build_range_index(&store, &region, &version.metadata, file.clone()).await?
539                else {
540                    // Defer series construction if a needed range is not ready.
541                    return Ok(None);
542                };
543                completed.push(entry.file_id);
544            }
545            build_series_index(&store, &region, &version.metadata, &bucket, &entry, &purger)
546                .await
547                .map(Some)
548        }
549        .await;
550        assert!(format!("{:?}", result.unwrap_err()).contains("injected index finish failure"));
551        assert_eq!(vec![files[0].file_id().file_id()], completed);
552        {
553            let states = states.lock().unwrap();
554            assert_eq!(2, states.len());
555            assert!(states.keys().all(|path| !path.contains("/series/")));
556            assert_eq!(
557                1,
558                states[&range_index_path(region.region_id, completed[0])].closed
559            );
560            assert_eq!(
561                1,
562                states[&range_index_path(region.region_id, files[1].file_id().file_id())].aborted
563            );
564        }
565        for (i, file) in files.iter().enumerate() {
566            assert_eq!(
567                i == 0,
568                store
569                    .exists(&range_index_path(
570                        region.region_id,
571                        file.file_id().file_id()
572                    ))
573                    .await
574                    .unwrap()
575            );
576        }
577        engine.stop().await.unwrap();
578    }
579
580    #[derive(Default)]
581    struct WriterState {
582        closed: usize,
583        aborted: usize,
584    }
585
586    type WriterStates = Arc<Mutex<BTreeMap<String, WriterState>>>;
587
588    enum WriterFailure {
589        None,
590        Finish,
591        Abort,
592    }
593
594    fn writer_layer(
595        states: &WriterStates,
596        failure: impl Fn(&str) -> WriterFailure + Send + Sync + 'static,
597    ) -> mock::MockLayer {
598        let states = states.clone();
599        MockLayerBuilder::default()
600            .writer_factory(Arc::new(move |path, _, inner| -> oio::Writer {
601                states
602                    .lock()
603                    .unwrap()
604                    .insert(path.to_string(), WriterState::default());
605                Box::new(RecordingWriter {
606                    inner,
607                    path: path.to_string(),
608                    states: states.clone(),
609                    failure: failure(path),
610                })
611            }))
612            .build()
613            .unwrap()
614    }
615
616    struct RecordingWriter {
617        inner: oio::Writer,
618        path: String,
619        states: WriterStates,
620        failure: WriterFailure,
621    }
622
623    impl mock::Write for RecordingWriter {
624        async fn write(&mut self, buffer: mock::Buffer) -> mock::Result<()> {
625            self.inner.write(buffer).await
626        }
627
628        async fn close(&mut self) -> mock::Result<mock::Metadata> {
629            if matches!(self.failure, WriterFailure::Finish) {
630                return Err(mock::Error::new(
631                    mock::ErrorKind::Unexpected,
632                    "injected index finish failure",
633                ));
634            }
635            let metadata = self.inner.close().await?;
636            self.states
637                .lock()
638                .unwrap()
639                .get_mut(&self.path)
640                .unwrap()
641                .closed += 1;
642            Ok(metadata)
643        }
644
645        async fn abort(&mut self) -> mock::Result<()> {
646            self.states
647                .lock()
648                .unwrap()
649                .get_mut(&self.path)
650                .unwrap()
651                .aborted += 1;
652            if matches!(self.failure, WriterFailure::Abort) {
653                return Err(mock::Error::new(
654                    mock::ErrorKind::Unexpected,
655                    "injected abort failure",
656                ));
657            }
658            self.inner.abort().await
659        }
660    }
661
662    struct FailingSstReader {
663        inner: oio::Reader,
664        fail: Arc<AtomicBool>,
665    }
666
667    impl mock::Read for FailingSstReader {
668        async fn read(
669            &self,
670            range: mock::BytesRange,
671        ) -> mock::Result<(mock::RpRead, mock::Buffer)> {
672            if self.fail.load(Ordering::Relaxed) {
673                return Err(mock::Error::new(
674                    mock::ErrorKind::Unexpected,
675                    "injected SST read failure",
676                ));
677            }
678            self.inner.read(range).await
679        }
680
681        async fn open(
682            &self,
683            range: mock::BytesRange,
684        ) -> mock::Result<(mock::RpRead, Box<dyn mock::ReadStreamDyn>)> {
685            if self.fail.load(Ordering::Relaxed) {
686                return Err(mock::Error::new(
687                    mock::ErrorKind::Unexpected,
688                    "injected SST read failure",
689                ));
690            }
691            self.inner.open(range).await
692        }
693    }
694
695    #[rstest::rstest]
696    #[case::range(false, false, false)]
697    #[case::range_abort_failure(false, false, true)]
698    #[case::series_setup(true, true, false)]
699    #[case::series_read(true, false, false)]
700    #[case::series_abort_failure(true, false, true)]
701    #[tokio::test]
702    async fn test_read_failure_aborts_unfinished_outputs(
703        #[case] series: bool,
704        #[case] fail_before_open: bool,
705        #[case] fail_abort: bool,
706    ) {
707        let fail = Arc::new(AtomicBool::new(false));
708        let read_fail = fail.clone();
709        let read_layer = MockLayerBuilder::default()
710            .reader_factory(Arc::new(move |path, _, inner| -> oio::Reader {
711                if path.ends_with(".parquet") {
712                    Box::new(FailingSstReader {
713                        inner,
714                        fail: read_fail.clone(),
715                    })
716                } else {
717                    inner
718                }
719            }))
720            .build()
721            .unwrap();
722        let mut env = TestEnv::with_prefix("series-builder-read-failure")
723            .await
724            .with_mock_layer(read_layer);
725        let (engine, region) = prepare_region(&mut env).await;
726        let version = region.version();
727        let (mut bucket, mut entry) = build_input(&version);
728        // A single source is first polled after opening the series writer.
729        bucket.files.truncate(1);
730        entry.source_file_ids = vec![bucket.files[0].file_id().file_id()];
731        let states = WriterStates::default();
732        let write_fail = fail.clone();
733        let write_layer = writer_layer(&states, move |_| {
734            write_fail.store(true, Ordering::Relaxed);
735            if fail_abort {
736                WriterFailure::Abort
737            } else {
738                WriterFailure::None
739            }
740        });
741        let store = ObjectStore::new(Memory::default())
742            .unwrap()
743            .layer(write_layer);
744        fail.store(fail_before_open, Ordering::Relaxed);
745        let result = if !series {
746            build_range_index(&store, &region, &version.metadata, bucket.files[0].clone())
747                .await
748                .map(|_| ())
749        } else {
750            let (purger, _receiver) = series_index_channel(store.clone());
751            build_series_index(&store, &region, &version.metadata, &bucket, &entry, &purger)
752                .await
753                .map(|_| ())
754        };
755        fail.store(false, Ordering::Relaxed);
756        let error = result.unwrap_err();
757        assert!(
758            format!("{error:?}").contains("injected SST read failure"),
759            "{error:?}"
760        );
761        {
762            let states = states.lock().unwrap();
763            assert_eq!(usize::from(!fail_before_open), states.len());
764            for (path, state) in states.iter() {
765                assert_eq!(0, state.closed, "{path}");
766                assert_eq!(1, state.aborted, "{path}");
767            }
768        }
769        assert!(store.list("/").await.unwrap().is_empty());
770        assert!(region.series_index_version().series_indexes.is_empty());
771        engine.stop().await.unwrap();
772    }
773}
mito2/series_index/builder.rs

mito2/series_index/
builder.rs
