HIVE-25948: Iceberg: Enable cost-based selection between Fanout and Clustered writers using column stats NDV

[deniskuzZ]

What changes were proposed in this pull request?

Cost-based selection between Fanout and Clustered writers

Why are the changes needed?

Perf optimization

Does this PR introduce any user-facing change?

No

How was this patch tested?

mvn test -Dtest=TestIcebergCliDriver -Dqfile=dynamic_partition_writes.q

┌───────────────────────┬───────────────────────────┐
│       Scenario        │         Expected          │
├───────────────────────┼───────────────────────────┤
│ threshold=0 (default) │ no sort (NDV<MAX_WRITERS) │
├───────────────────────┼───────────────────────────┤
│ threshold=-1          │ no sort                   │
├───────────────────────┼───────────────────────────┤
│ threshold=1           │ sort                      │
├───────────────────────┼───────────────────────────┤
│ threshold=2           │ sort (NDV>2)              │
├───────────────────────┼───────────────────────────┤
│ threshold=100         │ no sort (NDV<=100)        │
├───────────────────────┼───────────────────────────┤
│ fanout=false          │ sort                      │
└───────────────────────┴───────────────────────────┘

[sonarqubecloud]

Quality Gate passed

Issues
10 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[okumin]

Should this be located after all the bailouts are evaluated? It means we may mutate the context after if (!removeRSInsertedByEnforceBucketing(fsOp)) { finished.

[okumin]

I might be wrong. I presume this evaluates iceberg_bucket(ndv_100, 8 (=num buckets)) as 100. As iceberg_* always narrows the source column, the current implementation is likely to degrade performance for any workload.
This also means we have a chance to further optimize the cost-based optimizer. Since the cardinality of iceberg_bucket(x, 8) is obviously min(cardinality(x), 8), we can enable the optimization for more cases.

I guess we can achieve the optimization if we implement the following API in iceberg_* UDFs and resolve more accurate cardinality in this computePartCardinality method.
https://github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/stats/estimator/StatEstimator.java