Builds on #2

The key change here is that there is also an extra file written out.

Parquet files all have a metadata block in them at the end. This allows planners to identify whether or not the file needs to be used, as it contains things like min/max values for columns. It also contains sub-file level detail if there are enough rows.

However, to make use of this a system needs to process all of the files in storage to read the metadata block. Not an issue when all the files are local but a more significant one when they're on S3 as it'd need to crawl everything. In addition we have a non-standard partitioning scheme with multiple levels.

We can collect all of the metadata from each file and put it into one single metadata file that points off to the actual file locations where the data is stored. This should entirely remove the need to generate lists of required files in the other packages (the most common repeated code) and remove the need to pull the files locally before running queries. This is a standard feature in pyarrow, and should be usable outside of our code and outside of just python.

https://arrow.apache.org/docs/python/parquet.html#writing-metadata-and-common-metadata-files

When these are available, we can delete most of https://github.com/cardstack/cardstack/blob/main/packages/cardpay-reward-programs/cardpay_reward_programs/utils.py#L11-L78 and replace it with a single call to pyarrow dataset construction, which we can register in duckdb and remove the parquet_scan manual construction.

The current logic in the package when writing out new partitions is this:

• Generate a list of all partitions that should exist
• Iterate backwards, to find the latest one that has been written
• Write all newer ones

Example:

Existing partitions
* blocks 1-10
* blocks 10-20

Run again, partitions to exist are 1-10, 10-20, 20-30, 30-40

30-40 does not exist
20-30 does not exist
10-20 does exist - STOP

Write 30-40
Write 20-30

End

There is an issue here that in that final step it writes from newest to oldest. If the process fails during this, this happens:

Existing partitions
* blocks 1-10
* blocks 10-20

Run again, partitions to exist are 1-10, 10-20, 20-30, 30-40

30-40 does not exist
20-30 does not exist
10-20 does exist - STOP

Write 30-40
Write 20-30 - FAILS


Run again, partitions to exist are 1-10, 10-20, 20-30, 30-40

30-40 does exist - STOP

End

The partition for 20-30 is never written, causing downstream breakages.

We could write the partitions out in the opposite order, however there is a better approach. The partitions from the last successful run can be calculated based on what's in the latest.yaml file. We can then just look at the difference between the last successful run and the required partitions and write out the difference. This also saves having to check to see if lots of files exist.

This change adds some tests for running the thing end to end, and a more complicated test to check for the above scenario (which correctly fails on the old code and passes on this).

The generated _block_number column is inconsistently set in the output files, depending on whether there is any data or not.

This breaks anything that tries to read across multiple files when one of them is empty (possibly more cases).

We force uint32 instead.