Skip to content

Commit d7521d4

Browse files
pavanjavapavanjava
authored
bulk insert data (#10)
1 parent 9b5b2e1 commit d7521d4

8 files changed

Lines changed: 335 additions & 1 deletion

File tree

‎README.md‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,56 @@ INSERT INTO COLLECTION articles VALUES {'text': 'hello world'}
239239

240240
---
241241

242+
### INSERT BULK — batch insert multiple points
243+
244+
Inserts multiple documents in a single statement. Each item in the array must contain a `"text"` key. All items are embedded and upserted to Qdrant in **one batched call**, which is significantly faster than issuing one `INSERT` per record.
245+
246+
If the collection does not exist yet, it is **created automatically** on the first bulk insert.
247+
248+
**Syntax:**
249+
```
250+
INSERT BULK INTO COLLECTION <collection_name> VALUES [<dict>, <dict>, ...]
251+
INSERT BULK INTO COLLECTION <collection_name> VALUES [<dict>, ...] USING MODEL '<model_name>'
252+
INSERT BULK INTO COLLECTION <collection_name> VALUES [<dict>, ...] USING HYBRID
253+
INSERT BULK INTO COLLECTION <collection_name> VALUES [<dict>, ...] USING HYBRID DENSE MODEL '<model>' SPARSE MODEL '<model>'
254+
```
255+
256+
**Examples:**
257+
258+
Minimal bulk insert (text only):
259+
```sql
260+
INSERT BULK INTO COLLECTION articles VALUES [
261+
{'text': 'Qdrant supports cosine similarity search'},
262+
{'text': 'Sparse BM25 vectors enable keyword retrieval'},
263+
{'text': 'Hybrid search combines dense and sparse results via RRF'}
264+
]
265+
```
266+
267+
Bulk insert with metadata:
268+
```sql
269+
INSERT BULK INTO COLLECTION articles VALUES [
270+
{'text': 'Attention is all you need', 'author': 'vaswani', 'year': 2017},
271+
{'text': 'BERT: Pre-training of deep bidirectional transformers', 'author': 'devlin', 'year': 2018},
272+
{'text': 'Language models are few-shot learners', 'author': 'brown', 'year': 2020}
273+
]
274+
```
275+
276+
Bulk insert into a hybrid collection:
277+
```sql
278+
INSERT BULK INTO COLLECTION articles VALUES [
279+
{'text': 'Dense retrieval with FAISS', 'domain': 'ir'},
280+
{'text': 'Sparse retrieval with BM25', 'domain': 'ir'}
281+
] USING HYBRID
282+
```
283+
284+
**Rules:**
285+
- Every dict in the array must contain a `"text"` key. Missing `text` on any item raises an error with the offending index.
286+
- An empty array `[]` raises an error.
287+
- A UUID is auto-generated for each point — you do not provide IDs.
288+
- Supports all the same `USING` clauses as single `INSERT`.
289+
290+
---
291+
242292
### SEARCH — find similar points
243293

244294
Performs a **semantic similarity search**: your query text is embedded with the same model used during insert, then Qdrant finds the nearest vectors by cosine distance.

‎src/qql/ast_nodes.py‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,15 @@ class InsertStmt:
129129
sparse_model: str | None = None # sparse model; None → SparseEmbedder.DEFAULT_MODEL
130130

131131

132+
@dataclass(frozen=True)
133+
class InsertBulkStmt:
134+
collection: str
135+
values_list: tuple[dict[str, Any], ...] # each dict must contain "text"
136+
model: str | None # dense model; None → use config default
137+
hybrid: bool = False
138+
sparse_model: str | None = None
139+
140+
132141
@dataclass(frozen=True)
133142
class CreateCollectionStmt:
134143
collection: str
@@ -169,6 +178,7 @@ class DeleteStmt:
169178
# Union type for all top-level statement nodes
170179
ASTNode = (
171180
InsertStmt
181+
| InsertBulkStmt
172182
| CreateCollectionStmt
173183
| DropCollectionStmt
174184
| ShowCollectionsStmt

‎src/qql/cli.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,10 @@
2828
Optional: [yellow]USING MODEL[/yellow] '<model>'
2929
Optional: [yellow]USING HYBRID[/yellow] [DENSE MODEL '<model>'] [SPARSE MODEL '<model>']
3030
31+
[yellow]INSERT BULK INTO COLLECTION[/yellow] <name> [yellow]VALUES[/yellow] [{[yellow]'text'[/yellow]: '...', ...}, ...]
32+
Batch insert multiple points in a single call. Each dict must contain 'text'.
33+
Supports the same [yellow]USING[/yellow] clauses as INSERT.
34+
3135
[yellow]CREATE COLLECTION[/yellow] <name> [[yellow]HYBRID[/yellow]]
3236
Create a new collection. Add HYBRID for dense+sparse BM25 vectors.
3337
Optional: [yellow]USING MODEL[/yellow] '<model>'

‎src/qql/executor.py‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
DropCollectionStmt,
4343
FilterExpr,
4444
InExpr,
45+
InsertBulkStmt,
4546
InsertStmt,
4647
IsEmptyExpr,
4748
IsNotEmptyExpr,
@@ -77,6 +78,8 @@ def __init__(self, client: QdrantClient, config: QQLConfig) -> None:
7778
self._config = config
7879

7980
def execute(self, node: ASTNode) -> ExecutionResult:
81+
if isinstance(node, InsertBulkStmt):
82+
return self._execute_insert_bulk(node)
8083
if isinstance(node, InsertStmt):
8184
return self._execute_insert(node)
8285
if isinstance(node, CreateCollectionStmt):
@@ -170,6 +173,84 @@ def _execute_insert(self, node: InsertStmt) -> ExecutionResult:
170173
data={"id": point_id, "collection": node.collection},
171174
)
172175

176+
def _execute_insert_bulk(self, node: InsertBulkStmt) -> ExecutionResult:
177+
if not node.values_list:
178+
raise QQLRuntimeError("INSERT BULK VALUES list is empty")
179+
for i, vals in enumerate(node.values_list):
180+
if "text" not in vals:
181+
raise QQLRuntimeError(
182+
f"INSERT BULK: item at index {i} is missing required 'text' field"
183+
)
184+
185+
# ── Hybrid bulk INSERT: dense + sparse vectors ─────────────────────
186+
if node.hybrid:
187+
dense_model = node.model or self._config.default_model
188+
sparse_model_name = node.sparse_model or SparseEmbedder.DEFAULT_MODEL
189+
dense_embedder = Embedder(dense_model)
190+
sparse_embedder = SparseEmbedder(sparse_model_name)
191+
192+
points: list[PointStruct] = []
193+
for vals in node.values_list:
194+
dense_vector = dense_embedder.embed(vals["text"])
195+
sparse_obj = sparse_embedder.embed(vals["text"])
196+
sparse_vector = SparseVector(
197+
indices=sparse_obj["indices"], values=sparse_obj["values"]
198+
)
199+
point_id = str(uuid.uuid4())
200+
points.append(
201+
PointStruct(
202+
id=point_id,
203+
vector={"dense": dense_vector, "sparse": sparse_vector},
204+
payload=dict(vals),
205+
)
206+
)
207+
208+
if not self._client.collection_exists(node.collection):
209+
first_dense = dense_embedder.embed(node.values_list[0]["text"])
210+
self._client.create_collection(
211+
collection_name=node.collection,
212+
vectors_config={
213+
"dense": VectorParams(size=len(first_dense), distance=Distance.COSINE)
214+
},
215+
sparse_vectors_config={
216+
"sparse": SparseVectorParams(modifier=Modifier.IDF)
217+
},
218+
)
219+
220+
try:
221+
self._client.upsert(collection_name=node.collection, points=points)
222+
except UnexpectedResponse as e:
223+
raise QQLRuntimeError(f"Qdrant error during INSERT BULK: {e}") from e
224+
225+
return ExecutionResult(
226+
success=True,
227+
message=f"Inserted {len(points)} points (hybrid)",
228+
)
229+
230+
# ── Standard dense-only bulk INSERT ───────────────────────────────
231+
model_name = node.model or self._config.default_model
232+
embedder = Embedder(model_name)
233+
234+
points = []
235+
for vals in node.values_list:
236+
vector = embedder.embed(vals["text"])
237+
point_id = str(uuid.uuid4())
238+
points.append(
239+
PointStruct(id=point_id, vector=vector, payload=dict(vals))
240+
)
241+
242+
self._ensure_collection(node.collection, len(points[0].vector))
243+
244+
try:
245+
self._client.upsert(collection_name=node.collection, points=points)
246+
except UnexpectedResponse as e:
247+
raise QQLRuntimeError(f"Qdrant error during INSERT BULK: {e}") from e
248+
249+
return ExecutionResult(
250+
success=True,
251+
message=f"Inserted {len(points)} points",
252+
)
253+
173254
def _execute_create(self, node: CreateCollectionStmt) -> ExecutionResult:
174255
if self._client.collection_exists(node.collection):
175256
return ExecutionResult(

‎src/qql/lexer.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
class TokenKind(Enum):
88
# ── Statement keywords ────────────────────────────────────────────────
99
INSERT = auto()
10+
BULK = auto()
1011
INTO = auto()
1112
COLLECTION = auto()
1213
VALUES = auto()
@@ -71,6 +72,7 @@ class TokenKind(Enum):
7172
_KEYWORDS: dict[str, TokenKind] = {
7273
# Statement keywords
7374
"INSERT": TokenKind.INSERT,
75+
"BULK": TokenKind.BULK,
7476
"INTO": TokenKind.INTO,
7577
"COLLECTION": TokenKind.COLLECTION,
7678
"VALUES": TokenKind.VALUES,

‎src/qql/parser.py‎

Lines changed: 44 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
DropCollectionStmt,
1111
FilterExpr,
1212
InExpr,
13+
InsertBulkStmt,
1314
InsertStmt,
1415
IsEmptyExpr,
1516
IsNotEmptyExpr,
@@ -70,8 +71,12 @@ def parse(self) -> ASTNode:
7071

7172
# ── Statement parsers ─────────────────────────────────────────────────
7273

73-
def _parse_insert(self) -> InsertStmt:
74+
def _parse_insert(self) -> InsertStmt | InsertBulkStmt:
7475
self._expect(TokenKind.INSERT)
76+
if self._peek().kind == TokenKind.BULK:
77+
self._advance() # consume BULK
78+
return self._parse_insert_bulk_body()
79+
# ── Standard single INSERT ────────────────────────────────────────
7580
self._expect(TokenKind.INTO)
7681
self._expect(TokenKind.COLLECTION)
7782
collection = self._parse_identifier()
@@ -102,6 +107,44 @@ def _parse_insert(self) -> InsertStmt:
102107
hybrid=hybrid, sparse_model=sparse_model,
103108
)
104109

110+
def _parse_insert_bulk_body(self) -> InsertBulkStmt:
111+
self._expect(TokenKind.INTO)
112+
self._expect(TokenKind.COLLECTION)
113+
collection = self._parse_identifier()
114+
self._expect(TokenKind.VALUES)
115+
raw_list = self._parse_list()
116+
for i, item in enumerate(raw_list):
117+
if not isinstance(item, dict):
118+
raise QQLSyntaxError(
119+
f"INSERT BULK VALUES item at index {i} must be a dict, "
120+
f"got {type(item).__name__}",
121+
0,
122+
)
123+
values_list: tuple[dict, ...] = tuple(raw_list)
124+
model: str | None = None
125+
hybrid: bool = False
126+
sparse_model: str | None = None
127+
if self._peek().kind == TokenKind.USING:
128+
self._advance() # consume USING
129+
if self._peek().kind == TokenKind.HYBRID:
130+
self._advance() # consume HYBRID
131+
hybrid = True
132+
while self._peek().kind in (TokenKind.DENSE, TokenKind.SPARSE):
133+
sub = self._advance()
134+
self._expect(TokenKind.MODEL)
135+
m = self._expect(TokenKind.STRING).value
136+
if sub.kind == TokenKind.DENSE:
137+
model = m
138+
else:
139+
sparse_model = m
140+
else:
141+
self._expect(TokenKind.MODEL)
142+
model = self._expect(TokenKind.STRING).value
143+
return InsertBulkStmt(
144+
collection=collection, values_list=values_list,
145+
model=model, hybrid=hybrid, sparse_model=sparse_model,
146+
)
147+
105148
def _parse_create(self) -> CreateCollectionStmt:
106149
self._expect(TokenKind.CREATE)
107150
self._expect(TokenKind.COLLECTION)

‎tests/test_executor.py‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
CreateCollectionStmt,
55
DeleteStmt,
66
DropCollectionStmt,
7+
InsertBulkStmt,
78
InsertStmt,
89
SearchStmt,
910
SearchWith,
@@ -91,6 +92,79 @@ def test_insert_raises_on_dimension_mismatch(self, executor, mock_client):
9192
executor.execute(node)
9293

9394

95+
class TestInsertBulk:
96+
def test_bulk_insert_calls_upsert_once(self, executor, mock_client):
97+
node = InsertBulkStmt(
98+
collection="col",
99+
values_list=({"text": "hello"}, {"text": "world"}),
100+
model=None,
101+
)
102+
executor.execute(node)
103+
mock_client.upsert.assert_called_once()
104+
105+
def test_bulk_insert_upserts_correct_count(self, executor, mock_client):
106+
node = InsertBulkStmt(
107+
collection="col",
108+
values_list=({"text": "a"}, {"text": "b"}, {"text": "c"}),
109+
model=None,
110+
)
111+
executor.execute(node)
112+
call_args = mock_client.upsert.call_args.kwargs
113+
assert len(call_args["points"]) == 3
114+
115+
def test_bulk_insert_creates_collection_when_missing(self, executor, mock_client):
116+
node = InsertBulkStmt(
117+
collection="col",
118+
values_list=({"text": "hello"},),
119+
model=None,
120+
)
121+
executor.execute(node)
122+
mock_client.create_collection.assert_called_once()
123+
124+
def test_bulk_insert_skips_create_when_exists(self, executor, mock_client):
125+
mock_client.collection_exists.return_value = True
126+
mock_client.get_collection.return_value.config.params.vectors.size = 384
127+
node = InsertBulkStmt(
128+
collection="col",
129+
values_list=({"text": "hello"},),
130+
model=None,
131+
)
132+
executor.execute(node)
133+
mock_client.create_collection.assert_not_called()
134+
135+
def test_bulk_insert_raises_on_missing_text(self, executor):
136+
node = InsertBulkStmt(
137+
collection="col",
138+
values_list=({"text": "ok"}, {"author": "bob"}),
139+
model=None,
140+
)
141+
with pytest.raises(QQLRuntimeError, match="index 1"):
142+
executor.execute(node)
143+
144+
def test_bulk_insert_empty_list_raises(self, executor):
145+
node = InsertBulkStmt(collection="col", values_list=(), model=None)
146+
with pytest.raises(QQLRuntimeError, match="empty"):
147+
executor.execute(node)
148+
149+
def test_bulk_insert_result_message_contains_count(self, executor, mock_client):
150+
node = InsertBulkStmt(
151+
collection="col",
152+
values_list=({"text": "a"}, {"text": "b"}),
153+
model=None,
154+
)
155+
result = executor.execute(node)
156+
assert result.success is True
157+
assert "2" in result.message
158+
assert "points" in result.message
159+
160+
def test_single_insert_unaffected_by_bulk_dispatch(self, executor, mock_client):
161+
"""Ensure single INSERT still routes correctly after bulk dispatch added."""
162+
node = InsertStmt(collection="notes", values={"text": "hello"}, model=None)
163+
result = executor.execute(node)
164+
assert result.success is True
165+
assert "Inserted 1 point" in result.message
166+
167+
94168
class TestCreate:
95169
def test_create_new_collection(self, executor, mock_client):
96170
node = CreateCollectionStmt(collection="new_col")

0 commit comments

Comments
 (0)