Insert data

This page shows how to insert data into QuestDB using different programming languages and tools.

Overview

QuestDB supports the following data ingestion methods:

• InfluxDB Line Protocol: the recommended primary ingestion method in QuestDB for high-performance applications.

  • Dedicated InfluxDB Line Protocol client libraries available

• PostgreSQL wire protocol: interoperability with the PostgreSQL ecosystem.

  • SQL INSERT and COPY statements, including parameterized queries.
  • psql on the command line
  • Support for most PostgreSQL keywords and functions

• HTTP REST API: compatibility with a wide range of libraries and tools.

  • SQL INSERT for ad-hoc SQL queries
  • curl command and CSV file upload on the commend line
  • Accessing QuestDB via the Web Console:
    • Code editor for SQL INSERT queries
    • SQL COPY for one-off database migration
    • CSV file upload for uploading batches of CSV files

Recommended insert method

The table below outlines the general recommendation for data ingestion based on the shape of the data and different scenarios:

One-off data import

	CSV Upload	SQL COPY (Web Console)	InfluxDB Line Protocol	PostgreSQL
Sorted	✓	✓
Lightly out of order	✓
Heavily out of order		✓

Periodic batch ingest

	CSV Upload	SQL COPY (Web Console)	InfluxDB Line Protocol	PostgreSQL
Sorted	✓		✓	✓
Lightly out of order	✓		✓	✓
Heavily out of order	✓		✓	✓

Real-time ingest

	CSV Upload	SQL COPY (Web Console)	InfluxDB Line Protocol	PostgreSQL
Sorted			✓
Lightly out of order			✓
Heavily out of order			✓

Lightly out of order data refers to data with the following traits:

• The expected lag is usually within a few minutes.
• The data is mostly sorted. Timestamps are growing in time with occasional exceptions that are within the lag.

Heavily out of order data refers to data with the following traits:

• The data is mostly unsorted.
• The data belongs to different parts of different partitions in an arbitrary manner.

InfluxDB Line Protocol (ILP)

The InfluxDB Line Protocol is a text protocol over TCP on port 9009.

It is a one-way protocol to insert data, focusing on simplicity and performance.

This interface is the preferred ingestion method as it provides the following benefits:

• High-throughput ingestion
• Robust ingestion from multiple sources into tables with dedicated systems for reducing congestion
• Supports on-the-fly, concurrent schema changes

On the InfluxDB line protocol page, you may find additional details on the message format, ports and authentication.

Client libraries

The Client Libraries provide user-friendly InfluxDB Line Protocol clients for a growing number of languages.

Authentication

By default, Open Source InfluxDB Line Protocol Server is unauthenticated. To configure authentication on the server, follow our server configuration guide. To configure authentication on the client, follow the relevant documentation section in the Client Libraries overview.

QuestDB Cloud servers are configured for authentication already. Snippets for all the supported languages can be found at https://cloud.questdb.com under the instance "Connect" tab.

Examples

These examples send a few rows of input. These use client libraries as well as raw TCP socket connections, when a client library is not available.

Python

Python client library docs and repo.

See more examples, including ingesting data from Pandas dataframes.

python3 -m pip install questdb

from questdb.ingress import Sender, IngressError, TimestampNanos

import sys

import datetime

def example(host: str = 'localhost', port: int = 9009):

try:

with Sender(host, port) as sender:

# Record with provided designated timestamp (using the 'at' param)

# Notice the designated timestamp is expected in Nanoseconds,

# but timestamps in other columns are expected in Microseconds.

# The API provides convenient functions

sender.row(

'trades',

symbols={

'pair': 'USDGBP',

'type': 'buy'},

columns={

'traded_price': 0.83,

'limit_price': 0.84,

'qty': 100,

'traded_ts': datetime.datetime(

2022, 8, 6, 7, 35, 23, 189062,

tzinfo=datetime.timezone.utc)},

at=TimestampNanos.now())

# If no 'at' param is passed, the server will use its own timestamp.

sender.row(

'trades',

symbols={'pair': 'EURJPY'},

columns={

'traded_price': 135.97,

'qty': 400,

'limit_price': None}) # NULL columns can be passed as None,

# or simply be left out.

# We recommend flushing periodically, for example every few seconds.

# If you don't flush explicitly, the client will flush automatically

# once the buffer is reaches 63KiB and just before the connection

# is closed.

sender.flush()

except IngressError as e:

sys.stderr.write(f'Got error: {e}\n')

if __name__ == '__main__':

example()

Go

Go client library docs and repo.

package main

import (

"context"

"log"

"time"

qdb "github.com/questdb/go-questdb-client"

)

func main() {

ctx := context.TODO()

// Connect to QuestDB running on 127.0.0.1:9009

sender, err := qdb.NewLineSender(ctx)

if err != nil {

log.Fatal(err)

}

// Make sure to close the sender on exit to release resources.

defer sender.Close()

// Send a few ILP messages.

err = sender.

Table("trades").

Symbol("name", "test_ilp1").

Float64Column("value", 12.4).

At(ctx, time.Now().UnixNano())

if err != nil {

log.Fatal(err)

}

err = sender.

Table("trades").

Symbol("name", "test_ilp2").

Float64Column("value", 11.4).

At(ctx, time.Now().UnixNano())

if err != nil {

log.Fatal(err)

}

// Make sure that the messages are sent over the network.

err = sender.Flush(ctx)

if err != nil {

log.Fatal(err)

}

}

Java

Java client library docs and Maven artifact.

Maven

<dependency>
    <groupId>org.questdb</groupId>
    <artifactId>questdb</artifactId>
    <version>6.5</version>
</dependency>

Gradle

compile group: 'org.questdb', name: 'questdb', version: '6.5'

package com.example.sender;

import io.questdb.client.Sender;

public class BasicExample {

public static void main(String[] args) {

try (Sender sender = Sender.builder().address("localhost:9009").build()) {

sender.table("inventors")

.symbol("born", "Austrian Empire")

.longColumn("id", 0)

.stringColumn("name", "Nicola Tesla")

.atNow();

sender.table("inventors")

.symbol("born", "USA")

.longColumn("id", 1)

.stringColumn("name", "Thomas Alva Edison")

.atNow();

}

}

}

NodeJS

NodeJS client library repo.

const { Sender } = require("@questdb/nodejs-client");

async function run() {

// create a sender with a 4KB buffer

const sender = new Sender({ bufferSize: 4096 });

// connect to QuestDB

// host and port are required in connect options

await sender.connect({ port: 9009, host: "localhost" });

// add rows to the buffer of the sender

let bday = Date.parse("1856-07-10");

sender

.table("inventors")

.symbol("born", "Austrian Empire")

.timestampColumn("birthday", BigInt(bday) * 1000n) // epoch in micros (BigInt)

.intColumn("id", 0)

.stringColumn("name", "Nicola Tesla")

.at(BigInt(Date.now()) * 1000_000n); // epoch in nanos (BigInt)

bday = Date.parse("1847-02-11");

sender

.table("inventors")

.symbol("born", "USA")

.timestampColumn("birthday", BigInt(bday) * 1000n)

.intColumn("id", 1)

.stringColumn("name", "Thomas Alva Edison")

.atNow();

// flush the buffer of the sender, sending the data to QuestDB

// the buffer is cleared after the data is sent and the sender is ready to accept new data

await sender.flush();

// close the connection after all rows were sent

await sender.close();

return 0;

}

run()

.then(console.log)

.catch(console.error);

C#

.NET client library

using QuestDB;

using System;

using System.Threading.Tasks;

namespace QuestDBDemo

{

class Program

{

static async Task Main(string[] args)

{

using var sender = await LineTcpSender.ConnectAsync(

"localhost",

9009,

tlsMode: TlsMode.Disable);

sender.Table("trades")

.Symbol("pair", "USDGBP")

.Symbol("type", "buy")

.Column("traded_price", 0.83)

.Column("limit_price", 0.84)

.Column("qty", 100)

.Column("traded_ts", new DateTime(

2022, 8, 6, 7, 35, 23, 189, DateTimeKind.Utc))

.At(DateTime.UtcNow);

sender.Table("trades")

.Symbol("pair", "GBPJPY")

.Column("traded_price", 135.97)

.Column("qty", 400)

.AtNow();

await sender.SendAsync();

}

}

}

C

C client library docs

#include <questdb/ingress/line_sender.h>

#include <stdio.h>

#include <stdbool.h>

#include <string.h>

static bool example(const char* host, const char* port)

{

line_sender_error* err = NULL;

line_sender_opts* opts = NULL;

line_sender* sender = NULL;

line_sender_buffer* buffer = NULL;

line_sender_utf8 host_utf8 = { 0, NULL };

if (!line_sender_utf8_init(&host_utf8, strlen(host), host, &err))

goto on_error;

line_sender_utf8 port_utf8 = { 0, NULL };

if (!line_sender_utf8_init(&port_utf8, strlen(port), port, &err))

goto on_error;

// Call `line_sender_opts_new` if instead you have an integer port.

opts = line_sender_opts_new_service(host_utf8, port_utf8);

sender = line_sender_connect(opts, &err);

line_sender_opts_free(opts);

opts = NULL;

if (!sender)

goto on_error;

buffer = line_sender_buffer_new();

line_sender_buffer_reserve(buffer, 64 * 1024); // 64KB buffer initial size.

// We prepare all our table names and column names in advance.

// If we're inserting multiple rows, this allows us to avoid

// re-validating the same strings over and over again.

line_sender_table_name table_name = QDB_TABLE_NAME_LITERAL("c_cars");

line_sender_column_name id_name = QDB_COLUMN_NAME_LITERAL("id");

line_sender_column_name x_name = QDB_COLUMN_NAME_LITERAL("x");

line_sender_column_name y_name = QDB_COLUMN_NAME_LITERAL("y");

line_sender_column_name booked_name = QDB_COLUMN_NAME_LITERAL("booked");

line_sender_column_name passengers_name = QDB_COLUMN_NAME_LITERAL(

"passengers");

line_sender_column_name driver_name = QDB_COLUMN_NAME_LITERAL("driver");

if (!line_sender_buffer_table(buffer, table_name, &err))

goto on_error;

line_sender_utf8 id_value = QDB_UTF8_LITERAL(

"d6e5fe92-d19f-482a-a97a-c105f547f721");

if (!line_sender_buffer_symbol(buffer, id_name, id_value, &err))

goto on_error;

if (!line_sender_buffer_column_f64(buffer, x_name, 30.5, &err))

goto on_error;

if (!line_sender_buffer_column_f64(buffer, y_name, -150.25, &err))

goto on_error;

if (!line_sender_buffer_column_bool(buffer, booked_name, true, &err))

goto on_error;

if (!line_sender_buffer_column_i64(buffer, passengers_name, 3, &err))

goto on_error;

line_sender_utf8 driver_value = QDB_UTF8_LITERAL("John Doe");

if (!line_sender_buffer_column_str(buffer, driver_name, driver_value, &err))

goto on_error;

// 1997-07-04 04:56:55 UTC

int64_t designated_timestamp = 867992215000000000;

if (!line_sender_buffer_at_nanos(buffer, designated_timestamp, &err))

goto on_error;

//// If you want to get the current system timestamp as nanos, call:

// if (!line_sender_buffer_at_nanos(buffer, line_sender_now_nanos(), &err))

// goto on_error;

// To insert more records, call `line_sender_buffer_table(..)...` again.

// It's recommended to keep a timer and/or maximum buffer size to flush

// the buffer periodically with any accumulated records.

if (!line_sender_flush(sender, buffer, &err))

goto on_error;

line_sender_close(sender);

return true;

on_error: ;

line_sender_opts_free(opts);

size_t err_len = 0;

const char* err_msg = line_sender_error_msg(err, &err_len);

fprintf(stderr, "Error running example: %.*s\n", (int)err_len, err_msg);

line_sender_error_free(err);

line_sender_buffer_free(buffer);

line_sender_close(sender);

return false;

}

static bool displayed_help(int argc, const char* argv[])

{

for (int index = 1; index < argc; ++index)

{

const char* arg = argv[index];

if ((strncmp(arg, "-h", 2) == 0) || (strncmp(arg, "--help", 6) == 0))

{

fprintf(stderr, "Usage:\n");

fprintf(stderr, "line_sender_c_example: [HOST [PORT]]\n");

fprintf(stderr, " HOST: ILP host (defaults to \"localhost\").\n");

fprintf(stderr, " PORT: ILP port (defaults to \"9009\").\n");

return true;

}

}

return false;

}

int main(int argc, const char* argv[])

{

if (displayed_help(argc, argv))

return 0;

const char* host = "localhost";

if (argc >= 2)

host = argv[1];

const char* port = "9009";

if (argc >= 3)

port = argv[2];

return !example(host, port);

}

C++

C++ client library docs

#include <questdb/ingress/line_sender.hpp>

#include <iostream>

using namespace std::literals::string_view_literals;

using namespace questdb::ingress::literals;

static bool example(std::string_view host, std::string_view port)

{

try

{

// Connect.

questdb::ingress::line_sender sender{host, port};

// We prepare all our table names and column names in advance.

// If we're inserting multiple rows, this allows us to avoid

// re-validating the same strings over and over again.

const auto table_name = "cpp_cars"_tn;

const auto id_name = "id"_cn;

const auto x_name = "x"_cn;

const auto y_name = "y"_cn;

const auto booked_name = "booked"_cn;

const auto passengers_name = "passengers"_cn;

const auto driver_name = "driver"_cn;

questdb::ingress::line_sender_buffer buffer;

// 1997-07-04 04:56:55 UTC

questdb::ingress::timestamp_nanos designated_timestamp{867992215000000000};

buffer

.table(table_name)

.symbol(id_name, "d6e5fe92-d19f-482a-a97a-c105f547f721"_utf8)

.column(x_name, 30.5)

.column(y_name, -150.25)

.column(booked_name, true)

.column(passengers_name, int64_t{3})

.column(driver_name, "John Doe"_utf8)

.at(designated_timestamp);

// Call `.at(timestamp_nanos::now())` to use the current system time.

// To insert more records, call `buffer.table(..)...` again.

sender.flush(buffer);

// It's recommended to keep a timer and/or maximum buffer size to flush

// the buffer periodically with any accumulated records.

return true;

}

catch (const questdb::ingress::line_sender_error& err)

{

std::cerr

<< "Error running example: "

<< err.what()

<< std::endl;

return false;

}

}

static bool displayed_help(int argc, const char* argv[])

{

for (int index = 1; index < argc; ++index)

{

const std::string_view arg{argv[index]};

if ((arg == "-h"sv) || (arg == "--help"sv))

{

std::cerr

<< "Usage:\n"

<< "line_sender_c_example: [HOST [PORT]]\n"

<< " HOST: ILP host (defaults to \"localhost\").\n"

<< " PORT: ILP port (defaults to \"9009\")."

<< std::endl;

return true;

}

}

return false;

}

int main(int argc, const char* argv[])

{

if (displayed_help(argc, argv))

return 0;

auto host = "localhost"sv;

if (argc >= 2)

host = std::string_view{argv[1]};

auto port = "9009"sv;

if (argc >= 3)

port = std::string_view{argv[2]};

return !example(host, port);

}

Rust

Rust client library docs

use chrono::{TimeZone, Utc};

use questdb::{

ingress::{Buffer, SenderBuilder, TimestampNanos},

Result,

};

fn main() -> Result<()> {

let host: String = std::env::args().nth(1).unwrap_or("localhost".to_string());

let port: u16 = std::env::args()

.nth(2)

.unwrap_or("9009".to_string())

.parse()

.unwrap();

let mut sender = SenderBuilder::new(host, port).connect()?;

let mut buffer = Buffer::new();

let designated_timestamp =

TimestampNanos::from_datetime(Utc.with_ymd_and_hms(1997, 7, 4, 4, 56, 55).unwrap())?;

buffer

.table("sensors")?

.symbol("id", "toronto1")?

.column_f64("temperature", 20.0)?

.column_i64("humidity", 50)?

.at(designated_timestamp)?;

//// If you want to pass the current system timestamp, replace with:

// .at(TimestampNanos::now())?;

// You can add multiple rows before flushing.

// It's recommended to keep a timer and/or a buffer size before flushing.

sender.flush(&mut buffer)?;

// The buffer is now reusable. No need to reallocate a new one.

Ok(())

}

Ruby

require 'socket'

HOST = 'localhost'

PORT = 9009

# Returns the current time in nanoseconds

def time_in_nsec

now = Time.now

return now.to_i * (10 ** 9) + now.nsec

end

begin

s = TCPSocket.new HOST, PORT

# Single record insert

s.puts "trades,name=client_timestamp value=12.4 #{time_in_nsec}\n"

# Omitting the timestamp allows the server to assign one

s.puts "trades,name=client_timestamp value=12.4\n"

# Streams of readings must be newline-delimited

s.puts "trades,name=client_timestamp value=12.4\n" +

"trades,name=client_timestamp value=11.4\n"

rescue SocketError => ex

puts ex.inspect

ensure

s.close() if s

end

PHP

<?php

error_reporting(E_ALL);

/* Allow the script to hang around waiting for connections. */

set_time_limit(0);

/* Turn on implicit output flushing so we see what we're getting

* as it comes in. */

ob_implicit_flush();

$address = 'localhost';

$port = 9009;

/* Create a TCP/IP socket. */

$socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);

if ($socket === false) {

echo "socket_create() failed: reason: " . socket_strerror(socket_last_error()) . "\n";

} else {

echo "OK.\n";

}

echo "Attempting to connect to '$address' on port '$port'...";

$result = socket_connect($socket, $address, $port);

if ($result === false) {

echo "socket_connect() failed.\nReason: ($result) " . socket_strerror(socket_last_error($socket)) . "\n";

} else {

echo "OK.\n";

}

$row=utf8_encode("test_readings,city=London,make=Omron temperature=23.5,humidity=0.343 1465839830100400000\n");

echo "$row";

socket_write($socket, $row);

echo "\n";

socket_close($socket);

?>

Telegraf

The Telegraf guide helps you configure a Telegraf agent to collect and send metrics to QuestDB via InfluxDB Line Protocol.

PostgreSQL wire protocol

QuestDB also supports the same wire protocol as PostgreSQL, allowing you to connect and query the database with various third-party pre-existing client libraries and tools.

You can connect to TCP port 8812 and use both INSERT and SELECT SQL queries.

PostgreSQL wire protocol is better suited for applications inserting via SQL programmatically as it provides parameterized queries, which avoid SQL injection issues.

tip

InfluxDB Line Protocol is the recommended primary ingestion method in QuestDB. SQL INSERT statements over the PostgreSQL offer feedback and error reporting, but have worse overall performance.

Here are a few examples demonstrating SQL INSERT queries:

psql

note

If you using the QuestDB Cloud, your database requires TLS to connect. You can find host, port, and password configuration at https://cloud.questdb.com, on your instance "Connect" tab. To enable SSL from psql in the commands below, please follow this pattern:

psql -h {hostname} -p {port} -U admin "dbname=qdb sslmode=require" -c '{SQL_STATEMENT}'

Create the table:

psql -h localhost -p 8812 -U admin -d qdb \

-c "CREATE TABLE IF NOT EXISTS t1 (name STRING, value INT);"

Insert row:

psql -h localhost -p 8812 -U admin -d qdb -c "INSERT INTO t1 VALUES('a', 42)"

Query back:

psql -h localhost -p 8812 -U admin -d qdb -c "SELECT * FROM t1"

Note that you can also run psql from Docker without installing the client locally:

docker run -it --rm --network=host -e PGPASSWORD=quest \

postgres psql ....

Python

This example uses the psychopg3 adapter.

To install the client library, use pip:

python3 -m pip install "psycopg[binary]"

import psycopg as pg

import time

# Connect to an existing QuestDB instance

conn_str = 'user=admin password=quest host=127.0.0.1 port=8812 dbname=qdb'

with pg.connect(conn_str, autocommit=True) as connection:

# Open a cursor to perform database operations

with connection.cursor() as cur:

# Execute a command: this creates a new table

cur.execute('''

CREATE TABLE IF NOT EXISTS test_pg (

ts TIMESTAMP,

name STRING,

value INT

) timestamp(ts);

''')

print('Table created.')

# Insert data into the table.

for x in range(10):

# Converting datetime into millisecond for QuestDB

timestamp = time.time_ns() // 1000

cur.execute('''

INSERT INTO test_pg

VALUES (%s, %s, %s);

''',

(timestamp, 'python example', x))

print('Rows inserted.')

#Query the database and obtain data as Python objects.

cur.execute('SELECT * FROM test_pg;')

records = cur.fetchall()

for row in records:

print(row)

# the connection is now closed

Java

package com.myco;

import java.sql.*;

import java.util.Properties;

class App {

public static void main(String[] args) throws SQLException {

Properties properties = new Properties();

properties.setProperty("user", "admin");

properties.setProperty("password", "quest");

properties.setProperty("sslmode", "disable");

final Connection connection = DriverManager.getConnection(

"jdbc:postgresql://localhost:8812/qdb", properties);

connection.setAutoCommit(false);

final PreparedStatement statement = connection.prepareStatement(

"CREATE TABLE IF NOT EXISTS trades (" +

" ts TIMESTAMP, date DATE, name STRING, value INT" +

") timestamp(ts);");

statement.execute();

try (PreparedStatement preparedStatement = connection.prepareStatement(

"INSERT INTO TRADES VALUES (?, ?, ?, ?)")) {

preparedStatement.setTimestamp(

1,

new Timestamp(io.questdb.std.Os.currentTimeMicros()));

preparedStatement.setDate(2, new Date(System.currentTimeMillis()));

preparedStatement.setString(3, "abc");

preparedStatement.setInt(4, 123);

preparedStatement.execute();

}

System.out.println("Done");

connection.close();

}

}

NodeJS

This example uses the pg package which allows for quickly building queries using Postgres 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 process.hrtime.bigint() call can be used.

"use strict"

const { Client } = require("pg")

const start = async () => {

const client = new Client({

database: "qdb",

host: "127.0.0.1",

password: "quest",

port: 8812,

user: "admin",

})

await client.connect()

const createTable = await client.query(

"CREATE TABLE IF NOT EXISTS trades (" +

" ts TIMESTAMP, date DATE, name STRING, value INT" +

") timestamp(ts);",

)

console.log(createTable)

let now = new Date().toISOString()

const insertData = await client.query(

"INSERT INTO trades VALUES($1, $2, $3, $4);",

[now, now, "node pg example", 123],

)

await client.query("COMMIT")

console.log(insertData)

for (let rows = 0; rows < 10; rows++) {

// Providing a 'name' field allows for prepared statements / bind variables

now = new Date().toISOString()

const query = {

name: "insert-values",

text: "INSERT INTO trades VALUES($1, $2, $3, $4);",

values: [now, now, "node pg prep statement", rows],

}

await client.query(query)

}

await client.query("COMMIT")

const readAll = await client.query("SELECT * FROM trades")

console.log(readAll.rows)

await client.end()

}

start()

.then(() => console.log("Done"))

.catch(console.error)

Go

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

package main

import (

"context"

"fmt"

"log"

"time"

"github.com/jackc/pgx/v4"

)

var conn *pgx.Conn

var err error

func main() {

ctx := context.Background()

conn, _ = pgx.Connect(ctx, "postgresql://admin:quest@localhost:8812/qdb")

defer conn.Close(ctx)

// text-based query

_, err := conn.Exec(ctx,

("CREATE TABLE IF NOT EXISTS trades (" +

" ts TIMESTAMP, date DATE, name STRING, value INT" +

") timestamp(ts);"))

if err != nil {

log.Fatalln(err)

}

// Prepared statement given the name 'ps1'

_, err = conn.Prepare(ctx, "ps1", "INSERT INTO trades VALUES($1,$2,$3,$4)")

if err != nil {

log.Fatalln(err)

}

// Insert all rows in a single commit

tx, err := conn.Begin(ctx)

if err != nil {

log.Fatalln(err)

}

for i := 0; i < 10; i++ {

// Execute 'ps1' statement with a string and the loop iterator value

_, err = conn.Exec(

ctx,

"ps1",

time.Now(),

time.Now().Round(time.Millisecond),

"go prepared statement",

i + 1)

if err != nil {

log.Fatalln(err)

}

}

// Commit the transaction

err = tx.Commit(ctx)

if err != nil {

log.Fatalln(err)

}

// Read all rows from table

rows, err := conn.Query(ctx, "SELECT * FROM trades")

fmt.Println("Reading from trades table:")

for rows.Next() {

var name string

var value int64

var ts time.Time

var date time.Time

err = rows.Scan(&ts, &date, &name, &value)

fmt.Println(ts, date, name, value)

}

err = conn.Close(ctx)

}

Rust

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

use postgres::{Client, NoTls, Error};

use chrono::{Utc};

use std::time::SystemTime;

fn main() -> Result<(), Error> {

let mut client = Client::connect("postgresql://admin:quest@localhost:8812/qdb", NoTls)?;

// Basic query

client.batch_execute(

"CREATE TABLE IF NOT EXISTS trades ( \

ts TIMESTAMP, date DATE, name STRING, value INT \

) timestamp(ts);")?;

// Parameterized query

let name: &str = "rust example";

let val: i32 = 123;

let utc = Utc::now();

let sys_time = SystemTime::now();

client.execute(

"INSERT INTO trades VALUES($1,$2,$3,$4)",

&[&utc.naive_local(), &sys_time, &name, &val],

)?;

// Prepared statement

let mut txn = client.transaction()?;

let statement = txn.prepare("INSERT INTO trades VALUES ($1,$2,$3,$4)")?;

for value in 0..10 {

let utc = Utc::now();

let sys_time = SystemTime::now();

txn.execute(&statement, &[&utc.naive_local(), &sys_time, &name, &value])?;

}

txn.commit()?;

println!("import finished");

Ok(())

}

Web Console

QuestDB ships with an embedded Web Console running by default on port 9000.

Inserting data via SQL

SQL statements can be written in the code editor and executed by clicking the Run button. Note that the Web Console runs a single statement at a time.

There are two SQL keywords to insert data:

• INSERT:

  Creating a table and inserting some data

  CREATE TABLE takeaway_order (ts TIMESTAMP, id SYMBOL, status SYMBOL)

  TIMESTAMP(ts);

  INSERT INTO takeaway_order VALUES (now(), 'order1', 'placed');

  INSERT INTO takeaway_order VALUES (now(), 'order2', 'placed');

• COPY: For inserting bulk data or migrating data from other databases. See large CSV import.

Uploading CSV file

It is also possible to upload CSV files using the Import tab in the Web Console:

HTTP 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 insert-capable entrypoints:

Entrypoint	HTTP Method	Description	API Docs
/imp	POST	Import CSV data	Reference
/exec?query=..	GET	Run SQL Query returning JSON result set	Reference

For details such as content type, query parameters and more, refer to the REST API docs.

/imp: Uploading Tabular Data

tip

InfluxDB Line Protocol is the recommended primary ingestion method in QuestDB. CSV uploading offers insertion feedback and error reporting, but has worse overall performance.

See /imp's atomicity query parameter to customize behavior on error.

Let's assume you want to upload the following data via the /imp entrypoint:

CSV

data.csv

col1,col2,col3

a,10.5,True

b,100,False

c,,True

Table

col1	col2	col3
a	10.5	true
b	100	false
c	NULL	true

You can do so via the command line using cURL or programmatically via HTTP APIs in your scripts and applications.

By default, the response is designed to be human-readable. Use the fmt=json query argument to obtain a response in JSON. You can also specify the schema explicitly. See the second example in Python for these features.

cURL

This example imports a CSV file with automatic schema detection.

Basic import with table name

curl -F data=@data.csv http://localhost:9000/imp?name=table_name

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

Providing a user-defined schema

curl \

-F schema='[{"name":"ts", "type": "TIMESTAMP", "pattern": "yyyy-MM-dd - HH:mm:ss"}]' \

-F data=@weather.csv 'http://localhost:9000/imp?overwrite=true&timestamp=ts'

Python

This first example shows uploading the data.csv file with automatic schema detection.

import sys

import requests

csv = {'data': ('my_table', open('./data.csv', 'r'))}

host = 'http://localhost:9000'

try:

response = requests.post(host + '/imp', files=csv)

print(response.text)

except requests.exceptions.RequestException as e:

print(f'Error: {e}', file=sys.stderr)

The second example creates a CSV buffer from Python objects and uploads them with a custom schema. Note UTF-8 encoding.

The fmt=json parameter allows us to obtain a parsable response, rather than a tabular response designed for human consumption.

import io

import csv

import requests

import pprint

import json

def to_csv_str(table):

output = io.StringIO()

csv.writer(output, dialect='excel').writerows(table)

return output.getvalue().encode('utf-8')

def main():

table_name = 'example_table2'

table = [

['col1', 'col2', 'col3'],

['a', 10.5, True],

['b', 100, False],

['c', None, True]]

table_csv = to_csv_str(table)

print(table_csv)

schema = json.dumps([

{'name': 'col1', 'type': 'SYMBOL'},

{'name': 'col2', 'type': 'DOUBLE'},

{'name': 'col3', 'type': 'BOOLEAN'}])

response = requests.post(

'http://localhost:9000/imp',

params={'fmt': 'json'},

files={

'schema': schema,

'data': (table_name, table_csv)}).json()

# You can parse the `status` field and `error` fields

# of individual columns. See Reference/API/REST docs for details.

pprint.pprint(response)

if __name__ == '__main__':

main()

NodeJS

const fetch = require("node-fetch")

const FormData = require("form-data")

const fs = require("fs")

const HOST = "http://localhost:9000"

async function run() {

const form = new FormData()

form.append("data", fs.readFileSync(__dirname + "/data.csv"), {

filename: fileMetadata.name,

contentType: "application/octet-stream",

})

try {

const r = await fetch(`${HOST}/imp`, {

method: "POST",

body: form,

headers: form.getHeaders(),

})

console.log(r)

} catch (e) {

console.error(e)

}

}

run()

Go

package main

import (

"bytes"

"fmt"

"io"

"io/ioutil"

"log"

"mime/multipart"

"net/http"

"net/url"

"os"

)

func main() {

u, err := url.Parse("http://localhost:9000")

checkErr(err)

u.Path += "imp"

url := fmt.Sprintf("%v", u)

fileName := "/path/to/data.csv"

file, err := os.Open(fileName)

checkErr(err)

defer file.Close()

buf := new(bytes.Buffer)

writer := multipart.NewWriter(buf)

uploadFile, _ := writer.CreateFormFile("data", "data.csv")

_, err = io.Copy(uploadFile, file)

checkErr(err)

writer.Close()

req, err := http.NewRequest(http.MethodPut, url, buf)

checkErr(err)

req.Header.Add("Content-Type", writer.FormDataContentType())

client := &http.Client{}

res, err := client.Do(req)

checkErr(err)

defer res.Body.Close()

body, err := ioutil.ReadAll(res.Body)

checkErr(err)

log.Println(string(body))

}

func checkErr(err error) {

if err != nil {

panic(err)

}

}

/exec: SQL INSERT Query

The /exec entrypoint takes a SQL query and returns results as JSON.

We can use this for quick SQL inserts too, but note that there's no support for parameterized queries that are necessary to avoid SQL injection issues.

tip

Prefer the PostgreSQL interface if you are generating sql programmatically.

Prefer InfluxDB Line Protocol if you need high-performance inserts.

cURL

# Create Table

curl -G \

--data-urlencode "query=CREATE TABLE IF NOT EXISTS trades(name STRING, value INT)" \

http://localhost:9000/exec

# Insert a row

curl -G \

--data-urlencode "query=INSERT INTO trades VALUES('abc', 123456)" \

http://localhost:9000/exec

Python

import sys

import requests

import json

host = 'http://localhost:9000'

def run_query(sql_query):

query_params = {'query': sql_query, 'fmt' : 'json'}

try:

response = requests.get(host + '/exec', params=query_params)

json_response = json.loads(response.text)

print(json_response)

except requests.exceptions.RequestException as e:

print(f'Error: {e}', file=sys.stderr)

# create table

run_query("CREATE TABLE IF NOT EXISTS trades (name STRING, value INT)")

# insert row

run_query("INSERT INTO trades VALUES('abc', 123456)")

NodeJS

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

const fetch = require("node-fetch")

const HOST = "http://localhost:9000"

async function createTable() {

try {

const query = "CREATE TABLE IF NOT EXISTS trades (name STRING, value INT)"

const response = await fetch(

`${HOST}/exec?query=${encodeURIComponent(query)}`,

)

const json = await response.json()

console.log(json)

} catch (error) {

console.log(error)

}

}

async function insertData() {

try {

const query = "INSERT INTO trades VALUES('abc', 123456)"

const response = await fetch(

`${HOST}/exec?query=${encodeURIComponent(query)}`,

)

const json = await response.json()

console.log(json)

} catch (error) {

console.log(error)

}

}

createTable().then(insertData)

Go

package main

import (

"fmt"

"io/ioutil"

"log"

"net/http"

"net/url"

)

func main() {

u, err := url.Parse("http://localhost:9000")

checkErr(err)

u.Path += "exec"

params := url.Values{}

params.Add("query", `

CREATE TABLE IF NOT EXISTS

trades (name STRING, value INT);

INSERT INTO

trades

VALUES(

"abc",

123456

);

`)

u.RawQuery = params.Encode()

url := fmt.Sprintf("%v", u)

res, err := http.Get(url)

checkErr(err)

defer res.Body.Close()

body, err := ioutil.ReadAll(res.Body)

checkErr(err)

log.Println(string(body))

}

func checkErr(err error) {

if err != nil {

panic(err)

}

}