diff --git a/oracle/soda.py b/oracle/soda.py index 380bc37..7dea8f2 100644 --- a/oracle/soda.py +++ b/oracle/soda.py @@ -12,16 +12,15 @@ `SodaDocument` objects. SODA needs an Oracle 18c+ server (``DBMS_SODA``); `getSodaDatabase` raises -``NotSupportedError`` below that. This module covers collection management (#199) -and the document model — insert + read by key (#200). QBE find, update and -delete land in the follow-up sub-tickets of #163. +``NotSupportedError`` below that. The surface covers collection management, +documents (insert / read / upsert / bulk), query-by-example find with streaming, +update / delete, and indexing + data guide. A document's content is bound into ``DBMS_SODA`` as bytes; a value over the -32767-byte PL/SQL limit rides pyoracle's transparent temp-LOB bind (#91), so -*inserting* large documents works. *Reading* content back comes through a single -``DBMS_LOB.SUBSTR`` and is therefore capped at 32767 bytes — a document whose -stored content exceeds that raises rather than silently truncating; chunked -large-content reads are a follow-up. +32767-byte PL/SQL limit rides pyoracle's transparent temp-LOB bind (#91). +Reading content back returns the inline 32767-byte window from the batch read, +and slices anything larger out of the BLOB in 32767-byte chunks via a follow-up +round trip per oversized document. """ import json @@ -102,6 +101,19 @@ # getDocuments() materialises into host arrays, so it needs a capacity. With no # limit set, cap at this many and raise on overflow rather than truncate. _DEFAULT_FETCH_CAP = 1000 +# Read a single document's full content when it is larger than the 32767-byte +# inline window the batch read returns (#211): re-open by key and slice the BLOB +# into a RAW host array, one 32767-byte chunk per element, in one round trip. +# The caller sizes the array from the length the batch already reported. +_READ_CONTENT = ( + "DECLARE c SODA_COLLECTION_T; d SODA_DOCUMENT_T; b BLOB; len NUMBER; " + "off NUMBER := 1; i PLS_INTEGER := 0; BEGIN " + "c := DBMS_SODA.open_collection(:name); " + "d := c.find().key(:key).get_one(); b := d.get_blob(); " + "len := DBMS_LOB.GETLENGTH(b); " + "WHILE off <= len LOOP i := i + 1; " + ":chunks(i) := DBMS_LOB.SUBSTR(b, 32767, off); off := off + 32767; " + "END LOOP; :n := i; END;") # Update / delete / bulk (#202). replace_one returns a NUMBER (1 replaced / 0 # not); replace_one_and_get returns the new SODA_DOCUMENT_T (NULL if nothing # matched); remove returns the deleted count. DBMS_SODA's bulk insert_many and @@ -274,18 +286,6 @@ def _data_guide_doc(b: dict): return SodaDocument(content=text.encode("utf-8") if text else None) -def _content_or_raise(content, length): - # Guard against a silently truncated read: the single SUBSTR maxes at - # _MAX_INLINE bytes (#200 limitation). - if length is not None and int(length) > _MAX_INLINE: - raise NotSupportedError( - f"SODA document content is {int(length)} bytes; reading content " - f"over {_MAX_INLINE} bytes is not yet supported") - if content is None: - return None - return bytes(content) - - # -------------------------------------------------------------------------- # Sync # -------------------------------------------------------------------------- @@ -334,7 +334,7 @@ def getOne(self) -> "SodaDocument | None": b.update(binds) b["cap"] = 1 cur.execute(_GET_DOCS, b) - docs = _docs_from_arrays(b) + docs = self._build_docs(b) return docs[0] if docs else None def count(self) -> int: @@ -353,6 +353,23 @@ def _run_get_docs(self, skip: int, lim: int, cap: int) -> dict: cur.execute(_GET_DOCS, b) return b + def _read_full_content(self, key: str, clen: int) -> bytes: + nchunks = (int(clen) + _MAX_INLINE - 1) // _MAX_INLINE + cur = self._connection.cursor() + chunks = cur.arrayvar(DB_TYPE_RAW, nchunks) + cnt = cur.var(DB_TYPE_NUMBER) + cur.execute(_READ_CONTENT, {"name": self._collection.name, "key": key, + "chunks": chunks, "n": cnt}) + parts = chunks.getvalue() + return b"".join(bytes(p) for p in parts[:int(cnt.getvalue())] + if p is not None) + + def _build_docs(self, b: dict) -> list: + docs, large = _parse_doc_arrays(b) + for doc, key, clen in large: + doc._content = self._read_full_content(key, clen) + return docs + def getDocuments(self) -> "list[SodaDocument]": """Every matched document. Without `.limit(...)` this materialises up to a fixed cap and raises if more match, rather than truncating; use @@ -360,7 +377,7 @@ def getDocuments(self) -> "list[SodaDocument]": cap = self._limit if self._limit else _DEFAULT_FETCH_CAP b = self._run_get_docs(self._skip, self._limit, cap) _check_fetch_overflow(b, cap) - return _docs_from_arrays(b) + return self._build_docs(b) def getCursor(self, batchSize: int = 100): """Stream the matched documents in batches of `batchSize` (offset @@ -373,7 +390,7 @@ def getCursor(self, batchSize: int = 100): if self._limit and yielded >= self._limit: return lim = batch if not self._limit else min(batch, self._limit - yielded) - docs = _docs_from_arrays( + docs = self._build_docs( self._run_get_docs(self._skip + offset, lim, batch)) if not docs: return @@ -406,7 +423,7 @@ def replaceOneAndGet(self, doc) -> "SodaDocument | None": cur.execute(_REPLACE_ONE_AND_GET, b) if b["missing"].getvalue(): return None - return _doc_from_binds(b, with_content=False) + return _doc_from_binds(b) def remove(self) -> int: """Remove the matched documents; returns the number removed.""" @@ -471,7 +488,7 @@ def insertOneAndGet(self, doc) -> SodaDocument: b = _new_doc_out_binds(cur, content=False) b.update({"name": self.name, "key": key, "content": content, "mt": mt}) cur.execute(_INSERT_ONE_AND_GET, b) - return _doc_from_binds(b, with_content=False) + return _doc_from_binds(b) def insertMany(self, docs) -> None: """Insert several documents (each a `SodaDocument` or a bare value) in a @@ -491,7 +508,7 @@ def _save(self, doc) -> SodaDocument: b = _new_doc_out_binds(cur, content=False) b.update({"name": self.name, "key": key, "content": content, "mt": mt}) cur.execute(_SAVE, b) - return _doc_from_binds(b, with_content=False) + return _doc_from_binds(b) def save(self, doc) -> None: """Insert a document, or replace the existing one with the same key @@ -597,13 +614,11 @@ def _new_doc_out_binds(cur, content: bool = True) -> dict: return binds -def _doc_from_binds(b: dict, with_content: bool) -> SodaDocument: - content = None - if with_content: - content = _content_or_raise(b["content"].getvalue(), - b["clen"].getvalue()) +def _doc_from_binds(b: dict) -> SodaDocument: + # The metadata-only document an insert / replace / save returns (key, + # version, media type, timestamps — no content). return SodaDocument( - content=content, key=b["rkey"].getvalue(), version=b["rver"].getvalue(), + content=None, key=b["rkey"].getvalue(), version=b["rver"].getvalue(), mediaType=b["rmt"].getvalue() or _DEFAULT_MEDIA_TYPE, createdOn=b["rcreated"].getvalue(), lastModified=b["rmodified"].getvalue()) @@ -632,19 +647,28 @@ def _check_fetch_overflow(b: dict, cap: int) -> None: f"(streaming getCursor is a follow-up)") -def _docs_from_arrays(b: dict) -> list: +def _parse_doc_arrays(b: dict) -> tuple: + # Build the documents from a getDocuments batch. A document whose content + # fits the inline window gets it directly; one larger than that gets None + # content now and is returned in the second list as (doc, key, length) for + # the caller to backfill via a chunked read (#211). n = int(b["num"].getvalue()) keys, contents = b["keys"].getvalue(), b["contents"].getvalue() clens, vers = b["clens"].getvalue(), b["vers"].getvalue() mts, creates, mods = (b["mts"].getvalue(), b["creates"].getvalue(), b["mods"].getvalue()) - docs = [] + docs, large = [], [] for i in range(n): - docs.append(SodaDocument( - content=_content_or_raise(contents[i], clens[i]), key=keys[i], + clen = int(clens[i]) if clens[i] is not None else 0 + inline = bytes(contents[i]) if contents[i] is not None else None + doc = SodaDocument( + content=None if clen > _MAX_INLINE else inline, key=keys[i], version=vers[i], mediaType=mts[i] or _DEFAULT_MEDIA_TYPE, - createdOn=creates[i], lastModified=mods[i])) - return docs + createdOn=creates[i], lastModified=mods[i]) + docs.append(doc) + if clen > _MAX_INLINE: + large.append((doc, keys[i], clen)) + return docs, large # -------------------------------------------------------------------------- @@ -690,7 +714,7 @@ async def getOne(self) -> "SodaDocument | None": b.update(binds) b["cap"] = 1 await cur.execute(_GET_DOCS, b) - docs = _docs_from_arrays(b) + docs = await self._build_docs(b) return docs[0] if docs else None async def count(self) -> int: @@ -709,11 +733,29 @@ async def _run_get_docs(self, skip: int, lim: int, cap: int) -> dict: await cur.execute(_GET_DOCS, b) return b + async def _read_full_content(self, key: str, clen: int) -> bytes: + nchunks = (int(clen) + _MAX_INLINE - 1) // _MAX_INLINE + cur = self._connection.cursor() + chunks = cur.arrayvar(DB_TYPE_RAW, nchunks) + cnt = cur.var(DB_TYPE_NUMBER) + await cur.execute(_READ_CONTENT, + {"name": self._collection.name, "key": key, + "chunks": chunks, "n": cnt}) + parts = chunks.getvalue() + return b"".join(bytes(p) for p in parts[:int(cnt.getvalue())] + if p is not None) + + async def _build_docs(self, b: dict) -> list: + docs, large = _parse_doc_arrays(b) + for doc, key, clen in large: + doc._content = await self._read_full_content(key, clen) + return docs + async def getDocuments(self) -> "list[SodaDocument]": cap = self._limit if self._limit else _DEFAULT_FETCH_CAP b = await self._run_get_docs(self._skip, self._limit, cap) _check_fetch_overflow(b, cap) - return _docs_from_arrays(b) + return await self._build_docs(b) async def getCursor(self, batchSize: int = 100): """Async streaming counterpart to the sync `getCursor`: `async for doc in @@ -724,7 +766,7 @@ async def getCursor(self, batchSize: int = 100): if self._limit and yielded >= self._limit: return lim = batch if not self._limit else min(batch, self._limit - yielded) - docs = _docs_from_arrays( + docs = await self._build_docs( await self._run_get_docs(self._skip + offset, lim, batch)) if not docs: return @@ -754,7 +796,7 @@ async def replaceOneAndGet(self, doc) -> "SodaDocument | None": await cur.execute(_REPLACE_ONE_AND_GET, b) if b["missing"].getvalue(): return None - return _doc_from_binds(b, with_content=False) + return _doc_from_binds(b) async def remove(self) -> int: cur = self._connection.cursor() @@ -810,7 +852,7 @@ async def insertOneAndGet(self, doc) -> SodaDocument: b = _new_doc_out_binds(cur, content=False) b.update({"name": self.name, "key": key, "content": content, "mt": mt}) await cur.execute(_INSERT_ONE_AND_GET, b) - return _doc_from_binds(b, with_content=False) + return _doc_from_binds(b) async def insertMany(self, docs) -> None: contents = [_doc_to_bind(d)[1] for d in docs] @@ -828,7 +870,7 @@ async def _save(self, doc) -> SodaDocument: b = _new_doc_out_binds(cur, content=False) b.update({"name": self.name, "key": key, "content": content, "mt": mt}) await cur.execute(_SAVE, b) - return _doc_from_binds(b, with_content=False) + return _doc_from_binds(b) async def save(self, doc) -> None: await self._save(doc) diff --git a/tests/test_integration.py b/tests/test_integration.py index 16bfd29..3d49eb0 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2284,13 +2284,25 @@ def test_document_insert_and_read(self): col.insertOne({"name": "bob"}) self.assertIsNone(col.find().key("AABBCCDDEEFF00112233445566").getOne()) - def test_large_document_insert_and_read_guard(self): + def test_large_document_round_trip(self): col = self.soda.createCollection("it_big") - big = {"blob": "x" * 40000} # > 32767-byte inline read limit - saved = col.insertOneAndGet(big) # insert is fine (temp LOB) - self.assertTrue(saved.key) - with self.assertRaises(oracle.NotSupportedError): - col.find().key(saved.key).getOne() # read guards against truncation + # content past the 32767-byte inline window reads back whole, chunked + # out of the BLOB (#211); insert uses the temp-LOB bind. + payload = "x" * 120000 + k1 = col.insertOneAndGet({"tag": "A", "blob": payload}).key + col.insertOneAndGet({"tag": "B", "blob": "y" * 90000}) + self.assertEqual(col.find().key(k1).getOne().getContent()["blob"], + payload) + # a batch with two large documents keeps each one's content distinct + docs = {d.getContent()["tag"]: d.getContent()["blob"] + for d in col.find().getDocuments()} + self.assertEqual(docs["A"], payload) + self.assertEqual(docs["B"], "y" * 90000) + # boundary: exactly at and one past the inline window + for n in (32767, 32768): + k = col.insertOneAndGet({"p": "z" * n}).key + self.assertEqual(col.find().key(k).getOne().getContent()["p"], + "z" * n) def test_qbe_find(self): col = self.soda.createCollection("it_qbe") diff --git a/tests/test_soda.py b/tests/test_soda.py index b17f340..8ef9503 100644 --- a/tests/test_soda.py +++ b/tests/test_soda.py @@ -7,7 +7,7 @@ import unittest from oracle.exceptions import NotSupportedError -from oracle.soda import (SodaDocument, _check_soda_supported, _content_or_raise, +from oracle.soda import (SodaDocument, _check_soda_supported, _doc_to_bind, _encode_content, _names_query, _norm_filter, _norm_metadata) @@ -89,14 +89,6 @@ def test_norm_filter(self): '{"age": {"$gte": 30}}') # dict -> JSON self.assertEqual(_norm_filter('{"a":1}'), '{"a":1}') # str passthrough - def test_content_read_limit(self): - # bytes within the inline limit pass through; over it raises rather than - # silently truncating (#200). - self.assertEqual(_content_or_raise(b"abc", 3), b"abc") - self.assertIsNone(_content_or_raise(None, 0)) - with self.assertRaises(NotSupportedError): - _content_or_raise(b"x" * 32767, 40000) - if __name__ == "__main__": unittest.main()