Skip to content

dbsp: Change threshold for spilling to storage from rows to bytes. #1917

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
blp merged 2 commits into from
Jul 5, 2024
Merged

dbsp: Change threshold for spilling to storage from rows to bytes. #1917

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e1f17a4
dbsp: Change threshold for spilling to storage from rows to bytes.
blp Jun 14, 2024
dcf5bd0
Merge branch 'main' into byte-size
blp Jul 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions benchmark/feldera-sql/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,7 +291,7 @@ def main():
parser.add_argument('--output', action=argparse.BooleanOptionalAction, help='whether to write query output back to Kafka (default: --no-output)')
parser.add_argument('--merge', action=argparse.BooleanOptionalAction, help='whether to merge all the queries into one program (default: --no-merge)')
parser.add_argument('--storage', action=argparse.BooleanOptionalAction, help='whether to enable storage (default: --no-storage)')
parser.add_argument('--min-storage-rows', type=int, help='If storage is enabled, the minimum number of rows to write a batch to storage.')
parser.add_argument('--min-storage-bytes', type=int, help='If storage is enabled, the minimum number of bytes to write a batch to storage.')
parser.add_argument('--query', action='append', help='queries to run (by default, all queries), specify one or more of: ' + ','.join(sort_queries(QUERY_SQL.keys())))
parser.add_argument('--input-topic-suffix', help='suffix to apply to input topic names (by default, "")')
parser.add_argument('--csv', help='File to write results in .csv format')
Expand All @@ -313,9 +313,9 @@ def main():
queries = sort_queries(parse_queries(parser.parse_args().query))
cores = int(parser.parse_args().cores)
storage = parser.parse_args().storage
min_storage_rows = parser.parse_args().min_storage_rows
if min_storage_rows is not None:
min_storage_rows = int(min_storage_rows)
min_storage_bytes = parser.parse_args().min_storage_bytes
if min_storage_bytes is not None:
min_storage_bytes = int(min_storage_bytes)
suffix = parser.parse_args().input_topic_suffix or ''
csvfile = parser.parse_args().csv
csvmetricsfile = parser.parse_args().csv_metrics
Expand Down Expand Up @@ -375,7 +375,7 @@ def main():
"config": {
"workers": cores,
"storage": storage,
"min_storage_rows": min_storage_rows,
"min_storage_bytes": min_storage_bytes,
"cpu_profiler": True,
"resources": {
# "cpu_cores_min": 0,
Expand Down
8 changes: 4 additions & 4 deletions crates/adapters/src/controller/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,21 +447,21 @@ impl Controller {
-> Result<(Box<dyn DbspCircuitHandle>, Box<dyn CircuitCatalog>), ControllerError>,
{
let mut start: Option<Instant> = None;
let min_storage_rows = if controller.status.pipeline_config.global.storage {
let min_storage_bytes = if controller.status.pipeline_config.global.storage {
// This reduces the files stored on disk to a reasonable number.
controller
.status
.pipeline_config
.global
.min_storage_rows
.unwrap_or(1000)
.min_storage_bytes
.unwrap_or(1024 * 1024)
} else {
usize::MAX
};
let config = CircuitConfig {
layout: Layout::new_solo(controller.status.pipeline_config.global.workers as usize),
storage: controller.status.pipeline_config.storage_config.clone(),
min_storage_rows,
min_storage_bytes,
init_checkpoint: Uuid::nil(),
};
let mut circuit = match circuit_factory(config) {
Expand Down
12 changes: 6 additions & 6 deletions crates/dbsp/src/circuit/dbsp_handle.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,11 +215,11 @@ pub struct CircuitConfig {
pub layout: Layout,
/// Storage configuration (if storage is enabled).
pub storage: Option<StorageConfig>,
/// Minimum number of rows in a persistent trace to spill it to storage. If
/// this is 0, then all traces will be stored on disk; if it is
/// `usize::MAX`, then all traces will be kept in memory; and intermediate
/// Estimated minimum number of bytes in a data batch to spill it to
/// storage. If this is 0, then all batches will be stored on disk; if it is
/// `usize::MAX`, then all batches will be kept in memory; and intermediate
/// values specify a threshold.
pub min_storage_rows: usize,
pub min_storage_bytes: usize,
/// The initial checkpoint to start the circuit from.
///
/// In case of a new circuit, this should be `Uuid::nil()`.
Expand All @@ -239,7 +239,7 @@ impl CircuitConfig {
Self {
layout: Layout::new_solo(n),
storage: None,
min_storage_rows: usize::MAX,
min_storage_bytes: usize::MAX,
init_checkpoint: Uuid::nil(),
}
}
Expand Down Expand Up @@ -1050,7 +1050,7 @@ mod tests {
path: temp.path().to_str().unwrap().to_string(),
cache: StorageCacheConfig::default(),
}),
min_storage_rows: 0,
min_storage_bytes: 0,
init_checkpoint: Uuid::nil(),
};
(temp, cconf)
Expand Down
16 changes: 8 additions & 8 deletions crates/dbsp/src/circuit/runtime.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ struct RuntimeInner {
layout: Layout,
storage: PathBuf,
cache: StorageCacheConfig,
min_storage_rows: usize,
min_storage_bytes: usize,
store: LocalStore,
// Panic info collected from failed worker threads.
panic_info: Vec<RwLock<Option<WorkerPanicInfo>>>,
Expand Down Expand Up @@ -272,7 +272,7 @@ impl RuntimeInner {
layout: config.layout,
cache,
storage,
min_storage_rows: config.min_storage_rows,
min_storage_bytes: config.min_storage_bytes,
store: TypedDashMap::new(),
panic_info,
})
Expand Down Expand Up @@ -606,14 +606,14 @@ impl Runtime {
}
}

/// Returns the minimum number of rows of a trace to spill it to
/// Returns the minimum number of bytes in a batch to spill it to
/// storage. For threads that run without a runtime, this method returns
/// `usize::MAX`.
pub fn min_storage_rows() -> usize {
pub fn min_storage_bytes() -> usize {
RUNTIME.with(|rt| {
rt.borrow()
.as_ref()
.map_or(usize::MAX, |runtime| runtime.0.min_storage_rows)
.map_or(usize::MAX, |runtime| runtime.0.min_storage_bytes)
})
}

Expand Down Expand Up @@ -941,7 +941,7 @@ mod tests {
path: path.to_str().unwrap().to_string(),
cache: StorageCacheConfig::default(),
}),
min_storage_rows: usize::MAX,
min_storage_bytes: usize::MAX,
init_checkpoint: Uuid::nil(),
};

Expand All @@ -963,7 +963,7 @@ mod tests {
let cconf = CircuitConfig {
layout: Layout::new_solo(4),
storage: None,
min_storage_rows: usize::MAX,
min_storage_bytes: usize::MAX,
init_checkpoint: Uuid::nil(),
};
let storage_path_clone = storage_path.clone();
Expand All @@ -986,7 +986,7 @@ mod tests {
let cconf = CircuitConfig {
layout: Layout::new_solo(4),
storage: None,
min_storage_rows: usize::MAX,
min_storage_bytes: usize::MAX,
init_checkpoint: Uuid::nil(),
};
let storage_path_clone = storage_path.clone();
Expand Down
9 changes: 9 additions & 0 deletions crates/dbsp/src/storage/file/reader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1201,6 +1201,15 @@ where {
pub fn path(&self) -> PathBuf {
self.0.file.as_ref().path.clone()
}

/// Returns the size of the underlying file in bytes.
pub fn byte_size(&self) -> Result<u64, Error> {
Ok(self
.0
.file
.cache
.get_size(self.0.file.file_handle.as_ref().unwrap())?)
}
}

impl<T> Clone for Reader<T> {
Expand Down
13 changes: 13 additions & 0 deletions crates/dbsp/src/trace/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,19 @@ where
/// The number of updates in the batch.
fn len(&self) -> usize;

/// The memory or storage size of the batch in bytes.
///
/// This can be an approximation, such as the size of an on-disk file for a
/// stored batch.
///
/// Implementations of this function can be expensive because they might

@gz gz Jun 22, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's not obvious to me why it can't be computed and stored in the meta-data of the Vec* types when building the batch?

@blp blp Jun 24, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It can be. I just don't see enough benefit. It complicates the merger implementations, which are already hard enough for me to understand (and to have any confidence in those incremental implementations I'd have to write careful tests), and it adds a cost to every builder and merger. it would only have a benefit in a case where we're about to visit every key-value pair anyway (a merge) and in that case we're throwing away the source batches anyway (because the merge consumes them). Also, we know that in-memory source batches are relatively small by the standard of the threshold, because otherwise they wouldn't be in memory.

If this turns out to be a mistake, then it'll show up in a profile. (On the other hand, the cost of incrementally maintaining the size won't ever show up in a profile because it'll be hidden inside larger code.)

/// require iterating through all the data in a batch. Currently this is
/// only used to decide whether to keep the result of a merge in memory or
/// on storage. For this case, the merge will visit and copy all the data
/// in the batch. The batch will be discarded afterward, which means that
/// the implementation need not attempt to cache the return value.
fn approximate_byte_size(&self) -> usize;

/// True if the batch is empty.
fn is_empty(&self) -> bool {
self.len() == 0
Expand Down
Loading