Spark partial limit push down

[huaxingao]

For sql such as SELECT * FROM table LIMIT n, push down Spark's partial limit to Iceberg, so that Iceberg can stop reading data once the limit is reached.

[huaxingao]

cc @aokolnychyi @RussellSpitzer @szehon-ho

[RussellSpitzer]

Should we assert this is greater than 1? I assume and input 0 or negative is a bad call

[huaxingao]

I think we are OK because Spark has assert for limit. If the limit is negative, e.g. SELECT * FROM table limit -2, Spark will throw

org.apache.spark.sql.AnalysisException: [INVALID_LIMIT_LIKE_EXPRESSION.IS_NEGATIVE] The limit like expression "-2" is invalid. The limit expression must be equal to or greater than 0, but got -2.;

If the limit is 0, e.g. SELECT * FROM table limit 0, Spark changes this to an empty table scan and it won't reach here.

[RussellSpitzer]

This is public though, and anyone can call it so we can't rely on Spark to cover us.

[huaxingao]

Right, I forgot that. I've added a check to throw an IllegalArgumentException if the limit is <= 0

[RussellSpitzer]

Usually we do Preconditions.checkArgument
for things like this

[huaxingao]

Changed to Preconditions.checkArgument. Thanks

[RussellSpitzer]

"Cannot X (because Y) (recover by Z)" - > "Cannot read a negative number of values. numValsToRead = %D"

[huaxingao]

Changed. Thanks

[RussellSpitzer]

We are now potentially scanning the file scan tasks several times before beginning a query,

1. Push aggregates
2. Has Deletes
3. Actually creating tasks

I'm wondering if we should be caching this

[RussellSpitzer]

Probably save this for another PR

[huaxingao]

I will have a follow-up PR for this.

[RussellSpitzer]

I'm wondering if we should include this in SparkReadConf rather than having a separate argument. I have similar thoughts around SparkInputPartition. I'm not a big fan of having to plumb the new arguments all the way through the code base but those two options may not look great either since they aren't a great fit imho.

Ideally I think I would want something like SparkReadContext but I don't know how often more things like this will come up

[huaxingao]

Thanks @RussellSpitzer for sharing your concern and suggestion!
I think we can borrow @aokolnychyi 's idea of adding ParquetBatchReadConf and OrcBatchReadConf. We can have something like

@Value.Immutable
public interface ParquetBatchReadConf extends Serializable {
  ParquetReaderType readerType();  // this is for comet, we don't need this for now
  int batchSize();
  @Nullable
  Integer limit();
}

@Value.Immutable
public interface OrcBatchReadConf extends Serializable {
  int batchSize();
}

Similarly, we can also have ParquetRowReadConf and OrcRowReadConf.

I have changed the code to add ParquetBatchReadConf and OrcBatchReadConf. We still need to pass pushedLimit to the SparkPartitioningAwareScan, SparkScan and SparkBatch constructors, so pushedLimit can be passed from SparkScanBuilder to SparkBatch, this is because pushedLimit is not available in SparkReadConf, we have to call SparkScanBuilder.pushLimit to get pushedLimit.

Please let me know if this approach looks OK to you. Thanks!