This is an automated email from the ASF dual-hosted git repository.
dianfu 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 2e0056a483f [FLINK-40658][table] Preserve null ROW elements in UNNEST
(#29398)
2e0056a483f is described below
commit 2e0056a483f16fb4df91640271458b0063a978bb
Author: Beetle <[email protected]>
AuthorDate: Mon Oct 12 14:34:26 2026 +0800
[FLINK-40658][table] Preserve null ROW elements in UNNEST (#29398)
---
.../planner/runtime/batch/sql/UnnestITCase.scala | 79 ++++++++++++++++++++++
.../planner/runtime/stream/sql/UnnestITCase.scala | 58 ++++++++++++++++
.../functions/table/UnnestRowsFunction.java | 12 +++-
.../functions/table/UnnestRowsFunctionBase.java | 9 +++
.../table/UnnestRowsWithOrdinalityFunction.java | 14 ++--
5 files changed, 164 insertions(+), 8 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala
index 09587cc5f61..c1590d2402f 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala
@@ -19,6 +19,7 @@ package org.apache.flink.table.planner.runtime.batch.sql
import org.apache.flink.api.java.typeutils.{ObjectArrayTypeInfo, RowTypeInfo}
import org.apache.flink.table.legacy.api.Types
+import org.apache.flink.table.planner.factories.TestValuesTableFactory
import org.apache.flink.table.planner.runtime.utils.{BatchTestBase, TestData}
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.utils.DateTimeUtils.toLocalDateTime
@@ -315,6 +316,68 @@ class UnnestITCase extends BatchTestBase {
Seq(row('a', 1), row('a', 2), row('a', 3)))
}
+ @Test
+ def testUnnestArrayWithNullRow(): Unit = {
+ checkResult(
+ """
+ |SELECT *
+ |FROM UNNEST(ARRAY[
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(ROW(1, 'a') AS ROW<n INT, s STRING>),
+ | CAST(ROW(CAST(NULL AS INT), CAST(NULL AS STRING)) AS ROW<n INT, s
STRING>)])
+ |""".stripMargin,
+ Seq(row(null, null), row(null, null), row(1, "a"), row(null, null))
+ )
+ }
+
+ @Test
+ def testLeftUnnestArrayWithNullRow(): Unit = {
+ checkResult(
+ """
+ |SELECT id, n, s
+ |FROM (VALUES (1, ARRAY[
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(ROW(1, 'a') AS ROW<n INT, s STRING>)])) AS T(id, items)
+ |LEFT JOIN UNNEST(T.items) AS U(n, s) ON TRUE
+ |""".stripMargin,
+ Seq(row(1, null, null), row(1, null, null), row(1, 1, "a"))
+ )
+ }
+
+ @Test
+ def testUnnestMultisetWithNullRow(): Unit = {
+ registerNullRowMultiset()
+
+ checkResult(
+ "SELECT n, s FROM M, UNNEST(M.items) AS U(n, s)",
+ Seq(row(null, null), row(null, null)))
+ }
+
+ @Test
+ def testUnnestMultisetWithNullRowAndOrdinality(): Unit = {
+ registerNullRowMultiset()
+
+ checkResult(
+ "SELECT n, s, o FROM M, UNNEST(M.items) WITH ORDINALITY AS U(n, s, o)",
+ Seq(row(null, null, 1), row(null, null, 2)))
+ }
+
+ private def registerNullRowMultiset(): Unit = {
+ val items = new java.util.HashMap[Row, Integer]()
+ items.put(null, 2)
+ val dataId = TestValuesTableFactory.registerData(List(row(items)))
+ tEnv.executeSql(s"""
+ |CREATE TABLE M (items MULTISET<ROW<n INT, s STRING>>)
+ |WITH (
+ | 'connector' = 'values',
+ | 'data-id' = '$dataId',
+ | 'bounded' = 'true'
+ |)
+ |""".stripMargin)
+ }
+
@Test
def testUnnestWithOrdinalityWithValuesStream(): Unit = {
checkResult(
@@ -323,6 +386,22 @@ class UnnestITCase extends BatchTestBase {
)
}
+ @Test
+ def testUnnestArrayWithNullRowAndOrdinality(): Unit = {
+ checkResult(
+ """
+ |SELECT *
+ |FROM UNNEST(ARRAY[
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(ROW(1, 'a') AS ROW<n INT, s STRING>),
+ | CAST(ROW(CAST(NULL AS INT), CAST(NULL AS STRING)) AS ROW<n INT, s
STRING>)])
+ |WITH ORDINALITY
+ |""".stripMargin,
+ Seq(row(null, null, 1), row(null, null, 2), row(1, "a", 3), row(null,
null, 4))
+ )
+ }
+
@Test
def testUnnestArrayWithOrdinality(): Unit = {
val data = List(
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala
index 17f0093ec39..37458e6d8cc 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala
@@ -356,6 +356,44 @@ class UnnestITCase(mode: StateBackendMode) extends
StreamingWithStateTestBase(mo
)
}
+ @TestTemplate
+ def testUnnestArrayWithNullRow(): Unit = {
+ assertUnnest(
+ testData = List(1),
+ typeInfo = createTypeInformation[Int],
+ sqlQuery = """
+ |SELECT *
+ |FROM UNNEST(ARRAY[
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(ROW(1, 'a') AS ROW<n INT, s STRING>),
+ | CAST(ROW(CAST(NULL AS INT), CAST(NULL AS STRING)) AS
ROW<n INT, s STRING>)])
+ |""".stripMargin,
+ expectedResults = List("null,null", "null,null", "1,a", "null,null"),
+ isRetract = false,
+ fieldNames = 'dummy
+ )
+ }
+
+ @TestTemplate
+ def testLeftUnnestArrayWithNullRow(): Unit = {
+ assertUnnest(
+ testData = List(1),
+ typeInfo = createTypeInformation[Int],
+ sqlQuery = """
+ |SELECT id, n, s
+ |FROM (VALUES (1, ARRAY[
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(ROW(1, 'a') AS ROW<n INT, s STRING>)])) AS V(id,
items)
+ |LEFT JOIN UNNEST(V.items) AS U(n, s) ON TRUE
+ |""".stripMargin,
+ expectedResults = List("1,null,null", "1,null,null", "1,1,a"),
+ isRetract = false,
+ fieldNames = 'dummy
+ )
+ }
+
@TestTemplate
def testUnnestWithOrdinalityWithValuesStream(): Unit = {
assertUnnest(
@@ -368,6 +406,26 @@ class UnnestITCase(mode: StateBackendMode) extends
StreamingWithStateTestBase(mo
)
}
+ @TestTemplate
+ def testUnnestArrayWithNullRowAndOrdinality(): Unit = {
+ assertUnnest(
+ testData = List(1),
+ typeInfo = createTypeInformation[Int],
+ sqlQuery = """
+ |SELECT *
+ |FROM UNNEST(ARRAY[
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(NULL AS ROW<n INT, s STRING>),
+ | CAST(ROW(1, 'a') AS ROW<n INT, s STRING>),
+ | CAST(ROW(CAST(NULL AS INT), CAST(NULL AS STRING)) AS
ROW<n INT, s STRING>)])
+ |WITH ORDINALITY
+ |""".stripMargin,
+ expectedResults = List("null,null,1", "null,null,2", "1,a,3",
"null,null,4"),
+ isRetract = false,
+ fieldNames = 'dummy
+ )
+ }
+
@TestTemplate
def testUnnestArrayWithOrdinality(): Unit = {
val data = List(
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunction.java
index cece2e1e34a..db1096fe59f 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunction.java
@@ -60,6 +60,7 @@ public class UnnestRowsFunction extends
UnnestRowsFunctionBase {
private static final long serialVersionUID = 1L;
private final ArrayData.ElementGetter elementGetter;
+ private final int rowArity;
public CollectionUnnestFunction(
SpecializedContext context,
@@ -67,14 +68,21 @@ public class UnnestRowsFunction extends
UnnestRowsFunctionBase {
ArrayData.ElementGetter elementGetter) {
super(context, elementType, false);
this.elementGetter = elementGetter;
+ this.rowArity = getRowArity(elementType);
}
public void eval(ArrayData arrayData) {
- evalArrayData(arrayData, elementGetter, (element, position) ->
collect(element));
+ evalArrayData(
+ arrayData,
+ elementGetter,
+ (element, position) ->
collect(normalizeCollectionElement(element, rowArity)));
}
public void eval(MapData mapData) {
- evalMultisetData(mapData, elementGetter, (element, position) ->
collect(element));
+ evalMultisetData(
+ mapData,
+ elementGetter,
+ (element, position) ->
collect(normalizeCollectionElement(element, rowArity)));
}
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunctionBase.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunctionBase.java
index ea340adf2e7..ac9a4987131 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunctionBase.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsFunctionBase.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.functions.table;
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.GenericRowData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.UserDefinedFunction;
@@ -156,6 +157,14 @@ public abstract class UnnestRowsFunctionBase extends
BuiltInSpecializedFunction
return outputDataType;
}
+ protected static int getRowArity(LogicalType elementType) {
+ return elementType instanceof RowType ? ((RowType)
elementType).getFieldCount() : -1;
+ }
+
+ protected static Object normalizeCollectionElement(Object element, int
rowArity) {
+ return element == null && rowArity >= 0 ? new
GenericRowData(rowArity) : element;
+ }
+
protected void evalArrayData(
ArrayData arrayData,
ArrayData.ElementGetter elementGetter,
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsWithOrdinalityFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsWithOrdinalityFunction.java
index 6bd738a09e7..87ec87a75a7 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsWithOrdinalityFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/UnnestRowsWithOrdinalityFunction.java
@@ -45,8 +45,7 @@ public class UnnestRowsWithOrdinalityFunction extends
UnnestRowsFunctionBase {
SpecializedContext context,
LogicalType elementType,
ArrayData.ElementGetter elementGetter) {
- return new CollectionUnnestWithOrdinalityFunction(
- context, wrapWithOrdinality(elementType), elementGetter);
+ return new CollectionUnnestWithOrdinalityFunction(context,
elementType, elementGetter);
}
@Override
@@ -67,13 +66,15 @@ public class UnnestRowsWithOrdinalityFunction extends
UnnestRowsFunctionBase {
private static final long serialVersionUID = 1L;
private final ArrayData.ElementGetter elementGetter;
+ private final int rowArity;
public CollectionUnnestWithOrdinalityFunction(
SpecializedContext context,
LogicalType elementType,
ArrayData.ElementGetter elementGetter) {
- super(context, elementType, true);
+ super(context, wrapWithOrdinality(elementType), true);
this.elementGetter = elementGetter;
+ this.rowArity = getRowArity(elementType);
}
public void eval(ArrayData arrayData) {
@@ -85,11 +86,12 @@ public class UnnestRowsWithOrdinalityFunction extends
UnnestRowsFunctionBase {
}
private void collectWithOrdinality(Object element, int position) {
- if (element instanceof RowData) {
- RowData row = (RowData) element;
+ final Object normalizedElement =
normalizeCollectionElement(element, rowArity);
+ if (normalizedElement instanceof RowData) {
+ RowData row = (RowData) normalizedElement;
collect(new JoinedRowData(row.getRowKind(), row,
GenericRowData.of(position)));
} else {
- collect(GenericRowData.of(element, position));
+ collect(GenericRowData.of(normalizedElement, position));
}
}
}
