wuchong commented on code in PR #19709:
URL: https://github.com/apache/flink/pull/19709#discussion_r934288651
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala:
##########
@@ -311,22 +311,35 @@ object CodeGenUtils {
case DOUBLE => s"${className[JDouble]}.hashCode($term)"
case TIMESTAMP_WITHOUT_TIME_ZONE | TIMESTAMP_WITH_LOCAL_TIME_ZONE =>
s"$term.hashCode()"
- case TIMESTAMP_WITH_TIME_ZONE | ARRAY | MULTISET | MAP =>
+ case TIMESTAMP_WITH_TIME_ZONE =>
throw new UnsupportedOperationException(
s"Unsupported type($t) to generate hash code," +
s" the type($t) is not supported as a
GROUP_BY/PARTITION_BY/JOIN_EQUAL/UNION field.")
+ case ARRAY =>
+ val subCtx = new CodeGeneratorContext(ctx.tableConfig, ctx.classLoader)
+ val genHash =
+ HashCodeGenerator.generateArrayHash(
+ subCtx,
+ t.asInstanceOf[ArrayType].getElementType,
+ "SubHashArray")
+ genHashFunction(ctx, subCtx, genHash, term)
+ case MULTISET | MAP =>
+ val subCtx = new CodeGeneratorContext(ctx.tableConfig, ctx.classLoader)
+ val (keyType, valueType) = if (t.isInstanceOf[MultisetType]) {
+ (t.asInstanceOf[MultisetType].getElementType, new IntType())
+ } else {
+ (t.asInstanceOf[MapType].getKeyType,
t.asInstanceOf[MapType].getValueType)
+ }
Review Comment:
Can be replaced with Scala pattern match:
```scala
val (keyType, valueType) = t match {
case multiset: MultisetType =>
(multiset.getElementType, new IntType())
case map: MapType =>
(map.getKeyType, map.getValueType)
}
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/HashCodeGenerator.scala:
##########
@@ -75,6 +75,157 @@ object HashCodeGenerator {
return $resultTerm;
}
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for ArrayData.")}
+ }
+
+ @Override
+ public int hashCode($MAP_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for MapData.")}
+ }
+
+ ${ctx.reuseInnerClassDefinitionCode()}
+ }
+ """.stripMargin
+
+ new GeneratedHashFunction(className, code, ctx.references.toArray,
ctx.tableConfig)
+ }
+
+ def generateArrayHash(
+ ctx: CodeGeneratorContext,
+ elementType: LogicalType,
+ name: String): GeneratedHashFunction = {
+ val className = newName(name)
+ val baseClass = classOf[HashFunction]
+ val inputTerm = CodeGenUtils.DEFAULT_INPUT1_TERM
+
+ val typeTerm = primitiveTypeTermForType(elementType)
+ val isNull = newName("isNull")
+ val fieldTerm = newName("fieldTerm")
+ val hashIntTerm = newName("hashCode")
+ val i = newName("i")
+
+ // Generate element hash code firstly
+ val elementHashBody = hashCodeForType(ctx, elementType, fieldTerm)
+ val code =
+ j"""
+ public class $className implements ${baseClass.getCanonicalName} {
+
+ ${ctx.reuseMemberCode()}
+
+ public $className(Object[] references) throws Exception {
+ ${ctx.reuseInitCode()}
+ }
+
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ int $hashIntTerm = 0;
+ // This is inspired by hive & presto
+ for (int $i = 0; $i < $inputTerm.size(); $i++) {
+ boolean $isNull = $inputTerm.isNullAt($i);
+ if (!$isNull) {
+ $typeTerm $fieldTerm = ${rowFieldReadAccess(i, inputTerm,
elementType)};
+ $hashIntTerm = 31 * $hashIntTerm + $elementHashBody;
+ }
+ }
+
+ return $hashIntTerm;
+ }
+
+ @Override
+ public int hashCode($ROW_DATA $inputTerm) {
+ ${genThrowException("ArrayData hash function doesn't support to
generate hash code for RowData.")}
+ }
+
+ @Override
+ public int hashCode($MAP_DATA $inputTerm) {
+ ${genThrowException("ArrayData hash function doesn't support to
generate hash code for MapData.")}
+ }
+
+ ${ctx.reuseInnerClassDefinitionCode()}
+ }
+ """.stripMargin
+
+ new GeneratedHashFunction(className, code, ctx.references.toArray,
ctx.tableConfig)
+ }
+
+ def generateMapHash(
+ ctx: CodeGeneratorContext,
+ keyType: LogicalType,
+ valueType: LogicalType,
+ name: String): GeneratedHashFunction = {
+ val className = newName(name)
+ val baseClass = classOf[HashFunction]
+ val inputTerm = CodeGenUtils.DEFAULT_INPUT1_TERM
+
+ val keyTypeTerm = primitiveTypeTermForType(keyType)
+ val valueTypeTerm = primitiveTypeTermForType(valueType)
+ val keys = newName("keys")
+ val values = newName("values")
+ val keyIsNull = newName("keyIsNull")
+ val keyFieldTerm = newName("keyFieldTerm")
+ val valueIsNull = newName("valueIsNull")
+ val valueFieldTerm = newName("valueFieldTerm")
+ val keyHashTerm = newName("keyHashCode")
+ val valueHashTerm = newName("valueHashCode")
+ val hashIntTerm = newName("hashCode")
+ val i = newName("i")
+
+ // Generate key and value hash code body firstly
+ val keyElementHashBody = hashCodeForType(ctx, keyType, keyFieldTerm)
+ val valueElementHashBody = hashCodeForType(ctx, valueType, valueFieldTerm)
+ val code =
+ j"""
+ public class $className implements ${baseClass.getCanonicalName} {
+
+ ${ctx.reuseMemberCode()}
+
+ public $className(Object[] references) throws Exception {
+ ${ctx.reuseInitCode()}
+ }
+
+ @Override
+ public int hashCode($MAP_DATA $inputTerm) {
+ $ARRAY_DATA $keys = $inputTerm.keyArray();
+ $ARRAY_DATA $values = $inputTerm.valueArray();
+
+ int $keyHashTerm = 0;
+ int $valueHashTerm = 0;
+ int $hashIntTerm = 0;
+
+ // This is inspired by hive & presto
+ for (int $i = 0; $i < $inputTerm.size(); $i++) {
+ boolean $keyIsNull = $keys.isNullAt($i);
+ if (!$keyIsNull) {
+ $keyHashTerm = 0;
+ $keyTypeTerm $keyFieldTerm = ${rowFieldReadAccess(i, keys,
keyType)};
+ $keyHashTerm = $keyElementHashBody;
+ }
+
+ boolean $valueIsNull = $values.isNullAt($i);
+ if(!$valueIsNull) {
+ $valueHashTerm = 0;
Review Comment:
Should the `$keyHashTerm = 0;` and `$valueHashTerm = 0;` be executed before
if condition? Otherwise, if current value is null, the `valueHashTerm` would be
the hash of the last entry. Could you add a test for a map with multiple null
values?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/HashCodeGenerator.scala:
##########
@@ -75,6 +75,157 @@ object HashCodeGenerator {
return $resultTerm;
}
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for ArrayData.")}
+ }
+
+ @Override
+ public int hashCode($MAP_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for MapData.")}
+ }
+
+ ${ctx.reuseInnerClassDefinitionCode()}
+ }
+ """.stripMargin
+
+ new GeneratedHashFunction(className, code, ctx.references.toArray,
ctx.tableConfig)
+ }
+
+ def generateArrayHash(
+ ctx: CodeGeneratorContext,
+ elementType: LogicalType,
+ name: String): GeneratedHashFunction = {
+ val className = newName(name)
+ val baseClass = classOf[HashFunction]
+ val inputTerm = CodeGenUtils.DEFAULT_INPUT1_TERM
+
+ val typeTerm = primitiveTypeTermForType(elementType)
+ val isNull = newName("isNull")
+ val fieldTerm = newName("fieldTerm")
+ val hashIntTerm = newName("hashCode")
+ val i = newName("i")
+
+ // Generate element hash code firstly
+ val elementHashBody = hashCodeForType(ctx, elementType, fieldTerm)
+ val code =
+ j"""
+ public class $className implements ${baseClass.getCanonicalName} {
+
+ ${ctx.reuseMemberCode()}
+
+ public $className(Object[] references) throws Exception {
+ ${ctx.reuseInitCode()}
+ }
+
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ int $hashIntTerm = 0;
Review Comment:
Should the `int $hashIntTerm = 0;` start from `1`?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/HashCodeGenerator.scala:
##########
@@ -75,6 +75,157 @@ object HashCodeGenerator {
return $resultTerm;
}
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for ArrayData.")}
+ }
+
+ @Override
+ public int hashCode($MAP_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for MapData.")}
+ }
+
+ ${ctx.reuseInnerClassDefinitionCode()}
+ }
+ """.stripMargin
+
+ new GeneratedHashFunction(className, code, ctx.references.toArray,
ctx.tableConfig)
+ }
+
+ def generateArrayHash(
+ ctx: CodeGeneratorContext,
+ elementType: LogicalType,
+ name: String): GeneratedHashFunction = {
+ val className = newName(name)
+ val baseClass = classOf[HashFunction]
+ val inputTerm = CodeGenUtils.DEFAULT_INPUT1_TERM
+
+ val typeTerm = primitiveTypeTermForType(elementType)
+ val isNull = newName("isNull")
+ val fieldTerm = newName("fieldTerm")
+ val hashIntTerm = newName("hashCode")
+ val i = newName("i")
+
+ // Generate element hash code firstly
+ val elementHashBody = hashCodeForType(ctx, elementType, fieldTerm)
+ val code =
+ j"""
+ public class $className implements ${baseClass.getCanonicalName} {
+
+ ${ctx.reuseMemberCode()}
+
+ public $className(Object[] references) throws Exception {
+ ${ctx.reuseInitCode()}
+ }
+
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ int $hashIntTerm = 0;
+ // This is inspired by hive & presto
+ for (int $i = 0; $i < $inputTerm.size(); $i++) {
+ boolean $isNull = $inputTerm.isNullAt($i);
+ if (!$isNull) {
+ $typeTerm $fieldTerm = ${rowFieldReadAccess(i, inputTerm,
elementType)};
+ $hashIntTerm = 31 * $hashIntTerm + $elementHashBody;
+ }
+ }
+
+ return $hashIntTerm;
+ }
+
+ @Override
+ public int hashCode($ROW_DATA $inputTerm) {
+ ${genThrowException("ArrayData hash function doesn't support to
generate hash code for RowData.")}
+ }
+
+ @Override
+ public int hashCode($MAP_DATA $inputTerm) {
+ ${genThrowException("ArrayData hash function doesn't support to
generate hash code for MapData.")}
+ }
Review Comment:
Do not exceed 100 characters in a single line in Scala.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/HashCodeGenerator.scala:
##########
@@ -75,6 +75,157 @@ object HashCodeGenerator {
return $resultTerm;
}
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for ArrayData.")}
+ }
+
+ @Override
+ public int hashCode($MAP_DATA $inputTerm) {
+ ${genThrowException("RowData hash function doesn't support to
generate hash code for MapData.")}
+ }
+
+ ${ctx.reuseInnerClassDefinitionCode()}
+ }
+ """.stripMargin
+
+ new GeneratedHashFunction(className, code, ctx.references.toArray,
ctx.tableConfig)
+ }
+
+ def generateArrayHash(
+ ctx: CodeGeneratorContext,
+ elementType: LogicalType,
+ name: String): GeneratedHashFunction = {
+ val className = newName(name)
+ val baseClass = classOf[HashFunction]
+ val inputTerm = CodeGenUtils.DEFAULT_INPUT1_TERM
+
+ val typeTerm = primitiveTypeTermForType(elementType)
+ val isNull = newName("isNull")
+ val fieldTerm = newName("fieldTerm")
+ val hashIntTerm = newName("hashCode")
+ val i = newName("i")
+
+ // Generate element hash code firstly
+ val elementHashBody = hashCodeForType(ctx, elementType, fieldTerm)
+ val code =
+ j"""
+ public class $className implements ${baseClass.getCanonicalName} {
+
+ ${ctx.reuseMemberCode()}
+
+ public $className(Object[] references) throws Exception {
+ ${ctx.reuseInitCode()}
+ }
+
+ @Override
+ public int hashCode($ARRAY_DATA $inputTerm) {
+ int $hashIntTerm = 0;
+ // This is inspired by hive & presto
+ for (int $i = 0; $i < $inputTerm.size(); $i++) {
+ boolean $isNull = $inputTerm.isNullAt($i);
+ if (!$isNull) {
+ $typeTerm $fieldTerm = ${rowFieldReadAccess(i, inputTerm,
elementType)};
+ $hashIntTerm = 31 * $hashIntTerm + $elementHashBody;
+ }
Review Comment:
Should we consider null elements, e.g. use hash 0 for null elements instead
of skipping them?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/HashFunction.java:
##########
@@ -18,12 +18,23 @@
package org.apache.flink.table.runtime.generated;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RowData;
/**
- * Interface for code generated hash code of {@link RowData}, which will
select some fields to hash.
+ * Interface for code generated hash code of {@link RowData} which will select
some fields to hash
+ * or {@link ArrayData} or {@link MapData}.
+ *
+ * <p>Due to Janino's support for generic type is not very friendly, so here
can't introduce generic
+ * type for {@link HashFunction}, please see
https://github.com/janino-compiler/janino/issues/109
+ * for details.
Review Comment:
You can define generic types in the interface. The Janino limitation
requires generating the method parameters using `Object` types instead of
generic types. However, that's not a problem. You can manually cast the input
parameter to `RowData`/`ArrayData`/`MapData` type in the first line of the
method. See
`org.apache.flink.table.planner.codegen.FunctionCodeGenerator#generateFunction`
I don't like overloading many methods, but just one method is valid.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/HashCodeGeneratorTest.scala:
##########
@@ -41,16 +45,164 @@ class HashCodeGeneratorTest {
val hashFunc2 = HashCodeGenerator
.generateRowHash(
- new CodeGeneratorContext(new Configuration,
Thread.currentThread().getContextClassLoader),
+ new CodeGeneratorContext(new Configuration, classLoader),
RowType.of(new IntType(), new BigIntType(), new
VarBinaryType(VarBinaryType.MAX_LENGTH)),
"name",
Array(1, 2, 0)
)
.newInstance(classLoader)
val row = GenericRowData.of(ji(5), jl(8), Array[Byte](1, 5, 6))
- Assert.assertEquals(637, hashFunc1.hashCode(row))
- Assert.assertEquals(136516167, hashFunc2.hashCode(row))
+ assertEquals(637, hashFunc1.hashCode(row))
+ assertEquals(136516167, hashFunc2.hashCode(row))
+
+ // test row with nested array and map type
+ val hashFunc3 = HashCodeGenerator
+ .generateRowHash(
+ new CodeGeneratorContext(new Configuration, classLoader),
+ RowType.of(
+ new IntType(),
+ new ArrayType(new IntType()),
+ new MultisetType(new IntType()),
+ new MapType(new IntType(), new VarCharType())),
+ "name",
+ Array(1, 2, 0, 3)
+ )
+ .newInstance(classLoader)
+
+ val row3 = GenericRowData.of(
+ ji(5),
+ new GenericArrayData(Array(1, 5, 7)),
+ new GenericMapData(Map(1 -> null, 5 -> null, 10 -> null)),
+ new GenericMapData(
+ Map(1 -> StringData.fromString("ron"), 5 ->
StringData.fromString("danny"), 10 -> null))
+ )
+ assertEquals(1065781729, hashFunc3.hashCode(row3))
+
+ // test hash code for ArrayData
+ CommonTestUtils.assertThrows(
+ "RowData hash function doesn't support to generate hash code for
ArrayData.",
+ classOf[RuntimeException],
+ () => hashFunc3.hashCode(new GenericArrayData(Array(1)))
+ )
+
+ // test hash code for MapData
+ CommonTestUtils.assertThrows(
+ "RowData hash function doesn't support to generate hash code for
MapData.",
+ classOf[RuntimeException],
+ () => hashFunc3.hashCode(new GenericMapData(null)))
+ }
+
+ @Test
+ def testArrayHash(): Unit = {
+ // test primitive type
+ val hashFunc1 = HashCodeGenerator
+ .generateArrayHash(
+ new CodeGeneratorContext(new Configuration(), classLoader),
+ new IntType(),
+ "name")
+ .newInstance(classLoader)
+
+ val array1 = new GenericArrayData(Array(1, 5, 7))
+ assertEquals(1123, hashFunc1.hashCode(array1))
+
+ // test complex map type of element
+ val hashFunc2 = HashCodeGenerator
+ .generateArrayHash(
+ new CodeGeneratorContext(new Configuration(), classLoader),
+ new MapType(new IntType(), new VarCharType()),
+ "name")
+ .newInstance(classLoader)
+
+ val mapData = new GenericMapData(
+ Map(1 -> StringData.fromString("ron"), 5 ->
StringData.fromString("danny"), 10 -> null))
+ val array2 = new GenericArrayData(Array[AnyRef](mapData))
+ assertEquals(93178751, hashFunc2.hashCode(array2))
+
+ // test complex row type of element
+ val hashFunc3 = HashCodeGenerator
+ .generateArrayHash(
+ new CodeGeneratorContext(new Configuration(), classLoader),
+ RowType.of(new IntType(), new BigIntType()),
+ "name")
+ .newInstance(classLoader)
+
+ val array3 = new GenericArrayData(
+ Array[AnyRef](GenericRowData.of(ji(5), jl(8)), GenericRowData.of(ji(25),
jl(52))))
+ assertEquals(14520, hashFunc3.hashCode(array3))
+
+ // test hash code for RowData
+ CommonTestUtils.assertThrows(
+ "ArrayData hash function doesn't support to generate hash code for
RowData.",
+ classOf[RuntimeException],
+ () => hashFunc3.hashCode(GenericRowData.of(null))
+ )
+
+ // test hash code for MapData
+ CommonTestUtils.assertThrows(
+ "ArrayData hash function doesn't support to generate hash code for
MapData.",
+ classOf[RuntimeException],
+ () => hashFunc3.hashCode(new GenericMapData(null))
+ )
+ }
+
+ @Test
+ def testMapHash(): Unit = {
+ // test primitive type
+ val hashFunc1 = HashCodeGenerator
+ .generateMapHash(
+ new CodeGeneratorContext(new Configuration(), classLoader),
+ new IntType(),
+ new VarCharType(),
+ "name")
+ .newInstance(classLoader)
+
+ val map1 = new GenericMapData(
+ Map(1 -> StringData.fromString("ron"), 5 ->
StringData.fromString("danny"), 10 -> null))
Review Comment:
Please test hashcode of 2 maps are equal, which have entries in a different
order and contain null values.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/HashCodeGeneratorTest.scala:
##########
@@ -41,16 +45,164 @@ class HashCodeGeneratorTest {
val hashFunc2 = HashCodeGenerator
.generateRowHash(
- new CodeGeneratorContext(new Configuration,
Thread.currentThread().getContextClassLoader),
+ new CodeGeneratorContext(new Configuration, classLoader),
RowType.of(new IntType(), new BigIntType(), new
VarBinaryType(VarBinaryType.MAX_LENGTH)),
"name",
Array(1, 2, 0)
)
.newInstance(classLoader)
val row = GenericRowData.of(ji(5), jl(8), Array[Byte](1, 5, 6))
- Assert.assertEquals(637, hashFunc1.hashCode(row))
- Assert.assertEquals(136516167, hashFunc2.hashCode(row))
+ assertEquals(637, hashFunc1.hashCode(row))
+ assertEquals(136516167, hashFunc2.hashCode(row))
+
+ // test row with nested array and map type
+ val hashFunc3 = HashCodeGenerator
+ .generateRowHash(
+ new CodeGeneratorContext(new Configuration, classLoader),
+ RowType.of(
+ new IntType(),
+ new ArrayType(new IntType()),
+ new MultisetType(new IntType()),
+ new MapType(new IntType(), new VarCharType())),
+ "name",
+ Array(1, 2, 0, 3)
+ )
+ .newInstance(classLoader)
+
+ val row3 = GenericRowData.of(
+ ji(5),
+ new GenericArrayData(Array(1, 5, 7)),
+ new GenericMapData(Map(1 -> null, 5 -> null, 10 -> null)),
+ new GenericMapData(
+ Map(1 -> StringData.fromString("ron"), 5 ->
StringData.fromString("danny"), 10 -> null))
+ )
+ assertEquals(1065781729, hashFunc3.hashCode(row3))
+
+ // test hash code for ArrayData
+ CommonTestUtils.assertThrows(
+ "RowData hash function doesn't support to generate hash code for
ArrayData.",
+ classOf[RuntimeException],
+ () => hashFunc3.hashCode(new GenericArrayData(Array(1)))
+ )
+
+ // test hash code for MapData
+ CommonTestUtils.assertThrows(
+ "RowData hash function doesn't support to generate hash code for
MapData.",
+ classOf[RuntimeException],
+ () => hashFunc3.hashCode(new GenericMapData(null)))
+ }
+
+ @Test
+ def testArrayHash(): Unit = {
+ // test primitive type
+ val hashFunc1 = HashCodeGenerator
+ .generateArrayHash(
+ new CodeGeneratorContext(new Configuration(), classLoader),
+ new IntType(),
+ "name")
+ .newInstance(classLoader)
+
+ val array1 = new GenericArrayData(Array(1, 5, 7))
+ assertEquals(1123, hashFunc1.hashCode(array1))
Review Comment:
Asserting the hash int is hard to tell right or wrong. Would be better to
add assertions, e.g.
```
val array1 = new GenericArrayData(Array(1, 5, 7))
val array2 = new GenericArrayData(Array(1, 5, 7))
val array3 = new GenericArrayData(Array[AnyRef](1, null, 5, null, 7))
assertEquals(hashFunc1.hashCode(array1), hashFunc1.hashCode(array2))
assertNotEquals(hashFunc1.hashCode(array1), hashFunc1.hashCode(array3))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
