servers/postgres/copy_in/
influxdb.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! The `influxdb` line protocol format for COPY FROM STDIN.
16//!
17//! `COPY FROM STDIN WITH (FORMAT influxdb, precision 'ns')` ingests
18//! [InfluxDB line protocol](https://docs.influxdata.com/influxdb/v1.8/write_protocols/line_protocol_reference/)
19//! text. The format is table-less — each line names its own measurement
20//! — and does not parse here at all: the COPY layer buffers the whole
21//! payload (bounded by [`crate::postgres::copy_in::COPY_IN_MAX_PAYLOAD_BYTES`])
22//! and completes the copy by handing it to the frontend's line-protocol
23//! write pipeline as one body, exactly like the HTTP
24//! `/v1/influxdb/v1/write` endpoint receives one request body:
25//! conversion, permissions, interceptors, timestamp alignment, merge
26//! mode and inserts are the same path, and record boundaries (including
27//! newlines inside quoted field values and after escapes) are decided
28//! by the line-protocol parser alone — the COPY layer never splits
29//! records.
30//!
31//! One COPY statement is therefore one bounded ingestion request; a
32//! larger import is multiple COPY statements.
33//!
34//! The `precision` option carries the same values as the HTTP
35//! `precision` query parameter (`ns`, `us`, `ms`, `s`, ...), defaults to
36//! nanoseconds, and is resolved by the shared `parse_time_precision`.
37
38use common_grpc::precision::Precision;
39use pgwire::error::PgWireResult;
40
41use crate::postgres::copy_in::bad_copy_data;
42use crate::postgres::copy_in::codec::CopyFormat;
43use crate::postgres::copy_in::options::CopyOptionSet;
44
45/// The format's validated options. A `None` precision carries the
46/// pipeline's default (nanoseconds), exactly like an HTTP write without
47/// a `precision` parameter.
48#[derive(Debug, Clone)]
49pub(super) struct InfluxdbOptions {
50    pub(super) precision: Option<Precision>,
51}
52
53pub(super) fn build_influxdb_codec(mut options: CopyOptionSet) -> PgWireResult<CopyFormat> {
54    let precision = match options.take_string("precision")? {
55        Some(value) => Some(
56            crate::influxdb::parse_time_precision(&value)
57                .map_err(|e| bad_copy_data(format!("invalid precision option: {e}")))?,
58        ),
59        None => None,
60    };
61    options.reject_leftovers("influxdb")?;
62    Ok(CopyFormat::Influxdb(InfluxdbOptions { precision }))
63}
64
65#[cfg(test)]
66mod tests {
67    use super::*;
68    use crate::postgres::copy_in::options::CopyOptionValue;
69
70    #[test]
71    fn test_builder_options() {
72        let format = build_influxdb_codec(CopyOptionSet::default()).unwrap();
73        assert!(matches!(format, CopyFormat::Influxdb(_)));
74        assert!(matches!(format, CopyFormat::Influxdb(_)));
75
76        let mut set = CopyOptionSet::default();
77        set.insert("precision".into(), CopyOptionValue::String("ms".into()))
78            .unwrap();
79        assert!(matches!(
80            build_influxdb_codec(set).unwrap(),
81            CopyFormat::Influxdb(InfluxdbOptions {
82                precision: Some(Precision::Millisecond)
83            })
84        ));
85
86        let mut set = CopyOptionSet::default();
87        set.insert("precision".into(), CopyOptionValue::String("bogus".into()))
88            .unwrap();
89        assert!(build_influxdb_codec(set).is_err());
90
91        let mut set = CopyOptionSet::default();
92        set.insert("unknown".into(), CopyOptionValue::Flag).unwrap();
93        assert!(build_influxdb_codec(set).is_err());
94    }
95}
servers/postgres/copy_in/influxdb.rs

servers/postgres/copy_in/
influxdb.rs
