#3830 feat: Let streams yield native Arrow data for Arrow BATCH files Issue created by edgarrmondragon Feature scope Taps (catalog, state, tests, etc.) Description The

arrow

BATCH encoding builds each file with

pa.Table.from_pylist()

from the records that

get_records()

yields. Many sources can produce Arrow directly, for example ADBC drivers, `snowflake-connector-python`'s

fetch_arrow_batches()

, and PyMongoArrow. For them, building a Python dict per row and converting it back to Arrow throws away most of the speedup BATCH is meant to give. Today each tap works around this by overriding

get_batches

and carrying its own copy of the same plumbing: • re-slicing the driver's chunks to

batch_size

• writing Arrow IPC files • emitting one manifest per file • advancing the replication key bookmark after each file Getting that right isn't obvious. For example,

get_starting_replication_key_value()

returns

None

inside a custom

get_batches

unless the stream first calls the private

_write_starting_replication_value(context)

, because only the RECORD path initializes it. Proposal: add an optional

Stream.get_record_batches(context)

hook that returns an iterable of

pyarrow.Table

or

pyarrow.RecordBatch

, or

None

(the default) to keep using

get_records()

. When the

arrow

encoding is configured and the hook returns data, the SDK: • re-slices the tables into files of

batch_size

rows; • writes each file through

batch_config.storage

, so remote roots work too; • drops columns for deselected top-level properties; • advances the bookmark after each file is written: to the file's last replication key value if

is_sorted

, otherwise to its max.

_sync_batches

then emits STATE after each BATCH message, so an interrupted sync resumes from the last written file.

ArrowBatcher

would also gain public

write_table()

and

rechunk_tables()

helpers for taps that still need to override

get_batches

. A tap then only implements: def get_record_batches(self, context): start = self.get_starting_replication_key_value(context) yield from client.fetch_arrow(query_after(start)) A working implementation with tests and docs exists on a local branch, which I'll open as a PR after validating it against a real tap (MeltanoLabs/tap-mongodb). meltano/sdk