Flink: a few small fixes or tuning for range partitioner

[stevenzwu]

• handle empty stats (like no traffic)
• use struct comparator with transformed schema for SortKey
• also added more unit tests.

The next PR after this would be putting everything together in FlinkSink and document update.

[stevenzwu]

this refactoring is to make it easier to write unit test with the static util method

[stevenzwu]

this is an important fix. SortKey may have been serialized by subtask and deserialized by the coordinator. we need to use the schema of the transformed sort fields for struct comparator.

[pvary]

Is this code duplicated in the next method? Is there a way to extract the duplicated code?

[stevenzwu]

there are some duplications. but I think those two tests are different: scale up and scale down. I want to explicitly test each range.

The scale down case has this differently.

// > m
    // reassigns out-of-range partitions via mod (% 3 in this case)
    assertThat(
            SketchUtil.partition(
                CHAR_KEYS.get("n"), numPartitions, rangeBounds, SORT_ORDER_COMPARTOR))
        .isEqualTo(0);
    assertThat(
            SketchUtil.partition(
                CHAR_KEYS.get("z"), numPartitions, rangeBounds, SORT_ORDER_COMPARTOR))
        .isEqualTo(0);

[pvary]

Maybe a method like this would help to focus on the actual test and hide the common parameters?

private static assertPartition(SortKey key, int expected) {
   assertThat(
            SketchUtil.partition(
                key, numPartitions, rangeBounds, SORT_ORDER_COMPARTOR))
        .isEqualTo(expected);
}

The first 8 assertions seems like the same to me, but it is hard to check/understand as it is "hidden" by the long assertion statements.

WDYT?

[stevenzwu]

that is a good idea. Reducing to a single line would help readability.

[stevenzwu]

thanks @pvary for the review. will create the back port PR shortly