Does the partition field have to be converted to String type first?

[GXM2333]

I use kafka streaming to build my pinot realtime partition table.The partition key is studentID,which I defined it as Integer both in pinot table schema and kafka producer.Then I use Integer's HashCode function to make partitions, and I custom a partitioner class for kafka producer, the partition function is like this:

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        // I didn't convert `key` to String 
        return Utils.toPositive(key.hashCode() % numPartitions);
    }

I found I can't get result if I use where studentID = 111 as predicate, then I found the reason is the broker calculate partition number for partition column using this way:

pinot/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/PartitionSegmentPruner.java

Line 216 in 46ed731

partitionInfo._partitionFunction.getPartition(operands.get(1).getLiteral().getFieldValue().toString()));

which will convert field value to String value before doing the hashCode function.

In my opinion, both kafka and piont have type systems, and different types have their own hashCode function.So is it better using their own hashCode function to calculate partition number for partition columns, not have to convert to string value before?

[Jackie-Jiang]

kafka uses murmur2 partition function by default, which also converts value to string first before hashing it.

It is possible to use the hashCode function for individual value type, but it might also cause inconsistency if value is ingested as a different type (e.g. INT is ingested as DOUBLE). To get consistent result, currently we always use string literal to calculate the partition.

On the kafka side, the partition function should match the one within pinot:

  public int getPartition(Object value) {
    return abs(value.toString().hashCode()) % _numPartitions;
  }

  private static int abs(int n) {
    return (n == Integer.MIN_VALUE) ? 0 : Math.abs(n);
  }

[kkrugler]

Hi @GXM2333 - this is an interesting topic. I'd suggest posting this (and future questions) to Slack, e.g. the dev channel, as that's better for having discussions, and more people will see it there. If the result of the Slack discussion is that there's a bug or useful enhancement, then that would be a great time to file a GitHub issue.

[GXM2333]

kafka uses murmur2 partition function by default, which also converts value to string first before hashing it.

It is possible to use the hashCode function for individual value type, but it might also cause inconsistency if value is ingested as a different type (e.g. INT is ingested as DOUBLE). To get consistent result, currently we always use string literal to calculate the partition.

On the kafka side, the partition function should match the one within pinot:

  public int getPartition(Object value) {
    return abs(value.toString().hashCode()) % _numPartitions;
  }

  private static int abs(int n) {
    return (n == Integer.MIN_VALUE) ? 0 : Math.abs(n);
  }

[GXM2333]

Thanks for your reply, I've joined pinot slack, and I'll ask questions there firstly :)