[squash] [scala] [streaming] Join and Cross operator wip2