Minor fixes, formatting, dependency bump

Cargo.toml

@@ -34,12 +34,12 @@ nanorand = "0.7.0"
 derivative = "2.2.0"
 
 # serialization
-serde = { version = "1.0.197", features = ["derive"] }
-serde_json = "1.0.116"
+serde = { version = "1.0.203", features = ["derive"] }
+serde_json = "1.0.117"
 bincode = "1.3.3"
-toml = "0.8.12"
+toml = "0.8.13"
 
-thiserror = "1.0.58"
+thiserror = "1.0.61"
 
 # handy iterators functions
 
@@ -52,7 +52,7 @@ whoami = { version = "1.5.1", optional = true }
 shell-escape = { version = "0.1.5", optional = true }
 clap = { version = "4.5.4", features = ["derive"], optional = true }
 sha2 = { version = "0.10.8", optional = true }
-base64 = { version = "0.22.0", optional = true }
+base64 = { version = "0.22.1", optional = true }
 
 # channel implementation
 flume = "0.11.0"
@@ -69,7 +69,7 @@ coarsetime = "0.1.34"
 tokio = { version = "1.37.0", features = ["rt"], default-features = false, optional = true }
 futures = { version = "0.3.30", optional = true }
 
-parking_lot = "0.12.1"
+parking_lot = "0.12.3"
 
 wyhash = "0.5.0"
 fxhash = "0.2.1"
@@ -88,7 +88,7 @@ rand = { version = "0.8.5", features = ["small_rng"] }
 tempfile = "3.10.1"
 criterion = { version = "0.5.1", features = ["html_reports"] }
 fake = "2.9.2"
-mimalloc = { version = "0.1.39", default-features = false }
+mimalloc = { version = "0.1.42", default-features = false }
 tracing-subscriber = "0.3.18"
 itertools = "0.12.1"
 

examples/avro_rw.rs

@@ -7,10 +7,10 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Parser)]
 struct Options {
- #[clap(short,long)]
+ #[clap(short, long)]
  input: Option<PathBuf>,
 
- #[clap(short,long)]
+ #[clap(short, long)]
  output: PathBuf,
 }
 
@@ -30,12 +30,20 @@ fn main() {
  let source = if let Some(input) = opts.input {
  ctx.stream_avro(input).into_boxed()
  } else {
- ctx.stream_iter((0..100).map(|i| InputType{ s: format!("{i:o}"), num: i })).into_boxed()
+ ctx.stream_iter((0..100).map(|i| InputType {
+ s: format!("{i:o}"),
+ num: i,
+ }))
+ .into_boxed()
  };
 
- source.inspect(|e| eprintln!("{e:?}"))
- .map(|mut e| { e.num *= 2; e })
+ source
+ .inspect(|e| eprintln!("{e:?}"))
+ .map(|mut e| {
+ e.num *= 2;
+ e
+ })
  .write_avro(opts.output);
 
  ctx.execute_blocking();
-}
+}

src/operator/boxed.rs

@@ -48,7 +48,7 @@ where
 }
 
 pub struct BoxedOperator<O> {
- pub(crate) op: Box<dyn DynOperator<Out=O> + 'static + Send>,
+ pub(crate) op: Box<dyn DynOperator<Out = O> + 'static + Send>,
 }
 
 impl<T> Clone for BoxedOperator<T> {
@@ -66,10 +66,8 @@ impl<T> Display for BoxedOperator<T> {
 }
 
 impl<O: Data> BoxedOperator<O> {
- pub fn new<Op: Operator<Out=O> + 'static>(op: Op) -> Self {
- Self {
- op: Box::new(op),
- }
+ pub fn new<Op: Operator<Out = O> + 'static>(op: Op) -> Self {
+ Self { op: Box::new(op) }
  }
 }
 
@@ -95,7 +93,7 @@ where
  Op::Out: Clone + Send + 'static,
 {
  /// Erase operator type using dynamic dispatching.
- /// 
+ ///
  /// Use only when strictly necessary as it is decrimental for performance.
  pub fn into_boxed(self) -> Stream<BoxedOperator<Op::Out>> {
  self.add_operator(|prev| BoxedOperator::new(prev))

src/operator/mod.rs

@@ -6,7 +6,6 @@
 use std::fmt::Display;
 use std::hash::Hash;
 use std::ops::{AddAssign, Div};
-use std::path::PathBuf;
 
 use flume::{unbounded, Receiver};
 #[cfg(feature = "tokio")]
@@ -26,7 +25,6 @@ use crate::{BatchMode, KeyedStream, Stream};
 #[cfg(feature = "tokio")]
 use self::map_async::MapAsync;
 use self::map_memo::MapMemo;
-use self::sink::avro::AvroSink;
 use self::sink::collect::Collect;
 use self::sink::collect_channel::CollectChannelSink;
 use self::sink::collect_count::CollectCountSink;
@@ -60,6 +58,7 @@ use self::{
 #[cfg(feature = "timestamp")]
 mod add_timestamps;
 mod batch_mode;
+mod boxed;
 pub(crate) mod end;
 mod filter;
 mod filter_map;
@@ -88,7 +87,6 @@ pub mod source;
 mod start;
 pub mod window;
 mod zip;
-mod boxed;
 
 /// Marker trait that all the types inside a stream should implement.
 pub trait Data: Clone + Send + 'static {}

src/operator/sink/avro.rs

@@ -3,12 +3,10 @@ use serde::Serialize;
 use std::fmt::Display;
 use std::fs::File;
 use std::io::BufWriter;
-use std::marker::PhantomData;
 use std::path::PathBuf;
 
 use crate::block::{BlockStructure, OperatorKind, OperatorStructure};
-use crate::operator::sink::StreamOutputRef;
-use crate::operator::{ExchangeData, Operator, StreamElement};
+use crate::operator::{Operator, StreamElement};
 use crate::scheduler::ExecutionMetadata;
 use crate::Stream;
 
@@ -67,6 +65,7 @@ where
  .read(true)
  .write(true)
  .create(true)
+ .truncate(true)
  .open(&self.path)
  .unwrap_or_else(|err| {
  panic!(
@@ -75,7 +74,6 @@ where
  )
  });
 
-
  let buf_writer = BufWriter::new(file);
  self.writer = Some(buf_writer);
  }
@@ -113,11 +111,11 @@ where
  }
 }
 
-impl<Op: Operator> Stream<Op> where
+impl<Op: Operator> Stream<Op>
+where
  Op: 'static,
- Op::Out: AvroSchema + Serialize
+ Op::Out: AvroSchema + Serialize,
 {
-
  /// Apply the given function to all the elements of the stream, consuming the stream.
  ///
  /// ## Example
@@ -131,8 +129,7 @@ impl<Op: Operator> Stream<Op> where
  ///
  /// env.execute_blocking();
  /// ```
- pub fn write_avro<P: Into<PathBuf>>(self, path: P)
- {
+ pub fn write_avro<P: Into<PathBuf>>(self, path: P) {
  self.add_operator(|prev| AvroSink::new(prev, path))
  .finalize_block();
  }

src/operator/sink/mod.rs

@@ -5,12 +5,12 @@
 
 use std::sync::{Arc, Mutex};
 
+pub(super) mod avro;
 pub(super) mod collect;
 pub(super) mod collect_channel;
 pub(super) mod collect_count;
 pub(super) mod collect_vec;
 pub(super) mod for_each;
-pub(super) mod avro;
 
 pub(crate) type StreamOutputRef<Out> = Arc<Mutex<Option<Out>>>;
 

src/operator/source/avro.rs

@@ -1,7 +1,6 @@
 use std::fmt::Display;
 use std::fs::File;
-use std::io;
-use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
+use std::io::BufReader;
 use std::marker::PhantomData;
 use std::path::PathBuf;
 
@@ -85,7 +84,7 @@ impl<Out: Data + for<'a> Deserialize<'a>> Source for AvroSource<Out> {
 impl<Out: Data + for<'a> Deserialize<'a>> Operator for AvroSource<Out> {
  type Out = Out;
 
- fn setup(&mut self, metadata: &mut ExecutionMetadata) {
+ fn setup(&mut self, _metadata: &mut ExecutionMetadata) {
  // let global_id = metadata.global_id;
  // let instances = metadata.replicas.len();
 
@@ -118,7 +117,10 @@ impl<Out: Data + for<'a> Deserialize<'a>> Operator for AvroSource<Out> {
  match reader.next() {
  Some(Ok(el)) => {
  tracing::trace!("avro Value: {el:?}");
- StreamElement::Item(apache_avro::from_value(&el).expect("could not deserialize from avro Value to specified type"))
+ StreamElement::Item(
+ apache_avro::from_value(&el)
+ .expect("could not deserialize from avro Value to specified type"),
+ )
  }
  Some(Err(e)) => panic!("Error while reading Aveo file: {:?}", e),
  None => {
@@ -163,14 +165,14 @@ impl crate::StreamContext {
 
 #[cfg(test)]
 mod tests {
- use std::io::Write;
+ // use std::io::Write;
 
- use itertools::Itertools;
- use serde::{Deserialize, Serialize};
- use tempfile::NamedTempFile;
+ // use itertools::Itertools;
+ // use serde::{Deserialize, Serialize};
+ // use tempfile::NamedTempFile;
 
- use crate::config::RuntimeConfig;
- use crate::environment::StreamContext;
+ // use crate::config::RuntimeConfig;
+ // use crate::environment::StreamContext;
  // use crate::operator::source::AvroSource;
 
  // #[test]

src/operator/source/mod.rs

@@ -3,22 +3,22 @@
 pub use self::csv::*;
 #[cfg(feature = "tokio")]
 pub use async_stream::*;
+pub use avro::*;
 pub use channel::*;
 pub use file::*;
 pub use iterator::*;
 pub use parallel_iterator::*;
-pub use avro::*;
 
 use crate::{block::Replication, operator::Operator};
 
 #[cfg(feature = "tokio")]
 mod async_stream;
+mod avro;
 mod channel;
 mod csv;
 mod file;
 mod iterator;
 mod parallel_iterator;
-mod avro;
 
 /// This trait marks all the operators that can be used as sinks.
 pub trait Source: Operator {