Skip to content

[FLINK-40672][table] Preserve nullability when casting VARIANT values #29202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dylanhz wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-40672][table] Preserve nullability when casting VARIANT values #29202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions ...ble-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3479,8 +3479,7 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
.callSyntax("CAST", SqlCallSyntax.CAST)
.kind(SCALAR)
.inputTypeStrategy(SpecificInputTypeStrategies.CAST)
.outputTypeStrategy(
nullableIfArgs(ConstantArgumentCount.to(0), TypeStrategies.argument(1)))
.outputTypeStrategy(SpecificTypeStrategies.CAST)
.build();

public static final BuiltInFunctionDefinition TRY_CAST =
Expand Down
16 changes: 16 additions & 0 deletions ...c/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.table.types.inference.TypeStrategies;
import org.apache.flink.table.types.inference.TypeStrategy;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;

import java.util.List;
Expand All @@ -43,6 +44,21 @@ public final class SpecificTypeStrategies {
/** See {@link UnusedTypeStrategy}. */
public static final TypeStrategy UNUSED = new UnusedTypeStrategy();

/** Type strategy specific for {@link BuiltInFunctionDefinitions#CAST}. */
public static final TypeStrategy CAST =
callContext -> {
final LogicalType sourceType =
callContext.getArgumentDataTypes().get(0).getLogicalType();
final DataType targetType = callContext.getArgumentDataTypes().get(1);
final boolean nullable =
sourceType.isNullable()
|| (sourceType.is(LogicalTypeRoot.VARIANT)
&& !targetType
.getLogicalType()
.is(LogicalTypeRoot.VARIANT));
return Optional.of(nullable ? targetType.nullable() : targetType.notNull());
};

/** See {@link RowTypeStrategy}. */
public static final TypeStrategy ROW = new RowTypeStrategy();

Expand Down
24 changes: 24 additions & 0 deletions ...table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import static org.apache.flink.table.types.inference.TypeStrategies.nullableIfAllArgs;
import static org.apache.flink.table.types.inference.TypeStrategies.nullableIfArgs;
import static org.apache.flink.table.types.inference.TypeStrategies.varyingString;
import static org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.CAST;
import static org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.PERCENTILE;

/** Tests for built-in {@link TypeStrategies}. */
Expand All @@ -40,6 +41,29 @@ class TypeStrategiesTest extends TypeStrategiesTestBase {
@Override
protected Stream<TestSpec> testData() {
return Stream.of(
TestSpec.forStrategy("Casting a non-null VARIANT to a scalar", CAST)
.inputTypes(DataTypes.VARIANT().notNull(), DataTypes.INT())
.expectDataType(DataTypes.INT()),
TestSpec.forStrategy("Casting a nullable VARIANT to a scalar", CAST)
.inputTypes(DataTypes.VARIANT(), DataTypes.INT().notNull())
.expectDataType(DataTypes.INT()),
TestSpec.forStrategy("Casting VARIANT preserves nested nullability", CAST)
.inputTypes(
DataTypes.VARIANT().notNull(),
DataTypes.ARRAY(DataTypes.INT().notNull()).notNull())
.expectDataType(DataTypes.ARRAY(DataTypes.INT().notNull())),
TestSpec.forStrategy("Casting a non-null VARIANT to VARIANT", CAST)
.inputTypes(DataTypes.VARIANT().notNull(), DataTypes.VARIANT())
.expectDataType(DataTypes.VARIANT().notNull()),
TestSpec.forStrategy("Casting a nullable VARIANT to VARIANT", CAST)
.inputTypes(DataTypes.VARIANT(), DataTypes.VARIANT().notNull())
.expectDataType(DataTypes.VARIANT()),
TestSpec.forStrategy("Casting a non-null scalar", CAST)
.inputTypes(DataTypes.INT().notNull(), DataTypes.BIGINT())
.expectDataType(DataTypes.BIGINT().notNull()),
TestSpec.forStrategy("Casting a nullable scalar", CAST)
.inputTypes(DataTypes.INT(), DataTypes.BIGINT().notNull())
.expectDataType(DataTypes.BIGINT()),
// missing strategy with arbitrary argument
TypeStrategiesTestBase.TestSpec.forStrategy(MISSING)
.inputTypes(DataTypes.INT())
Expand Down
33 changes: 33 additions & 0 deletions ...k-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,11 @@
* #checkOperandTypes(SqlCallBinding, boolean)} and new method {@link #canCastFrom(RelDataType,
* RelDataType)}.
*
* <p>Flink's {@link #deriveType(RelDataTypeFactory, RelDataType, RelDataType, boolean)} preserves
* the target type's nested nullability instead of delegating to {@link
* #createTypeWithNullabilityFromExpr(RelDataTypeFactory, RelDataType, RelDataType, boolean)}. The
* VARIANT rules introduced in Calcite 1.39.0 are also applied in {@code deriveType}.
*
* @see SqlCastOperator
*/
public class SqlCastFunction extends SqlFunction {
Expand Down Expand Up @@ -148,6 +153,21 @@ public static RelDataType deriveType(
RelDataType expressionType,
RelDataType targetType,
boolean safe) {
// Flink modification: apply the same VARIANT nullability rules as
// createTypeWithNullabilityFromExpr.
if (targetType.getSqlTypeName() == SqlTypeName.VARIANT) {
// A variant can be cast from any other type, and it inherits
// the nullability of the source.
// Note that the order of this test and the next one is important.
return typeFactory.createTypeWithNullability(targetType, expressionType.isNullable());
}

if (expressionType.getSqlTypeName() == SqlTypeName.VARIANT) {
// A variant can be cast to any other type, but the result
// is always nullable, like in the case of a safe cast.
return typeFactory.createTypeWithNullability(targetType, true);
}

return typeFactory.createTypeWithNullability(
targetType, expressionType.isNullable() || safe);
}
Expand All @@ -159,6 +179,19 @@ private static RelDataType createTypeWithNullabilityFromExpr(
boolean safe) {
boolean isNullable = expressionType.isNullable() || safe;

if (targetType.getSqlTypeName() == SqlTypeName.VARIANT) {
// A variant can be cast from any other type, and it inherits
// the nullability of the source.
// Note that the order of this test and the next one is important.
return typeFactory.createTypeWithNullability(targetType, expressionType.isNullable());
}

if (expressionType.getSqlTypeName() == SqlTypeName.VARIANT) {
// A variant can be cast to any other type, but the result
// is always nullable, like in the case of a safe cast.
return typeFactory.createTypeWithNullability(targetType, true);
}

if (isCollection(expressionType)) {
RelDataType expressionElementType = expressionType.getComponentType();
RelDataType targetElementType = targetType.getComponentType();
Expand Down
10 changes: 8 additions & 2 deletions ...n/java/org/apache/flink/table/planner/expressions/converter/converters/CastConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
import org.apache.flink.table.expressions.TypeLiteralExpression;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.planner.expressions.converter.CallExpressionConvertRule;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;

import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexNode;
Expand All @@ -36,10 +38,14 @@ public RexNode convert(CallExpression call, CallExpressionConvertRule.ConvertCon

final RexNode child = context.toRexNode(call.getChildren().get(0));
final TypeLiteralExpression targetType = (TypeLiteralExpression) call.getChildren().get(1);
final DataType inputType = call.getResolvedChildren().get(0).getOutputDataType();
final DataType resultType =
inputType.getLogicalType().is(LogicalTypeRoot.VARIANT)
? call.getOutputDataType()
: targetType.getOutputDataType();
final RelDataType targetRelDataType =
context.getTypeFactory()
.createFieldTypeFromLogicalType(
targetType.getOutputDataType().getLogicalType());
.createFieldTypeFromLogicalType(resultType.getLogicalType());

return context.getRelBuilder()
.getRexBuilder()
Expand Down
Loading