You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "raminqaf (via GitHub)" <gi...@apache.org> on 2026/10/14 15:29:02 UTC
[PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
raminqaf opened a new pull request, #29481:
URL: https://github.com/apache/flink/pull/29481
## What is the purpose of the change
Comparing two MAPs or MULTISETs with `=` returns FALSE when they hold equal values, because the generated code compares the boxed values by reference.
```sql
-- f0 and f1 both hold {'a': 1.5, 'b': 2.5} as MAP<STRING, DOUBLE>
SELECT f0 = f1 FROM t
```
| Equal values in both maps | Before | After |
|----------------------------------------------------|--------|-------|
| `MAP<STRING, DOUBLE>` `{a: 1.5, b: 2.5}` | FALSE | TRUE |
| `MAP<STRING, INT>` `{a: 7, b: 5000}` | FALSE | TRUE |
| `MAP<STRING, DATE>` `{a: 1985-11-04, b: 2018-07-26}` | FALSE | TRUE |
| `MULTISET<STRING>` `{a: 200, b: 300}` | FALSE | TRUE |
`generateMapComparison` passes the boxed map values to `generateEquals`, which emits `==` for numeric, DATE, TIME and interval types. INT values only compare correctly inside the Integer cache. The fix unboxes the values first, so `=` on map values behaves like `=` on the scalar type. For DOUBLE that means NaN is not equal to NaN and 0.0 equals -0.0. The null handling is unchanged.
I did not switch to `.equals()`. `Double.equals` compares bits, so it would disagree with `DOUBLE =` on NaN and 0.0.
The same code is used by `IS [NOT] DISTINCT FROM` and by the field comparison of the `RecordEqualiser`. Rows with MAP or MULTISET fields therefore also compare by value in the stateful operators that use the equaliser, for example deduplication and group aggregation. Those operators now skip updates that only differ in the identity of a box.
Not part of this PR:
- BINARY and UUID map keys still compare by array identity.
- ARRAY and ROW over binary data still take a byte comparison shortcut that depends on the entry order.
- A map with a NULL value still compares unequal to every map, including a copy of itself.
## Brief change log
- `ScalarOperatorGens#generateMapComparison` unboxes the map values with `generateInputFieldUnboxing` before it calls `generateEquals`.
- `MapFunctionITCase` gets equality cases for maps and multisets.
## Verifying this change
This change added tests and can be verified as follows:
- `MapFunctionITCase` has new equality cases for DOUBLE, INT, DATE, DECIMAL, nested maps, maps with different keys or size, NULL values, MULTISET with counts above 127 and maps built from columns. Each case checks `=` and `<>` in the Table API and in SQL.
- Without the fix, 18 of the 27 new test items fail with `expected: +I[true] but was: +I[false]`. The DECIMAL, different keys or size and NULL cases pass either way.
- `CollectionFunctionsITCase`, `CastFunctionITCase`, `JsonFunctionsITCase`, `ComparisonFunctionITCase`, `ScalarOperatorsTest` and `EqualiserCodeGeneratorTest` pass.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes. The generated map comparison adds a null check and an assignment per compared value.
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code 2.1.299 (Claude Sonnet 5.5)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4274615551
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1950,56 +1950,78 @@ object ScalarOperatorGens {
val resultTerm = newName(ctx, "compareResult")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
|if ($leftTerm.size() == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $rightKeysTerm.size(); $indexTerm++) {
Review Comment:
since it is hotcode: extract into a var
```scala
$rightKeysTerm.size()
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4293015033
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,60 +1948,98 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val leftKeyIndexTerm = newName(ctx, "leftKeyIndex")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val entryTerm = newName(ctx, "entry")
+ val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
+ // A map can hold a key more than once, and then its last entry counts. Each key index keeps
+ // the last position of every key, so equal maps have key indexes of the same size whose
+ // values match at those positions.
val stmt =
s"""
|boolean $resultTerm;
- |if ($leftTerm.size() == $rightTerm.size()) {
+ |int $sizeTerm = $leftTerm.size();
+ |if ($sizeTerm == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
+ |
+ | $mapCls $leftKeyIndexTerm = new $hashMapCls();
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
Review Comment:
I switched the key lookup to SQL `=` in this PR. Each left key now scans the right keys with `generateEquals`, so no HashMap and no boxing are left. BINARY keys and 0.0 vs -0.0 keys now match, with tests for both. This is quadratic in the map size, like ARRAY_DISTINCT and MAP_UNION. FLINK-41013 tracks hash-based matching with a hash that agrees with SQL `=`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4284191371
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,60 +1948,98 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val leftKeyIndexTerm = newName(ctx, "leftKeyIndex")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val entryTerm = newName(ctx, "entry")
+ val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
+ // A map can hold a key more than once, and then its last entry counts. Each key index keeps
+ // the last position of every key, so equal maps have key indexes of the same size whose
+ // values match at those positions.
val stmt =
s"""
|boolean $resultTerm;
- |if ($leftTerm.size() == $rightTerm.size()) {
+ |int $sizeTerm = $leftTerm.size();
+ |if ($sizeTerm == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
+ |
+ | $mapCls $leftKeyIndexTerm = new $hashMapCls();
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
Review Comment:
Agreed, the key lookup uses Java `equals`. `BINARY` keys never match, and 0.0 and -0.0 are different keys. Master does the same lookup through `convertToJavaMap`. Would a follow-up ticket be OK?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4273473249
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1955,16 +1955,15 @@ object ScalarOperatorGens {
val leftMapTerm = newName(ctx, "leftMap")
val leftKeyTerm = newName(ctx, "leftKey")
- val leftValueTerm = newName(ctx, "leftValue")
- val leftValueNullTerm = newName(ctx, "leftValueIsNull")
+ // Map values are boxed. Unbox them, otherwise generateEquals compares the boxes by reference.
+ val leftBoxedValueTerm = newName(ctx, "leftBoxedValue")
val leftValueExpr =
Review Comment:
why do we need to do autoboxing/unboxing here?
can we instead have map key to index and then retrieve values by index and use for instance `getInt` or other types?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4274570009
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1955,16 +1955,15 @@ object ScalarOperatorGens {
val leftMapTerm = newName(ctx, "leftMap")
val leftKeyTerm = newName(ctx, "leftKey")
- val leftValueTerm = newName(ctx, "leftValue")
- val leftValueNullTerm = newName(ctx, "leftValueIsNull")
+ // Map values are boxed. Unbox them, otherwise generateEquals compares the boxes by reference.
+ val leftBoxedValueTerm = newName(ctx, "leftBoxedValue")
val leftValueExpr =
Review Comment:
I went through the `generateArrayComparison` and did a similar implementation for both arryas
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4275879157
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java:
##########
@@ -889,4 +891,190 @@ private Stream<TestSetSpec> mapUnionTestCases() {
"Invalid input arguments. Expected signatures are:\n"
+ "MAP_UNION(<COMMON>, <COMMON>...)"));
}
+
+ private Stream<TestSetSpec> mapEqualsTestCases() {
+ // In f1 the entries of f0 come in the opposite order, so the planner cannot merge or fold
+ // the two sides. f2 differs from f0 in one value.
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS, "MAP<STRING, DOUBLE>")
+ .onFieldsWithData(
+ orderedMap(A, 1.5, B, 2.5),
+ orderedMap(B, 2.5, A, 1.5),
+ orderedMap(A, 1.5, B, 1.5))
+ .andDataTypes(
+ MAP(STRING(), DOUBLE()),
+ MAP(STRING(), DOUBLE()),
+ MAP(STRING(), DOUBLE()))
+ .testResult(
+ resultSpec($("f0").isEqual($("f1")), "f0 = f1", true, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f1")), "f0 <> f1", false, BOOLEAN()),
+ resultSpec($("f0").isEqual($("f2")), "f0 = f2", false, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f2")), "f0 <> f2", true, BOOLEAN())),
+ // 7 is a cached Integer box, 5000 is not
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS, "MAP<STRING, INT>")
+ .onFieldsWithData(
+ orderedMap(A, 7, B, 5000),
+ orderedMap(B, 5000, A, 7),
+ orderedMap(A, 7, B, 6000))
+ .andDataTypes(
+ MAP(STRING(), INT()), MAP(STRING(), INT()), MAP(STRING(), INT()))
+ .testResult(
+ resultSpec($("f0").isEqual($("f1")), "f0 = f1", true, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f1")), "f0 <> f1", false, BOOLEAN()),
+ resultSpec($("f0").isEqual($("f2")), "f0 = f2", false, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f2")), "f0 <> f2", true, BOOLEAN())),
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS, "MAP<STRING, DATE>")
+ .onFieldsWithData(
+ orderedMap(A, TEST_DATE_1, B, TEST_DATE_2),
+ orderedMap(B, TEST_DATE_2, A, TEST_DATE_1),
+ orderedMap(A, TEST_DATE_1, B, TEST_DATE_1))
+ .andDataTypes(
+ MAP(STRING(), DATE()), MAP(STRING(), DATE()), MAP(STRING(), DATE()))
+ .testResult(
+ resultSpec($("f0").isEqual($("f1")), "f0 = f1", true, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f1")), "f0 <> f1", false, BOOLEAN()),
+ resultSpec($("f0").isEqual($("f2")), "f0 = f2", false, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f2")), "f0 <> f2", true, BOOLEAN())),
+ // a reference type, so unboxing leaves the values as they are
Review Comment:
Nothing is unboxed anymore, so this comment can go.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,58 +1948,83 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
val leftKeyTerm = newName(ctx, "leftKey")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
- |if ($leftTerm.size() == $rightTerm.size()) {
+ |int $sizeTerm = $leftTerm.size();
+ |if ($sizeTerm == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $sizeTerm; $indexTerm++) {
+ | $rightKeyIndexTerm.put(
+ | $keyGetterTerm.getElementOrNull($rightKeysTerm, $indexTerm),
+ | $indexTerm);
+ | }
|
- | ${valueEqualsExpr.code}
- | if (!${valueEqualsExpr.resultTerm}) {
- | $resultTerm = false;
- | break;
- | }
- | } else {
+ | for (int $leftIndexTerm = 0; $leftIndexTerm < $sizeTerm; $leftIndexTerm++) {
+ | Object $leftKeyTerm = $keyGetterTerm.getElementOrNull($leftKeysTerm, $leftIndexTerm);
+ | Integer $rightBoxedIndexTerm = (Integer) $rightKeyIndexTerm.get($leftKeyTerm);
Review Comment:
Duplicate keys break this: `MAP[a, 1, b, 2] = MAP[c, 1, d, 2]` with all keys 'x' is TRUE on master, FALSE here. Can you also index the left keys, compare the index sizes and add a test?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4277564572
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java:
##########
@@ -889,4 +891,190 @@ private Stream<TestSetSpec> mapUnionTestCases() {
"Invalid input arguments. Expected signatures are:\n"
+ "MAP_UNION(<COMMON>, <COMMON>...)"));
}
+
+ private Stream<TestSetSpec> mapEqualsTestCases() {
+ // In f1 the entries of f0 come in the opposite order, so the planner cannot merge or fold
+ // the two sides. f2 differs from f0 in one value.
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS, "MAP<STRING, DOUBLE>")
+ .onFieldsWithData(
+ orderedMap(A, 1.5, B, 2.5),
+ orderedMap(B, 2.5, A, 1.5),
+ orderedMap(A, 1.5, B, 1.5))
+ .andDataTypes(
+ MAP(STRING(), DOUBLE()),
+ MAP(STRING(), DOUBLE()),
+ MAP(STRING(), DOUBLE()))
+ .testResult(
+ resultSpec($("f0").isEqual($("f1")), "f0 = f1", true, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f1")), "f0 <> f1", false, BOOLEAN()),
+ resultSpec($("f0").isEqual($("f2")), "f0 = f2", false, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f2")), "f0 <> f2", true, BOOLEAN())),
+ // 7 is a cached Integer box, 5000 is not
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS, "MAP<STRING, INT>")
+ .onFieldsWithData(
+ orderedMap(A, 7, B, 5000),
+ orderedMap(B, 5000, A, 7),
+ orderedMap(A, 7, B, 6000))
+ .andDataTypes(
+ MAP(STRING(), INT()), MAP(STRING(), INT()), MAP(STRING(), INT()))
+ .testResult(
+ resultSpec($("f0").isEqual($("f1")), "f0 = f1", true, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f1")), "f0 <> f1", false, BOOLEAN()),
+ resultSpec($("f0").isEqual($("f2")), "f0 = f2", false, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f2")), "f0 <> f2", true, BOOLEAN())),
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.EQUALS, "MAP<STRING, DATE>")
+ .onFieldsWithData(
+ orderedMap(A, TEST_DATE_1, B, TEST_DATE_2),
+ orderedMap(B, TEST_DATE_2, A, TEST_DATE_1),
+ orderedMap(A, TEST_DATE_1, B, TEST_DATE_1))
+ .andDataTypes(
+ MAP(STRING(), DATE()), MAP(STRING(), DATE()), MAP(STRING(), DATE()))
+ .testResult(
+ resultSpec($("f0").isEqual($("f1")), "f0 = f1", true, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f1")), "f0 <> f1", false, BOOLEAN()),
+ resultSpec($("f0").isEqual($("f2")), "f0 = f2", false, BOOLEAN()),
+ resultSpec(
+ $("f0").isNotEqual($("f2")), "f0 <> f2", true, BOOLEAN())),
+ // a reference type, so unboxing leaves the values as they are
Review Comment:
removed
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4274720282
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1950,56 +1950,78 @@ object ScalarOperatorGens {
val resultTerm = newName(ctx, "compareResult")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
|if ($leftTerm.size() == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $rightKeysTerm.size(); $indexTerm++) {
Review Comment:
Addressed
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1950,56 +1950,78 @@ object ScalarOperatorGens {
val resultTerm = newName(ctx, "compareResult")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
|if ($leftTerm.size() == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $rightKeysTerm.size(); $indexTerm++) {
+ | $rightKeyIndexTerm.put(
+ | $keyGetterTerm.getElementOrNull($rightKeysTerm, $indexTerm),
+ | $indexTerm);
+ | }
|
- | ${valueEqualsExpr.code}
- | if (!${valueEqualsExpr.resultTerm}) {
- | $resultTerm = false;
- | break;
- | }
- | } else {
+ | for (int $leftIndexTerm = 0; $leftIndexTerm < $leftKeysTerm.size(); $leftIndexTerm++) {
Review Comment:
Correct and addressed
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4284155633
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,60 +1948,98 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val leftKeyIndexTerm = newName(ctx, "leftKeyIndex")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val entryTerm = newName(ctx, "entry")
+ val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
+ // A map can hold a key more than once, and then its last entry counts. Each key index keeps
+ // the last position of every key, so equal maps have key indexes of the same size whose
+ // values match at those positions.
Review Comment:
Done. The left map is now walked from the end
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4280150179
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,60 +1948,98 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val leftKeyIndexTerm = newName(ctx, "leftKeyIndex")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val entryTerm = newName(ctx, "entry")
+ val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
+ // A map can hold a key more than once, and then its last entry counts. Each key index keeps
+ // the last position of every key, so equal maps have key indexes of the same size whose
+ // values match at those positions.
val stmt =
s"""
|boolean $resultTerm;
- |if ($leftTerm.size() == $rightTerm.size()) {
+ |int $sizeTerm = $leftTerm.size();
+ |if ($sizeTerm == $rightTerm.size()) {
Review Comment:
can we have fast return and less indentation if sizes do not match?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4274649927
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1950,56 +1950,78 @@ object ScalarOperatorGens {
val resultTerm = newName(ctx, "compareResult")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
|if ($leftTerm.size() == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $rightKeysTerm.size(); $indexTerm++) {
+ | $rightKeyIndexTerm.put(
+ | $keyGetterTerm.getElementOrNull($rightKeysTerm, $indexTerm),
+ | $indexTerm);
+ | }
|
- | ${valueEqualsExpr.code}
- | if (!${valueEqualsExpr.resultTerm}) {
- | $resultTerm = false;
- | break;
- | }
- | } else {
+ | for (int $leftIndexTerm = 0; $leftIndexTerm < $leftKeysTerm.size(); $leftIndexTerm++) {
+ | Integer $rightBoxedIndexTerm = (Integer) $rightKeyIndexTerm.get(
+ | $keyGetterTerm.getElementOrNull($leftKeysTerm, $leftIndexTerm));
Review Comment:
why is it element or null?
it is index, correct?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4274627740
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1950,56 +1950,78 @@ object ScalarOperatorGens {
val resultTerm = newName(ctx, "compareResult")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
|if ($leftTerm.size() == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $rightKeysTerm.size(); $indexTerm++) {
+ | $rightKeyIndexTerm.put(
+ | $keyGetterTerm.getElementOrNull($rightKeysTerm, $indexTerm),
+ | $indexTerm);
+ | }
|
- | ${valueEqualsExpr.code}
- | if (!${valueEqualsExpr.resultTerm}) {
- | $resultTerm = false;
- | break;
- | }
- | } else {
+ | for (int $leftIndexTerm = 0; $leftIndexTerm < $leftKeysTerm.size(); $leftIndexTerm++) {
Review Comment:
if we are here `$leftKeysTerm.size()` should be equal to `$rightKeysTerm.size()` and we can reuse var correct?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4280131921
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,60 +1948,98 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val leftKeyIndexTerm = newName(ctx, "leftKeyIndex")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val entryTerm = newName(ctx, "entry")
+ val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
+ // A map can hold a key more than once, and then its last entry counts. Each key index keeps
+ // the last position of every key, so equal maps have key indexes of the same size whose
+ // values match at those positions.
Review Comment:
do we need this at all?
can we instead iterate from the end?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4274816795
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1950,56 +1950,78 @@ object ScalarOperatorGens {
val resultTerm = newName(ctx, "compareResult")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
|if ($leftTerm.size() == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $rightKeysTerm.size(); $indexTerm++) {
+ | $rightKeyIndexTerm.put(
+ | $keyGetterTerm.getElementOrNull($rightKeysTerm, $indexTerm),
+ | $indexTerm);
+ | }
|
- | ${valueEqualsExpr.code}
- | if (!${valueEqualsExpr.resultTerm}) {
- | $resultTerm = false;
- | break;
- | }
- | } else {
+ | for (int $leftIndexTerm = 0; $leftIndexTerm < $leftKeysTerm.size(); $leftIndexTerm++) {
+ | Integer $rightBoxedIndexTerm = (Integer) $rightKeyIndexTerm.get(
+ | $keyGetterTerm.getElementOrNull($leftKeysTerm, $leftIndexTerm));
Review Comment:
What if the key is a `NULL` key (Flink allows this)? Then we need o return `NULL`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #29481:
URL: https://github.com/apache/flink/pull/29481
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4277782820
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,58 +1948,83 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val leftIndexTerm = newName(ctx, "leftIndex")
val leftKeyTerm = newName(ctx, "leftKey")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
val stmt =
s"""
|boolean $resultTerm;
- |if ($leftTerm.size() == $rightTerm.size()) {
+ |int $sizeTerm = $leftTerm.size();
+ |if ($sizeTerm == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
|
- | for ($entryCls $entryTerm : $leftMapTerm.entrySet()) {
- | $keyCls $leftKeyTerm = ($keyCls) $entryTerm.getKey();
- | if ($rightMapTerm.containsKey($leftKeyTerm)) {
- | $valueCls $leftValueTerm = ($valueCls) $entryTerm.getValue();
- | $valueCls $rightValueTerm = ($valueCls) $rightMapTerm.get($leftKeyTerm);
- | boolean $leftValueNullTerm = ($leftValueTerm == null);
- | boolean $rightValueNullTerm = ($rightValueTerm == null);
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
+ | for (int $indexTerm = 0; $indexTerm < $sizeTerm; $indexTerm++) {
+ | $rightKeyIndexTerm.put(
+ | $keyGetterTerm.getElementOrNull($rightKeysTerm, $indexTerm),
+ | $indexTerm);
+ | }
|
- | ${valueEqualsExpr.code}
- | if (!${valueEqualsExpr.resultTerm}) {
- | $resultTerm = false;
- | break;
- | }
- | } else {
+ | for (int $leftIndexTerm = 0; $leftIndexTerm < $sizeTerm; $leftIndexTerm++) {
+ | Object $leftKeyTerm = $keyGetterTerm.getElementOrNull($leftKeysTerm, $leftIndexTerm);
+ | Integer $rightBoxedIndexTerm = (Integer) $rightKeyIndexTerm.get($leftKeyTerm);
Review Comment:
Good catch! Introduced a new index map and a check between the two index map of the keys. Added the test to cover the edge case.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4280659214
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,60 +1948,98 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val leftKeyIndexTerm = newName(ctx, "leftKeyIndex")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val entryTerm = newName(ctx, "entry")
+ val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
+ // A map can hold a key more than once, and then its last entry counts. Each key index keeps
+ // the last position of every key, so equal maps have key indexes of the same size whose
+ // values match at those positions.
val stmt =
s"""
|boolean $resultTerm;
- |if ($leftTerm.size() == $rightTerm.size()) {
+ |int $sizeTerm = $leftTerm.size();
+ |if ($sizeTerm == $rightTerm.size()) {
| $resultTerm = true;
- | $mapCls $leftMapTerm = $mapDataUtil
- | .convertToJavaMap($leftTerm, $keyTypeTerm, $valueTypeTerm);
- | $mapCls $rightMapTerm = $mapDataUtil
- | .convertToJavaMap($rightTerm, $keyTypeTerm, $valueTypeTerm);
+ | $ARRAY_DATA $leftKeysTerm = $leftTerm.keyArray();
+ | $ARRAY_DATA $leftValuesTerm = $leftTerm.valueArray();
+ | $ARRAY_DATA $rightKeysTerm = $rightTerm.keyArray();
+ | $ARRAY_DATA $rightValuesTerm = $rightTerm.valueArray();
+ |
+ | $mapCls $leftKeyIndexTerm = new $hashMapCls();
+ | $mapCls $rightKeyIndexTerm = new $hashMapCls();
Review Comment:
having maps here is the main issue
it will use java `equals` instead of SQL `equals`
we need to use `Equaliser`
see `generateRowComparison` for more details
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29481:
URL: https://github.com/apache/flink/pull/29481#discussion_r4284160183
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1948,60 +1948,98 @@ object ScalarOperatorGens {
val rightTerm = args(1)
val resultTerm = newName(ctx, "compareResult")
+ val sizeTerm = newName(ctx, "size")
val mapCls = className[java.util.Map[_, _]]
- val keyCls = boxedTypeTermForType(keyType)
- val valueCls = boxedTypeTermForType(valueType)
+ val hashMapCls = className[java.util.HashMap[_, _]]
+ val valueCls = primitiveTypeTermForType(valueType)
+ val valueDefault = primitiveDefaultValue(valueType)
- val leftMapTerm = newName(ctx, "leftMap")
- val leftKeyTerm = newName(ctx, "leftKey")
+ val keyGetterCls = classOf[ArrayData.ElementGetter].getCanonicalName
+ val keyGetterTerm =
+ ctx.addReusableObject(ArrayData.createElementGetter(keyType), "keyGetter", keyGetterCls)
+
+ val leftKeysTerm = newName(ctx, "leftKeys")
+ val leftValuesTerm = newName(ctx, "leftValues")
+ val rightKeysTerm = newName(ctx, "rightKeys")
+ val rightValuesTerm = newName(ctx, "rightValues")
+ val leftKeyIndexTerm = newName(ctx, "leftKeyIndex")
+ val rightKeyIndexTerm = newName(ctx, "rightKeyIndex")
+
+ val indexTerm = newName(ctx, "index")
+ val entryTerm = newName(ctx, "entry")
+ val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
+ val leftIndexTerm = newName(ctx, "leftIndex")
+ val rightIndexTerm = newName(ctx, "rightIndex")
+ val rightBoxedIndexTerm = newName(ctx, "rightBoxedIndex")
+
+ // The values are read in place from the value arrays. Reading them from a java.util.Map
+ // would box them, and generateEquals compares boxes by reference.
val leftValueTerm = newName(ctx, "leftValue")
val leftValueNullTerm = newName(ctx, "leftValueIsNull")
- val leftValueExpr =
- GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
+ val leftValueExpr = GeneratedExpression(leftValueTerm, leftValueNullTerm, "", valueType)
- val rightMapTerm = newName(ctx, "rightMap")
val rightValueTerm = newName(ctx, "rightValue")
val rightValueNullTerm = newName(ctx, "rightValueIsNull")
- val rightValueExpr =
- GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
+ val rightValueExpr = GeneratedExpression(rightValueTerm, rightValueNullTerm, "", valueType)
- val entryTerm = newName(ctx, "entry")
- val entryCls = classOf[java.util.Map.Entry[AnyRef, AnyRef]].getCanonicalName
val valueEqualsExpr =
generateEquals(ctx, leftValueExpr, rightValueExpr, new BooleanType(valueType.isNullable))
- val internalTypeCls = classOf[LogicalType].getCanonicalName
- val keyTypeTerm = ctx.addReusableObject(keyType, "keyType", internalTypeCls)
- val valueTypeTerm = ctx.addReusableObject(valueType, "valueType", internalTypeCls)
- val mapDataUtil = className[MapDataUtil]
-
+ // A map can hold a key more than once, and then its last entry counts. Each key index keeps
+ // the last position of every key, so equal maps have key indexes of the same size whose
+ // values match at those positions.
val stmt =
s"""
|boolean $resultTerm;
- |if ($leftTerm.size() == $rightTerm.size()) {
+ |int $sizeTerm = $leftTerm.size();
+ |if ($sizeTerm == $rightTerm.size()) {
Review Comment:
Done, I am breaking out faster
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40989][table-planner] Compare MAP values by value instead of by reference [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29481:
URL: https://github.com/apache/flink/pull/29481#issuecomment-6172196115
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "11e2a08aee2136243dc0c478aba990c41fa19146",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "11e2a08aee2136243dc0c478aba990c41fa19146",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 11e2a08aee2136243dc0c478aba990c41fa19146 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org