Fix

Author: trueleo

server/src/handlers/livetail.rs

@@ -6,6 +6,7 @@ use futures::future;
 use futures::stream::BoxStream;
 use futures_util::{Future, StreamExt, TryFutureExt, TryStreamExt};
 use http_auth_basic::Credentials;
+use rand::distributions::{Alphanumeric, DistString};
 use tonic::metadata::MetadataMap;
 use tonic::transport::Server;
 use tonic::{Request, Response, Status, Streaming};
@@ -22,6 +23,7 @@ use crate::livetail::{Message, LIVETAIL};
 use crate::metadata::STREAM_INFO;
 use crate::rbac::map::SessionKey;
 use crate::rbac::{self, Users};
+use crate::utils;
 
 use super::SESSION_COOKIE_NAME;
 
@@ -90,10 +92,15 @@ impl FlightService for FlightServiceImpl {
             .schema(stream)
             .map_err(|err| Status::failed_precondition(err.to_string()))?;
 
-        let rx = LIVETAIL.new_pipe("a".to_string(), stream.to_string());
-        let rx = rx.filter_map(|x| {
+        let rx = LIVETAIL.new_pipe(
+            Alphanumeric.sample_string(&mut rand::thread_rng(), 32),
+            stream.to_string(),
+        );
+
+        let adapter_schema = schema.clone();
+        let rx = rx.filter_map(move |x| {
             future::ready(match x {
-                Message::Record(t) => Some(Ok(t)),
+                Message::Record(t) => Some(Ok(utils::arrow::adapt_batch(&adapter_schema, &t))),
                 _ => None,
             })
         });