ClientClientCoodinatorCoodinatorEngineEnginetransform(dataset)prepare_transform_batches(metadata)loop[foreach batch]execute_query(transaction_id, input_slices, transform, prev_checkpoint)data_slice, new_checkpointcompute_hash(data_slice)add_system_columns(data_slice)commit(metadata_block)