[FLINK-5184] Error result of compareSerialized in RowComparator class

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.2.0
• Fix Version/s: 1.1.4, 1.2.0
• Component/s: Table SQL / API
• Labels: pull-request-available
• Assignee: godfrey he
• Reporter: godfrey he
• Created: 29/Nov/16 06:58
• Updated: 22/Apr/20 11:46
• Resolved: 29/Nov/16 14:52

Description

RowSerializer will write null mask for all fields in a row before serialize row data to DataOutputView.

RowSerializer.scala

override def serialize(value: Row, target: DataOutputView) {
    val len = fieldSerializers.length

    if (value.productArity != len) {
      throw new RuntimeException("Row arity of value does not match serializers.")
    }

    // write a null mask
    writeNullMask(len, value, target)

......
}

RowComparator will deserialize a row data from DataInputView when call compareSerialized method. However, the first parameter value of readIntoNullMask method is wrong, which should be the count of all fields, rather than the length of serializers (to deserialize the first n fields for comparison).

RowComparator.scala

override def compareSerialized(firstSource: DataInputView, secondSource: DataInputView): Int = {
    val len = serializers.length
    val keyLen = keyPositions.length

    readIntoNullMask(len, firstSource, nullMask1)
    readIntoNullMask(len, secondSource, nullMask2)
......
}

Issue Links

• links to: GitHub Pull Request #2894