[python] Keep reading an Avro file after a fully-filtered batch

[jackylee-ch]

Purpose

FormatAvroReader.read_arrow_batch reads the fastavro generator batch_size (default 1024) rows at a time and applies the pushed-down predicate in Python. When a whole batch matched nothing it returned None — but a RecordBatchReader returns None only at end of input, and ConcatBatchReader treats None as "reader exhausted" and advances to the next file.

So a filtered read of an Avro file with a full non-matching batch silently dropped every remaining row: an append-only file.format = avro table of 3000 rows read with user_id >= 2000 returns 0 rows instead of 1000 (the leading 1024-row block fails the predicate → None → EOF).

The reader now loops to the next batch on an empty filtered batch, returning None only when the generator is exhausted. A loop (not recursion) avoids a RecursionError on long filtered runs; the no-predicate path is unchanged. format_row_reader already handles this the same way.

Tests

reader_append_only_test.test_avro_ao_reader_filter_keeps_rows_past_first_batch: a 3000-row Avro table read with user_id >= 2000 asserts the 1000 matching rows. Fails on master ([]); passes here.

[Akash3121]

Nit: With the default batch size of 1024, this threshold makes only the first batch fully filtered; the second batch already contains matching rows. Since the implementation intentionally uses a loop to support arbitrarily many consecutive empty batches, could we use  2500  here instead? With 3000 input rows, that would skip two complete batches before returning the final 500 rows and would distinguish the loop from an implementation that retries only once.

I think, the assertion should become:

 read_builder = table.new_read_builder().with_filter(
     pb.greater_or_equal('user_id', 2500))
 result = self._read_test_table(read_builder)
 
 self.assertEqual(
     sorted(result.column('user_id').to_pylist()), list(range(2500, n)))