You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/06/08 07:12:29 UTC
[flink] branch master updated: [FLINK-25699][table] Use HashMap for MAP value constructors
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 872fa84dbbd [FLINK-25699][table] Use HashMap for MAP value constructors
872fa84dbbd is described below
commit 872fa84dbbd9c61122d55c6c0986edca21a5841a
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Tue Jan 25 14:17:49 2022 +0100
[FLINK-25699][table] Use HashMap for MAP value constructors
---
.../planner/codegen/calls/ScalarOperatorGens.scala | 18 +-
.../planner/functions/CastFunctionITCase.java | 6 +-
.../table/planner/functions/MapFunctionITCase.java | 210 +++++++++++++++++++++
.../table/planner/expressions/MapTypeTest.scala | 62 ++----
4 files changed, 246 insertions(+), 50 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
index 8a3b69cba5a..1e2040becef 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
@@ -1359,13 +1359,27 @@ object ScalarOperatorGens {
val baseMap = newName("map")
// prepare map key array
- val keyElements = elements.grouped(2).map { case Seq(key, _) => key }.toSeq
+ val keyElements = elements
+ .grouped(2)
+ .map { case Seq(key, value) => (key, value) }
+ .toSeq
+ .groupBy(_._1)
+ .map(_._2.last)
+ .keys
+ .toSeq
val keyType = mapType.getKeyType
val keyExpr = generateArray(ctx, new ArrayType(keyType), keyElements)
val isKeyFixLength = isPrimitive(keyType)
// prepare map value array
- val valueElements = elements.grouped(2).map { case Seq(_, value) => value }.toSeq
+ val valueElements = elements
+ .grouped(2)
+ .map { case Seq(key, value) => (key, value) }
+ .toSeq
+ .groupBy(_._1)
+ .map(_._2.last)
+ .values
+ .toSeq
val valueType = mapType.getValueType
val valueExpr = generateArray(ctx, new ArrayType(valueType), valueElements)
val isValueFixLength = isPrimitive(valueType)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
index e737366d450..82ea5c5d024 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
@@ -1074,9 +1074,9 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase {
.fromCase(INTERVAL(DAY()), Duration.ofHours(36), "+1 12:00:00.000")
.fromCase(ARRAY(INT().nullable()), new Integer[] {null, 456}, "[NULL, 456]")
.fromCase(
- MAP(STRING(), INTERVAL(MONTH()).nullable()),
- map(entry("a", -123), entry("b", null)),
- "{a=-10-03, b=NULL}")
+ MAP(STRING(), INTERVAL(MONTH())),
+ map(entry("a", -123)),
+ "{a=-10-03}")
.fromCase(
ROW(FIELD("f0", INT().nullable()), FIELD("f1", STRING())),
Row.of(null, "abc"),
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java
new file mode 100644
index 00000000000..b02a78472fb
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java
@@ -0,0 +1,210 @@
+/*
+ * 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.planner.functions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.util.CollectionUtil;
+
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.Period;
+import java.util.Collections;
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.BOOLEAN;
+import static org.apache.flink.table.api.DataTypes.DATE;
+import static org.apache.flink.table.api.DataTypes.DECIMAL;
+import static org.apache.flink.table.api.DataTypes.DOUBLE;
+import static org.apache.flink.table.api.DataTypes.FLOAT;
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.INTERVAL;
+import static org.apache.flink.table.api.DataTypes.MAP;
+import static org.apache.flink.table.api.DataTypes.MONTH;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.DataTypes.TIME;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.map;
+import static org.apache.flink.util.CollectionUtil.entry;
+
+/** Test {@link BuiltInFunctionDefinitions#MAP} and its return type. */
+public class MapFunctionITCase extends BuiltInFunctionTestBase {
+ private static final LocalDate TEST_DATE_1 = LocalDate.of(1985, 11, 4);
+ private static final LocalDate TEST_DATE_2 = LocalDate.of(2018, 7, 26);
+ private static final LocalTime TEST_TIME_1 = LocalTime.of(17, 18, 19);
+ private static final LocalTime TEST_TIME_2 = LocalTime.of(14, 15, 16);
+ private static final LocalDateTime TEST_DATE_TIME_1 = LocalDateTime.of(1985, 11, 4, 17, 18, 19);
+ private static final LocalDateTime TEST_DATE_TIME_2 = LocalDateTime.of(2018, 7, 26, 14, 15, 16);
+ private static final String A = "a";
+ private static final String B = "b";
+ private static final int INTERVAL_1 = -123;
+ private static final Integer INTERVAL_NULL = null;
+
+ @Override
+ Stream<TestSetSpec> getTestSetSpecs() {
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.MAP)
+ .onFieldsWithData(
+ 1,
+ 2,
+ 3,
+ 4,
+ new BigDecimal("1.2345"),
+ new BigDecimal("1.2346"),
+ true)
+ .andDataTypes(
+ INT().notNull(),
+ INT().notNull(),
+ INT().notNull(),
+ INT().notNull(),
+ DECIMAL(10, 4).notNull(),
+ DECIMAL(10, 4).notNull(),
+ BOOLEAN().notNull())
+ .testResult(
+ resultSpec(
+ map($("f0"), $("f0"), $("f0"), $("f1")),
+ "MAP[f0, f1]",
+ Collections.singletonMap(1, 2),
+ DataTypes.MAP(INT().notNull(), INT().notNull()).notNull()),
+ resultSpec(
+ map($("f4"), $("f5")),
+ "MAP[f4, f5]",
+ Collections.singletonMap(
+ new BigDecimal("1.2345"), new BigDecimal("1.2346")),
+ DataTypes.MAP(
+ DECIMAL(10, 4).notNull(),
+ DECIMAL(10, 4).notNull())
+ .notNull()),
+ resultSpec(
+ map(
+ $("f0").plus($("f1")),
+ $("f2").times($("f2")),
+ $("f2").minus($("f1")),
+ $("f3").minus($("f0"))),
+ "MAP[f0 + f1, f2 * f2, f2 - f1, f3 - f0]",
+ CollectionUtil.map(
+ entry(1 + 2, 3 * 3), entry(3 - 2, 4 - 1)),
+ DataTypes.MAP(INT().notNull(), INT().notNull()).notNull()),
+ resultSpec(
+ map(
+ $("f0"),
+ $("f1").cast(BIGINT().notNull()),
+ $("f2"),
+ $("f3").cast(BIGINT().notNull())),
+ "MAP[f0, CAST(f1 AS BIGINT), f2, CAST(f3 AS BIGINT)]",
+ CollectionUtil.map(entry(1, 2L), entry(3, 4L)),
+ DataTypes.MAP(INT().notNull(), BIGINT().notNull())
+ .notNull()),
+ resultSpec(
+ map($("f6"), $("f6")),
+ "MAP[f6, f6]",
+ Collections.singletonMap(true, true),
+ DataTypes.MAP(BOOLEAN().notNull(), BOOLEAN().notNull())
+ .notNull()),
+ resultSpec(
+ map(
+ $("f0"),
+ $("f1").cast(DOUBLE().notNull()),
+ $("f2"),
+ $("f3").cast(FLOAT().notNull())),
+ "MAP[f0, CAST(f1 AS DOUBLE), f2, CAST(f3 AS FLOAT)]",
+ CollectionUtil.map(entry(1, 2d), entry(3, 4.0)),
+ DataTypes.MAP(INT().notNull(), DOUBLE().notNull())
+ .notNull()),
+ resultSpec(
+ map($("f4"), $("f5")),
+ "MAP[f4, f5]",
+ Collections.singletonMap(
+ new BigDecimal("1.2345"), new BigDecimal("1.2346")),
+ DataTypes.MAP(
+ DECIMAL(10, 4).notNull(),
+ DECIMAL(10, 4).notNull())
+ .notNull()),
+ resultSpec(
+ map(map($("f0"), $("f1")), map($("f2"), $("f3"))),
+ "MAP[MAP[f0, f1], MAP[f2, f3]]",
+ Collections.singletonMap(
+ Collections.singletonMap(1, 2),
+ Collections.singletonMap(3, 4)),
+ DataTypes.MAP(
+ MAP(
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull())
+ .notNull(),
+ MAP(
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull())
+ .notNull())
+ .notNull())),
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.MAP)
+ .onFieldsWithData(
+ TEST_DATE_1,
+ TEST_DATE_2,
+ TEST_TIME_1,
+ TEST_TIME_2,
+ TEST_DATE_TIME_1,
+ TEST_DATE_TIME_2)
+ .andDataTypes(
+ DATE().notNull(),
+ DATE().notNull(),
+ TIME().notNull(),
+ TIME().notNull(),
+ TIMESTAMP().notNull(),
+ TIMESTAMP().notNull())
+ .testResult(
+ resultSpec(
+ map($("f0"), $("f2"), $("f1"), $("f3")),
+ "MAP[f0, f2, f1, f3]",
+ CollectionUtil.map(
+ entry(TEST_DATE_1, TEST_TIME_1),
+ entry(TEST_DATE_2, TEST_TIME_2)),
+ DataTypes.MAP(DATE().notNull(), TIME().notNull())
+ .notNull()),
+ resultSpec(
+ map($("f2"), $("f4"), $("f3"), $("f5")),
+ "MAP[f2, f4, f3, f5]",
+ CollectionUtil.map(
+ entry(TEST_TIME_1, TEST_DATE_TIME_1),
+ entry(TEST_TIME_2, TEST_DATE_TIME_2)),
+ DataTypes.MAP(TIME().notNull(), TIMESTAMP().notNull())
+ .notNull())),
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.MAP)
+ .onFieldsWithData(A, B, INTERVAL_1, INTERVAL_NULL)
+ .andDataTypes(
+ STRING().notNull(),
+ STRING().notNull(),
+ INTERVAL(MONTH()),
+ INTERVAL(MONTH()).nullable())
+ .testResult(
+ resultSpec(
+ map($("f0"), $("f2"), $("f1"), $("f3")),
+ "MAP[f0, f2, f1, f3]",
+ CollectionUtil.map(
+ entry(A, Period.ofMonths(INTERVAL_1)),
+ entry(B, INTERVAL_NULL)),
+ DataTypes.MAP(
+ STRING().notNull(),
+ INTERVAL(MONTH()).nullable())
+ .notNull())));
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/MapTypeTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/MapTypeTest.scala
index 13aaa6d5ddb..c2aaa4aa4da 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/MapTypeTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/MapTypeTest.scala
@@ -30,10 +30,7 @@ class MapTypeTest extends MapTypeTestBase {
@Test
def testInputTypeGeneralization(): Unit = {
- testAllApis(
- map(1, "ABC", 2.0, "D"),
- "MAP[1, 'ABC', cast(2.0 AS DOUBLE), 'D']",
- "{1.0=ABC, 2.0=D}")
+ testAllApis(map(1d, "ABC"), "MAP[CAST(1 AS DOUBLE), 'ABC']", "{1.0=ABC}")
}
@Test
@@ -58,67 +55,42 @@ class MapTypeTest extends MapTypeTestBase {
testAllApis(map(map(1, 2), map(3, 4)), "MAP[MAP[1, 2], MAP[3, 4]]", "{{1=2}={3=4}}")
- testAllApis(map(1 + 2, 3 * 3, 3 - 6, 4 - 2), "map[1 + 2, 3 * 3, 3 - 6, 4 - 2]", "{3=9, -3=2}")
-
testAllApis(map(1, nullOf(DataTypes.INT)), "map[1, NULLIF(1,1)]", "{1=NULL}")
// explicit conversion
- testAllApis(map(1, 2L, 3, 4L), "MAP[1, CAST(2 AS BIGINT), 3, CAST(4 AS BIGINT)]", "{1=2, 3=4}")
+ testAllApis(map(1, 2L), "MAP[1, CAST(2 AS BIGINT)]", "{1=2}")
testAllApis(
- map(
- valueLiteral(localDate("1985-04-11")),
- valueLiteral(gLocalTime("14:15:16")),
- valueLiteral(localDate("2018-07-26")),
- valueLiteral(gLocalTime("17:18:19"))),
- "MAP[DATE '1985-04-11', TIME '14:15:16', DATE '2018-07-26', TIME '17:18:19']",
- "{1985-04-11=14:15:16, 2018-07-26=17:18:19}"
- )
+ map(valueLiteral(localDate("1985-04-11")), valueLiteral(gLocalTime("14:15:16"))),
+ "MAP[DATE '1985-04-11', TIME '14:15:16']",
+ "{1985-04-11=14:15:16}")
// There is no timestamp literal function in Java String Table API,
// toTimestamp is casting string to TIMESTAMP(3) which is not the same to timestamp literal.
testTableApi(
- map(
- valueLiteral(gLocalTime("14:15:16")),
- valueLiteral(localDateTime("1985-04-11 14:15:16")),
- valueLiteral(gLocalTime("17:18:19")),
- valueLiteral(localDateTime("2018-07-26 17:18:19"))
- ),
- "{14:15:16=1985-04-11 14:15:16, 17:18:19=2018-07-26 17:18:19}"
- )
+ map(valueLiteral(gLocalTime("14:15:16")), valueLiteral(localDateTime("1985-04-11 14:15:16"))),
+ "{14:15:16=1985-04-11 14:15:16}")
testSqlApi(
- "MAP[TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16', " +
- "TIME '17:18:19', TIMESTAMP '2018-07-26 17:18:19']",
- "{14:15:16=1985-04-11 14:15:16, 17:18:19=2018-07-26 17:18:19}"
- )
+ "MAP[TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16']",
+ "{14:15:16=1985-04-11 14:15:16}")
testAllApis(
map(
valueLiteral(gLocalTime("14:15:16")),
- valueLiteral(localDateTime("1985-04-11 14:15:16.123")),
- valueLiteral(gLocalTime("17:18:19")),
- valueLiteral(localDateTime("2018-07-26 17:18:19.123"))
- ),
- "MAP[TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16.123', " +
- "TIME '17:18:19', TIMESTAMP '2018-07-26 17:18:19.123']",
- "{14:15:16=1985-04-11 14:15:16.123, 17:18:19=2018-07-26 17:18:19.123}"
+ valueLiteral(localDateTime("1985-04-11 14:15:16.123"))),
+ "MAP[TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16.123']",
+ "{14:15:16=1985-04-11 14:15:16.123}"
)
testTableApi(
map(
valueLiteral(gLocalTime("14:15:16")),
- valueLiteral(JLocalTimestamp.of(1985, 4, 11, 14, 15, 16, 123456000)),
- valueLiteral(gLocalTime("17:18:19")),
- valueLiteral(JLocalTimestamp.of(2018, 7, 26, 17, 18, 19, 123456000))
- ),
- "{14:15:16=1985-04-11 14:15:16.123456, 17:18:19=2018-07-26 17:18:19.123456}"
- )
+ valueLiteral(JLocalTimestamp.of(1985, 4, 11, 14, 15, 16, 123456000))),
+ "{14:15:16=1985-04-11 14:15:16.123456}")
testSqlApi(
- "MAP[TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16.123456', " +
- "TIME '17:18:19', TIMESTAMP '2018-07-26 17:18:19.123456']",
- "{14:15:16=1985-04-11 14:15:16.123456, 17:18:19=2018-07-26 17:18:19.123456}"
- )
+ "MAP[TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16.123456']",
+ "{14:15:16=1985-04-11 14:15:16.123456}")
testAllApis(
map(BigDecimal(2.0002), BigDecimal(2.0003)),
@@ -126,7 +98,7 @@ class MapTypeTest extends MapTypeTestBase {
"{2.0002=2.0003}")
// implicit type cast only works on SQL API
- testSqlApi("MAP['k1', CAST(1 AS DOUBLE), 'k2', CAST(2 AS FLOAT)]", "{k1=1.0, k2=2.0}")
+ testSqlApi("MAP['k1', CAST(1 AS DOUBLE)]", "{k1=1.0}")
}
@Test