This page shows how to insert data into QuestDB using different programming languages and tools. To ingest data to a running instance, there are three main methods that can be used:

  • InfluxDB line protocol which provides flexibility, ease of use, and high ingestion rates
  • Postgres wire protocol for compatibility with a range of clients
  • Rest API which can be used for importing datasets from CSV

Prerequisites

This page assumes that QuestDB is running and accessible. QuestDB can be run using either Docker, the binaries or Homebrew for macOS users.

InfluxDB line protocol

QuestDB implements InfluxDB line protocol which is accessible by default on TCP port 9009. This allows using QuestDB as a drop-in replacement for InfluxDB and others implementing the protocol. Configuration settings for ingestion using this protocol can be set for for Influx line over TCP and Influx line over UDP.

More information on the InfluxDB line protocol implementation with details on message format, ports, authentication and examples can be found on the InfluxDB API reference page. Additionally, a guide on the Telegraf agent for collecting and sending metrics to QuestDB via this protocol can be found on the Telegraf guide.

<Tabs defaultValue=”nodejs” values={[ { label: “NodeJS”, value: “nodejs” }, { label: “Go”, value: “go” }, { label: “Java”, value: “java” } ]}>

  1. const net = require("net")
  3. const client = new net.Socket()
  5. const HOST = "localhost"
  6. const PORT = 9009
  8. function run() {
  9.   client.connect(PORT, HOST, () => {
  10.     const rows = [
  11.       `trades,name=test_ilp1 value=12.4 ${Date.now() * 1e6}`,
  12.       `trades,name=test_ilp2 value=11.4 ${Date.now() * 1e6}`,
  13.     ]
  15.     rows.forEach((row) => {
  16.       client.write(`${row}\n`)
  17.     })
  19.     client.destroy()
  20.   })
  22.   client.on("data", function (data) {
  23.     console.log("Received: " + data)
  24.   })
  26.   client.on("close", function () {
  27.     console.log("Connection closed")
  28.   })
  29. }
  31. run()
  1. package main
  3. import (
  4.   "fmt"
  5.   "io/ioutil"
  6.   "net"
  7.   "time"
  8. )
  10. func main() {
  11.   host := "127.0.0.1:9009"
  12.   tcpAddr, err := net.ResolveTCPAddr("tcp4", host)
  13.   checkErr(err)
  14.   rows := [2]string{
  15.     fmt.Sprintf("trades,name=test_ilp1 value=12.4 %d", time.Now().UnixNano()),
  16.     fmt.Sprintf("trades,name=test_ilp2 value=11.4 %d", time.Now().UnixNano()),
  17.   }
  19.   conn, err := net.DialTCP("tcp", nil, tcpAddr)
  20.   checkErr(err)
  21.   defer conn.Close()
  23.   for _, s := range rows {
  24.     _, err = conn.Write([]byte(fmt.Sprintf("%s\n", s)))
  25.     checkErr(err)
  26.   }
  28.   result, err := ioutil.ReadAll(conn)
  29.   checkErr(err)
  31.   fmt.Println(string(result))
  32. }
  34. func checkErr(err error) {
  35.   if err != nil {
  36.     panic(err)
  37.   }
  38. }
  1. import io.questdb.cutlass.line.LineProtoSender;
  2. import io.questdb.cutlass.line.tcp.LineTCPProtoSender;
  3. import io.questdb.network.Net;
  4. import io.questdb.std.Os;
  6. public class LineTCPSenderMain {
  7.     /*
  8.         Maven:
  10.             <dependency>
  11.                 <groupId>org.questdb</groupId>
  12.                 <artifactId>questdb</artifactId>
  13.                 <version>{@version@}</version>
  14.             </dependency>
  16.         Gradle:
  18.             compile group: 'org.questdb', name: 'questdb', version: '{@version@}'
  20.      */
  21.     public static void main(String[] args) {
  22.         String hostIPv4 = "127.0.0.1";
  23.         int port = 9009;
  24.         int bufferCapacity = 256 * 1024;
  26.         try (LineProtoSender sender = new LineTCPProtoSender(Net.parseIPv4(hostIPv4), port, bufferCapacity)) {
  27.             sender
  28.                     .metric("trades")
  29.                     .tag("name", "test_ilp1")
  30.                     .field("value", 12.4)
  31.                     .$(Os.currentTimeNanos());
  32.             sender
  33.                     .metric("trades")
  34.                     .tag("name", "test_ilp2")
  35.                     .field("value", 11.4)
  36.                     .$(Os.currentTimeNanos());
  38.             sender.flush();
  39.         }
  40.     }
  41. }

Postgres compatibility

You can query data using the Postgres endpoint that QuestDB exposes. This is accessible via port 8812 by default. More information on the Postgres wire protocol implementation with details on supported features can be found on the Postgres API reference page.

<Tabs defaultValue=”nodejs” values={[ { label: “NodeJS”, value: “nodejs” }, { label: “Go”, value: “go” }, { label: “Rust”, value: “rust” }, { label: “Java”, value: “java” }, { label: “Python”, value: “python” }, ]}>

This example uses the pg package which allows for quickly building queries using PostgreSQL wire protocol. Details on the use of this package can be found on the node-postgres documentation.

This example uses naive Date.now() * 1000 inserts for Timestamp types in microsecond resolution. For accurate microsecond timestamps, the node-microtime package can be used which makes system calls to tv_usec from C++.

  1. const { Client } = require("pg")
  3. const start = async () => {
  4.   try {
  5.     const client = new Client({
  6.       database: "qdb",
  7.       host: "127.0.0.1",
  8.       password: "quest",
  9.       port: 8812,
  10.       user: "admin",
  11.     })
  12.     await client.connect()
  14.     const createTable = await client.query(
  15.       "CREATE TABLE IF NOT EXISTS trades (ts TIMESTAMP, date DATE, name STRING, value INT) timestamp(ts);",
  16.     )
  17.     console.log(createTable)
  19.     const insertData = await client.query(
  20.       "INSERT INTO trades VALUES($1, $2, $3, $4);",
  21.       [Date.now() * 1000, Date.now(), "node pg example", 123],
  22.     )
  23.     await client.query("COMMIT")
  25.     console.log(insertData)
  27.     for (let rows = 0; rows < 10; rows++) {
  28.       // Providing a 'name' field allows for prepared statements / bind variables
  29.       const query = {
  30.         name: "insert-values",
  31.         text: "INSERT INTO trades VALUES($1, $2, $3, $4);",
  32.         values: [Date.now() * 1000, Date.now(), "node pg prep statement", rows],
  33.       }
  34.       const preparedStatement = await client.query(query)
  35.     }
  37.     await client.query("COMMIT")
  39.     const readAll = await client.query("SELECT * FROM trades")
  40.     console.log(readAll.rows)
  42.     await client.end()
  43.   } catch (e) {
  44.     console.log(e)
  45.   }
  46. }
  48. start()

This example uses the pgx driver and toolkit for postgres in Go. More details on the use of this toolkit can be found on the GitHub repository for pgx.

  1. package main
  3. import (
  4.   "context"
  5.   "fmt"
  6.   "log"
  7.   "time"
  9.   "github.com/jackc/pgx/v4"
  10. )
  12. var conn *pgx.Conn
  13. var err error
  15. func main() {
  16.   ctx := context.Background()
  17.   conn, _ = pgx.Connect(ctx, "postgresql://admin:quest@localhost:8812/qdb")
  18.   defer conn.Close(ctx)
  20.   // text-based query
  21.   _, err := conn.Exec(ctx, "CREATE TABLE IF NOT EXISTS trades (ts TIMESTAMP, date DATE, name STRING, value INT) timestamp(ts);")
  22.   if err != nil {
  23.     log.Fatalln(err)
  24.   }
  26.   // Prepared statement given the name 'ps1'
  27.   _, err = conn.Prepare(ctx, "ps1", "INSERT INTO trades VALUES($1,$2,$3,$4)")
  28.   if err != nil {
  29.     log.Fatalln(err)
  30.   }
  31.   for i := 0; i < 10; i++ {
  32.     // Execute 'ps1' statement with a string and the loop iterator value
  33.     _, err = conn.Exec(ctx, "ps1", time.Now(), time.Now().Round(time.Millisecond), "go prepared statement", i+1)
  34.     if err != nil {
  35.       log.Fatalln(err)
  36.     }
  37.   }
  39.   // Read all rows from table
  40.   rows, err := conn.Query(ctx, "SELECT * FROM trades")
  41.   fmt.Println("Reading from trades table:")
  42.   for rows.Next() {
  43.     var name string
  44.     var value int64
  45.     var ts time.Time
  46.     var date time.Time
  47.     err = rows.Scan(&ts, &date, &name, &value)
  48.     fmt.Println(ts, date, name, value)
  49.   }
  51.   err = conn.Close(ctx)
  52. }

The following example shows how to use parameterized queries and prepared statements using the rust-postgres client.

  1. use postgres::{Client, NoTls, Error};
  2. use chrono::{Utc};
  3. use std::time::SystemTime;
  5. fn main() -> Result<(), Error> {
  6.     let mut client = Client::connect("postgresql://admin:quest@localhost:8812/qdb", NoTls)?;
  8.     // Basic query
  9.     client.batch_execute("CREATE TABLE IF NOT EXISTS trades (ts TIMESTAMP, date DATE, name STRING, value INT) timestamp(ts);")?;
  11.     // Parameterized query
  12.     let name: &str = "rust example";
  13.     let val: i32 = 123;
  14.     let utc = Utc::now();
  15.     let sys_time = SystemTime::now();
  16.     client.execute(
  17.         "INSERT INTO trades VALUES($1,$2,$3,$4)",
  18.         &[&utc.naive_local(), &sys_time, &name, &val],
  19.     )?;
  21.     // Prepared statement
  22.     let mut txn = client.transaction()?;
  23.     let statement = txn.prepare("insert into trades values ($1,$2,$3,$4)")?;
  24.     for value in 0..10 {
  25.         let utc = Utc::now();
  26.         let sys_time = SystemTime::now();
  27.         txn.execute(&statement, &[&utc.naive_local(), &sys_time, &name, &value])?;
  28.     }
  29.     txn.commit()?;
  31.     println!("import finished");
  32.     Ok(())
  33. }
  1. package com.myco;
  3. import java.sql.*;
  4. import java.util.Properties;
  6. class App {
  7.   public static void main(String[] args) throws SQLException {
  8.     Properties properties = new Properties();
  9.     properties.setProperty("user", "admin");
  10.     properties.setProperty("password", "quest");
  11.     properties.setProperty("sslmode", "disable");
  13.     final Connection connection = DriverManager.getConnection("jdbc:postgresql://localhost:8812/qdb", properties);
  14.     connection.setAutoCommit(false);
  16.     final PreparedStatement statement = connection.prepareStatement("CREATE TABLE IF NOT EXISTS trades (ts TIMESTAMP, date DATE, name STRING, value INT) timestamp(ts);");
  17.     statement.execute();
  19.     try (PreparedStatement preparedStatement = connection.prepareStatement("INSERT INTO TRADES  VALUES (?, ?, ?, ?)")) {
  20.       preparedStatement.setTimestamp(1, new Timestamp(io.questdb.std.Os.currentTimeMicros()));
  21.       preparedStatement.setDate(2, new Date(System.currentTimeMillis()));
  22.       preparedStatement.setString(3, "abc");
  23.       preparedStatement.setInt(4, 123);
  24.       preparedStatement.execute();
  25.     }
  26.     System.out.println("Done");
  27.     connection.close();
  28.   }
  29. }

This example uses the psychopg2 database adapter which does not support prepared statements (bind variables). This functionality is on the roadmap for the antecedent psychopg3 adapter.

  1. import psycopg2 as pg
  2. import datetime as dt
  4. try:
  5.     connection = pg.connect(user="admin",
  6.                             password="quest",
  7.                             host="127.0.0.1",
  8.                             port="8812",
  9.                             database="qdb")
  10.     cursor = connection.cursor()
  12.     # text-only query
  13.     cursor.execute("CREATE TABLE IF NOT EXISTS trades (ts TIMESTAMP, date DATE, name STRING, value INT) timestamp(ts);")
  15.     # insert 10 records
  16.     for x in range(10):
  17.       now = dt.datetime.utcnow()
  18.       date = dt.datetime.now().date()
  19.       cursor.execute("""
  20.         INSERT INTO trades
  21.         VALUES (%s, %s, %s, %s);
  22.         """, (now, date, "python example", x))
  23.     # commit records
  24.     connection.commit()
  26.     cursor.execute("SELECT * FROM trades;")
  27.     records = cursor.fetchall()
  28.     for row in records:
  29.         print(row)
  31. finally:
  32.     if (connection):
  33.         cursor.close()
  34.         connection.close()
  35.         print("PostgreSQL connection is closed")

REST API

QuestDB exposes a REST API for compatibility with a wide range of libraries and tools. The REST API is accessible on port 9000 and has the following entrypoints:

  • /imp - import data
  • /exec - execute an SQL statement

More details on the use of these entrypoints can be found on the REST API reference page.

/imp endpoint

The /imp endpoint allows for importing a CSV file directly.

import Tabs from “@theme/Tabs” import TabItem from “@theme/TabItem”

<Tabs defaultValue=”curl” values={[ { label: “cURL”, value: “curl” }, { label: “NodeJS”, value: “nodejs” }, { label: “Go”, value: “go” }, ]}>

This example imports a CSV file with automatic schema detection.

  1. curl -F data=@data.csv http://localhost:9000/imp

This example overwrites an existing table, specifies a timestamp format and a designated timestamp column. For more information on the optional parameters for specifying timestamp formats, partitioning and renaming tables, see the REST API documentation.

  1. curl \
  2. -F schema='[{"name":"ts", "type": "TIMESTAMP", "pattern": "yyyy-MM-dd - HH:mm:ss"}]' \
  3. -F data=@weather.csv 'http://localhost:9000/imp?overwrite=true&timestamp=ts'
  1. const fetch = require("node-fetch")
  2. const FormData = require("form-data")
  3. const fs = require("fs")
  4. const qs = require("querystring")
  6. const HOST = "http://localhost:9000"
  8. async function run() {
  9.   const form = new FormData()
  11.   form.append("data", fs.readFileSync(__dirname + "/data.csv"), {
  12.     filename: fileMetadata.name,
  13.     contentType: "application/octet-stream",
  14.   })
  16.   try {
  17.     const r = await fetch(`${HOST}/imp`, {
  18.       method: "POST",
  19.       body: form,
  20.       headers: form.getHeaders(),
  21.     })
  23.     console.log(r)
  24.   } catch (e) {
  25.     console.error(e)
  26.   }
  27. }
  29. run()
  1. package main
  3. import (
  4.   "bytes"
  5.   "fmt"
  6.   "io"
  7.   "io/ioutil"
  8.   "log"
  9.   "mime/multipart"
  10.   "net/http"
  11.   "net/url"
  12.   "os"
  13. )
  15. func main() {
  16.   u, err := url.Parse("http://localhost:9000")
  17.   checkErr(err)
  18.   u.Path += "imp"
  19.   url := fmt.Sprintf("%v", u)
  20.   fileName := "/path/to/data.csv"
  21.   file, err := os.Open(fileName)
  22.   checkErr(err)
  24.   defer file.Close()
  26.   buf := new(bytes.Buffer)
  27.   writer := multipart.NewWriter(buf)
  28.   uploadFile, _ := writer.CreateFormFile("data", "data.csv")
  29.   _, err = io.Copy(uploadFile, file)
  30.   checkErr(err)
  31.   writer.Close()
  33.   req, err := http.NewRequest(http.MethodPut, url, buf)
  34.   checkErr(err)
  35.   req.Header.Add("Content-Type", writer.FormDataContentType())
  37.   client := &http.Client{}
  38.   res, err := client.Do(req)
  39.   checkErr(err)
  41.   defer res.Body.Close()
  43.   body, err := ioutil.ReadAll(res.Body)
  44.   checkErr(err)
  46.   log.Println(string(body))
  47. }
  49. func checkErr(err error) {
  50.   if err != nil {
  51.     panic(err)
  52.   }
  53. }

/exec endpoint

Alternatively, the /exec endpoint can be used to create a table and the INSERT statement can be used to populate it with values:

<Tabs defaultValue=”curl” values={[ { label: “cURL”, value: “curl” }, { label: “NodeJS”, value: “nodejs” }, { label: “Go”, value: “go” }, ]}>

  1. # Create Table
  2. curl -G \
  3.   --data-urlencode "query=CREATE TABLE IF NOT EXISTS trades(name STRING, value INT)" \
  4.   http://localhost:9000/exec
  6. # Insert a row
  7. curl -G \
  8.   --data-urlencode "query=INSERT INTO trades VALUES('abc', 123456);" \
  9.   http://localhost:9000/exec

Note that these two queries can be combined into a single curl request:

  1. curl -G \
  2.   --data-urlencode "query=CREATE TABLE IF NOT EXISTS trades(name STRING, value INT);\
  3.   INSERT INTO trades VALUES('abc', 123456);" \
  4.   http://localhost:9000/exec

The node-fetch package can be installed using npm i node-fetch.

  1. const fetch = require("node-fetch")
  2. const qs = require("querystring")
  4. const HOST = "http://localhost:9000"
  6. async function createTable() {
  7.   try {
  8.     const queryData = {
  9.       query: "CREATE TABLE IF NOT EXISTS trades (name STRING, value INT);",
  10.     }
  12.     const response = await fetch(`${HOST}/exec?${qs.encode(queryData)}`)
  13.     const json = await response.json()
  15.     console.log(json)
  16.   } catch (error) {
  17.     console.log(error)
  18.   }
  19. }
  21. async function insertData() {
  22.   try {
  23.     const queryData = {
  24.       query: "INSERT INTO trades VALUES('abc', 123456);",
  25.     }
  27.     const response = await fetch(`${HOST}/exec?${qs.encode(queryData)}`)
  28.     const json = await response.json()
  30.     console.log(json)
  31.   } catch (error) {
  32.     console.log(error)
  33.   }
  34. }
  36. createTable()
  37. insertData()
  1. package main
  3. import (
  4.   "fmt"
  5.   "io/ioutil"
  6.   "log"
  7.   "net/http"
  8.   "net/url"
  9. )
  11. func main() {
  12.   u, err := url.Parse("http://localhost:9000")
  13.   checkErr(err)
  15.   u.Path += "exec"
  16.   params := url.Values{}
  17.   params.Add("query", `
  18.     CREATE TABLE IF NOT EXISTS
  19.       trades (name STRING, value INT);
  20.     INSERT INTO
  21.       trades
  22.     VALUES(
  23.       "abc",
  24.       123456
  25.     );
  26.   `)
  27.   u.RawQuery = params.Encode()
  28.   url := fmt.Sprintf("%v", u)
  30.   res, err := http.Get(url)
  31.   checkErr(err)
  33.   defer res.Body.Close()
  35.   body, err := ioutil.ReadAll(res.Body)
  36.   checkErr(err)
  38.   log.Println(string(body))
  39. }
  41. func checkErr(err error) {
  42.   if err != nil {
  43.     panic(err)
  44.   }
  45. }

Web Console

By default, QuestDB has an embedded Web Console running at http://\[server-address\]:9000. When running locally, this is accessible at http://localhost:9000. The Web Console can be used to explore table schemas, visualizing query results as tables or graphs, and importing datasets from CSV files. For details on these components, refer to the Web Console reference page.