[core] Support upper bound in dynamic bucket mode

[liyubin117]

Purpose

In dynamic bucket mode, unlimited buckets lead to an unpredicable number of small files, which lead to stability problems. so we should support upper bound in dynamic bucket mode.

Linked issue: close #4942

Tests

HashBucketAssignerTest
SimpleHashBucketAssignerTest

• Use new feature, Produce 3 buckets after inserting 12 rows

• Not use new feature, Produce 6 buckets after inserting 12 rows

API and Format

Documentation

docs/layouts/shortcodes/generated/core_configuration.html

[liyubin117]

@JingsongLi CI passed, PTAL, thanks!

[JingsongLi]

I'm not sure if these modifications are effective, so let me give you my suggestion:
We only need to modify PartitionIndex.asign inside:

Only code:

// 3. create a new bucket
for (int i = 0; i < Short.MAX_VALUE; i++) {
    if (bucketFilter.test(i) && !totalBucket.contains(i)) {
        hash2Bucket.put(hash, (short) i);
        nonFullBucketInformation.put(i, 1L);
        totalBucket.add(i);
        return i;
    }
}

// 4. too many buckets, throw exception
@SuppressWarnings("OptionalGetWithoutIsPresent")
int maxBucket = totalBucket.stream().mapToInt(Integer::intValue).max().getAsInt();
throw new RuntimeException(
        String.format(
                "Too more bucket %s, you should increase target bucket row number %s.",
                maxBucket, targetBucketRowNumber));

New code:

// 3. create a new bucket
for (int i = 0; i < max_buckets; i++) {
    if (bucketFilter.test(i) && !totalBucket.contains(i)) {
        hash2Bucket.put(hash, (short) i);
        nonFullBucketInformation.put(i, 1L);
        totalBucket.add(i);
        return i;
    }
}

// 4. exceed max_buckets, just pick a bucket for record.
pick a min bucket (belongs to this task) to the record.

[liyubin117]

After offline discussion with @JingsongLi , we have reached a consesus: We can't just update PartitionIndex logic because it doesn't handle SimpleHashBucketAssigner; When the buckets are full, a random bucket is selected for writing.

[liyubin117]

@JingsongLi The feature has been completed as discussed, Looking forward your review !

[JingsongLi]

You should find a bucket for own... Not randomly from all buckets.

[liyubin117]

Thanks for your kind remind.

1. Only when a new hash arrives and the bucket limit is exceeded, a bucket is randomly selected from task-owned buckets (totalBucket instance maintains buckets owned by itself task).
2. Then, the key-value pair of this hash and bucket will be put into the cache.
3. After that, whenever this hash arrives, it will always find its own bucket.

        // 1. is it a key that has appeared before
        if (hash2Bucket.containsKey(hash)) {
            return hash2Bucket.get(hash);
        }

[JingsongLi]

Here should not invoke the stream every time, you should cache the maxBucketId.

[liyubin117]

Wise consideration :)

[JingsongLi]

+1