Don't group files with different partition specs in BaseRewriteManifests

[aokolnychyi]

I think we should not combine files with potentially different partition specs into the same manifest in BaseRewriteManifests as we in do in MergingSnapshotProducer.

[aokolnychyi]

@bryanck @rdblue what do you think?

[bryanck]

The downside is that this could potentially result in many small manifests. For example, we have a bucketed table with 10k buckets, which is additionally partitioned by date and hour. So for a week of data we'd have 10k * 24 * 7 = 1.68 million manifests. What we do with RewriteManifests today is combine files into groups of four buckets and ignore date and hour, which results in 10k/4 = 2.5k manifests for a week of data. Performance with file pruning is good even though the manifests have multiple partition specs.

[aokolnychyi]

@bryanck I mean a partitioning strategy in general by partition spec, which seems to be the same in your case. What will happen in case of partition schema evolution? Manifests are supposed to have files written with the same partition spec:

Manifests store data files from any partition, as long as the partition spec is the same for the data files

[aokolnychyi]

I also raised a question similar to what you describe in #481.

[bryanck]

Right, I was thinking partition, not partition spec. In that case, then I agree it makes sense.

[rdblue]

I agree with this. A manifest can only be written for one partition spec, so this is definitely a bug!

[manishmalhotrawork]

@aokolnychyi trying to understand the problem and thinking loud for solution.

so, would the fix be to group manifests by partition spec before cluster by clusterFunction.
so for group of each partition spec's manifest files, the cluster by logic will be applied, where currently its applied for every manifests irrespective of the underlying manifests partition spec.

                     ManifestReader.read(ops.io().newInputFile(manifest.path()), ops.current().specsById())) {
                FilteredManifest filteredManifest = reader.select(Arrays.asList("*"));
                filteredManifest.liveEntries().forEach(
                    entry -> appendEntry(entry, clusterByFunc.apply(entry.file()))
                );

Manifest files grouped by partition spec is done in MergingSnapshotProducer at this line. So similar logic can be applied.

One more interesting point,
as per this line ManifestFileWriter is created using the TableOperations.current.spec which will be used to create all ManifestWriters, though ManifestReader is created based on the partitionSpec belongs to the Manifest File, as per this line

Not sure if my understanding is correct or not, So that is also part of the problem ?

[aokolnychyi]

@manishmalhotrawork, I think you got the problem correctly. We need to take into account that there might be multiple partition specs in a table. We can get a map of partition specs from Table and use that to correctly initialize ManifestWriter.

[aokolnychyi]

@manishmalhotrawork, do you want to submit a PR?

[manishmalhotrawork]

@aokolnychyi yes, let me submit PR. thanks.