`NativeCollectLimit` submits one job per scanned partition

[lyne7-sc]

Describe the bug

NativeCollectLimitBase.executeCollect() uses RDD.toLocalIterator to collect rows from its input partitions.

toLocalIterator submits a separate Spark job for each scanned partition. For a LIMIT query with empty or highly selective input, Auron may therefore submit many jobs before finding enough rows or exhausting all partitions.

To Reproduce

Run a limit query whose filter removes all rows:

spark
  .read
  .parquet(path)
  .where("id < 0")
  .limit(1)
  .collect()

Inspecting the Spark jobs shows that each scanned partition is handled by a separate job.

Expected behavior

Partitions should be scanned in batches, using the same strategy as Spark's executeTake. The number of submitted jobs should be lower than the number of scanned partitions.

Screenshots

Additional context