Function test_compaction

src/storage/src/source/reclock.rs:381–472

error: unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`

()

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`
    async fn test_compaction() {
        let persist_location = PersistLocation {
            blob_uri: SensitiveUrl::from_str("mem://").expect("invalid URL"),
            consensus_uri: SensitiveUrl::from_str("mem://").expect("invalid URL"),
        };

        let remap_shard = ShardId::new();

        let persist_client = PERSIST_CACHE
            .open(persist_location)
            .await
            .expect("error creating persist client");

        let mut remap_read_handle = persist_client
            .open_critical_since::<SourceData, (), Timestamp, StorageDiff>(
                remap_shard,
                PersistClient::CONTROLLER_CRITICAL_SINCE,
                Opaque::encode(&0u64),
                Diagnostics::from_purpose("test_since_hold"),
            )
            .await
            .expect("error opening persist shard");

        let (mut operator, _batch) =
            make_test_operator(remap_shard, Antichain::from_elem(0.into())).await;

        // Ming a few bindings
        let source_upper = partitioned_frontier([(0, MzOffset::from(3))]);
        operator
            .mint(
                1000.into(),
                Antichain::from_elem(1001.into()),
                source_upper.borrow(),
            )
            .await;

        let source_upper = partitioned_frontier([(0, MzOffset::from(5))]);
        operator
            .mint(
                2000.into(),
                Antichain::from_elem(2001.into()),
                source_upper.borrow(),
            )
            .await;

        // Compact enough so that offsets >= 3 remain uncompacted
        remap_read_handle
            .compare_and_downgrade_since(
                &Opaque::encode(&0u64),
                (&Opaque::encode(&0u64), &Antichain::from_elem(1000.into())),
            )
            .await
            .unwrap();

        // Starting a new operator with an `as_of` is the same as having compacted
        let (_operator, mut initial_batch) =
            make_test_operator(remap_shard, Antichain::from_elem(1000.into())).await;