Issues with passing unsized and dynamic dispatch based traits as arguments

Question

I am trying to program a network of transducers (called Query below) which feed into each other.

A Query<A, B> takes in input items of type A, and outputs values in B. It does so by feeding them to a sink Sink<B> which exposes appropriate methods to consume elements.

pub trait Sink<A> {
    fn init(&mut self);
    fn next(&mut self, item: A);
    fn end(&mut self);
}

pub trait Query<A,B>: {
    fn init(&mut self, sink: &mut dyn Sink<B>);
    fn next(&mut self, item: A, sink: &mut dyn Sink<B>);
    fn end(&mut self, sink: &mut dyn Sink<B>);
}

Now, given a Query<A, B> and a Query<B, C>, one can compose them to form a Query<A, C>. I will call this operation Pipeline. Below, I have defined a constructor for Pipeline for convenience.

pub struct Pipeline<'a, A, B, C, QAB: Query<A, B>, QBC: Query<B, C> > {
    q_ab: &'a mut QAB,
    q_bc: &'a mut QBC,
    _abc : PhantomData<(A, B, C)>
}

impl<'a, A, B, C, QAB: Query<A, B>, QBC: Query<B, C>> Pipeline<'a, A, B, C, QAB, QBC>{
    pub fn new(q1: &'a mut QAB, q2: &'a mut QBC) -> Self{
       Self {q_ab : q1, q_bc: q2, _abc:PhantomData}
    }
}

To implement the Pipeline, we need an intermediate sink whose sole job is to push the output of Query<A, B> to that of Query<B, C>

struct PushSink<'a, B, C, QBC : Query<B, C>, S : Sink<C> + ?Sized> {
    q_bc : &'a mut QBC,
    s_c : &'a mut S,
    _bc : PhantomData<(B, C)>
}

impl <'a, B, C, QBC: Query<B, C>, S: Sink<C> + ?Sized> PushSink<'a, B, C, QBC, S>{
    pub fn new(q_bc: &'a mut QBC, s_c : &'a mut S) -> Self{
        PushSink {q_bc : q_bc, s_c : s_c, _bc : PhantomData}
    }
}

impl<'a, B, C, QBC: Query<B, C>, S: Sink<C> + ?Sized> Sink<B> for PushSink<'a, B, C, QBC, S>{
    fn init(&mut self) {
        self.q_bc.init(self.s_c);
    }
    fn next(&mut self, item: B) {
        self.q_bc.next(item, self.s_c);
    }
    fn end(&mut self) {
        self.q_bc.end(self.s_c);
    }
}

Finally, we can use this definition to define how Pipeline acts as a Query<A, C>

impl<'a, A, B, C, QAB: Query<A, B>, QBC: Query<B, C>> Query<A, C> for Pipeline<'a, A, B, C, QAB, QBC>{
    fn init(&mut self, sink: &mut dyn Sink<C>){
        self.q_ab.init(&mut PushSink::new(self.q_bc, sink))
    }
    fn next(&mut self, item: A, sink: &mut dyn Sink<C>){
        self.q_ab.next(item, &mut PushSink::new(self.q_bc, sink))
    }
    fn end(&mut self, sink: &mut dyn Sink<C>){
        self.q_ab.end(&mut PushSink::new(self.q_bc, sink))
    }
}

However, this does not work because

error[E0277]: the size for values of type `S` cannot be known at compilation time
  --> test.rs:47:23
   |
39 | impl<'a, B, C, QBC: Query<B, C>, S: Sink<C> + ?Sized> Sink<B> for PushSink<'a, B, C, QBC, S>{
   |                                  - this type parameter needs to be `std::marker::Sized`
...
47 |         self.q_bc.end(self.s_c);
   |                       ^^^^^^^^ doesn't have a size known at compile-time
   |
   = help: the trait `std::marker::Sized` is not implemented for `S`
   = note: to learn more, visit <https://doc.rust-lang.org/book/ch19-04-advanced-types.html#dynamically-sized-types-and-the-sized-trait>
   = note: required for the cast to the object type `dyn Sink<C>`

How can I fix this situation?

I know how to implement the same ideas in Java where I use interfaces instead of traits. It seems to work with no problems in Java.

Also, I want a trait based solution here. I want to have Query<A, B> and Sink<B>, etc as traits rather than something else (for example, an enumerated data type).

I would also want to avoid boxes, if possible. I would much prefer a solution using generics and traits instead of boxes.

I have also tried changing the signature of Query<,> to

pub trait Query<A,B>: {
    fn init(&mut self, sink: &mut (dyn Sink<B> + ?Sized));
    fn next(&mut self, item: A, sink: &mut (dyn Sink<B> + ?Sized));
    fn end(&mut self, sink: &mut dyn Sink<B>);
}

but apparently, this is not allowed

Full source code here, for convenience

EDIT: Playground link, thanks to @eggyal