///|
/// Parameters for finite-window content-defined chunking.
pub(all) struct CdcConfig {
min_size : Int
average_size : Int
max_size : Int
window_size : Int
} derive(Debug, Eq)
///|
pub(all) struct CdcSummary {
bytes : Int
chunks : Int
min_chunk : Int
max_chunk : Int
average_chunk : Double
} derive(Debug)
///|
/// Stateful finite-window CDC processor for streaming or batched input.
///
/// The rolling window remains continuous across emitted chunks, matching the
/// boundary semantics of `content_defined_chunks` while avoiding the need to
/// retain the complete input in memory.
pub(all) struct CdcStream {
config : CdcConfig
pending : Array[Int]
ring : Array[Int]
mut ring_start : Int
mut rolling_value : Int
mut next_start : Int
highest_power : Int
} derive(Debug)
///|
pub fn CdcConfig::new(
min_size : Int,
average_size : Int,
max_size : Int,
window_size : Int,
) -> CdcConfig {
let safe_min = max_int(1, min_size)
let safe_average = max_int(safe_min, average_size)
let safe_max = max_int(safe_average, max_size)
{
min_size: safe_min,
average_size: safe_average,
max_size: safe_max,
window_size: max_int(1, min_int(window_size, safe_min)),
}
}
///|
pub fn CdcStream::new(config : CdcConfig) -> CdcStream {
let highest_power = modular_power(257, config.window_size - 1, 1000000007)
{
config,
pending: [],
ring: [],
ring_start: 0,
rolling_value: 0,
next_start: 0,
highest_power,
}
}
///|
fn CdcStream::push_byte(self : CdcStream, byte : Int) -> Unit {
let normalized = normalize_byte(byte)
self.pending.push(normalized)
if self.ring.length() < self.config.window_size {
self.ring.push(normalized)
if self.ring.length() == self.config.window_size {
self.rolling_value = polynomial_hash_value(self.ring)
}
} else {
let removed = self.ring[self.ring_start] + 1
self.ring[self.ring_start] = normalized
self.ring_start = (self.ring_start + 1) % self.config.window_size
let modulus = 1000000007L
let without_old = (
self.rolling_value.to_int64() -
removed.to_int64() * self.highest_power.to_int64() % modulus +
modulus
) %
modulus
self.rolling_value = ((without_old * 257L + (normalized + 1).to_int64()) %
modulus).to_int()
}
}
///|
fn CdcStream::should_cut(self : CdcStream) -> Bool {
let length = self.pending.length()
let has_window = self.ring.length() == self.config.window_size
let content_boundary = length >= self.config.min_size &&
has_window &&
self.rolling_value % self.config.average_size == 0
content_boundary || length >= self.config.max_size
}
///|
fn CdcStream::emit_pending(self : CdcStream) -> ChunkFingerprint {
let chunk = {
start: self.next_start,
length: self.pending.length(),
hash: polynomial_hash_value(self.pending),
}
self.next_start = self.next_start + self.pending.length()
self.pending.clear()
chunk
}
///|
/// Feeds any number of bytes and returns only fully decided chunks.
pub fn CdcStream::push(
self : CdcStream,
bytes : Array[Int],
) -> Array[ChunkFingerprint] {
let completed : Array[ChunkFingerprint] = []
for byte in bytes {
self.push_byte(byte)
if self.should_cut() {
completed.push(self.emit_pending())
}
}
completed
}
///|
/// Finishes a stream and returns its final partial chunk, if present.
pub fn CdcStream::finish(self : CdcStream) -> Array[ChunkFingerprint] {
let completed : Array[ChunkFingerprint] = []
if self.pending.length() > 0 {
completed.push(self.emit_pending())
}
completed
}
///|
pub fn CdcStream::pending_bytes(self : CdcStream) -> Int {
self.pending.length()
}
///|
/// Splits data at content-derived boundaries.
///
/// Boundary decisions use a finite rolling window, allowing chunk boundaries
/// to resynchronize after local insertions or deletions.
pub fn content_defined_chunks(
bytes : Array[Int],
config : CdcConfig,
) -> Array[ChunkFingerprint] {
let result : Array[ChunkFingerprint] = []
if bytes.length() == 0 {
return result
}
let rolling = window_hashes(bytes, config.window_size)
let mut start = 0
for end = 1; end <= bytes.length(); end = end + 1 {
let length = end - start
let enough_context = end >= config.window_size
let fingerprint = if enough_context {
rolling[end - config.window_size].value
} else {
0
}
let content_boundary = length >= config.min_size &&
enough_context &&
fingerprint % config.average_size == 0
let forced_boundary = length >= config.max_size
if content_boundary || forced_boundary || end == bytes.length() {
result.push(fingerprint_range(bytes, start, end))
start = end
}
}
result
}
///|
fn fingerprint_range(
bytes : Array[Int],
start : Int,
end : Int,
) -> ChunkFingerprint {
let chunk : Array[Int] = []
for i = start; i < end; i = i + 1 {
chunk.push(bytes[i])
}
{ start, length: end - start, hash: polynomial_hash_value(chunk) }
}
///|
pub fn summarize_cdc(chunks : Array[ChunkFingerprint]) -> CdcSummary {
if chunks.length() == 0 {
return {
bytes: 0,
chunks: 0,
min_chunk: 0,
max_chunk: 0,
average_chunk: 0.0,
}
}
let mut bytes = 0
let mut min_chunk = chunks[0].length
let mut max_chunk = chunks[0].length
for chunk in chunks {
bytes = bytes + chunk.length
min_chunk = min_int(min_chunk, chunk.length)
max_chunk = max_int(max_chunk, chunk.length)
}
{
bytes,
chunks: chunks.length(),
min_chunk,
max_chunk,
average_chunk: bytes.to_double() / chunks.length().to_double(),
}
}
///|
pub fn CdcSummary::to_json(self : CdcSummary) -> String {
"{\"bytes\":\{self.bytes},\"chunks\":\{self.chunks},\"min_chunk\":\{self.min_chunk},\"max_chunk\":\{self.max_chunk},\"average_chunk\":\{self.average_chunk}}"
}
///|
fn min_int(a : Int, b : Int) -> Int {
if a < b {
a
} else {
b
}
}
///|
fn max_int(a : Int, b : Int) -> Int {
if a > b {
a
} else {
b
}
}