Add support for preferred insert or create table layouts

[sopel39]

No description provided.

[sopel39]

this is ready to go

[findepi]

Apply partitioning by partition columns also when the table is bucketed.

[sopel39]

When table is bucketed then data partitioning on partition columns should be required, but this is outside of scope of this PR.
For example,

• currently: if you have 10 buckets per partition, then we partition data by bucket columns. This means only 10 workers will write data for all partitions
• better: if you partition data by bucket columns AND partition columns, then still 10 workers will write data per partition. However, set of workers per partition changes therefore improving parallelism.

[findepi]

This adds use-preferred-write-partitioning configuration / use_preferred_write_partitioning session toggle.
Shouldn't we have a Hive- (catalog-) specific toggle instead?

[sopel39]

Shouldn't we have a Hive- (catalog-) specific toggle instead?

Some more context on engine vs connector approach. While connector specific toggle could be some stop-gap here, I don't think it's a good approach. This PR opens a way for CBO to make decision on using preferred insert layout. For example there could be very basic CBO rule that chooses proffered partitioning if number of groups if greater than some constant. I think such rule would work good enough in practice. This PR is step toward it. I don't like too many manual toggles
and insert queries are often simpler so stats are easier to estimate.

The SPI needs to be changed regardless of approach as connectors need to be able to return FIXED_HASH ConnectorPartitioningHandle. This is system partitioning handle and reimplementing it in connectors (there are various methods associated with it) doesn't make much sense.

[sopel39]

@dain mentioned there was some PR to add support for arbitrary partitioning in local exchanges.

[martint]

Not sure I understand what the commit message is referring to. Whats "evenly partitioning"?

[martint]

This is strange. Why would partioningColumns be missing when partitioningColumns is present?

[sopel39]

This is strange. Why would partitioning be missing when partitioningColumns is present?

If you just care that data is partitioned on columns, but don't care how exactly, then you don't need to specify partitioning

[martint]

I find the fact we're looking at getLayout().getPartitioning() to decide whether to call getPartitioning() very confusing, and probably a sign of an API design issue.

It'd be cleaner to make getPartitioning() for NewTableLayout return optional and just do:

PartitioningHandle partitioningHandle = writeTableLayout.get()
    .getPartitioning()
    .orElse(FIXED_HASH_DISTRIBUTION)

[martint]

Maybe...

boolean hasFixedHashDistribution = node.getPartitioningScheme()
        .map(scheme -> scheme.getPartitioning().getHandle())
        .filter(isEqual(FIXED_HASH_DISTRIBUTION))
        .isPresent();

or

boolean hasFixedHashDistribution = node.getPartitioningScheme()
        .filter(scheme -> scheme.getPartitioning().getHandle().equals(FIXED_HASH_DISTRIBUTION))
        .isPresent();

[martint]

This can move inside the if (getTaskWriterCount(session) > 1) { block

[sopel39]

ac

[sopel39]

This is strange. Why would partitioning be missing when partitioningColumns is present?

If you just care that data is partitioned on columns, but don't care how exactly, then you don't need to specify partitioning