// Copyright 2026 Leo Cheng
// SPDX-License-Identifier: Apache-2.0
// Bulk copy: rows travel in the protocol's own load format instead of as one INSERT per
// row. The statement `INSERT BULK` declares the shape of the stream and arms the table;
// what follows is a message of column metadata and row tokens, each row written straight
// into the stream. A thousand rows cost one round trip rather than a thousand.
///|
/// One column of a bulk stream: the name the `INSERT BULK` statement addresses it by, and
/// the TDS type its cells are written as.
priv struct BulkColumn {
name : String
ty : Int32
}
///|
/// The type a column is written as, inferred from its first non-`NULL` value.
///
/// The declaration in `INSERT BULK` is what the server converts from, so it does not have
/// to be the table's own type: `bigint` for every integer, `nvarchar(max)` for every
/// string, and the server casts into the destination column. What it must be is *wide
/// enough*, which is why an integer always travels as eight bytes and a string always as
/// `(max)` — a later row cannot then overflow a length chosen by an earlier one.
fn bulk_type_of(value : @moondb.Value) -> Int32 {
match value {
@moondb.Bool(_) => 0x68 // BITN
@moondb.Int(_) | @moondb.Int64(_) => 0x26 // INTN
@moondb.Double(_) => 0x6D // FLTN
@moondb.Blob(_) => 0xA5 // BIGVARBINARY
@moondb.Uuid(_) => 0x24 // GUID
_ => 0xE7 // NVARCHAR: text, and everything that travels as text
}
}
///|
/// The T-SQL type name a bulk column is declared with. It pairs with [`bulk_type_of`] and
/// with the cell encoding below; the three have to agree or the server reads the stream as
/// a different shape than it was written in.
fn bulk_declaration(ty : Int32) -> String {
if ty == 0x68 {
"bit"
} else if ty == 0x26 {
"bigint"
} else if ty == 0x6D {
"float"
} else if ty == 0xA5 {
"varbinary(max)"
} else if ty == 0x24 {
"uniqueidentifier"
} else {
"nvarchar(max)"
}
}
///|
/// The type of each column, taken from the first row that has a value in it. A column of
/// nothing but `NULL`s is declared `nvarchar(max)` and carries `NULL`s, which converts into
/// any destination column that accepts one.
fn bulk_columns(
columns : Array[String],
rows : Array[Array[@moondb.Value]],
) -> Array[BulkColumn] {
let out : Array[BulkColumn] = []
for i = 0; i < columns.length(); i = i + 1 {
let mut ty = 0xE7
let mut found = false
for row in rows {
if !found && i < row.length() && !row[i].is_null() {
ty = bulk_type_of(row[i])
found = true
}
}
out.push({ name: columns[i], ty, })
}
out
}
///|
/// One cell in a row token. Unlike an RPC parameter a cell carries no type byte: the column
/// metadata said what it is, so the cell is a length and its bytes.
fn put_cell(out : Buffer, ty : Int32, value : @moondb.Value) -> Unit {
match ty {
0x26 =>
match value {
@moondb.Int(v) => {
out.write_byte(b'\x08')
put_u64(out, v.to_int64())
}
@moondb.Int64(v) => {
out.write_byte(b'\x08')
put_u64(out, v)
}
@moondb.Bool(v) => {
out.write_byte(b'\x08')
put_u64(out, if v { 1L } else { 0L })
}
_ => out.write_byte(b'\x00') // NULL
}
0x68 =>
match value {
@moondb.Bool(v) => {
out.write_byte(b'\x01')
out.write_byte((if v { 1 } else { 0 }).to_byte())
}
_ => out.write_byte(b'\x00')
}
0x6D =>
match value {
@moondb.Double(v) => {
out.write_byte(b'\x08')
put_u64(out, v.reinterpret_as_int64())
}
_ => out.write_byte(b'\x00')
}
0xA5 =>
match value {
@moondb.Blob(v) => plp(out, v)
_ => plp_null(out)
}
0x24 =>
match value {
@moondb.Uuid(v) => {
out.write_byte(b'\x10')
out.write_bytes(v)
}
_ => out.write_byte(b'\x00')
}
_ =>
match value {
@moondb.Null => plp_null(out)
@moondb.Array(_) => plp(out, @utf16.encode(value.json_text()))
@moondb.Text(v)
| @moondb.Time(v)
| @moondb.Numeric(v)
| @moondb.Json(v) => plp(out, @utf16.encode(v))
@moondb.Blob(v) => plp(out, v)
@moondb.Bool(v) =>
plp(out, @utf16.encode(if v { "true" } else { "false" }))
@moondb.Int(v) => plp(out, @utf16.encode(v.to_string()))
@moondb.Int64(v) => plp(out, @utf16.encode(v.to_string()))
@moondb.Double(v) => plp(out, @utf16.encode(v.to_string()))
@moondb.Uuid(v) => plp(out, v)
}
}
}
///|
/// One column of the stream's `COLMETADATA`, type information and name included.
fn put_bulk_column(out : Buffer, column : BulkColumn) -> Unit {
put_u32(out, 0) // user type
put_u16(out, 0x01) // nullable
out.write_byte(column.ty.to_byte())
if column.ty == 0x26 || column.ty == 0x68 || column.ty == 0x6D {
out.write_byte(b'\x08')
} else if column.ty == 0x24 {
out.write_byte(b'\x10')
} else if column.ty == 0xE7 {
put_u16(out, 0xFFFF)
for b in [0x09, 0x04, 0xD0, 0x00, 0x34] {
out.write_byte(b.to_byte())
}
} else {
put_u16(out, 0xFFFF)
}
let name = @utf16.encode(column.name)
out.write_byte((name.length() / 2).to_byte())
out.write_bytes(name)
}
///|
/// The whole bulk stream: the column metadata, one row token per row, and the `DONE` that
/// ends the message.
fn build_bulk_load(
columns : Array[BulkColumn],
rows : Array[Array[@moondb.Value]],
) -> Bytes {
let out = Buffer()
out.write_byte(b'\x81') // COLMETADATA
put_u16(out, columns.length())
for column in columns {
put_bulk_column(out, column)
}
for row in rows {
out.write_byte(b'\xd1') // ROW
for i = 0; i < columns.length(); i = i + 1 {
put_cell(out, columns[i].ty, row[i])
}
}
out.write_byte(b'\xfd') // DONE
put_u16(out, 0)
put_u16(out, 0)
put_u64(out, 0L)
out.to_bytes()
}
///|
/// Copy `rows` into `table`, one cell per name in `columns`, and report how many rows the
/// server took.
///
/// The rows travel in the protocol's bulk-load format: `INSERT BULK` declares the shape,
/// then one message carries the column metadata and every row. Values are bound into that
/// stream, never spliced into the statement, so a string containing a quote is a string.
///
/// ```moonbit nocheck
/// let copied = conn.bulk_copy("hero", ["id", "name"], [
/// [@moondb.Int(1), @moondb.Text("Nova")],
/// [@moondb.Int(2), @moondb.Text("Ivan")],
/// ])
/// ```
pub async fn SqlServerConn::bulk_copy(
self : SqlServerConn,
table : String,
columns : Array[String],
rows : Array[Array[@moondb.Value]],
) -> Int64 raise @moondb.DbError {
self.check_open()
if columns.length() == 0 {
raise @moondb.TypeError("bulk_copy needs at least one column")
}
for i = 0; i < rows.length(); i = i + 1 {
if rows[i].length() != columns.length() {
raise @moondb.TypeError(
"row " +
i.to_string() +
" has " +
rows[i].length().to_string() +
" values for " +
columns.length().to_string() +
" columns",
)
}
}
if rows.length() == 0 {
return 0L
}
let bulk = bulk_columns(columns, rows)
let defs = Buffer()
for i = 0; i < bulk.length(); i = i + 1 {
if i > 0 {
defs.write_bytes(@utf8.encode(", "))
}
defs.write_bytes(@utf8.encode("[" + bulk[i].name + "] "))
defs.write_bytes(@utf8.encode(bulk_declaration(bulk[i].ty)))
}
let statement = "INSERT BULK " +
table +
" (" +
decode_lossy(defs.to_bytes()) +
")"
self.run(statement, []) |> ignore
self.wire.write_message(0x07, build_bulk_load(bulk, rows))
let (_, payload) = self.wire.read_message()
let response = parse_response(payload)
match response.txn {
Some(descriptor) => self.txn = descriptor
None => ()
}
match response.failure {
Some(detail) => raise error_of(detail)
None => ()
}
response.rows_affected
}