[FLINK-17321][table] Add support casting of map to map and multiset to multiset #18287

snuyanzin · 2022-01-06T13:00:03Z

What is the purpose of the change

This PR adds support of casting maps to maps and multisets to multisets

Brief change log

Casting rule org.apache.flink.table.planner.functions.casting.MapToMapAndMultisetToMultisetCastRule
Tests in org.apache.flink.table.planner.functions.casting.CastRulesTest

Verifying this change

This change added tests and can be verified as follows:

Added tests in CastRulesTest that validate casting of maps to maps, multisets to multisets

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): (no)
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)

…o multiset

flinkbot · 2022-01-06T13:02:52Z

CI report:

9909643 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flinkbot · 2022-01-06T13:06:01Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit a65fb5e (Thu Jan 06 13:06:01 UTC 2022)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

slinkydeveloper

I did a quick scan, and it looks good. Will do a proper review later.

Just one question, can you add one single test case for map to map and one for multiset to multiset in CastFunctionITCase, as it tests that it works through the stack? Because I'm pretty sure it doesn't, as you need to change LogicalTypeCasts as well.

.../org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java

…Case

snuyanzin · 2022-01-06T15:01:11Z

I added test for map and multiset to CastFunctionITCase and for map it is ok while for multiset it fails with

rg.apache.flink.table.api.ValidationException: Could not cast the value of the 1 column: [ map(1, 1) ] of a row: [ null, map(1, 1) ] to the requested type: MULTISET<INT>

	at org.apache.flink.table.operations.utils.ValuesOperationFactory.lambda$null$4(ValuesOperationFactory.java:130)
	at java.util.Optional.orElseThrow(Optional.java:290)
	at org.apache.flink.table.operations.utils.ValuesOperationFactory.lambda$convertTopLevelExpressionToExpectedRowType$5(ValuesOperationFactory.java:127)
	at java.util.stream.IntPipeline$4$1.accept(IntPipeline.java:250)
	at java.util.stream.Streams$RangeIntSpliterator.forEachRemaining(Streams.java:110)

It seems the reason is not LogicalTypeCasts but absent converter for multiset here

flink/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/ValuesOperationFactory.java

Line 190 in c5581b8

if (sourceExpression instanceof CallExpression) {

I will have a look at it later today

slinkydeveloper · 2022-01-06T15:07:07Z

Hey @snuyanzin have you checked this issue? There might be something helpful for you in the patch i posted there https://issues.apache.org/jira/browse/FLINK-25428.

If the multiset test is a blocker for you, and there's no easy fix for that, you can ignore it now, as it's still not well supported by the whole stack, so perhaps you can just leave it the test there commented.

snuyanzin · 2022-01-06T23:59:12Z

@slinkydeveloper thanks for pointing the issue, I had a look however it seems it does not help with current multiset issue.
After some debugging I noticed there is a lack of support of multisets during resolving of multiset expressions.
I added some things to make casting working.
Not sure if it makes sense to have under this issue or under a separate one

slinkydeveloper · 2022-01-07T10:02:14Z

@snuyanzin I rather prefer that the multiset e2e casting gets worked out in a separate issue, as it requires significant more testing to check that it works throughout the stack, and as you have shown in the commits, it also requires new function definitions to expose it to the user. That's something we should rather address in a separate context.

As a goal for this task i would say we want:

CastRulesTest tests for array to array, row to row, map to map and multiset to multiset
CastFunctionITCase tests for array to array, row to row and map to map.

snuyanzin · 2022-01-07T14:18:27Z

ok, I removed multiset related fix from this PR and created a separate issue for that https://issues.apache.org/jira/browse/FLINK-25567

Now it should match the goal, please let me know if not

slinkydeveloper

I think we're good, but there's a mystery here worth to be investigated before merging 😄

slinkydeveloper · 2022-01-07T15:11:14Z

...table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java

+                CastTestSpecBuilder.testCastTo(MAP(STRING(), STRING()))
+                        .fromCase(MAP(FLOAT(), DOUBLE()), null, null)
+                        .fromCase(
+                                MAP(INT(), INT()),
+                                Collections.singletonMap(1, 2),
+                                Collections.singletonMap("1", "2"))


🤔 I'm surprised that this works without changing anything in LogicalTypeCasts... Can you add a test case for failure? For example:

CastTestSpecBuilder .testCastTo(MAP(STRING(), STRING())) .fail(MAP(STRING(), ROW(INT())), whateverValue)

Same for ROW to ROW and ARRAY to ARRAY. Just take whatever inner invalid tuple and see what exception you get. If it fails within the single CastRule implementations, then something is wrong with LogicalTypeCasts.

I didn't get, should it fail?

SELECT CAST(MAP['a', row(1)] AS MAP<STRING, STRING>);

I thought it should return something like that

{a=(1)}

It should not fail, the goal of this issue is to allow such casting. The thing is that I'm actually surprised it doesn't fail without touching LogicalTypeCasts, which is the class that checks valid/invalid casts.

EDIT: I've done some debugging, LogicalTypeCasts#supportsConstructedCasting does the job and just checks for the equality of the type root and for the castability of the children. So it works 😄

twalthr

Sorry, for jumping in so late.

twalthr · 2022-01-18T09:46:13Z

.../org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java

+                                                    index, valueArrayTerm, innerInputValueType),
+                                            "false",
+                                            // Null check is done at the array access level
+                                            innerInputValueType.copy(false),


@slinkydeveloper didn't we change this behavior recently and use a new method that deals with NullType?

yep, perhaps can you fix it when merging?

twalthr · 2022-01-18T09:57:44Z

.../org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java

+        final String value = newName("value");
+
+        return new CastRuleUtils.CodeWriter()
+                .declStmt(className(Map.class), map, constructorCall(HashMap.class))


I'm not sure if a HashMap is safe to use because not all internal data structures have proper hashCode/equals. By looking at org.apache.flink.table.planner.codegen.calls.ScalarOperatorGens#generateMap, I also don't see any HashMap. A safer solution sounds like reusing the array to array casting logic and apply it keyArray and valueArray? In general, we don't guarantee unique keys for maps.

I think this is safe at this level, as the input map is assumed to be valid and should not have multiple values for the same key anyway.

A cast to MAP<CHAR(1), X> could produce multiple keys. But after a offline discussion, I think it is safe to use a HashMap here and the problem is rather in the ScalarOperatorGens#generateMap.

twalthr

I will apply the remaining issue while merging. Thanks @snuyanzin.

twalthr · 2022-01-18T10:27:58Z

...-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala

-      case MAP_VALUE_CONSTRUCTOR | MULTISET_VALUE =>
-        generateMapOrMultiset(ctx, resultType, operands)
+      // maps
+      case MAP_VALUE_CONSTRUCTOR =>


@snuyanzin Looking forward to a PR for this commit. Maybe you can also address the comment about the value constructor in this PR (using HashMap for deduplication of key).

hi @twalthr , thanks for highlighting this
yes I will try to handle this

…o multiset This closes apache#18287.

[FLINK-17321][table] Add support casting of map to map and multiset t…

a65fb5e

…o multiset

slinkydeveloper reviewed Jan 6, 2022

View reviewed changes

.../org/apache/flink/table/planner/functions/casting/MapToMapAndMultisetToMultisetCastRule.java Outdated Show resolved Hide resolved

[FLINK-17321][table] fixup

6f663ca

rmetzger added the component=TableSQL/Runtime label Jan 6, 2022

[FLINK-17321][table] Add tests for map and multiset to CastFunctionIT…

c6b7bc3

…Case

[FLINK-17321][table] Fix multisets

2d16266

[FLINK-17321] Multisets fix should be in a separate pr

9909643

slinkydeveloper suggested changes Jan 7, 2022

View reviewed changes

slinkydeveloper approved these changes Jan 10, 2022

View reviewed changes

twalthr requested changes Jan 18, 2022

View reviewed changes

twalthr approved these changes Jan 18, 2022

View reviewed changes

twalthr reviewed Jan 18, 2022

View reviewed changes

twalthr closed this in f5c99c6 Jan 18, 2022

niklassemmler pushed a commit to niklassemmler/flink that referenced this pull request Feb 3, 2022

[FLINK-17321][table] Add support casting of map to map and multiset t…

7042f96

…o multiset This closes apache#18287.

snuyanzin mentioned this pull request May 4, 2022

[FLINK-25567][table] Add support casting of multisets to multisets #19637

Open

snuyanzin deleted the flink17321 branch May 5, 2022 11:13

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17321][table] Add support casting of map to map and multiset to multiset #18287

[FLINK-17321][table] Add support casting of map to map and multiset to multiset #18287

snuyanzin commented Jan 6, 2022

flinkbot commented Jan 6, 2022 •

edited

flinkbot commented Jan 6, 2022

slinkydeveloper left a comment

snuyanzin commented Jan 6, 2022

slinkydeveloper commented Jan 6, 2022

snuyanzin commented Jan 6, 2022 •

edited

slinkydeveloper commented Jan 7, 2022 •

edited

snuyanzin commented Jan 7, 2022

slinkydeveloper left a comment

slinkydeveloper Jan 7, 2022

snuyanzin Jan 7, 2022

slinkydeveloper Jan 10, 2022

twalthr left a comment

twalthr Jan 18, 2022

slinkydeveloper Jan 18, 2022

twalthr Jan 18, 2022

slinkydeveloper Jan 18, 2022 •

edited

twalthr Jan 18, 2022 •

edited

twalthr left a comment

twalthr Jan 18, 2022

snuyanzin Jan 19, 2022

[FLINK-17321][table] Add support casting of map to map and multiset to multiset #18287

[FLINK-17321][table] Add support casting of map to map and multiset to multiset #18287

Conversation

snuyanzin commented Jan 6, 2022

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jan 6, 2022 • edited

CI report:

flinkbot commented Jan 6, 2022

Automated Checks

Review Progress

slinkydeveloper left a comment

Choose a reason for hiding this comment

snuyanzin commented Jan 6, 2022

slinkydeveloper commented Jan 6, 2022

snuyanzin commented Jan 6, 2022 • edited

slinkydeveloper commented Jan 7, 2022 • edited

snuyanzin commented Jan 7, 2022

slinkydeveloper left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

twalthr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

slinkydeveloper Jan 18, 2022 • edited

Choose a reason for hiding this comment

twalthr Jan 18, 2022 • edited

Choose a reason for hiding this comment

twalthr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Jan 6, 2022 •

edited

snuyanzin commented Jan 6, 2022 •

edited

slinkydeveloper commented Jan 7, 2022 •

edited

slinkydeveloper Jan 18, 2022 •

edited

twalthr Jan 18, 2022 •

edited