dynoxide-rs 3.1.0

//! Tests for Fix #6: Parallel Scan (Segment/TotalSegments)
//!
//! Tests that parallel scan segments return disjoint, complete subsets
//! with SQLite-level filtering via FNV-1a hash.

use dynoxide::Database;
use dynoxide::actions::create_table::CreateTableRequest;
use dynoxide::actions::delete_item::DeleteItemRequest;
use dynoxide::actions::put_item::PutItemRequest;
use dynoxide::actions::scan::ScanRequest;
use dynoxide::types::*;
use std::collections::{HashMap, HashSet};

fn make_db_with_items(count: usize) -> Database {
    let db = Database::memory().unwrap();
    db.create_table(CreateTableRequest {
        table_name: "Tbl".to_string(),
        key_schema: vec![KeySchemaElement {
            attribute_name: "pk".to_string(),
            key_type: KeyType::HASH,
        }],
        attribute_definitions: vec![AttributeDefinition {
            attribute_name: "pk".to_string(),
            attribute_type: ScalarAttributeType::S,
        }],
        ..Default::default()
    })
    .unwrap();

    for i in 0..count {
        let mut item = HashMap::new();
        item.insert("pk".to_string(), AttributeValue::S(format!("item-{i:04}")));
        item.insert("data".to_string(), AttributeValue::S(format!("value-{i}")));
        db.put_item(PutItemRequest {
            table_name: "Tbl".to_string(),
            item,
            ..Default::default()
        })
        .unwrap();
    }
    db
}

fn scan_req(table: &str, segment: Option<u32>, total: Option<u32>) -> ScanRequest {
    ScanRequest {
        table_name: table.to_string(),
        segment,
        total_segments: total,
        ..Default::default()
    }
}

#[test]
fn test_parallel_scan_returns_subset() {
    let db = make_db_with_items(20);
    let resp = db.scan(scan_req("Tbl", Some(0), Some(4))).unwrap();
    // Should return a subset, not all 20
    assert!(resp.count < 20, "segment should return a subset");
    assert!(resp.count > 0, "segment should not be empty");
}

#[test]
fn test_all_segments_return_all_items() {
    let db = make_db_with_items(50);
    let total_segments = 4;
    let mut all_pks = HashSet::new();

    for segment in 0..total_segments {
        let resp = db
            .scan(scan_req("Tbl", Some(segment), Some(total_segments)))
            .unwrap();
        for item in resp.items.unwrap_or_default() {
            if let Some(AttributeValue::S(pk)) = item.get("pk") {
                let was_new = all_pks.insert(pk.clone());
                assert!(was_new, "duplicate pk across segments: {pk}");
            }
        }
    }

    assert_eq!(
        all_pks.len(),
        50,
        "all segments combined should return all 50 items"
    );
}

#[test]
fn test_segment_ge_total_segments_rejected() {
    let db = make_db_with_items(5);
    let err = db.scan(scan_req("Tbl", Some(4), Some(4))).unwrap_err();
    let msg = format!("{err}");
    assert!(
        msg.contains("Segment") && msg.contains("less than") && msg.contains("TotalSegments"),
        "unexpected error: {msg}"
    );
}

#[test]
fn test_total_segments_zero_rejected() {
    let db = make_db_with_items(5);
    let err = db.scan(scan_req("Tbl", Some(0), Some(0))).unwrap_err();
    let msg = format!("{err}");
    assert!(
        msg == "1 validation error detected: Value at 'TotalSegments' failed to satisfy constraint: Member must have value greater than or equal to 1",
        "unexpected error: {msg}"
    );
}

#[test]
fn test_segment_without_total_segments_rejected() {
    let db = make_db_with_items(5);
    let err = db.scan(scan_req("Tbl", Some(0), None)).unwrap_err();
    let msg = format!("{err}");
    assert!(
        msg.contains("TotalSegments parameter is required"),
        "unexpected error: {msg}"
    );
}

#[test]
fn test_paginated_parallel_scan() {
    let db = make_db_with_items(30);
    let total_segments = 2;
    let mut all_pks = HashSet::new();

    for segment in 0..total_segments {
        let mut start_key: Option<HashMap<String, AttributeValue>> = None;
        loop {
            let req = ScanRequest {
                table_name: "Tbl".to_string(),
                segment: Some(segment),
                total_segments: Some(total_segments),
                limit: Some(5),
                exclusive_start_key: start_key.clone(),
                ..Default::default()
            };
            let resp = db.scan(req).unwrap();

            for item in resp.items.unwrap_or_default() {
                if let Some(AttributeValue::S(pk)) = item.get("pk") {
                    all_pks.insert(pk.clone());
                }
            }

            if resp.last_evaluated_key.is_none() {
                break;
            }
            start_key = resp.last_evaluated_key;
        }
    }

    assert_eq!(
        all_pks.len(),
        30,
        "paginated parallel scan should return all 30 items"
    );
}

/// With more segments than the 4096 hash buckets, some segments hold no bucket
/// and return nothing, so a full parallel scan still sees each item exactly
/// once. It used to hand every item to the segments past the last bucket.
#[test]
fn more_segments_than_buckets_still_see_each_item_once() {
    let db = make_db_with_items(7);
    for total in [4097, 8192] {
        let mut seen = Vec::new();
        for segment in 0..total {
            let resp = db
                .scan(scan_req("Tbl", Some(segment), Some(total)))
                .unwrap();
            for item in resp.items.unwrap_or_default() {
                if let Some(AttributeValue::S(pk)) = item.get("pk") {
                    seen.push(pk.clone());
                }
            }
        }
        seen.sort();
        let expected: Vec<String> = (0..7).map(|i| format!("item-{i:04}")).collect();
        assert_eq!(seen, expected, "TotalSegments {total}");
    }
    // At the largest TotalSegments, the last segment holds no bucket.
    let resp = db
        .scan(scan_req("Tbl", Some(999_999), Some(1_000_000)))
        .unwrap();
    assert_eq!(resp.count, 0);
}

/// A worker that deletes each page as it scans its segment must still see the
/// whole segment: the next page starts from the deleted item's place. It used
/// to start from nowhere and return nothing more.
#[test]
fn a_segment_scanned_while_its_pages_are_deleted_is_seen_whole() {
    let db = make_db_with_items(100);
    let total = 2;
    let mut seen = Vec::new();
    for segment in 0..total {
        let mut start = None;
        loop {
            let resp = db
                .scan(ScanRequest {
                    exclusive_start_key: start.take(),
                    limit: Some(10),
                    ..scan_req("Tbl", Some(segment), Some(total))
                })
                .unwrap();
            for item in resp.items.unwrap_or_default() {
                db.delete_item(DeleteItemRequest {
                    table_name: "Tbl".to_string(),
                    key: HashMap::from([("pk".to_string(), item["pk"].clone())]),
                    ..Default::default()
                })
                .unwrap();
                if let Some(AttributeValue::S(pk)) = item.get("pk") {
                    seen.push(pk.clone());
                }
            }
            match resp.last_evaluated_key {
                Some(key) => start = Some(key),
                None => break,
            }
        }
    }
    seen.sort();
    let expected: Vec<String> = (0..100).map(|i| format!("item-{i:04}")).collect();
    assert_eq!(seen, expected);
}