Bulk upsert data
YDB supports bulk upsert of a large number of rows without atomicity guarantees. Data writes are split into several independent transactions, each affecting a single partition, and executed in parallel. As a result, this approach is more efficient than YQL. On success, the BulkUpsert method guarantees insertion of all data passed in the request.
Warning
When using BulkUpsert to insert data into columnar tables, you must provide values for all columns, including NULL values.
Below are code examples that use the built‑in YDB SDK bulk upsert facilities:
C++
Go
Java
Python
C#
JavaScript
Rust
PHP
Native SDK
userver
#include <ydb-cpp-sdk/client/table/table.h>
void BulkUpsertLogs(const NYdb::TDriver& driver) {
NYdb::NTable::TTableClient client(driver);
constexpr int kBatchSize = 1000;
NYdb::TValueBuilder rowsBuilder;
rowsBuilder.BeginList();
for (int i = 0; i < kBatchSize; ++i) {
rowsBuilder.AddListItem()
.BeginStruct()
.AddMember("App").Utf8("App_" + std::to_string(i / 256))
.AddMember("Host").Utf8("192.168.0." + std::to_string(i % 256))
.AddMember("Timestamp")
.Timestamp(TInstant::Now() + TDuration::Seconds(i))
.AddMember("HttpCode").Uint32(static_cast<uint32_t>(i % 113 == 0 ? 404 : 200))
.AddMember("Message")
.Utf8(i % 3 == 0 ? "GET / HTTP/1.1" : "GET /images/logo.png HTTP/1.1")
.EndStruct();
}
rowsBuilder.EndList();
NYdb::TValue rows = rowsBuilder.Build();
NYdb::NStatusHelpers::ThrowOnError(client.RetryOperationSync(
[&rows](NYdb::NTable::TTableClient& client) {
return client.BulkUpsert("/local/bulk_upsert_example", NYdb::TValue{rows}).GetValueSync();
},
NYdb::NTable::TRetryOperationSettings()
.Idempotent(true)
));
}
#include <userver/ydb/io/supported_types.hpp>
#include <userver/ydb/table.hpp>
struct LogMessage final {
ydb::Utf8 App;
ydb::Utf8 Host;
std::chrono::system_clock::time_point Timestamp;
std::uint32_t HttpCode;
ydb::Utf8 Message;
};
void BulkUpsertLogs(ydb::TableClient& client) {
constexpr int kBatchSize = 1000;
std::vector<LogMessage> rows;
rows.reserve(kBatchSize);
for (int i = 0; i < kBatchSize; ++i) {
rows.push_back({
.App = ydb::Utf8{"App_" + std::to_string(i / 256)},
.Host = ydb::Utf8{"192.168.0." + std::to_string(i % 256)},
.Timestamp = std::chrono::system_clock::now() + std::chrono::seconds{i},
.HttpCode = static_cast<std::uint32_t>(i % 113 == 0 ? 404 : 200),
.Message = ydb::Utf8{
i % 3 == 0 ? "GET / HTTP/1.1" : "GET /images/logo.png HTTP/1.1"
},
});
}
client.BulkUpsert(
"/local/bulk_upsert_example",
rows,
ydb::OperationSettings{
.is_idempotent = true,
}
);
}
Native SDK
database/sql
Bulk upsert native YDB data
package main
import (
"context"
"os"
"github.com/ydb-platform/ydb-go-sdk/v3"
"github.com/ydb-platform/ydb-go-sdk/v3/table"
"github.com/ydb-platform/ydb-go-sdk/v3/table/types"
)
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
db, err := ydb.Open(ctx,
os.Getenv("YDB_CONNECTION_STRING"),
ydb.WithAccessTokenCredentials(os.Getenv("YDB_TOKEN")),
)
if err != nil {
panic(err)
}
defer db.Close(ctx)
type logMessage struct {
App string
Host string
Timestamp time.Time
HTTPCode uint32
Message string
}
// prepare native go data
const batchSize = 10000
logs := make([]logMessage, 0, batchSize)
for i := 0; i < batchSize; i++ {
message := logMessage{
App: fmt.Sprintf("App_%d", i/256),
Host: fmt.Sprintf("192.168.0.%d", i%256),
Timestamp: time.Now().Add(time.Millisecond * time.Duration(i%1000)),
HTTPCode: 200,
}
if i%2 == 0 {
message.Message = "GET / HTTP/1.1"
} else {
message.Message = "GET /images/logo.png HTTP/1.1"
}
logs = append(logs, message)
}
// execute bulk upsert with native ydb data
err = db.Table().Do( // Do retry operation on errors with best effort
ctx, // context manage exiting from Do
func(ctx context.Context, s table.Session) (err error) { // retry operation
rows := make([]types.Value, 0, len(logs))
for _, msg := range logs {
rows = append(rows, types.StructValue(
types.StructFieldValue("App", types.UTF8Value(msg.App)),
types.StructFieldValue("Host", types.UTF8Value(msg.Host)),
types.StructFieldValue("Timestamp", types.TimestampValueFromTime(msg.Timestamp)),
types.StructFieldValue("HTTPCode", types.Uint32Value(msg.HTTPCode)),
types.StructFieldValue("Message", types.UTF8Value(msg.Message)),
))
}
return s.BulkUpsert(ctx, "/local/bulk_upsert_example", types.ListValue(rows...))
},
)
if err != nil {
fmt.Printf("unexpected error: %v", err)
}
}
Bulk upsert CSV
package main
import (
"context"
"fmt"
"os"
"github.com/ydb-platform/ydb-go-sdk/v3"
"github.com/ydb-platform/ydb-go-sdk/v3/table"
)
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
db, err := ydb.Open(ctx,
os.Getenv("YDB_CONNECTION_STRING"),
ydb.WithAccessTokenCredentials(os.Getenv("YDB_TOKEN")),
)
if err != nil {
panic(err)
}
defer db.Close(ctx)
csv := `skip row
id,val
42,"text42"
43,"text43"
44,hello
`
err = db.Table().BulkUpsert(ctx, "/local/bulk_upsert_example", table.BulkUpsertDataCsv(
[]byte(csv),
table.WithCsvHeader(),
table.WithCsvSkipRows(2),
table.WithCsvNullValue([]byte("hello")), // the string "hello" will be interpreted as NULL
))
if err != nil {
fmt.Printf("unexpected error: %v", err)
}
}
Bulk upsert Apache Arrow
The following example uses the arrow package to prepare data.
package main
import (
"bytes"
"context"
"fmt"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/ipc"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/ydb-platform/ydb-go-sdk/v3"
"github.com/ydb-platform/ydb-go-sdk/v3/table"
)
func main() {
ctx := context.Background()
db, err := ydb.Open(ctx,
os.Getenv("YDB_CONNECTION_STRING"),
ydb.WithAccessTokenCredentials(os.Getenv("YDB_TOKEN")),
)
if err != nil {
panic(err)
}
defer db.Close(ctx) // cleanup resources
mem := memory.NewGoAllocator()
schema := arrow.NewSchema([]arrow.Field{
{Name: "id", Type: arrow.PrimitiveTypes.Int64},
{Name: "val", Type: arrow.BinaryTypes.String},
}, nil)
b := array.NewRecordBuilder(mem, schema)
defer b.Release()
b.Field(0).(*array.Int64Builder).AppendValues(
[]int64{123, 234}, nil)
b.Field(1).(*array.StringBuilder).AppendValues(
[]string{"data1", "data2"}, nil)
rec := b.NewRecordBatch()
defer rec.Release()
schemaPayload := ipc.GetSchemaPayload(rec.Schema(), mem)
defer schemaPayload.Release()
dataPayload, err := ipc.GetRecordBatchPayload(rec)
if err != nil {
panic(err)
}
defer dataPayload.Release()
var schemaBuf bytes.Buffer
_, err = schemaPayload.WritePayload(&schemaBuf)
if err != nil {
panic(err)
}
var dataBuf bytes.Buffer
_, err = dataPayload.WritePayload(&dataBuf)
if err != nil {
panic(err)
}
err = db.Table().BulkUpsert(ctx, "/local/bulk_upsert_example", table.BulkUpsertDataArrow(
dataBuf.Bytes(),
table.WithArrowSchema(schemaBuf.Bytes()), // schema is required
))
if err != nil {
fmt.Printf("unexpected error: %v", err)
}
}
The database/sql driver implementation for YDB does not support non‑transactional bulk upsert.
For bulk upsert you should use transactional upsert.
Bulk upsert via BulkUpsert is more efficient than transactional YQL for large data volumes. For small row sets see UPSERT. The table schema is described in the Tables section.
Native SDK
JDBC
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import tech.ydb.auth.NopAuthProvider;
import tech.ydb.common.transaction.TxMode;
import tech.ydb.core.grpc.GrpcTransport;
import tech.ydb.query.QueryClient;
import tech.ydb.query.result.ResultSetReader;
import tech.ydb.query.tools.QueryReader;
import tech.ydb.query.tools.SessionRetryContext;
import tech.ydb.table.TableClient;
import tech.ydb.table.query.Params;
import tech.ydb.table.settings.BulkUpsertSettings;
import tech.ydb.table.values.ListType;
import tech.ydb.table.values.ListValue;
import tech.ydb.table.values.PrimitiveType;
import tech.ydb.table.values.PrimitiveValue;
import tech.ydb.table.values.StructType;
import tech.ydb.table.values.Value;
public class BulkUpsertExample {
private static final String TABLE_NAME = "bulk_upsert";
private static final int BATCH_SIZE = 1000;
public static void main(String[] args) {
String connectionString = System.getenv().getOrDefault(
"YDB_CONNECTION_STRING", "grpc://localhost:2136/local");
try (GrpcTransport transport = GrpcTransport.forConnectionString(connectionString)
.withAuthProvider(NopAuthProvider.INSTANCE)
.build();
QueryClient queryClient = QueryClient.newClient(transport).build();
TableClient tableClient = TableClient.newClient(transport).build()) {
SessionRetryContext queryRetry = SessionRetryContext.create(queryClient).build();
SessionRetryContext tableRetry = SessionRetryContext.create(tableClient).build();
// Creating a table for bulk upsert
queryRetry.supplyResult(session -> QueryReader.readFrom(session.createQuery("""
CREATE TABLE IF NOT EXISTS bulk_upsert (
app Text,
timestamp Timestamp,
host Text,
http_code Uint32,
message Text,
PRIMARY KEY (app, timestamp, host)
);
""", TxMode.NONE, Params.empty())
)).join().getValue();
// Full path to the table: /local/bulk_upsert
String tablePath = transport.getDatabase() + "/" + TABLE_NAME;
StructType rowType = StructType.of(
"app", PrimitiveType.Text,
"timestamp", PrimitiveType.Timestamp,
"host", PrimitiveType.Text,
"http_code", PrimitiveType.Uint32,
"message", PrimitiveType.Text
);
// Generating a batch of records
List<Value<?>> rows = new ArrayList<>(BATCH_SIZE);
for (int i = 0; i < BATCH_SIZE; i++) {
rows.add(rowType.newValue(
"app", PrimitiveValue.newText("App_" + i / 256),
"timestamp", PrimitiveValue.newTimestamp(Instant.now().plusSeconds(i)),
"host", PrimitiveValue.newText("192.168.0." + i % 256),
"http_code", PrimitiveValue.newUint32(i % 113 == 0 ? 404 : 200),
"message", PrimitiveValue.newText(
i % 3 == 0 ? "GET / HTTP/1.1" : "GET /images/logo.png HTTP/1.1")
));
}
ListValue batch = ListType.of(rowType).newValue(rows);
// Bulk upsert without atomicity of the whole batch
tableRetry.supplyStatus(session ->
session.executeBulkUpsert(tablePath, batch, new BulkUpsertSettings())
).join().expectSuccess("bulk upsert failed");
// Checking the number of rows
QueryReader countReader = queryRetry.supplyResult(session -> QueryReader.readFrom(
session.createQuery(
"SELECT COUNT(*) AS cnt FROM bulk_upsert", TxMode.NONE, Params.empty())
)).join().getValue();
ResultSetReader rs = countReader.getResultSet(0);
if (rs.next()) {
System.out.println("Строк в таблице bulk_upsert: " + rs.getColumn("cnt").getUint64());
}
}
}
}
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.time.Instant;
public class JdbcBulkUpsertExample {
private static final int BATCH_SIZE = 1000;
public static void main(String[] args) throws SQLException {
String url = System.getenv().getOrDefault(
"YDB_JDBC_URL", "jdbc:ydb:grpc://localhost:2136/local");
try (Connection conn = DriverManager.getConnection(url)) {
try (Statement stmt = conn.createStatement()) {
stmt.execute("""
CREATE TABLE IF NOT EXISTS bulk_upsert (
app Text,
timestamp Timestamp,
host Text,
http_code Uint32,
message Text,
PRIMARY KEY (app, timestamp, host)
);
""");
}
String bulkSql = """
BULK UPSERT INTO bulk_upsert (app, timestamp, host, http_code, message)
VALUES (?, ?, ?, ?, ?);
""";
try (PreparedStatement ps = conn.prepareStatement(bulkSql)) {
for (int i = 0; i < BATCH_SIZE; i++) {
ps.setString(1, "App_" + i / 256);
ps.setTimestamp(2, Timestamp.from(Instant.now().plusSeconds(i)));
ps.setString(3, "192.168.0." + i % 256);
ps.setLong(4, i % 113 == 0 ? 404 : 200);
ps.setString(5, i % 3 == 0 ? "GET / HTTP/1.1" : "GET /images/logo.png HTTP/1.1");
ps.addBatch();
}
ps.executeBatch();
}
try (Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("SELECT COUNT(*) AS cnt FROM bulk_upsert")) {
if (rs.next()) {
System.out.println("Строк в таблице bulk_upsert: " + rs.getLong("cnt"));
}
}
}
}
}
In Spring Boot, Hibernate, JOOQ, and other frameworks built on top of ORM over JDBC you can execute native YQL (including from repositories and @Query). The driver aims to optimize large inserts; operations UPDATE, INSERT, DELETE, UPSERT that go through JDBC are automatically batched on the driver side when needed.
Native SDK
Native SDK (Asyncio)
SQLAlchemy
import posixpath
import ydb
def bulk_upsert(driver: ydb.Driver, path: str):
column_types = (
ydb.BulkUpsertColumns()
.add_column("id", ydb.PrimitiveType.Uint64)
.add_column("val", ydb.OptionalType(ydb.PrimitiveType.Utf8))
)
rows = [
{"id": 1, "val": "1"},
{"id": 2, "val": "2"},
{"id": 3, "val": "3"},
]
driver.table_client.bulk_upsert(posixpath.join(path, "tablename"), rows, column_types)
import os
import posixpath
import ydb
import asyncio
async def bulk_upsert(driver: ydb.aio.Driver, path: str):
column_types = (
ydb.BulkUpsertColumns()
.add_column("id", ydb.PrimitiveType.Uint64)
.add_column("val", ydb.OptionalType(ydb.PrimitiveType.Utf8))
)
rows = [
{"id": 1, "val": "1"},
{"id": 2, "val": "2"},
{"id": 3, "val": "3"},
]
await driver.table_client.bulk_upsert(
posixpath.join(path, "tablename"), rows, column_types
)
async def main():
async with ydb.aio.Driver(
connection_string=os.environ["YDB_CONNECTION_STRING"],
credentials=ydb.credentials_from_env_variables(),
) as driver:
await driver.wait()
await bulk_upsert(driver, "/local")
asyncio.run(main())
import os
import sqlalchemy as sa
import ydb
engine = sa.create_engine(os.environ["YDB_SQLALCHEMY_URL"])
with engine.connect() as connection:
dbapi_conn = connection.connection
column_types = (
ydb.BulkUpsertColumns()
.add_column("id", ydb.PrimitiveType.Uint64)
.add_column("val", ydb.OptionalType(ydb.PrimitiveType.Utf8))
)
rows = [
{"id": 1, "val": "1"},
{"id": 2, "val": "2"},
{"id": 3, "val": "3"},
]
dbapi_conn.bulk_upsert("tablename", rows, column_types)
This functionality is not currently supported.
This functionality is not currently supported.
use ydb::{ydb_struct, AccessTokenCredentials, ClientBuilder, Value, YdbResult};
#[tokio::main]
async fn main() -> YdbResult<()> {
let client = ClientBuilder::new_from_connection_string(
"grpc://localhost:2136?database=local",
)?
.with_credentials(AccessTokenCredentials::from("..."))
.client()?;
client.wait().await?;
let rows: Vec<Value> = vec![
ydb_struct!(
"id" => 1_u64,
"val" => Value::Text("1".into()),
),
ydb_struct!(
"id" => 2_u64,
"val" => Value::Text("2".into()),
),
ydb_struct!(
"id" => 3_u64,
"val" => Value::Text("3".into()),
),
];
client
.table_client()
.bulk_upsert("/local/tablename", rows)
.idempotent(true)
.await?;
Ok(())
}
<?php
use YdbPlatform\Ydb\Ydb;
$ydb = new Ydb($config);
$rows = [
['id' => 1, 'val' => '1'],
['id' => 2, 'val' => '2'],
['id' => 3, 'val' => '3'],
];
$ydb->table()->bulkUpsert('tablename', $rows, [
'id' => 'UINT64',
'val' => 'UTF8',
]);