Repository: influxdata/influxdb
File: influxdb3_server/src/http.rs (Rust)

Code before:
                    }
                }
            }

            let mut future = JsonFuture {
                state: State::FirstPoll,
                stream,
            };
            Ok(Body::wrap_stream(futures::stream::poll_fn(move |ctx| {
                future.poll_unpin(ctx)
            })))
        }
        QueryFormat::JsonLines => {
            let stream = futures::stream::poll_fn(move |ctx| match stream.poll_next_unpin(ctx) {
                Poll::Ready(Some(batch)) => {
                    let batch = match batch {
                        Ok(batch) => batch,
                        Err(e) => return Poll::Ready(Some(Err(e))),
                    };
                    let mut writer = arrow_json::LineDelimitedWriter::new(Vec::new());
                    writer.write(&batch).unwrap();
                    writer.finish().unwrap();
                    Poll::Ready(Some(Ok(Bytes::from(writer.into_inner()))))
                }
                Poll::Ready(None) => Poll::Ready(None),
                Poll::Pending => Poll::Pending,
            });
            Ok(Body::wrap_stream(stream))
        }
    }
}

// This is a hack around the fact that bool default is false not true
const fn true_fn() -> bool {
    true
}
#[derive(Debug, Deserialize)]
pub(crate) struct WriteParams {
    pub(crate) db: String,
    #[serde(default = "true_fn")]
    pub(crate) accept_partial: bool,
    #[serde(default)]
    pub(crate) precision: Precision,
    #[serde(default)]
    pub(crate) no_sync: bool,
}

impl From<iox_http::write::WriteParams> for WriteParams {
    fn from(legacy: iox_http::write::WriteParams) -> Self {
        Self {
            db: legacy.namespace.to_string(),

Review comment (performance):
Would it be possible (or make sense if it is possible) to store the `ArrayWriter` in the state enum so it doesn't have to be reconstructed for every `RecordBatch`? Also, does the `ArrayWriter` automatically take care of adding the opening square bracket? Just curious because I see the closing square bracket written in this future but not the opening one when the stream is non-empty.

Code after:
                    is_first_row: bool,
                ) -> std::result::Result<(), arrow_schema::ArrowError> {
                    if !is_first_row {
                        writer.write_all(b",")?;
                    }

                    Ok(())
                }
            }

            impl Future for JsonFuture {
                type Output = Option<Result<Bytes, DataFusionError>>;

                fn poll(
                    mut self: Pin<&mut Self>,
                    ctx: &mut std::task::Context<'_>,
                ) -> Poll<Self::Output> {
                    match dbg!(self.stream.poll_next_unpin(ctx)) {
                        Poll::Ready(Some(batch)) => {
                            let batch = match batch {
                                Ok(batch) => batch,
                                Err(e) => return Poll::Ready(Some(Err(e))),
                            };
                            match self.state {
                                State::FirstPoll => {
                                    let mut writer = arrow_json::ArrayWriter::new(Vec::new());
                                    if let Err(err) = writer.write(&batch) {
                                        Poll::Ready(Some(Err(err.into())))
                                    } else {
                                        self.state = State::Body;
                                        Poll::Ready(Some(Ok(Bytes::from(writer.into_inner()))))
                                    }
                                }
                                State::Body => {
                                    let mut writer = arrow_json::WriterBuilder::new()
                                        .build::<Vec<_>, JsonMiddle>(Vec::new());
                                    if let Err(err) = writer.write(&batch) {
                                        Poll::Ready(Some(Err(err.into())))
                                    } else {
                                        Poll::Ready(Some(Ok(Bytes::from(writer.into_inner()))))
                                    }
                                }
                                _ => unreachable!(),
                            }
                        }
                        Poll::Ready(None) => {
                            match self.state {
                                // We've never written anything so just close the
                                // stream with an empty array
                                State::FirstPoll => {
                                    self.state = State::Done;