You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "hanyuzheng7 (via GitHub)" <gi...@apache.org> on 2023/06/21 15:03:27 UTC
[GitHub] [flink] hanyuzheng7 opened a new pull request, #22842: [FLINK-32261]-table-Add-MAP_UNION-function
hanyuzheng7 opened a new pull request, #22842:
URL: https://github.com/apache/flink/pull/22842
Description:
The current implementation of the map_union method in the Flink library does not provide a way to combine two dictionaries into a single dictionary. This enhancement aims to add this functionality, allowing users to merge dictionaries efficiently.
Syntax:
map_union[map1: map, map2: map] -> map
Arguments:
map1:The first map to be merged.
map2:The second map to be merged.
Returns: A new map that contains the combined key-value pairs from map1 and map2. If there are any overlapping keys, the value from map2 will overwrite the value from map1.
Examples:
Merging maps with unique keys:
map1 = ['a': 1, 'b': 2] map2 = ['c': 3, 'd': 4] result = map_union(map1, map2) Output: ['a': 1, 'b': 2, 'c': 3, 'd': 4]
Merging maps with overlapping keys:
map1 = ['a': 1, 'b': 2] dict2 = ['b': 3, 'c': 4] result = map_union[dict1, dict2] Output: ['a': 1, 'b': 3, 'c': 4] See also:
prestodb: https://prestodb.io/docs/current/functions/aggregate.html
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)*
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
- The S3 file system connector: (yes / no / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242783033
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
`HashMap` under the hood uses java's `equals` which is not 100% same as SQL.
We need to use SQL with `equalityEvaluator` like for instance done for `ARRAY_DISTINCT`
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1629714547
@snuyanzin support support varargs now
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1643174801
@flinkbot run azure
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1622492472
@flinkbot run azure
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252368137
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
`Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have a common map type. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1621039450
@snuyanzin , do you know why test_ci misc report error?
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1620637342
@flinkbot run azure
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #22842:
URL: https://github.com/apache/flink/pull/22842
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 closed pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
URL: https://github.com/apache/flink/pull/22842
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1503890234
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java:
##########
@@ -368,6 +369,14 @@ public static InputTypeStrategy commonMultipleArrayType(int minCount) {
/** @see ItemAtIndexArgumentTypeStrategy */
public static final ArgumentTypeStrategy ITEM_AT_INDEX = new ItemAtIndexArgumentTypeStrategy();
+ /**
+ * An {@link InputTypeStrategy} that expects {@code count} arguments that have a common map
+ * type.
+ */
+ public static InputTypeStrategy commonMapType(int count) {
Review Comment:
looks like `count` is not used here
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1507128713
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
Review Comment:
Ok, I will extracted it.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ if (key == null) {
+ isKeyNullExist = true;
+ }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(map2.valueArray(), i));
Review Comment:
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1244442659
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
Review Comment:
ok. Thank you.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242763657
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
Review Comment:
no need for new map creation if one of the maps is empty
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242767973
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1528,6 +1528,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ This two maps should have same data structure. If there are overlapping keys,
Review Comment:
```suggestion
These two maps should have same data structure. If there are overlapping keys,
```
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242863699
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
Can you please elaborate where have you seen guarantees that binaryData's equals is in sync with SQL's sync?
Also there is no limitations to use non binarydata base on current method signature, so we can not rely on it
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252368976
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
you mean generate a CommonCollectionInputTypeStrategy and let CommonMapInputTypeStrategy and CommonArrayInputTypeStrategy become static class, and we can call them through CommonCollectionInputTypeStrategy?
but CommonMapInputTypeStrategy is a normal class which doesn't need implements InputTypeStrategy, why we don't call it CommonCollectionInputTypeStrategyUtils?
And we just move two class in a new class, do we really need to do this? I think it is ok to maintain two class now.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252365645
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
I would change this sentence ` These two maps are common map type. ` in a way
` These two maps should have a common map type. `
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242867857
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
To be honest, I don't know API and SQL there are different. Why binaryData's equals is not sync with SQL's sync.
But If we transfer the input Data to the Expression, the data we received in MapUnionFunction are binaryMap.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242768691
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -1423,6 +1424,16 @@ public OutType mapEntries() {
return toApiSpecificExpression(unresolvedCall(MAP_ENTRIES, toExpr()));
}
+ /**
+ * Returns a map created by merging two maps, 'map1' and 'map2'. This two maps should have same
Review Comment:
```suggestion
* Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same
```
since it is plural
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242839050
##########
docs/data/sql_functions.yml:
##########
@@ -649,6 +649,9 @@ collection:
- sql: MAP_KEYS(map)
table: MAP.mapKeys()
description: Returns the keys of the map as array. No order guaranteed.
+ SQL: MAP_UNION(map1, map2)
+ Table: map1.mapUnion(map2)
+ Description: Returns a map created by merging two maps, 'map1' and 'map2'. This two maps should have same data structure. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps are null, return null.
Review Comment:
I will fix it. Thank you.
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1652083503
@flinkbot run azure
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1621163864
@snuyanzin solve all comments.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1251168614
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ try {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ if (map1.size() == 0 && map2.size() == 0) {
+ return map1;
+ }
+ if (map1.size() == 0) {
+ return map2;
+ }
+ if (map2.size() == 0) {
+ return map1;
+ }
Review Comment:
i guess it could be simplified with reducing number of checks like
```suggestion
if (map1.size() == 0) {
if (map2.size() == 0) {
return map1;
}
return map2;
}
if (map2.size() == 0) {
return map1;
}
```
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252558140
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
ok, I will try it.
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1984944407
@snuyanzin Ci test pass now.
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1503826435
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ if (key == null) {
+ isKeyNullExist = true;
+ }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(map2.valueArray(), i));
+ }
+
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key1 = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value1 = valueElementGetter.getElementOrNull(map1.valueArray(), i);
Review Comment:
same
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1644769938
@flinkbot run azure
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242849867
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
Review Comment:
if both are empty just return any of them.
Returning a new empty map creates new object, so this is not the best option
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242823684
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
but if use equalityEvaluator, the time complexity will become to n square. because we cannot use hashmap. Is that 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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242838442
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -1423,6 +1424,16 @@ public OutType mapEntries() {
return toApiSpecificExpression(unresolvedCall(MAP_ENTRIES, toExpr()));
}
+ /**
+ * Returns a map created by merging two maps, 'map1' and 'map2'. This two maps should have same
Review Comment:
I will fix it. Thank you.
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1528,6 +1528,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ This two maps should have same data structure. If there are overlapping keys,
Review Comment:
I will fix it. Thank you.
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1610617097
@snuyanzin I have already solve all the comments, could you please do the second code review. Thank you.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252355890
##########
docs/data/sql_functions_zh.yml:
##########
@@ -760,7 +760,9 @@ collection:
- sql: MAP_FROM_ARRAYS(array_of_keys, array_of_values)
table: mapFromArrays(array_of_keys, array_of_values)
description: Returns a map created from an arrays of keys and values. Note that the lengths of two arrays should be the same.
-
+ - sql: MAP_UNION(map1, map2)
+ table: map1.mapUnion(map2)
+ description: 返回一个通过合并两个图 'map1' 和 'map2' 创建的图。这两个图应该有相同的数据结构。如果有重叠的键，'map2' 的值将覆盖 'map1' 的值。如果任一图为空，则返回 null。
Review Comment:
there should be an empty line before next key which is `json`
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252343611
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
`Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same type for their key and value individually. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, return null.`
What I want to say is that two maps' key and value should have same data type.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252516691
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
it could be implemented in different ways, the main idea is to extract common logic into one place and reuse it
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252336736
##########
docs/data/sql_functions.yml:
##########
@@ -658,6 +658,9 @@ collection:
- sql: MAP_KEYS(map)
table: MAP.mapKeys()
description: Returns the keys of the map as array. No order guaranteed.
+ - sql: MAP_UNION(map1, map2)
+ table: map1.mapUnion(map2)
+ description: Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same data structure. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, return null.
Review Comment:
it's not clear what is same datastructure in SQL context...
i guess same type (or not same, however common)
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -1468,6 +1469,16 @@ public OutType mapEntries() {
return toApiSpecificExpression(unresolvedCall(MAP_ENTRIES, toExpr()));
}
+ /**
+ * Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same
+ * data structure. If there are overlapping keys, the value from 'map2' will overwrite the value
Review Comment:
it's not clear what is same datastructure in SQL context...
i guess same type (or not same, however common)
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252647333
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
ok, change it.
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1963080316
I will try to have a look this week
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1507132219
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ if (key == null) {
+ isKeyNullExist = true;
+ }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(map2.valueArray(), i));
+ }
+
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key1 = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value1 = valueElementGetter.getElementOrNull(map1.valueArray(), i);
Review Comment:
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1985133263
@snuyanzin by mistake.
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1984943907
> @hanyuzheng7 could you please have at build failure? it seems it is related to the recent changes
>
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=57976&view=logs&j=0c940707-2659-5648-cbe6-a1ad63045f0a&t=075c2716-8010-5565-fe08-3c4bb45824a4
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1989110545
Ok, Thank you @snuyanzin
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1622756809
@snuyanzin do you know why Test - table report error?
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252374401
##########
docs/data/sql_functions_zh.yml:
##########
@@ -760,7 +760,9 @@ collection:
- sql: MAP_FROM_ARRAYS(array_of_keys, array_of_values)
table: mapFromArrays(array_of_keys, array_of_values)
description: Returns a map created from an arrays of keys and values. Note that the lengths of two arrays should be the same.
-
+ - sql: MAP_UNION(map1, map2)
+ table: map1.mapUnion(map2)
+ description: 返回一个通过合并两个图 'map1' 和 'map2' 创建的图。这两个图应该有相同的数据结构。如果有重叠的键，'map2' 的值将覆盖 'map1' 的值。如果任一图为空，则返回 null。
Review Comment:
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1625929844
@snuyanzin do you think it is ok now? I let CommonMapInputTypeStrategy and CommonArrayInputTypeStrategy as subclass extends CommonCollectionInputTypeStrategy.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242823684
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
But using equalityEvaluator, the time complexity will become to n square, because we cannot use hashmap. Is that 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1613986096
@snuyanzin can you help me do another code review. Thank you.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242836983
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
Review Comment:
I can add a if statement, if one map is empty, I can directly return the non-empty one.
if two maps are empty, I will return the first map. or if two maps are empty, I return a new empty map.
which one is better?
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242809920
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
Because Two map should have same data structure. their key type should same, their value type also should same. we cannot union two maps which data structure are different.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242856919
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value = valueElementGetter.getElementOrNull(map1.valueArray(), i);
+ this.map.put(key, value);
+ }
+ for (int i = 0; i < map2.size(); i++) {
+ final Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ final Object value = valueElementGetter.getElementOrNull(map2.valueArray(), i);
+ this.map.put(key, value);
+ }
+ this.updateArrays();
+ }
+
+ @Override
+ public int size() {
+ return map.size();
+ }
+
+ @Override
+ public ArrayData keyArray() {
+ return keysArray;
+ }
+
+ @Override
+ public ArrayData valueArray() {
+ return valuesArray;
+ }
+
+ private void updateArrays() {
+ this.keysArray = new GenericArrayData(map.keySet().toArray());
Review Comment:
If we use equalityEvaluator, we will update map many times. So I think we should keep it now.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
Review Comment:
I will fix it.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242850416
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
And I check the source code. if the data can pass to this function, they are all binaryData, so we can use java's equal.
--
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
[GitHub] [flink] hanyuzheng7 closed pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 closed pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
URL: https://github.com/apache/flink/pull/22842
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252343611
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
`Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same type. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, return null.`
What I want to say is that two maps' key and value should have same data type.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252222157
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ try {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ if (map1.size() == 0) {
+ if (map2.size() == 0) {
+ return map1;
+ }
+ return map2;
+ }
+ if (map2.size() == 0) {
+ return map1;
+ }
+ return new MapDataForMapUnion(map1, map2);
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+
+ for (int i = 0; i < map1.size(); i++) {
+ keysList.add(keyElementGetter.getElementOrNull(map1.keyArray(), i));
+ valuesList.add(valueElementGetter.getElementOrNull(map1.valueArray(), i));
+ }
+
+ for (int i = 0; i < map2.size(); i++) {
+ final Object key2 = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ final Object value2 = valueElementGetter.getElementOrNull(map2.valueArray(), i);
+
+ boolean keyExists = false;
+ if (key2 != null) {
+ for (int j = 0; j < keysList.size(); j++) {
+ final Object key1 = keysList.get(j);
+ if (key1 != null && (boolean) keyEqualityHandle.invoke(key1, key2)) {
+ // If key exists, update the value with the value from the second map
+ valuesList.set(j, value2);
Review Comment:
good suggestion, thank you, I will change it.
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1620966114
@flinkbot run azure
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252368976
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
you mean generate a CommonCollectionInputTypeStrategy and let CommonMapInputTypeStrategy and CommonArrayInputTypeStrategy become static class, and we can call them through CommonCollectionInputTypeStrategy?
but CommonMapInputTypeStrategy is a normal class which doesn't need implements InputTypeStrategy, why we don't call it CommonCollectionInputTypeStrategyUtils?
--
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
[GitHub] [flink] snuyanzin commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1639962674
@hanyuzheng7 could you please make ci for this PR green?
right now it is still failing
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1984099455
@flinkbot run azure
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1503885866
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ if (key == null) {
+ isKeyNullExist = true;
+ }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(map2.valueArray(), i));
+ }
+
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key1 = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value1 = valueElementGetter.getElementOrNull(map1.valueArray(), i);
Review Comment:
it seems we need `value1` only if key is not present
should we move it under corresponding `if` to make it lazy?
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1507136138
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java:
##########
@@ -368,6 +369,14 @@ public static InputTypeStrategy commonMultipleArrayType(int minCount) {
/** @see ItemAtIndexArgumentTypeStrategy */
public static final ArgumentTypeStrategy ITEM_AT_INDEX = new ItemAtIndexArgumentTypeStrategy();
+ /**
+ * An {@link InputTypeStrategy} that expects {@code count} arguments that have a common map
+ * type.
+ */
+ public static InputTypeStrategy commonMapType(int count) {
Review Comment:
change it to minCount.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242773854
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
This class looks same as `org.apache.flink.table.types.inference.strategies.CommonInputTypeStrategy` (except class name).
Why do we need it?
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242874089
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
I check the isEqual method in ARRAY_DISTINCT, but I cannot find it's source code. So I don't know the background logical of IsEqual.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1244606082
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value = valueElementGetter.getElementOrNull(map1.valueArray(), i);
+ this.map.put(key, value);
+ }
+ for (int i = 0; i < map2.size(); i++) {
+ final Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ final Object value = valueElementGetter.getElementOrNull(map2.valueArray(), i);
+ this.map.put(key, value);
+ }
+ this.updateArrays();
+ }
+
+ @Override
+ public int size() {
+ return map.size();
+ }
+
+ @Override
+ public ArrayData keyArray() {
+ return keysArray;
+ }
+
+ @Override
+ public ArrayData valueArray() {
+ return valuesArray;
+ }
+
+ private void updateArrays() {
+ this.keysArray = new GenericArrayData(map.keySet().toArray());
Review Comment:
fixed it.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242829756
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
IMHO first we should care about correctness and only then about speed
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1251168614
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ try {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ if (map1.size() == 0 && map2.size() == 0) {
+ return map1;
+ }
+ if (map1.size() == 0) {
+ return map2;
+ }
+ if (map2.size() == 0) {
+ return map1;
+ }
Review Comment:
i guess it could be simplified with reducing number of checks like
```suggestion
if (map1.size() == 0) {
if (map1.size() == 0) {
return map2;
}
return map1;
}
if (map2.size() == 0) {
return map1;
}
```
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1251547228
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ try {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ if (map1.size() == 0) {
+ if (map2.size() == 0) {
+ return map1;
+ }
+ return map2;
+ }
+ if (map2.size() == 0) {
+ return map1;
+ }
+ return new MapDataForMapUnion(map1, map2);
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+
+ for (int i = 0; i < map1.size(); i++) {
+ keysList.add(keyElementGetter.getElementOrNull(map1.keyArray(), i));
+ valuesList.add(valueElementGetter.getElementOrNull(map1.valueArray(), i));
+ }
+
+ for (int i = 0; i < map2.size(); i++) {
+ final Object key2 = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ final Object value2 = valueElementGetter.getElementOrNull(map2.valueArray(), i);
+
+ boolean keyExists = false;
+ if (key2 != null) {
+ for (int j = 0; j < keysList.size(); j++) {
+ final Object key1 = keysList.get(j);
+ if (key1 != null && (boolean) keyEqualityHandle.invoke(key1, key2)) {
+ // If key exists, update the value with the value from the second map
+ valuesList.set(j, value2);
Review Comment:
nit: if you start `MapDataForMapUnion` with `map2` you don't neet update opeartion
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ try {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ if (map1.size() == 0) {
+ if (map2.size() == 0) {
+ return map1;
+ }
+ return map2;
+ }
+ if (map2.size() == 0) {
+ return map1;
+ }
+ return new MapDataForMapUnion(map1, map2);
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+
+ for (int i = 0; i < map1.size(); i++) {
+ keysList.add(keyElementGetter.getElementOrNull(map1.keyArray(), i));
+ valuesList.add(valueElementGetter.getElementOrNull(map1.valueArray(), i));
+ }
+
+ for (int i = 0; i < map2.size(); i++) {
+ final Object key2 = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ final Object value2 = valueElementGetter.getElementOrNull(map2.valueArray(), i);
+
+ boolean keyExists = false;
+ if (key2 != null) {
+ for (int j = 0; j < keysList.size(); j++) {
+ final Object key1 = keysList.get(j);
+ if (key1 != null && (boolean) keyEqualityHandle.invoke(key1, key2)) {
+ // If key exists, update the value with the value from the second map
+ valuesList.set(j, value2);
Review Comment:
nit: if you start `MapDataForMapUnion` with `map2` you don't need update opeartion
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252357094
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
Can we create a `CommonCollectionInputTypeStrategy` class with this logic
and 2 subclasses `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` which customize it a bit.
I'm asking since `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` are almost identical
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
Can we create a `CommonCollectionInputTypeStrategy` class with this logic
and 2 subclasses `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` which customize it a bit?
I'm asking since `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` are almost identical
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252363975
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
`Returns a map created by merging two maps, 'map1' and 'map2'. These two maps are common map type. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, return null.
is this 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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252368976
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
you mean generate a CommonCollectionInputTypeStrategy and let CommonMapInputTypeStrategy and CommonArrayInputTypeStrategy become static class, and we can call them through CommonCollectionInputTypeStrategy?
but CommonMapInputTypeStrategy is a normal class which doesn't need implements InputTypeStrategy, why we don't call it CommonCollectionInputTypeStrategyUtils?
And we just move two class in a new class, do we really need to do this? I think it is ok to maintain two class now.
@snuyanzin
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1256335928
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
> Can we create a `CommonCollectionInputTypeStrategy` class with this logic and 2 subclasses `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` which customize it a bit?
>
> I'm asking since `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` are almost identical
@snuyanzin I have already achieved it. Do you think the change is 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1652682664
@snuyanzin ci test pass now
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1985127583
@hanyuzheng7 did you close it by mistake or there is a reason?
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1503804242
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
Review Comment:
first element also could be null, see trace in comments
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1503819763
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
Review Comment:
I wonder whether we need this check?
If in doc we mention that it should return null if any of maps is null then IMHO once we encountered the first null map we don't need to check the rest and can immediatly return null. Or did I miss anything?
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1959740514
@snuyanzin can you help me do a code review again, Thank you!
--
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
[GitHub] [flink] flinkbot commented on pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1601024848
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4f26f9375e99db7765ce6a4bf6d161bc8ef4899e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4f26f9375e99db7765ce6a4bf6d161bc8ef4899e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4f26f9375e99db7765ce6a4bf6d161bc8ef4899e 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1618981779
>
Is it legal for the key of the map to be 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1622069560
@snuyanzin , do you think the code is ok now? If ok, I can remove CommonMapInputTypeStrategy and CommonArrayInputTypeStrategy these two Class.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252368976
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
you mean generate a CommonCollectionInputTypeStrategy and let CommonMapInputTypeStrategy and CommonArrayInputTypeStrategy become static class, and we can call them through CommonCollectionInputTypeStrategy?
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252363975
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
`Returns a map created by merging two maps, 'map1' and 'map2'. These two maps are common map type. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, return null.`
is this 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1619462048
> This query fails with NullPointerException
>
> ```sql
> SELECT map_union(MAP[1, 2], MAP[CAST(NULL AS INT), 3]);
> ```
key = CAST(NULL AS INT) is ok now.
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252558140
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
ok, I will try it, by the way, I have a question about pyflink function defintion, the ci test said
```
@property
1562: def map_union(self, map2) -> 'Expression':
"""
Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have a
common map type. If there are overlapping keys, the value from 'map2' will overwrite the
value from 'map1'. If any of maps is null, return null.
.. seealso:: :py:attr:`~Expression.map_union`
"""
return _binary_op("mapUnion")(self, map2)
```
` pyflink/table/expression.py:1562: error: Too many arguments`
do you know the reason?
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242769137
##########
docs/data/sql_functions.yml:
##########
@@ -649,6 +649,9 @@ collection:
- sql: MAP_KEYS(map)
table: MAP.mapKeys()
description: Returns the keys of the map as array. No order guaranteed.
+ SQL: MAP_UNION(map1, map2)
+ Table: map1.mapUnion(map2)
+ Description: Returns a map created by merging two maps, 'map1' and 'map2'. This two maps should have same data structure. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps are null, return null.
Review Comment:
```suggestion
Description: Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same data structure. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps are null, 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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1251458008
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ try {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ if (map1.size() == 0 && map2.size() == 0) {
+ return map1;
+ }
+ if (map1.size() == 0) {
+ return map2;
+ }
+ if (map2.size() == 0) {
+ return map1;
+ }
Review Comment:
ok, thank you, fixed it.
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1619461436
@snuyanzin key = CAST(NULL AS INT) is ok now.
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1620575480
@snuyanzin There are some problems with rebase, this pr is closed for some reason, please go to the new rebased PR.[
](https://github.com/apache/flink/pull/22950)
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252368976
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
remove CommonArrayInputTypeStrategy may affect other code?
--
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
[GitHub] [flink] snuyanzin commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1621047910
looks like because of this https://issues.apache.org/jira/browse/FLINK-32539
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1256335928
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CommonMapInputTypeStrategy.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** An {@link InputTypeStrategy} that expects that all arguments have a common map type. */
+@Internal
+public final class CommonMapInputTypeStrategy implements InputTypeStrategy {
Review Comment:
> Can we create a `CommonCollectionInputTypeStrategy` class with this logic and 2 subclasses `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` which customize it a bit?
>
> I'm asking since `CommonMapInputTypeStrategy` and `CommonArrayInputTypeStrategy` are almost identical
@snuyanzin I have already achieved it. Do you think the change is 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1625673575
@snuyanzin do you think it's ok now?
--
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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1644803737
@snuyanzin It seem that e2e_1_ci has error, it report
```
Jul 20 23:47:21 * TCP_NODELAY set
Jul 20 23:47:21 * connect to ::1 port 8081 failed: Connection refused
Jul 20 23:47:21 * Trying 127.0.0.1:8081...
Jul 20 23:47:21 * TCP_NODELAY set
Jul 20 23:47:21 * connect to 127.0.0.1 port 8081 failed: Connection refused
Jul 20 23:47:21 * Failed to connect to localhost port 8081: Connection refused
Jul 20 23:47:21
```
I don't know how to solve it.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242765635
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value = valueElementGetter.getElementOrNull(map1.valueArray(), i);
+ this.map.put(key, value);
+ }
+ for (int i = 0; i < map2.size(); i++) {
+ final Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ final Object value = valueElementGetter.getElementOrNull(map2.valueArray(), i);
+ this.map.put(key, value);
+ }
+ this.updateArrays();
+ }
+
+ @Override
+ public int size() {
+ return map.size();
+ }
+
+ @Override
+ public ArrayData keyArray() {
+ return keysArray;
+ }
+
+ @Override
+ public ArrayData valueArray() {
+ return valuesArray;
+ }
+
+ private void updateArrays() {
+ this.keysArray = new GenericArrayData(map.keySet().toArray());
Review Comment:
do we need a separate method for that if it's used only once and contains only 2 lines?
--
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
[GitHub] [flink] twalthr commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1243252360
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
> Why binaryData's equals is not sync with SQL's sync?
Because the input might not only be binary data but also generic data. And the two cannot be compared. Also SQL has specifics such as leading/trailing zeros and many other specifics that the code generated equals/hashcodes can handle.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242829756
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
first we should care about correctness and only then about speed
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242867857
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
To be honest, I don't know under the hood uses java's equals which is not 100% same as SQL.
Why binaryData's equals is not sync with SQL's sync?
But If we transfer the input Data to the Expression, the data we received in MapUnionFunction are binaryMap.
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1503885866
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ if (key == null) {
+ isKeyNullExist = true;
+ }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(map2.valueArray(), i));
+ }
+
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key1 = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value1 = valueElementGetter.getElementOrNull(map1.valueArray(), i);
Review Comment:
it seems we need `value1` only if key is not present
should we move it under corresponding `if`?
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1506962857
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
Review Comment:
Yes, if map == null, we should 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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1506962507
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
Review Comment:
Ok, I will fix it.
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1507129381
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ if (key == null) {
+ isKeyNullExist = true;
+ }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(map2.valueArray(), i));
+ }
+
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key1 = keyElementGetter.getElementOrNull(map1.keyArray(), i);
+ final Object value1 = valueElementGetter.getElementOrNull(map1.valueArray(), i);
Review Comment:
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1503825971
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
Review Comment:
nit: depending on implementation it could generate new objects
so probably `map2.keyArray()` could be extracted into a var before the loop
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ private final SpecializedFunction.ExpressionEvaluator keyEqualityEvaluator;
+ private transient MethodHandle keyEqualityHandle;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ final DataType keyDataType = outputType.getKeyDataType();
+ final DataType valueDataType = outputType.getValueDataType();
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ keyEqualityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityHandle = keyEqualityEvaluator.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable MapData... maps) {
+ try {
+ if (maps == null || maps.length == 0) {
+ return null;
+ }
+ if (maps.length == 1) {
+ return maps[0];
+ }
+ MapData result = maps[0];
+ for (int i = 1; i < maps.length; ++i) {
+ MapData map = maps[i];
+ if (map != null && map.size() > 0) {
+ result = new MapDataForMapUnion(result, map);
+ }
+ }
+ return result;
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private final GenericArrayData keysArray;
+ private final GenericArrayData valuesArray;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(map2.keyArray(), i);
+ if (key == null) {
+ isKeyNullExist = true;
+ }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(map2.valueArray(), i));
Review Comment:
same here
--
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-32261][table] Add built-in MAP_UNION function. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1983757510
@hanyuzheng7 could you please have at build failure?
it seems it is related to the recent changes
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=57976&view=logs&j=0c940707-2659-5648-cbe6-a1ad63045f0a&t=075c2716-8010-5565-fe08-3c4bb45824a4
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252342408
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -1468,6 +1469,16 @@ public OutType mapEntries() {
return toApiSpecificExpression(unresolvedCall(MAP_ENTRIES, toExpr()));
}
+ /**
+ * Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same
+ * data structure. If there are overlapping keys, the value from 'map2' will overwrite the value
Review Comment:
ok
##########
docs/data/sql_functions.yml:
##########
@@ -658,6 +658,9 @@ collection:
- sql: MAP_KEYS(map)
table: MAP.mapKeys()
description: Returns the keys of the map as array. No order guaranteed.
+ - sql: MAP_UNION(map1, map2)
+ table: map1.mapUnion(map2)
+ description: Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same data structure. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, return null.
Review Comment:
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252343611
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
`Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same type for their key and value individually. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, return null.`
how about this one?
What I want to say is that two maps' key and value should have same data type.
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252335930
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
it's not clear what is `same datastructure` in SQL context...
i guess same type (or not same, however common)
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252335930
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
it's not clear what is `same datastructure` in SQL context...
i guess same type
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252342119
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
ok, I will change the description
--
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
[GitHub] [flink] snuyanzin commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1618976415
This query fails with NullPointerException
```sql
SELECT map_union(MAP[1, 2], MAP[CAST(NULL AS INT), 3]);
```
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1252354619
##########
flink-python/pyflink/table/expression.py:
##########
@@ -1558,6 +1558,18 @@ def map_keys(self) -> 'Expression':
"""
return _unary_op("mapKeys")(self)
+ @property
+ def map_union(self, map2) -> 'Expression':
+ """
+ Returns a map created by merging two maps, 'map1' and 'map2'.
+ These two maps should have same data structure. If there are overlapping keys,
Review Comment:
you can not say `same` since one could be nullable and another not...
so it should be common type
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242766634
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
Review Comment:
it could be final i guess
--
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
[GitHub] [flink] snuyanzin commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1242769137
##########
docs/data/sql_functions.yml:
##########
@@ -649,6 +649,9 @@ collection:
- sql: MAP_KEYS(map)
table: MAP.mapKeys()
description: Returns the keys of the map as array. No order guaranteed.
+ SQL: MAP_UNION(map1, map2)
+ Table: map1.mapUnion(map2)
+ Description: Returns a map created by merging two maps, 'map1' and 'map2'. This two maps should have same data structure. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps are null, return null.
Review Comment:
```suggestion
Description: Returns a map created by merging two maps, 'map1' and 'map2'. These two maps should have same data structure. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1'. If any of maps is null, 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1626496365
Returns a map created by merging at least one map. These maps should have a common map type. If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1', the value from 'map3' will overwrite the value from 'map2', the value from 'mapn' will overwrite the value from 'map(n-1)'. If any of maps is null, 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
[GitHub] [flink] hanyuzheng7 commented on pull request #22842: [FLINK-32261][table] Add built-in MAP_UNION function.
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #22842:
URL: https://github.com/apache/flink/pull/22842#issuecomment-1622117991
@flinkbot run azure
--
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
[GitHub] [flink] hanyuzheng7 commented on a diff in pull request #22842: [FLINK-32261]-table-Add-MAP_UNION-function
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #22842:
URL: https://github.com/apache/flink/pull/22842#discussion_r1244608398
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.KeyValueDataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_UNION}. */
+@Internal
+public class MapUnionFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ArrayData.ElementGetter valueElementGetter;
+
+ public MapUnionFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_UNION, context);
+ KeyValueDataType outputType =
+ ((KeyValueDataType) context.getCallContext().getOutputDataType().get());
+ keyElementGetter =
+ ArrayData.createElementGetter(outputType.getKeyDataType().getLogicalType());
+ valueElementGetter =
+ ArrayData.createElementGetter(outputType.getValueDataType().getLogicalType());
+ }
+
+ public @Nullable MapData eval(@Nullable MapData map1, @Nullable MapData map2) {
+ if (map1 == null || map2 == null) {
+ return null;
+ }
+ MapDataForMapUnion resultMap = new MapDataForMapUnion(map1, map2);
+ return resultMap;
+ }
+
+ private class MapDataForMapUnion implements MapData {
+ private ArrayData keysArray;
+ private ArrayData valuesArray;
+ private final Map<Object, Object> map;
+
+ public MapDataForMapUnion(MapData map1, MapData map2) {
+
+ this.map = new HashMap<>();
Review Comment:
fixed it.
--
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