delta-arrow-reader 0.7.0

Read-only Delta Lake to Apache Arrow reader
Documentation 
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

//! Numerical costs for ordinary and pipelined range reads, in byte-equivalent units.
//! These helpers use the supplied network measurements and never perform I/O.

use std::{ops::Range, time::Duration};

/// Recent transport conditions used to compare physical range-read plans.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(in super::super) struct TransportEstimate {
    /// Typical time for a request to reach payload availability.
    pub(in super::super) request_latency: Duration,
    /// Typical aggregate payload throughput across concurrent requests.
    pub(in super::super) shared_throughput_bytes_per_second: u64,
}

/// Returns the bytes covered by a physical range plan.
pub(in super::super) fn range_bytes(plan: &[Range<u64>]) -> u128 {
    plan.iter()
        .map(|range| u128::from(range.end - range.start))
        .sum()
}

/// Returns how many rounds of requests a plan needs at the given concurrency.
pub(in super::super) fn request_waves(request_count: usize, max_concurrent_reads: usize) -> usize {
    request_count.div_ceil(max_concurrent_reads.max(1))
}

/// Scores a plan as transferred bytes plus one bandwidth-delay cost per request wave.
pub(in super::super) fn range_cost_bytes(
    plan: &[Range<u64>],
    estimate: TransportEstimate,
    max_concurrent_reads: usize,
) -> u128 {
    let bandwidth_delay_bytes = bandwidth_delay_bytes(estimate);
    range_bytes(plan).saturating_add(
        (request_waves(plan.len(), max_concurrent_reads) as u128)
            .saturating_mul(bandwidth_delay_bytes),
    )
}

/// Scores pipelined partial reads in shared-bandwidth byte units.
///
/// Requests waiting for their first byte can overlap other responses. One
/// request, one wave, or concurrency one still pays latency plus transfer time.
pub(in super::super) fn partial_read_cost_bytes(
    bytes: u128,
    requests: usize,
    estimate: TransportEstimate,
    concurrency: usize,
    request_overhead_bytes: u128,
) -> u128 {
    let latency = bandwidth_delay_bytes(estimate);
    let waves = request_waves(requests, concurrency) as u128;
    let latency_cost = latency.saturating_mul(waves);
    // ponytail: approximate steady pipelining; unequal range sizes can reduce overlap.
    let overlap = (bytes.saturating_mul(concurrency.saturating_sub(1) as u128)
        / concurrency.max(1) as u128)
        .min(latency_cost.saturating_sub(latency));
    let transport_cost = bytes.saturating_add(latency_cost).saturating_sub(overlap);
    // Request handling and transport run concurrently. Completion intervals
    // already include small-response delivery, so adding the costs counts it twice.
    let processing_cost = request_overhead_bytes
        .saturating_mul(requests as u128)
        .saturating_add(if requests == 0 { 0 } else { latency });
    transport_cost.max(processing_cost)
}

/// Adds the cost of dependent rounds, each described by its bytes and request count.
/// Requests within one round can overlap; successive rounds must pay their own latency.
pub(in super::super) fn dependent_read_cost_bytes(
    rounds: impl IntoIterator<Item = (u128, usize)>,
    estimate: TransportEstimate,
    concurrency: usize,
    request_overhead_bytes: u128,
) -> u128 {
    rounds.into_iter().fold(0_u128, |cost, (bytes, requests)| {
        cost.saturating_add(partial_read_cost_bytes(
            bytes,
            requests,
            estimate,
            concurrency,
            request_overhead_bytes,
        ))
    })
}

/// Returns the bytes transferable during one typical request-latency interval.
pub(in super::super) fn bandwidth_delay_bytes(estimate: TransportEstimate) -> u128 {
    duration_cost_bytes(
        estimate.request_latency,
        estimate.shared_throughput_bytes_per_second,
    )
}

/// Expresses a duration in the same byte-equivalent units as the read costs.
pub(in super::super) fn duration_cost_bytes(duration: Duration, bytes_per_second: u64) -> u128 {
    duration
        .as_nanos()
        .saturating_mul(u128::from(bytes_per_second))
        / 1_000_000_000
}

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

    #[test]
    fn partial_cost_accounts_for_overlap_without_hiding_serial_latency() {
        let estimate = TransportEstimate {
            request_latency: Duration::from_millis(100),
            shared_throughput_bytes_per_second: 1_000,
        };
        assert_eq!(partial_read_cost_bytes(0, 0, estimate, 4, 0), 0);
        assert_eq!(partial_read_cost_bytes(400, 1, estimate, 4, 0), 500);
        assert_eq!(partial_read_cost_bytes(400, 4, estimate, 4, 0), 500);
        assert_eq!(partial_read_cost_bytes(400, 8, estimate, 1, 0), 1_200);
        assert_eq!(partial_read_cost_bytes(400, 8, estimate, 4, 0), 500);
        assert_eq!(partial_read_cost_bytes(4, 4, estimate, 2, 0), 202);
        assert_eq!(partial_read_cost_bytes(4, 4, estimate, 2, 10), 202);
        assert_eq!(partial_read_cost_bytes(4, 4, estimate, 2, 100), 500);
    }

    #[test]
    fn dependent_rounds_do_not_share_their_latency() {
        let estimate = TransportEstimate {
            request_latency: Duration::from_millis(100),
            shared_throughput_bytes_per_second: 1_000,
        };
        assert_eq!(
            dependent_read_cost_bytes([(400, 4), (400, 4)], estimate, 4, 0),
            1_000,
        );
        assert_eq!(partial_read_cost_bytes(800, 8, estimate, 4, 0), 900);
        assert_eq!(dependent_read_cost_bytes([], estimate, 4, 0), 0);
    }
}