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]

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!

[RussellSpitzer]

orcBatchSize?

[RussellSpitzer]

Suggested change

	assertThat(pushedLimit).as("Pushed down limit should be " + limit).isEqualTo(null);
	assertThat(pushedLimit).as("Pushed down limit should be " + limit).isNull();

[RussellSpitzer]

nit

[huaxingao]

Done. Thanks

[RussellSpitzer]

should probably extend TestBase

[huaxingao]

Done. Thanks!

[RussellSpitzer]

I've left my last few little nits on the PR, I want to make sure we get @aokolnychyi to take a peak too though since it was his suggestion to add in the Immutable Parquet and ORC Conf Classes. I wasn't sure how far he wanted to take them down the Spark Hierarchy.

[aokolnychyi]

Ack, will try to get to this asap. I am partially off this week.

[aokolnychyi]

This is too expensive. We should be using snapshot summary.

[RussellSpitzer]

That's optional in the spec, and delete content is also optional even if the snapshot summary is present. Can we rely on that here?

I think we would have to fall back to the scan anyway if they are missing

[aokolnychyi]

In a few places, we default to what's safe. In this case it would be assuming there are deletes. We can also check if there are delete manifests in the snapshot we read, but doing an extra scan and iterating through all files seems too expensive.

[RussellSpitzer]

I agree that the scan is probably too expensive here and we probably should only be doing it once for the various requirements we have. I also think we should codify some of the keys and elements of the Summary, But for now I think no one should be relying on the summary as a primary source of truth.

[huaxingao]

I moved the deletes checking to SparkPartitioningAwareScan.tasks() so we don't need to do an extra planFiles

[aokolnychyi]

Why do we push it all the way to the reader instead of simply reading just enough files to match the limit based on the number of records we have in the metadata for each data file?

[aokolnychyi]

My initial thought is that we can use ExpressionUtil$selectsPartitions to test if the filter selects entire files. If so, we can simply find matching files so that the sum of recordCount exceeds the pushed limit. It may be more complicated for filters that only select some records within files. Let me explore this PR more for that.

[huaxingao]

I think pushing it all the way to the reader could yield a bit more performance gain.

Actually if filter exists, Spark side doesn't push down limit.