DeleteVectorIterator::advance_to mishandles positions at the 2^32 boundary #3400

#[cfg(test)]
mod repro_advance_to {
    use super::*;

    const BUCKET_1: u64 = 1 << 32;

    #[test]
    fn advance_to_past_last_bucket_leaves_no_stale_positions() {
        let mut dv = DeleteVector::default();
        dv.insert(5);
        dv.insert(7);
        let mut it = dv.iter();
        assert_eq!(it.next(), Some(5));
        it.advance_to(BUCKET_1 + 10);
        assert_eq!(it.collect::<Vec<u64>>(), Vec::<u64>::new());
    }

    #[test]
    fn advance_to_lower_target_keeps_later_positions_in_current_bucket() {
        let mut dv = DeleteVector::default();
        dv.insert(BUCKET_1 + 10);
        dv.insert(BUCKET_1 + 20);
        let mut it = dv.iter();
        assert_eq!(it.next(), Some(BUCKET_1 + 10));
        it.advance_to(50);
        assert_eq!(it.collect::<Vec<u64>>(), vec![BUCKET_1 + 20]);
    }

    #[test]
    fn advance_to_on_fresh_iterator_skips_positions_below_target() {
        let mut dv = DeleteVector::default();
        dv.insert(5);
        dv.insert(200);
        let mut it = dv.iter();
        it.advance_to(100);
        assert_eq!(it.collect::<Vec<u64>>(), vec![200]);
    }
}

#[cfg(test)]
mod repro_advance_to {
    use std::sync::Arc;
    use std::sync::mpsc;
    use std::time::Duration;

    use parquet::arrow::arrow_reader::{RowSelection, RowSelector};
    use parquet::file::metadata::{ColumnChunkMetaData, RowGroupMetaData};
    use parquet::schema::types::{SchemaDescPtr, SchemaDescriptor, Type as SchemaType};

    use super::ArrowReader;
    use crate::delete_vector::DeleteVector;

    const BUCKET_1: i64 = 1 << 32;

    fn schema_descr() -> SchemaDescPtr {
        let field = SchemaType::primitive_type_builder("a", parquet::basic::Type::INT32)
            .build()
            .unwrap();
        let schema = SchemaType::group_type_builder("schema")
            .with_fields(vec![Arc::new(field)])
            .build()
            .unwrap();
        Arc::new(SchemaDescriptor::new(Arc::new(schema)))
    }

    fn row_group(schema: &SchemaDescPtr, num_rows: i64, ordinal: i16) -> RowGroupMetaData {
        let columns: Vec<ColumnChunkMetaData> = schema
            .columns()
            .iter()
            .map(|c| ColumnChunkMetaData::builder(c.clone()).build().unwrap())
            .collect();
        RowGroupMetaData::builder(schema.clone())
            .set_num_rows(num_rows)
            .set_column_metadata(columns)
            .set_ordinal(ordinal)
            .build()
            .unwrap()
    }

    // Runs `build_deletes_row_selection` on a worker thread so a non-terminating loop fails the test
    // instead of hanging the test binary.
    fn build_selection_bounded(
        row_groups: Vec<RowGroupMetaData>,
        selected_row_groups: Option<Vec<usize>>,
        deletes: DeleteVector,
    ) -> Option<RowSelection> {
        let (tx, rx) = mpsc::channel();
        std::thread::spawn(move || {
            let selection = ArrowReader::build_deletes_row_selection(
                &row_groups,
                &selected_row_groups,
                &deletes,
            )
            .unwrap();
            let _ = tx.send(selection);
        });
        rx.recv_timeout(Duration::from_secs(10)).ok()
    }

    fn deletes_of(positions: &[i64]) -> DeleteVector {
        let mut dv = DeleteVector::default();
        for &pos in positions {
            dv.insert(pos as u64);
        }
        dv
    }

    #[test]
    fn skipped_row_group_crossing_2_32_then_selected_row_group_terminates() {
        let schema = schema_descr();
        let row_groups = vec![
            row_group(&schema, BUCKET_1 + 10, 0),
            row_group(&schema, 10, 1),
        ];
        let selection = build_selection_bounded(
            row_groups,
            Some(vec![1]),
            deletes_of(&[3, 7]),
        )
        .expect("build_deletes_row_selection did not terminate within 10s");
        assert_eq!(
            selection,
            RowSelection::from(vec![RowSelector::select(10)])
        );
    }

    #[test]
    fn skipped_row_group_ending_at_50_keeps_delete_in_bucket_1() {
        let schema = schema_descr();
        let row_groups = vec![
            row_group(&schema, 50, 0),
            row_group(&schema, BUCKET_1 + 50, 1),
        ];
        let selection = build_selection_bounded(
            row_groups,
            Some(vec![1]),
            deletes_of(&[BUCKET_1 + 10, BUCKET_1 + 20]),
        )
        .expect("build_deletes_row_selection did not terminate within 10s");
        assert_eq!(
            selection,
            RowSelection::from(vec![
                RowSelector::select((BUCKET_1 - 40) as usize),
                RowSelector::skip(1),
                RowSelector::select(9),
                RowSelector::skip(1),
                RowSelector::select(79),
            ])
        );
    }
}