auroflow commented on code in PR #28924:
URL: https://github.com/apache/flink/pull/28924#discussion_r3795777433
##########
flink-python/src/main/java/org/apache/flink/table/utils/python/PythonTableUtils.java:
##########
@@ -127,6 +136,214 @@ public static Table createTableFromElement(
dataCollection,
InternalSerializers.create(dataType.getLogicalType()));
}
+ /**
+ * Creates a literal from a value received through Py4J.
+ *
+ * <p>Py4J represents Python numeric values as {@link Integer}, {@link
Long}, or {@link Double},
+ * which does not preserve the boxed Java classes required by some {@link
DataType}s. This
+ * method adapts the value to the data type's external representation and
creates the literal in
+ * the same JVM call so that the adapted value is not converted by Py4J
again. If {@code
+ * dataType} is absent, Java literal inference remains authoritative.
Constructed values are
+ * represented by constructor expressions because raw constructed value
literals cannot be
+ * planned.
+ *
+ * @param value the literal value received through Py4J
+ * @param dataType the declared data type, or {@code null} for type
inference
+ * @return the literal expression
+ * @throws ValidationException if the constructed value has no plannable
literal expression
+ */
+ public static ApiExpression createLiteral(
+ final Object value, @Nullable final DataType dataType) {
+ if (dataType != null) {
+ return createTypedLiteral(value, dataType);
+ }
+
+ final Object inferredValue = materializeInferredArrays(value);
+ final ApiExpression literal = Expressions.lit(inferredValue);
+ final DataType inferredDataType =
+ ((ValueLiteralExpression)
literal.toExpr()).getOutputDataType();
+ // Raw array literals can be inferred but not planned. Rebuild the
array as a constructor
+ // expression while preserving the data type inferred by Java.
+ if (inferredDataType.getLogicalType() instanceof ArrayType) {
+ return createTypedLiteral(inferredValue, inferredDataType);
+ }
+ return literal;
+ }
+
+ private static ApiExpression createTypedLiteral(final Object value, final
DataType dataType) {
+ if (value == null) {
+ // A typed null carries no composite payload and is directly
plannable.
+ return Expressions.lit(value, dataType);
+ }
+ if (dataType.getLogicalType().isNullable()) {
+ // Delegate the invalid non-null value/nullable type combination
to Java validation.
+ return Expressions.lit(value, dataType);
+ }
+ if (!usesDefaultLiteralConversion(dataType)) {
+ // Custom conversion classes are opaque to this bridge; use native
literal handling.
+ return Expressions.lit(value, dataType);
+ }
+
+ if (dataType.getLogicalType() instanceof ArrayType) {
+ if (!(value instanceof List) && !value.getClass().isArray()) {
+ // Delegate incompatible ARRAY representations to standard
literal validation.
+ return Expressions.lit(value, dataType);
+ }
+ final int length = getLiteralArrayLength(value);
+ if (length == 0) {
+ return createEmptyArray(dataType);
+ }
+ final DataType elementDataType = dataType.getChildren().get(0);
+ final Object[] tail = new Object[length - 1];
+ for (int pos = 1; pos < length; pos++) {
+ tail[pos - 1] =
+ createNestedLiteral(getLiteralArrayElement(value,
pos), elementDataType);
+ }
+ return Expressions.array(
+ createNestedLiteral(getLiteralArrayElement(value,
0), elementDataType),
+ tail)
+ .cast(dataType);
+ }
+ if (dataType.getLogicalType() instanceof RowType) {
+ if (!isLiteralRow(value)) {
+ // Delegate incompatible ROW representations to standard
literal validation.
+ return Expressions.lit(value, dataType);
+ }
+ final RowType rowType = (RowType) dataType.getLogicalType();
+ final List<DataType> fieldDataTypes = dataType.getChildren();
+ if (fieldDataTypes.isEmpty()) {
+ throw new ValidationException("Non-null empty ROW literals are
not supported.");
+ }
+ if (!(value instanceof Map)) {
+ final int valueArity = getLiteralRowArity(value);
+ if (valueArity != fieldDataTypes.size()) {
+ throw new ValidationException(
+ String.format(
+ "ROW literal has arity %d but the data
type has arity %d.",
+ valueArity, fieldDataTypes.size()));
+ }
+ }
+ final List<String> fieldNames = rowType.getFieldNames();
+ final Object[] tail = new Object[fieldDataTypes.size() - 1];
+ for (int pos = 1; pos < fieldDataTypes.size(); pos++) {
+ tail[pos - 1] =
+ createNestedLiteral(
+ getLiteralRowField(value, pos,
fieldNames.get(pos)),
+ fieldDataTypes.get(pos));
+ }
+ return Expressions.row(
Review Comment:
Agreed, I rejected non-INSERT rows 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
