Insert data

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

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

For transactional data inserts, use the PostgreSQL wire protocol.

For operational (ad-hoc) data ingestion, the Web Console makes it easy to upload CSV files and insert via SQL statements. You can also perform these same actions via the HTTP REST API. For large CSV import (database migrations), use SQL COPY.

In summary, these are the different options:

• InfluxDB Line Protocol
  • High performance.
  • Optional automatic timestamps.
  • Optional integrated authentication.
  • Client libraries in various programming languages.
• PostgreSQL wire protocol
  • SQL INSERT statements, including parameterized queries.
  • Use psql on the command line.
  • Interoperability with third-party tools and libraries.
• Web Console
  • CSV upload.
  • SQL INSERT statements.
  • SQL COPY for large CSV import.
• HTTP REST API
  • CSV upload.
  • SQL INSERT statements.
  • Use curl on the command line.

Here is a summary table comparing the different ways to insert data we support:

Protocol	Record Insertion Reporting	Data Insertion Performance
InfluxDB Line Protocol	Server logs; Disconnect on error	Best
CSV upload via HTTP REST	Configurable	Very Good
SQL INSERT statements	Transaction-level	Good
SQL COPY statements	Transaction-level	Suitable for one-off data migration

InfluxDB Line Protocol

The InfluxDB Line Protocol (ILP) 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
• Configurable commit-lag for out-of-order data via server configuration settings

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

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

tip

The ILP client libraries provide more user-friendly ILP clients for a growing number of languages.

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 4k 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

sender

.table("prices")

.symbol("instrument", "EURUSD")

.floatColumn("bid", 1.0195)

.floatColumn("ask", 1.0221)

.atNow();

sender

.table("prices")

.symbol("instrument", "GBPUSD")

.floatColumn("bid", 1.2076)

.floatColumn("ask", 1.2082)

.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();

// add rows to the buffer again and send it to the server

sender

.table("prices")

.symbol("instrument", "EURUSD")

.floatColumn("bid", 1.0197)

.floatColumn("ask", 1.0224)

.atNow();

await sender.flush();

// close the connection after all rows ingested

await sender.close();

return 0;

}

run()

.then((value) => console.log(value))

.catch((err) => console.log(err));

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/ilp/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;

if (!line_sender_buffer_at_now(buffer, &err))

goto on_error;

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

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/ilp/line_sender.hpp>

#include <iostream>

using namespace std::literals::string_view_literals;

using namespace questdb::ilp::literals;

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

{

try

{

// Connect.

questdb::ilp::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::ilp::line_sender_buffer buffer;

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_now();

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

sender.flush(buffer);

return true;

}

catch (const questdb::ilp::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 questdb::{

Result,

ingress::{

Buffer,

SenderBuilder}};

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();

buffer

.table("sensors")?

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

.column_f64("temperature", 20.0)?

.column_i64("humidity", 50)?

.at_now()?;

sender.flush(&mut buffer)?;

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);

?>

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

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 trades_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.

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');

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.

For inserting bulk data or migrating data from other databases, see large CSV import.

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 ILP 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)

}

}