All-null Arrow cursor batch prematurely closes bounded incremental source

[mattfaltyn]

dlt version

Reproduced on upstream devel at 14e22bc93acf3c7c842cbecb34da9f75224dcb28 (dlt.__version__ == "1.29.0").

Describe the problem

A bounded ascending incremental resource closes its source generator when an Arrow batch contains only null cursor values, even when later batches contain valid in-range rows.

This silently skips the remainder of a paginated or streaming source. The same behavior reproduces with typed pandas and Polars dataframes because they use the Arrow incremental transform.

Steps to reproduce

import dlt
import pyarrow as pa


requested_pages = []


@dlt.resource
def pages(
    ts=dlt.sources.incremental(
        "ts",
        initial_value=0,
        end_value=10,
        row_order="asc",
        on_cursor_value_missing="exclude",
    ),
):
    requested_pages.append(1)
    yield pa.table(
        {
            "id": [1, 2],
            "ts": pa.array([None, None], type=pa.int64()),
        }
    )

    requested_pages.append(2)
    yield pa.table({"id": [3, 4], "ts": [5, 6]})


result = list(pages())
print("requested_pages:", requested_pages)
print("result:", result)

Actual output:

requested_pages: [1]
result: []

Expected behavior

The null-only first batch should be filtered according to on_cursor_value_missing="exclude" without being treated as evidence that the upper bound was crossed. Page 2 should be requested and its rows with cursor values 5 and 6 should be returned.

Impact

Bounded backfills and other ordered incremental extractions can silently lose all data after a null-only Arrow/dataframe page. There is no exception or warning indicating that the source was closed early.

Controls:

• Removing row_order="asc" requests both pages.
• A first Arrow batch containing a non-null in-range cursor requests both pages.
• Object/JSON null rows do not prematurely close the source.

Root cause

ArrowIncremental.__call__ computes the aggregate cursor value before removing null cursor rows:

dlt/dlt/extract/incremental/transform.py

Lines 469 to 480 in 14e22bc

	# The new max/min value
	row_value_scalar = self.compute(tbl[cursor_path])
	row_value = from_arrow_scalar(row_value_scalar)
	# correct tz-awareness
	if not self.seen_data and self.start_value_is_datetime and isinstance(row_value, datetime):
	# NOTE: we are making sure that last_value == start_value here (self.seen_data)
	assert self.last_value == self.start_value
	self.start_value = self.last_value = self._adapt_timezone(
	row_value, self.last_value, "last_value", self.resource_name
	)
	if tbl.schema.field(cursor_path).nullable:

For an all-null cursor column, the aggregate and the Arrow comparison produce a null scalar. At the end-bound check, .as_py() therefore returns None, and not None sets end_out_of_range to True:

dlt/dlt/extract/incremental/transform.py

Lines 488 to 511 in 14e22bc

	not self.seen_data
	and self.end_value_is_datetime
	and isinstance(row_value, datetime)
	):
	self.end_value = self._adapt_timezone(
	row_value, self.end_value, "end_value", self.resource_name
	)
	try:
	end_value_scalar = to_arrow_scalar(self.end_value, cursor_data_type)
	except Exception as ex:
	raise IncrementalCursorInvalidCoercion(
	self.resource_name,
	cursor_path,
	self.end_value,
	"end_value",
	"<arrow column>",
	cursor_data_type,
	str(ex),
	) from ex
	# Is max row value higher than end value?
	# NOTE: pyarrow bool *always* evaluates to python True. `as_py()` is necessary
	end_out_of_range = not self.end_compare(row_value_scalar, end_value_scalar).as_py()
	if end_out_of_range:
	tbl = tbl.filter(self.end_compare(tbl[cursor_path], end_value_scalar))

Because row_order is ascending, Incremental.can_close() interprets that flag as a real upper-bound crossing and closes the source pipe:

dlt/dlt/extract/incremental/__init__.py

Lines 464 to 479 in 14e22bc

	self.resource_name,
	)
	def _transform_item(
	self, transformer: IncrementalTransform, row: TDataItem
	) -> Optional[TDataItem]:
	row, self.start_out_of_range, self.end_out_of_range = transformer(row)
	# if we know that rows are ordered we can close the generator automatically
	# mind that closing pipe will not immediately close processing. it only closes the
	# generator so this page will be fully processed
	# TODO: we cannot close partially evaluated transformer gen. to implement that
	# we'd need to pass the source gen along with each yielded item and close this particular gen
	# NOTE: with that implemented we could implement add_limit as a regular transform having access to gen
	if self.can_close() and not self._bound_pipe.has_parent:
	self._bound_pipe.close()
	return row

Suggested validation

Add a regression test with a null-only typed table/dataframe batch followed by an in-range batch. The later batch must be requested and returned for a bounded ascending incremental resource. The end-bound comparison should treat a null aggregate as “no comparable cursor value in this batch,” not as an out-of-range value.