This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 2c425ba1a51 [FLINK-40839][table] Support `VARIANT` columns omitted
from an `INSERT` column list
2c425ba1a51 is described below
commit 2c425ba1a518f6eda1f3fe1e850aea7f26878757
Author: Ramin Gharib <[email protected]>
AuthorDate: Wed Sep 30 13:08:04 2026 +0200
[FLINK-40839][table] Support `VARIANT` columns omitted from an `INSERT`
column list
---
.../org/apache/calcite/sql/type/SqlTypeUtil.java | 12 +++--
.../table/planner/calcite/SqlTypeUtilTest.java | 61 +++++++++++++++++-----
.../nodes/exec/stream/VariantSemanticTest.java | 25 ++++++++-
3 files changed, 80 insertions(+), 18 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java
b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java
index d33e27c81ab..e92babdf68b 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java
@@ -77,9 +77,10 @@ import static org.apache.calcite.util.Static.RESOURCE;
* <p>FLINK modifications are at lines
*
* <ol>
- * <li>We should use ExtendedSqlCollectionTypeNameSpec for rows: Lines
1291-1300
- * <li>We should use ExtendedSqlRowTypeNameSpec for rows: Lines 1312-1316
- * <li>Should be removed after fixing CALCITE-7062: Lines 1336
+ * <li>We should use ExtendedSqlCollectionTypeNameSpec for rows: Lines
1292-1301
+ * <li>We should use ExtendedSqlRowTypeNameSpec for rows: Lines 1313-1317
+ * <li>Should be removed after fixing CALCITE-7062: Lines 1337
+ * <li>Should be removed after upgrading to Calcite 1.42.0, see
CALCITE-7293: Lines 1994-2001
* </ol>
*/
public abstract class SqlTypeUtil {
@@ -1990,11 +1991,14 @@ public abstract class SqlTypeUtil {
if (typeName == null) {
return false;
}
+ // FLINK MODIFICATION BEGIN
return SqlTypeUtil.isDatetime(type)
|| SqlTypeUtil.isNumeric(type)
|| SqlTypeUtil.isString(type)
|| SqlTypeUtil.isBoolean(type)
- || typeName == SqlTypeName.UUID;
+ || typeName == SqlTypeName.UUID
+ || typeName == SqlTypeName.VARIANT;
+ // FLINK MODIFICATION END
}
/** Returns a DECIMAL type with the maximum precision for the current type
system. */
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/SqlTypeUtilTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/SqlTypeUtilTest.java
index 58560bdee9a..19ead0c0b28 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/SqlTypeUtilTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/SqlTypeUtilTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.planner.calcite;
import org.apache.flink.table.planner.typeutils.LogicalRelDataTypeConverter;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
@@ -30,11 +31,34 @@ import org.apache.calcite.sql.SqlWriter;
import org.apache.calcite.sql.pretty.SqlPrettyWriter;
import org.apache.calcite.sql.type.SqlTypeUtil;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.DataTypes.ARRAY;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.MAP;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.VARIANT;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link SqlTypeUtil}. */
class SqlTypeUtilTest {
+
+ private final FlinkTypeFactory typeFactory =
+ new FlinkTypeFactory(
+ Thread.currentThread().getContextClassLoader(),
FlinkTypeSystem.INSTANCE);
+
+ private final SqlWriter writer =
+ new SqlPrettyWriter(
+ SqlPrettyWriter.config()
+ .withAlwaysUseParentheses(false)
+ .withSelectListItemsOnSeparateLines(false)
+ .withIndentation(0));
+
/**
* Test case for <a
href="https://issues.apache.org/jira/browse/FLINK-38913";>[FLINK-38913]
* ArrayIndexOutOfBoundsException when creating a table with computed rows
including casts to
@@ -42,25 +66,36 @@ class SqlTypeUtilTest {
*/
@Test
void testConvertRowTypeToSpecAndUnparse() {
- FlinkTypeFactory typeFactory =
- new FlinkTypeFactory(
- Thread.currentThread().getContextClassLoader(),
FlinkTypeSystem.INSTANCE);
RowType rowType =
RowType.of(
new LogicalType[] {new IntType(), new VarCharType(1)},
new String[] {"a", "b"});
- RelDataType relDataType =
LogicalRelDataTypeConverter.toRelDataType(rowType, typeFactory);
+ assertThat(convertToSpecAndUnparse(rowType))
+ .hasToString("ROW(\"a\" INTEGER, \"b\" VARCHAR(1) CHARACTER
SET \"UTF-16LE\")");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("variantTypes")
+ void testConvertTypeToSpecAndUnparse(DataType dataType, String expected) {
+
assertThat(convertToSpecAndUnparse(dataType.getLogicalType())).isEqualTo(expected);
+ }
+
+ private static Stream<Arguments> variantTypes() {
+ return Stream.of(
+ Arguments.of(VARIANT(), "VARIANT"),
+ Arguments.of(VARIANT().notNull(), "VARIANT"),
+ Arguments.of(ARRAY(VARIANT()), "VARIANT ARRAY"),
+ Arguments.of(ROW(FIELD("v", VARIANT())), "ROW(\"v\" VARIANT)"),
+ Arguments.of(MAP(INT(), VARIANT()), "MAP< INTEGER, VARIANT
>"));
+ }
+
+ private String convertToSpecAndUnparse(LogicalType logicalType) {
+ RelDataType relDataType =
+ LogicalRelDataTypeConverter.toRelDataType(logicalType,
typeFactory);
SqlDataTypeSpec typeSpec = SqlTypeUtil.convertTypeToSpec(relDataType);
- SqlWriter writer =
- new SqlPrettyWriter(
- SqlPrettyWriter.config()
- .withAlwaysUseParentheses(false)
- .withSelectListItemsOnSeparateLines(false)
- .withIndentation(0));
+ writer.reset();
// unparse that will end up passing no comments through
typeSpec.unparse(writer, 0, 0);
- String result = writer.toSqlString().getSql();
- assertThat(result)
- .hasToString("ROW(\"a\" INTEGER, \"b\" VARCHAR(1) CHARACTER
SET \"UTF-16LE\")");
+ return writer.toSqlString().getSql();
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
index d9f49877bdd..c3653a2e194 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
@@ -111,6 +111,28 @@ public class VariantSemanticTest extends SemanticTestBase {
.runSql("INSERT INTO sink_t SELECT TRY_PARSE_JSON(v) FROM
t")
.build();
+ static final TableTestProgram VARIANT_OMITTED_FROM_INSERT_COLUMN_LIST =
+ TableTestProgram.of(
+ "variant-omitted-from-insert-column-list",
+ "validates that VARIANT columns omitted from an
INSERT column list are filled with NULL")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema("k INT")
+ .producedValues(Row.of(1))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "k INT",
+ "v VARIANT",
+ "a ARRAY<VARIANT>",
+ "r ROW<f VARIANT>",
+ "m MAP<STRING, VARIANT>")
+ .consumedValues(Row.of(1, null, null,
null, null))
+ .build())
+ .runSql("INSERT INTO sink_t (k) SELECT k FROM t")
+ .build();
+
static final Variant V1 = BUILDER.of(1);
static final Variant V2 = BUILDER.of(2);
@@ -495,7 +517,8 @@ public class VariantSemanticTest extends SemanticTestBase {
VARIANT_OBJECT_ACCESS,
VARIANT_NESTED_ACCESS,
VARIANT_ARRAY_ERROR_ACCESS,
- VARIANT_OBJECT_ERROR_ACCESS);
+ VARIANT_OBJECT_ERROR_ACCESS,
+ VARIANT_OMITTED_FROM_INSERT_COLUMN_LIST);
}
public static class VariantIdentity extends ScalarFunction {
