Bind a property's texts beside a long operand on Oracle, so a text past 32 KB a text function or a sum reads beside one leaves the session open

drivine/cypher/sql/oracle_cypher_compiler.py

@@ -28,6 +28,7 @@
     ROW,
     ROWS,
     SAVED,
+    SHARED_ABOVE,
     VALUE,
     Body,
     SqlCypherCompiler,
@@ -316,21 +317,96 @@ def texts(self, jsons: Sequence[str], body: Body) -> str:
         it whole. The body gives null where a text it is given is null, and where what it gives would not fit in a
         `VARCHAR2`: see `replace_text` and `split`. A short text is not read as a CLOB, which is slower: `COALESCE`
         reads its second value only where the first is null. The body knows which it is given by `_long`.
-        """
 
-        def both(xs: list[str]) -> str:
-            # Within a body given CLOBs, as CLOBs alone.
-            if self._long:
-                return body([self._text_of(it, clob=True) for it in xs])
-            read = body([self._text_of(it, clob=False) for it in xs])
-            self._long = True
-            try:
-                whole = body([self._text_of(it, clob=True) for it in xs])
-            finally:
-                self._long = False
-            return f"COALESCE({read}, {whole})"
+        A property beside a long value, which is bound in a subquery, has its texts bound beside it, each read by the
+        property's own path (see `_direct`): Oracle closed the session (ORA-03113) where a text past 32 KB was read as a
+        CLOB, and searched or compared, beside such a subquery for each of two rows or more (drivine4j#370), where the
+        property was left where it is read, and where it was bound as JSON and read as a CLOB from that.
+        """
+        properties = [_beside_long(it) for it in jsons]
+        if not self.binds_long(jsons) or not any(properties):
+            return self.shared(jsons, lambda xs: self._texts_of([self._texts_of_json(it) for it in xs], body))
+        # Within a body given CLOBs, a property's CLOB alone: its VARCHAR2 is not read.
+        columns: list[str] = []
+        for json, property_ in zip(jsons, properties, strict=True):
+            if not property_:
+                columns.append(json)
+            elif self._long:
+                columns.append(self._text_of(json, clob=True))
+            else:
+                columns += [self._text_of(json, clob=False), self._text_of(json, clob=True)]
+
+        def bound(names: list[str]) -> str:
+            rest = iter(names)
+            texts: list[_Texts] = []
+            for property_ in properties:
+                if not property_:
+                    texts.append(self._texts_of_json(next(rest)))
+                elif self._long:
+                    whole = next(rest)
+                    texts.append(_Texts(whole, whole))
+                else:
+                    texts.append(_Texts(next(rest), next(rest)))
+            return self._texts_of(texts, body)
+
+        return self.shared_bound(columns, bound)
+
+    def _texts_of_json(self, json: str) -> _Texts:
+        """The JSON's texts: see `_text_of`."""
+        return _Texts(self._text_of(json, clob=False), self._text_of(json, clob=True))
+
+    def _texts_of(self, texts: Sequence[_Texts], body: Body) -> str:
+        """The body over the texts as `VARCHAR2`s, and where it gives null, as CLOBs: see `texts`."""
+        # Within a body given CLOBs, as CLOBs alone.
+        if self._long:
+            return body([it.whole for it in texts])
+        read = body([it.short for it in texts])
+        self._long = True
+        try:
+            whole = body([it.whole for it in texts])
+        finally:
+            self._long = False
+        return f"COALESCE({read}, {whole})"
+
+    def shared_sum(self, a: str, b: str, body: Body) -> str:
+        """
+        A side of a `+` that is a property, beside a long value, is bound by the property's own path, as `texts` binds
+        its texts: its kind and its CLOB, each a column the body's reads of them are rewritten to, and the property as
+        JSON only where it is no text past _BOUND_TEXT characters. Bound as JSON whole and read as a CLOB from that, a
+        text past 32 KB closed the session on two rows or more (ORA-03113, drivine4j#370); left where it is read beside
+        the subquery, as `_direct` reads it, a sum of short texts closed it on one, `a.id + '->' + b.id`. A text past
+        _BOUND_TEXT characters is summed with a text or a number from its CLOB, and is null added to a list, which takes
+        it as JSON: read as JSON beside the subquery, it closed the session.
+        """
+        sides = [a, b]
+        properties = [_beside_long(it) for it in sides]
+        if not self.binds_long(sides) or not any(properties):
+            return self.shared(sides, body)
+        columns: list[str] = []
+        for json, property_ in zip(sides, properties, strict=True):
+            if not property_:
+                columns.append(json)
+                continue
+            props, path = cast(tuple[str, str], _property_read(json))
+            length, type_ = f"{path[:-1]}.length()'", f"{path[:-1]}.type()'"
+            columns += [
+                f"CASE WHEN JSON_VALUE({props}, {length} RETURNING NUMBER) > {_BOUND_TEXT} THEN NULL ELSE {json} END",
+                f"JSON_VALUE({props}, {type_})",
+                f"JSON_VALUE({props}, {path} RETURNING CLOB)",
+            ]
+
+        def bound(names: list[str]) -> str:
+            rest = iter(names)
+            sides_bound = [[next(rest), next(rest), next(rest)] if it else [next(rest)] for it in properties]
+            sql = body([it[0] for it in sides_bound])
+            # A property's kind and CLOB are read from their own columns: its JSON is null past _BOUND_TEXT characters.
+            for side in sides_bound:
+                if len(side) == 3:
+                    json, kind, clob = side
+                    sql = sql.replace(_kind(json), kind).replace(f"JSON_VALUE({json}, '$' RETURNING CLOB)", clob)
+            return sql
 
-        return self.shared(jsons, both)
+        return self.shared_bound(columns, bound)
 
     def _text_of(self, json: str, clob: bool) -> str:
         """The JSON as text, where it is a string: a CLOB where `clob`, else as `text` reads it; a property by its own
@@ -1381,6 +1457,11 @@ def _keyed_as_extended(value: Any) -> bool:
 _SHORT_TEXT = 4000
 """The most bytes a `VARCHAR2` holds in SQL, as a database is made by default (`MAX_STRING_SIZE = STANDARD`)."""
 
+_BOUND_TEXT = 32_767
+"""The most characters of a text `shared_sum` binds as JSON: one of 40,001, bound as JSON beside a long value, closed
+the session (drivine4j#370), where one of 12,000 three-byte characters, 36 KB, did not, so a text is counted in
+characters, as `.length()` counts them. One as long or shorter is bound as it was."""
+
 _REVERSED_PIECE = 1000
 """How many UTF-16 units of a CLOB `_reversed_pieces` reverses as one `VARCHAR2`: with one more, where a pair is not
 cut, and the `x` the piece is put together after, at most 3,002 bytes, as a unit is three bytes at most and a pair
@@ -1402,6 +1483,20 @@ def _property_read(json: str) -> tuple[str, str] | None:
     return (match[1], match[2]) if match is not None and _balanced(match[1]) else None
 
 
+def _beside_long(json: str) -> bool:
+    """Whether the JSON is a property as `property_of` reads it, short beside a long value: see `texts` and
+    `shared_sum`."""
+    return len(json) <= SHARED_ABOVE and _property_read(json) is not None
+
+
+@dataclasses.dataclass(frozen=True)
+class _Texts:
+    """A text as a `VARCHAR2`, null past 4,000 bytes, and as a CLOB: see `texts`."""
+
+    short: str
+    whole: str
+
+
 def _balanced(sql: str) -> bool:
     """Whether the SQL's parentheses outside its literals close each other: so it is the one argument, not two
     calls."""

drivine/cypher/sql/sql_cypher_compiler.py

@@ -519,11 +519,23 @@ def shared(self, values: Sequence[str], body: Body) -> str:
         one another do not grow the statement exponentially. Short values are written where they are read. Oracle
         groups by no subquery, so a projection that would group by a value so bound works it out first.
         """
-        if not self._sharing or all(len(it) <= SHARED_ABOVE for it in values):
+        if not self.binds_long(values):
             return body(list(values))
+        return self.shared_bound(values, body)
+
+    def shared_bound(self, values: Sequence[str], body: Body) -> str:
+        """The body over the values `bound` as `shared` binds them, and counted so: see `_binds`."""
         self._shared_bindings += 1
         return self.bound(values, body)
 
+    def binds_long(self, values: Sequence[str]) -> bool:
+        """Whether `shared` binds the values: where one is long, and values are bound here at all."""
+        return self._sharing and any(len(it) > SHARED_ABOVE for it in values)
+
+    def shared_sum(self, a: str, b: str, body: Body) -> str:
+        """The body over the sides of a `+`, as `shared` binds them: see Oracle's."""
+        return self.shared([a, b], body)
+
     def shared1(self, value: str, body: Callable[[str], str]) -> str:
         """`shared`, of one value."""
         return self.shared([value], lambda it: body(it[0]))
@@ -3838,14 +3850,15 @@ def of(text: str, kind: Temporal | None) -> Json:
 
             # A side read from the store: the temporal of the two texts, null where neither is of a kind it reads; then,
             # in an addition, the two added as any two values are, as Cypher adds a temporal value to a list or to a
-            # text.
+            # text. Bound as a sum's sides are, so a stored text past 32 KB beside a long duration leaves Oracle's
+            # session open (drivine4j#370).
             def read(kind: Temporal | None, temporal: Callable[[str, str], str]) -> Json:
                 def both(values: list[str]) -> str:
                     x, y = values
                     value = self.scalar(self.shared([self.text(x), self.text(y)], lambda it: temporal(it[0], it[1])))
                     return f"COALESCE({value}, {self.add(x, y)})" if sign > 0 else value
 
-                return Json(self.shared([json_a, json_b], both), temporal=kind)
+                return Json(self.shared_sum(json_a, json_b, both), temporal=kind)
 
             if l_ is duration and r is duration:
                 return of(t.sum(a, b, sign), duration)
@@ -4108,7 +4121,7 @@ def ordered(values: list[str]) -> str:
                 MADE_TEXT if given_right is None and text_right else given_right,
             )
             return Json(
-                self.shared([a, b], lambda it: self.add(it[0], it[1], given)),
+                self.shared_sum(a, b, lambda it: self.add(it[0], it[1], given)),
                 items=items if items == _items_of(right_value) else None,
                 item_items=item_items if item_items == _item_items_of(right_value) else None,
                 temporals=_has_temporals(left_value) or _has_temporals(right_value),

tests/integration/test_cypher_compiled.py

@@ -839,6 +839,60 @@ async def test_an_optional_call_in_a_writes_subquery_counts_what_it_finds_and_th
         assert await cypher_pm.query(spec("MATCH (n:CyWrite {id: 1}) RETURN [n.o, n.f, n.b]")) == [[0, 1, 1]]
 
 
+async def test_a_stored_text_past_32_kb_a_text_function_or_a_sum_reads_beside_a_long_value_on_two_rows_is_read_whole(
+    cypher_pm: PersistenceManager,
+):
+    # On Oracle a value past SHARED_ABOVE's length is bound in a subquery, and a property beside it was bound too, taken
+    # out as JSON and read as a CLOB: a text past 32 KB so closed the session from the second run on (ORA-03113,
+    # drivine4j#370), as #144's did. A node of another label holds the same text, and is not read. drivine4j's
+    # CypherQueryContract of liberation-data/drivine4j#370.
+    await clear(cypher_pm)
+    bio = "a" * 40_000 + "x"
+    # A text past 4,000 bytes but within 32 KB, too, which a list is added to; and one of fewer than 32,767 characters
+    # that is past 32 KB, each of three bytes.
+    mid = "b" * 20_000
+    euros = "€" * 12_000
+    props = "{name: 'x', first: 'A', bio: $bio, mid: $mid, euros: $euros, span: 'p1d'}"
+    given = {"bio": bio, "mid": mid, "euros": euros}
+    await cypher_pm.query(spec(f"CREATE (:CyBio {props}), (:CyBio {props}), (:CyBioOther {props})", given))
+
+    def long(text: str) -> str:
+        # Long to write as SQL, so bound: a coalesce of a parameter's key, trimmed and made lower case.
+        return f"toLower(trim(coalesce($p.q, {text})))"
+
+    def each(value: Any) -> list[Any]:
+        return [value] * 4
+
+    statements: list[tuple[str, list[Any]]] = [
+        (f"MATCH (n:CyBio), (m:CyBio) WHERE n.bio CONTAINS {long('m.name')} RETURN count(*)", [4]),
+        (f"MATCH (n:CyBio), (m:CyBio) WHERE n.bio STARTS WITH {long('m.first')} RETURN count(*)", [4]),
+        (f"MATCH (n:CyBio), (m:CyBio) WHERE n.bio ENDS WITH {long('m.name')} RETURN count(*)", [4]),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + {long('m.name')})", each(40_002)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size({long('m.name')} + n.bio)", each(40_002)),
+        # A sum of short texts too, which closed the session on Oracle where the property was left where it is read.
+        ("MATCH (n:CyBio), (m:CyBio) RETURN n.name + '->' + m.first", each("x->A")),
+        ("MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + '->' + m.name)", each(40_004)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.mid + {long('m.name')})", each(20_001)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(last([{long('m.name')}] + n.mid))", each(20_000)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.euros + {long('m.name')})", each(12_001)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + size({long('m.name')}))", each(40_002)),
+        # A duration, added to a text as any value is.
+        (
+            "MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + duration(toUpper(trim(coalesce($p.q, m.span)))))",
+            each(40_004),
+        ),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(split(n.bio, {long('m.name')}))", each(2)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(replace(n.bio, {long('m.name')}, 'yy'))", each(40_002)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN right(replace(n.bio, 'x', {long('m.first')} + 'b'), 3)", each("aab")),
+        (f"MATCH (n:CyBio) WITH n WHERE n.bio CONTAINS {long('n.name')} RETURN count(*)", [2]),
+    ]
+    async with one_session(cypher_pm):
+        # Oracle closed the session from the second run on.
+        for run in range(1, 4):
+            for cypher, expected in statements:
+                assert await asked_again(cypher_pm, cypher, {"p": {"q": None}}) == expected, f"{cypher}, run {run}"
+
+
 async def test_collect_distinct_beside_keys_other_aggregates_an_order_a_page_and_a_variable_outside_its_aggregates(
     cypher_pm: PersistenceManager,
 ):

tests/sync_integration/test_cypher_compiled.py

@@ -835,6 +835,60 @@ def test_an_optional_call_in_a_writes_subquery_counts_what_it_finds_and_the_row_
         assert cypher_pm.query(spec("MATCH (n:CyWrite {id: 1}) RETURN [n.o, n.f, n.b]")) == [[0, 1, 1]]
 
 
+def test_a_stored_text_past_32_kb_a_text_function_or_a_sum_reads_beside_a_long_value_on_two_rows_is_read_whole(
+    cypher_pm: PersistenceManager,
+):
+    # On Oracle a value past SHARED_ABOVE's length is bound in a subquery, and a property beside it was bound too, taken
+    # out as JSON and read as a CLOB: a text past 32 KB so closed the session from the second run on (ORA-03113,
+    # drivine4j#370), as #144's did. A node of another label holds the same text, and is not read. drivine4j's
+    # CypherQueryContract of liberation-data/drivine4j#370.
+    clear(cypher_pm)
+    bio = "a" * 40_000 + "x"
+    # A text past 4,000 bytes but within 32 KB, too, which a list is added to; and one of fewer than 32,767 characters
+    # that is past 32 KB, each of three bytes.
+    mid = "b" * 20_000
+    euros = "€" * 12_000
+    props = "{name: 'x', first: 'A', bio: $bio, mid: $mid, euros: $euros, span: 'p1d'}"
+    given = {"bio": bio, "mid": mid, "euros": euros}
+    cypher_pm.query(spec(f"CREATE (:CyBio {props}), (:CyBio {props}), (:CyBioOther {props})", given))
+
+    def long(text: str) -> str:
+        # Long to write as SQL, so bound: a coalesce of a parameter's key, trimmed and made lower case.
+        return f"toLower(trim(coalesce($p.q, {text})))"
+
+    def each(value: Any) -> list[Any]:
+        return [value] * 4
+
+    statements: list[tuple[str, list[Any]]] = [
+        (f"MATCH (n:CyBio), (m:CyBio) WHERE n.bio CONTAINS {long('m.name')} RETURN count(*)", [4]),
+        (f"MATCH (n:CyBio), (m:CyBio) WHERE n.bio STARTS WITH {long('m.first')} RETURN count(*)", [4]),
+        (f"MATCH (n:CyBio), (m:CyBio) WHERE n.bio ENDS WITH {long('m.name')} RETURN count(*)", [4]),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + {long('m.name')})", each(40_002)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size({long('m.name')} + n.bio)", each(40_002)),
+        # A sum of short texts too, which closed the session on Oracle where the property was left where it is read.
+        ("MATCH (n:CyBio), (m:CyBio) RETURN n.name + '->' + m.first", each("x->A")),
+        ("MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + '->' + m.name)", each(40_004)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.mid + {long('m.name')})", each(20_001)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(last([{long('m.name')}] + n.mid))", each(20_000)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.euros + {long('m.name')})", each(12_001)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + size({long('m.name')}))", each(40_002)),
+        # A duration, added to a text as any value is.
+        (
+            "MATCH (n:CyBio), (m:CyBio) RETURN size(n.bio + duration(toUpper(trim(coalesce($p.q, m.span)))))",
+            each(40_004),
+        ),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(split(n.bio, {long('m.name')}))", each(2)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN size(replace(n.bio, {long('m.name')}, 'yy'))", each(40_002)),
+        (f"MATCH (n:CyBio), (m:CyBio) RETURN right(replace(n.bio, 'x', {long('m.first')} + 'b'), 3)", each("aab")),
+        (f"MATCH (n:CyBio) WITH n WHERE n.bio CONTAINS {long('n.name')} RETURN count(*)", [2]),
+    ]
+    with one_session(cypher_pm):
+        # Oracle closed the session from the second run on.
+        for run in range(1, 4):
+            for cypher, expected in statements:
+                assert asked_again(cypher_pm, cypher, {"p": {"q": None}}) == expected, f"{cypher}, run {run}"
+
+
 def test_collect_distinct_beside_keys_other_aggregates_an_order_a_page_and_a_variable_outside_its_aggregates(
     cypher_pm: PersistenceManager,
 ):