airbyte_cdk.sources.declarative.parsers.stop_condition_safety

  1#
  2# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
  3#
  4
  5"""
  6Config-time analysis of a `CursorPagination` `stop_condition` against a page size reduction.
  7
  8A condition that infers "this was the last page" from a short page is only correct relative to the page size
  9that was actually requested. When `page_size_reduction` shrinks that size, a *full* page at the reduced size
 10reads as a short page against the configured size, the pagination ends, and the rest of the partition is
 11dropped without anything failing.
 12
 13Matching the expression as a string cannot tell the reduction-aware `page_size` variable apart from a
 14config-derived constant - `\\bpage_size\\b` matches inside `config['page_size']` just as well - and it cannot
 15tell an inequality, which a reduction can invalidate, apart from `last_page_size == 0`, which no reduction can.
 16So the expression is parsed with Jinja and the comparisons involving `last_page_size` are inspected on the AST,
 17where a bare `Name` node is distinguishable from a `Getitem` on `config`.
 18
 19The safety rule is a single question: can the condition be true for a page that is *full at the size that was
 20requested*? A full page holds exactly `requested_page_size` records, and a reduction only ever lowers that
 21number, so:
 22
 23- `last_page_size == 0` and `last_page_size != 0` cannot change verdict: a full page is never empty, because
 24  `minimum_page_size` is at least 1.
 25- `last_page_size > x` and `last_page_size >= x` can only stop being true as the page shrinks, so a reduction
 26  cannot introduce a stop that would not have happened anyway.
 27- `last_page_size < x` and `last_page_size <= x` are the dangerous shape. They are safe only when `x` is the
 28  `page_size` variable, which follows the reduction, or when `x` is a literal at or below `minimum_page_size`,
 29  which makes the comparison a rewrite of "the page is empty". An expression merely built from `page_size`,
 30  such as `[page_size, 100] | max` or `page_size + 50`, does not count: it can hold the configured size again.
 31
 32Each verdict above reads the comparison on its own, which is only sound while the comparison decides the
 33condition in its own polarity. `{{ not (last_page_size >= 100) }}` means `last_page_size < 100`, and
 34`{{ last_page_size - 100 < 0 }}` means the same again, so the operator and the operands of the comparison no
 35longer say what the condition does. A comparison is therefore only classified when the truth of the condition follows its own: reached from the
 36root through `and`/`or`, or as the test of a `{% if %}` that renders truthy text and nothing else, and with
 37`last_page_size` compared bare rather than transformed first. Anything else is unclassifiable.
 38
 39A condition that never names `last_page_size` is not safe by default either: the response body carries the
 40same count, so `{{ response['data'] | length < 100 }}` truncates in exactly the same way. Which of a
 41connector's own response fields counts the records of a page is not knowable here, so any comparison that
 42bounds something from above is reported as unclassifiable. A condition with no such comparison, such as
 43`{{ not response.next }}`, cannot read a page length by comparison and is safe.
 44
 45Anything else is reported as unclassifiable rather than as a truncation: this analysis rejects a manifest at
 46stream construction, so a shape it does not understand must not be treated as a defect.
 47"""
 48
 49from enum import Enum
 50from typing import Iterator, List, Optional, Tuple
 51
 52from jinja2 import nodes
 53from jinja2.environment import Environment
 54from jinja2.exceptions import TemplateSyntaxError
 55
 56from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import FALSE_VALUES
 57
 58LAST_PAGE_SIZE_VARIABLE = "last_page_size"
 59REQUESTED_PAGE_SIZE_VARIABLE = "page_size"
 60
 61# Parsing is all this environment is used for: no filter, test or global is resolved, so the plain environment
 62# parses everything the interpolation environment accepts.
 63_PARSING_ENVIRONMENT = Environment()
 64
 65_MIRRORED_OPERATORS = {
 66    "lt": "gt",
 67    "lteq": "gteq",
 68    "gt": "lt",
 69    "gteq": "lteq",
 70    "eq": "eq",
 71    "ne": "ne",
 72}
 73
 74
 75class StopConditionSafety(Enum):
 76    """Whether a page size reduction can change what a `stop_condition` decides."""
 77
 78    SAFE = "SAFE"
 79    """No reduction can make the condition stop the pagination earlier than it already would."""
 80
 81    TRUNCATES = "TRUNCATES"
 82    """A full page at a reduced size satisfies the condition, so the partition would be cut short."""
 83
 84    UNKNOWN = "UNKNOWN"
 85    """The condition uses `last_page_size` in a shape this analysis cannot reason about."""
 86
 87
 88def classify_stop_condition(
 89    stop_condition: str, minimum_page_size: int
 90) -> Tuple[StopConditionSafety, str]:
 91    """
 92    :param stop_condition: the raw `stop_condition` template from the manifest
 93    :param minimum_page_size: the smallest page size the reduction is allowed to request
 94    :return: the verdict and a human readable reason for it
 95    """
 96    try:
 97        template = _PARSING_ENVIRONMENT.parse(stop_condition)
 98    except TemplateSyntaxError as exception:
 99        return (
100            StopConditionSafety.UNKNOWN,
101            f"it is not a valid Jinja expression ({exception.message})",
102        )
103
104    if not _references(template, LAST_PAGE_SIZE_VARIABLE):
105        return _classify_without_last_page_size(template)
106
107    unclassifiable: List[str] = []
108    understood = 0
109    for left, operator, right, decides_condition in _comparisons(template):
110        if not _references(left, LAST_PAGE_SIZE_VARIABLE) and not _references(
111            right, LAST_PAGE_SIZE_VARIABLE
112        ):
113            continue
114        if not decides_condition:
115            # The comparison is negated or is an operand of a larger expression, so its own operator no longer
116            # says what the condition does and neither verdict below would be about the right question.
117            unclassifiable.append(
118                f"it uses `{LAST_PAGE_SIZE_VARIABLE}` in a comparison that does not decide the condition on "
119                f"its own, such as one under a `not` or inside a larger expression"
120            )
121            continue
122        verdict, reason = _classify_comparison(left, operator, right, minimum_page_size)
123        if verdict is StopConditionSafety.TRUNCATES:
124            return verdict, reason
125        if verdict is StopConditionSafety.UNKNOWN:
126            unclassifiable.append(reason)
127        else:
128            understood += 1
129
130    if unclassifiable:
131        return StopConditionSafety.UNKNOWN, unclassifiable[0]
132    if understood:
133        return (
134            StopConditionSafety.SAFE,
135            f"every comparison it makes against `{LAST_PAGE_SIZE_VARIABLE}` holds whatever page size was requested",
136        )
137    return (
138        StopConditionSafety.UNKNOWN,
139        f"it uses `{LAST_PAGE_SIZE_VARIABLE}` outside of a comparison",
140    )
141
142
143def _classify_without_last_page_size(
144    template: nodes.Template,
145) -> Tuple[StopConditionSafety, str]:
146    """
147    Classify a condition that never names `last_page_size`.
148
149    `last_page_size` is not the only way to observe how many records a page held: the response body carries
150    the same number, and `{{ response['data'] | length < 100 }}` is the comparison `{{ last_page_size < 100 }}`
151    counted one layer out. The CDK cannot tell which of a connector's own response fields counts the records
152    of the page, so a condition that bounds *anything* from above is reported as unclassifiable and warned
153    about rather than assumed to be safe. A condition that makes no such comparison - the common
154    `{{ not response.next }}` shape - cannot read a page length by comparison at all, so it stays safe.
155    """
156    for left, operator, right, decides_condition in _comparisons(template):
157        bounded = _bounded_from_above(left, operator, right, decides_condition)
158        if bounded is not None:
159            return (
160                StopConditionSafety.UNKNOWN,
161                f"it stops when {_describe(bounded)} stays below a threshold. `{LAST_PAGE_SIZE_VARIABLE}` is "
162                f"not the only way to count the records of a page - a count read from the response body is "
163                f"another - and the connector cannot tell whether this one does, so a full page at a reduced "
164                f"size may read as a short page here",
165            )
166    return (
167        StopConditionSafety.SAFE,
168        f"it does not use `{LAST_PAGE_SIZE_VARIABLE}` and makes no comparison that could be bounding the "
169        f"number of records in a page from above",
170    )
171
172
173def _bounded_from_above(
174    left: nodes.Node, operator: str, right: nodes.Node, decides_condition: bool
175) -> Optional[nodes.Node]:
176    """
177    :return: the operand the comparison bounds from above, or None when it bounds nothing from above
178
179    An upper bound is the shape a reduction can invalidate, for the same reason it can invalidate
180    `last_page_size < 100`: the reduction lowers what a full page holds, so the bound starts holding for pages
181    that are not short at all. A lower bound can only stop being satisfied as pages get smaller. When the
182    comparison does not decide the condition in its own polarity, which of the two it is cannot be read off
183    the operator, so it counts as an upper bound.
184
185    Every comparison with one of these four operators bounds one of its operands from above - `a < b` bounds
186    `a`, `b > a` bounds `a` too - and a literal cannot be a page length, so the bounded side only escapes when
187    it is one. The two operator directions are therefore mirrored, the way `_MIRRORED_OPERATORS` mirrors them
188    on the `last_page_size` path: reading only `lt`/`lteq` in full left
189    `{{ config['page_size'] > response['data'] | length }}` classified as safe while
190    `{{ response['data'] | length < config['page_size'] }}` was warned about.
191    """
192    if operator not in ("lt", "lteq", "gt", "gteq"):
193        return None
194    if not decides_condition:
195        return left if not isinstance(left, nodes.Const) else right
196    if operator in ("lt", "lteq"):
197        # `x < 100` bounds x from above; `100 < x` bounds the literal, which is not a page length.
198        return None if isinstance(left, nodes.Const) else left
199    # `100 > x` is the mirror of `x < 100`; `x > 100` bounds the literal again.
200    return None if isinstance(right, nodes.Const) else right
201
202
203def _comparisons(
204    node: nodes.Node, decides_condition: bool = True
205) -> Iterator[Tuple[nodes.Node, str, nodes.Node, bool]]:
206    """
207    Flatten every comparison, including the chained ones, into (left, operator, right, decides_condition).
208
209    `decides_condition` is true only for a comparison the truth of the whole condition follows directly:
210    reached from the root through `and`/`or` alone. Under a `not`, inside a conditional expression, piped
211    through a filter or used as an operand of another expression, the comparison's own operator says nothing
212    about what the condition decides, so it is flagged and left unclassified.
213    """
214    if isinstance(node, nodes.Compare):
215        left = node.expr
216        for operand in node.ops:
217            yield left, operand.op, operand.expr, decides_condition
218            left = operand.expr
219        # A comparison nested inside an operand of this one is an ordinary sub-expression, not a decider.
220        for operand_node in [node.expr, *(operand.expr for operand in node.ops)]:
221            yield from _comparisons(operand_node, False)
222        return
223
224    if isinstance(node, nodes.If):
225        # `{% if last_page_size < 100 %}true{% endif %}` renders truthy text exactly when its test holds, so
226        # the test decides the condition. That only follows while the branch it guards is the whole story:
227        # an `else`, an `elif`, or a body rendering something the CDK reads as false breaks the equivalence.
228        test_decides = decides_condition and _if_follows_its_test(node)
229        yield from _comparisons(node.test, test_decides)
230        for branch in [*node.body, *node.elif_, *node.else_]:
231            yield from _comparisons(branch, False)
232        return
233
234    children_decide = decides_condition and _passes_truth_through(node)
235    for child in node.iter_child_nodes():
236        yield from _comparisons(child, children_decide)
237
238
239def _passes_truth_through(node: nodes.Node) -> bool:
240    """Whether the truth of the condition follows the truth of this node's children."""
241    if isinstance(node, (nodes.And, nodes.Or)):
242        return True
243    if isinstance(node, nodes.Template):
244        # Anything rendered next to the comparison is text of its own, which makes the rendered condition
245        # truthy whatever the comparison decided.
246        return len(node.body) == 1
247    if isinstance(node, nodes.Output):
248        return len(node.nodes) == 1
249    return False
250
251
252def _if_follows_its_test(node: nodes.If) -> bool:
253    if node.elif_ or node.else_:
254        return False
255    rendered = ""
256    for statement in node.body:
257        if not isinstance(statement, nodes.Output):
258            return False
259        for output in statement.nodes:
260            if isinstance(output, nodes.TemplateData):
261                rendered += output.data
262            elif isinstance(output, nodes.Const):
263                rendered += str(output.value)
264            else:
265                return False
266    return rendered not in FALSE_VALUES
267
268
269def _classify_comparison(
270    left: nodes.Node, operator: str, right: nodes.Node, minimum_page_size: int
271) -> Tuple[StopConditionSafety, str]:
272    if _references(right, LAST_PAGE_SIZE_VARIABLE) and not _references(
273        left, LAST_PAGE_SIZE_VARIABLE
274    ):
275        left, right, operator = right, left, _MIRRORED_OPERATORS.get(operator, operator)
276    elif _references(left, LAST_PAGE_SIZE_VARIABLE) and _references(right, LAST_PAGE_SIZE_VARIABLE):
277        return (
278            StopConditionSafety.UNKNOWN,
279            f"it compares `{LAST_PAGE_SIZE_VARIABLE}` against itself",
280        )
281
282    is_bare = isinstance(left, nodes.Name) and left.name == LAST_PAGE_SIZE_VARIABLE
283
284    if operator in ("eq", "ne"):
285        if not is_bare:
286            return (
287                StopConditionSafety.UNKNOWN,
288                f"`{LAST_PAGE_SIZE_VARIABLE}` is transformed before being compared",
289            )
290        if isinstance(right, nodes.Const) and right.value == 0:
291            # A page that is full at the requested size holds at least `minimum_page_size` records, which is
292            # at least 1, so no reduction can make an emptiness test fire.
293            return (
294                StopConditionSafety.SAFE,
295                f"`{LAST_PAGE_SIZE_VARIABLE}` is only tested for emptiness",
296            )
297        return (
298            StopConditionSafety.UNKNOWN,
299            f"it tests `{LAST_PAGE_SIZE_VARIABLE}` for equality against {_describe(right)} rather than against 0",
300        )
301
302    if operator in ("gt", "gteq"):
303        if is_bare:
304            # A reduction only lowers the size of a full page, so a lower bound can only stop being satisfied.
305            return (
306                StopConditionSafety.SAFE,
307                f"a smaller page can only make `{LAST_PAGE_SIZE_VARIABLE} {'>' if operator == 'gt' else '>='} "
308                f"{_describe(right)}` less true, never more",
309            )
310        return (
311            StopConditionSafety.UNKNOWN,
312            f"`{LAST_PAGE_SIZE_VARIABLE}` is transformed before being compared",
313        )
314
315    if operator in ("lt", "lteq"):
316        if isinstance(left, nodes.BinExpr):
317            # `last_page_size - 100 < 0` is `last_page_size < 100` in disguise: with arithmetic on the left the
318            # threshold on the right is no longer the page size the condition really stops at, so neither the
319            # `page_size` reading nor the `minimum_page_size` reading below applies.
320            return (
321                StopConditionSafety.UNKNOWN,
322                f"`{LAST_PAGE_SIZE_VARIABLE}` takes part in arithmetic before being compared",
323            )
324        if _references(right, REQUESTED_PAGE_SIZE_VARIABLE):
325            if not _is_requested_page_size(right):
326                # `[page_size, 100] | max` and `page_size + 50` both mention the variable while holding a
327                # value a reduction does not lower, so the threshold does not follow the reduction after all.
328                return (
329                    StopConditionSafety.UNKNOWN,
330                    f"it compares `{LAST_PAGE_SIZE_VARIABLE}` against an expression built from "
331                    f"`{REQUESTED_PAGE_SIZE_VARIABLE}` rather than against `{REQUESTED_PAGE_SIZE_VARIABLE}` "
332                    f"itself, so the threshold may not follow the reduction",
333                )
334            return (
335                StopConditionSafety.SAFE,
336                f"it compares `{LAST_PAGE_SIZE_VARIABLE}` against `{REQUESTED_PAGE_SIZE_VARIABLE}`, "
337                f"which follows the reduction",
338            )
339        if isinstance(right, nodes.Const) and _is_whole_number(right.value):
340            # `last_page_size < k` is false for every full page exactly when k is at or below the smallest page
341            # the connector is allowed to request, which makes it another way of writing "the page is empty".
342            threshold = right.value if operator == "lt" else right.value + 1
343            if threshold <= minimum_page_size:
344                return (
345                    StopConditionSafety.SAFE,
346                    f"no page the connector is allowed to request is smaller than {threshold}",
347                )
348        return (
349            StopConditionSafety.TRUNCATES,
350            f"it stops as soon as a page is smaller than {_describe(right)}, which does not follow the "
351            f"reduction, so a full page at a reduced size reads as a short page",
352        )
353
354    return (
355        StopConditionSafety.UNKNOWN,
356        f"it uses `{LAST_PAGE_SIZE_VARIABLE}` with the `{operator}` operator",
357    )
358
359
360def _is_requested_page_size(node: nodes.Node) -> bool:
361    """Whether the node is the `page_size` variable itself, filters aside."""
362    unfiltered: Optional[nodes.Node] = node
363    while isinstance(unfiltered, nodes.Filter):
364        unfiltered = unfiltered.node
365    return isinstance(unfiltered, nodes.Name) and unfiltered.name == REQUESTED_PAGE_SIZE_VARIABLE
366
367
368def references_requested_page_size(template: str) -> bool:
369    """
370    Whether an expression reads the `page_size` interpolation variable.
371
372    The variable is only bound to a number when the strategy declares a `page_size`, and a comparison against
373    an unbound one does not fail: Jinja raises, the interpolation falls back to the raw template string, and a
374    non-empty string is truthy - so `{{ last_page_size < page_size }}` would stop the pagination after the
375    first page. Reading it off the AST is what tells the variable apart from `config['page_size']`, which is a
376    `Getitem` on `config` rather than a `Name` and is bound whatever the strategy declares.
377    """
378    try:
379        parsed = _PARSING_ENVIRONMENT.parse(template)
380    except TemplateSyntaxError:
381        # An expression that does not parse is not this function's to report on. The interpolation layer
382        # renders it as the literal string it is, and it names no variable either way.
383        return False
384    return _references(parsed, REQUESTED_PAGE_SIZE_VARIABLE)
385
386
387def _references(node: nodes.Node, name: str) -> bool:
388    if isinstance(node, nodes.Name):
389        return bool(node.name == name)
390    return any(referenced.name == name for referenced in node.find_all(nodes.Name))
391
392
393def _is_whole_number(value: object) -> bool:
394    return isinstance(value, int) and not isinstance(value, bool)
395
396
397def _describe(node: nodes.Node) -> str:
398    """Render the expression a `last_page_size` comparison is made against, for an error message."""
399    if isinstance(node, nodes.Const):
400        return repr(node.value)
401    if isinstance(node, nodes.Name):
402        return f"`{node.name}`"
403    if isinstance(node, nodes.Getitem) and isinstance(node.arg, nodes.Const):
404        return f"`{_describe(node.node).strip('`')}[{node.arg.value!r}]`"
405    if isinstance(node, nodes.Getattr):
406        return f"`{_describe(node.node).strip('`')}.{node.attr}`"
407    return "a value the connector computes itself"