twalthr commented on a change in pull request #18063:
URL: https://github.com/apache/flink/pull/18063#discussion_r766698498
##########
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecks.java
##########
@@ -90,6 +90,10 @@
private static final FieldNamesExtractor FIELD_NAMES_EXTRACTOR = new
FieldNamesExtractor();
+ public static boolean isStringType(LogicalType type) {
Review comment:
can we drop this method? It is not really a shortcut and thus does not
provide much benefit.
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
Review comment:
`BINARY_STRING`?
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
+ *
+ * <p>This rule is used for casting from any of the {@link
LogicalTypeFamily#PREDEFINED} types to
+ * {@link LogicalTypeRoot#CHAR} or {@link LogicalTypeRoot#VARCHAR}. It calls
the underlying concrete
+ * matching rule, i.e.: {@link NumericToStringCastRule} to do the actual
conversion and then
+ * performs any necessary trimming or padding so that the length of the result
string value matches
+ * the one specified by the precision of the target {@link
LogicalTypeRoot#CHAR} or {@link
+ * LogicalTypeRoot#VARCHAR} type.
+ */
+class CharacterFamilyTrimmingAndPaddingCastRule
+ extends AbstractNullAwareCodeGeneratorCastRule<StringData, StringData>
{
+
+ static final CharacterFamilyTrimmingAndPaddingCastRule INSTANCE =
+ new CharacterFamilyTrimmingAndPaddingCastRule();
+
+ private CharacterFamilyTrimmingAndPaddingCastRule() {
+ super(
+ CastRulePredicate.builder()
+ .predicate(
+ (inputType, targetType) ->
+
targetType.is(LogicalTypeFamily.CHARACTER_STRING)
+ && !isStringType(targetType))
+ .build());
+ }
+
+ /* Example generated code for STRING() -> CHAR(6) cast
+
+ isNull$0 = _myInputIsNull;
+ if (!isNull$0) {
+ if (_myInput.numChars() > 6) {
+ result$1 = _myInput.substring(0, 6);
+ } else {
+ if (_myInput.numChars() < 6) {
+ int padLength$1;
+ padLength$1 = 6 - _myInput.numChars();
+ org.apache.flink.table.data.binary.BinaryStringData
padString$2;
+ padString$2 =
org.apache.flink.table.data.binary.BinaryStringData.blankString(padLength$1);
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringDataUtil.concat(_myInput,
padString$2);
+ } else {
+ result$1 = _myInput;
+ }
+ }
+ isNull$0 = result$1 == null;
+ } else {
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8;
+ }
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final int precision = LogicalTypeChecks.getLength(targetLogicalType);
+ CastRule<?, ?> castRule =
+ CastRuleProvider.resolve(inputLogicalType,
VarCharType.STRING_TYPE);
+
+ // Only used for non-Constructed types - for constructed, the
trimming/padding is applied
+ // on each individual rule, i.e.: ArrayToStringCastRule
+ if (castRule instanceof ExpressionCodeGeneratorCastRule) {
+ @SuppressWarnings("rawtypes")
+ final String stringExpr =
+ ((ExpressionCodeGeneratorCastRule) castRule)
+ .generateExpression(
+ context, inputTerm, inputLogicalType,
targetLogicalType);
+
+ CastRuleUtils.CodeWriter writer = new CastRuleUtils.CodeWriter();
+ if (context.legacyBehaviour()
+ || !(shouldPossiblyTrim(precision)
+ || shouldPossiblyPad(targetLogicalType,
precision))) {
+ return writer.assignStmt(returnVariable,
stringExpr).toString();
+ }
+ return writer.ifStmt(
+ methodCall(stringExpr, "numChars") + " > " +
precision,
+ thenWriter ->
+ thenWriter.assignStmt(
+ returnVariable,
+ methodCall(stringExpr,
"substring", 0, precision)),
+ elseWriter -> {
+ if (shouldPossiblyPad(targetLogicalType,
precision)) {
+ final String padLength =
newName("padLength");
+ final String padString =
newName("padString");
+ elseWriter.ifStmt(
+ methodCall(stringExpr, "numChars")
+ " < " + precision,
+ thenInnerWriter ->
+ thenInnerWriter
+
.declStmt(int.class, padLength)
+ .assignStmt(
+ padLength,
+ precision
+ +
" - "
+ +
methodCall(
+
stringExpr,
+
"numChars"))
+ .declStmt(
+
BinaryStringData.class,
+ padString)
+ .assignStmt(
+ padString,
+ staticCall(
+
BinaryStringData.class,
+
"blankString",
+
padLength))
+ .assignStmt(
+
returnVariable,
+ staticCall(
+
BinaryStringDataUtil
+
.class,
+
"concat",
+
stringExpr,
+
padString)),
+ elseInnerWriter ->
+ elseInnerWriter.assignStmt(
+ returnVariable,
stringExpr));
+ } else {
+ elseWriter.assignStmt(returnVariable,
stringExpr).toString();
+ }
+ })
+ .toString();
+ } else {
+ throw new IllegalStateException("This is a bug. Please file an
issue.");
+ }
+ }
+
+ static String lengthExceedsPrecision(String strTerm, int precision) {
+ return methodCall(strTerm, "length") + " > " + precision;
+ }
+
+ static String precisionExceedsLength(String strTerm, int precision) {
+ return methodCall(strTerm, "length") + " < " + precision;
+ }
+
+ static boolean shouldPossiblyTrim(int precision) {
+ return precision < VarCharType.MAX_LENGTH;
+ }
+
+ static boolean shouldPossiblyPad(LogicalType targetType, int precision) {
+ return targetType.is(LogicalTypeRoot.CHAR) && precision <
VarCharType.MAX_LENGTH;
+ }
Review comment:
add a comment to highlight the implicit code section
```
// ---------------
// Shared methods
// ---------------
```
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
+ *
+ * <p>This rule is used for casting from any of the {@link
LogicalTypeFamily#PREDEFINED} types to
+ * {@link LogicalTypeRoot#CHAR} or {@link LogicalTypeRoot#VARCHAR}. It calls
the underlying concrete
+ * matching rule, i.e.: {@link NumericToStringCastRule} to do the actual
conversion and then
+ * performs any necessary trimming or padding so that the length of the result
string value matches
+ * the one specified by the precision of the target {@link
LogicalTypeRoot#CHAR} or {@link
+ * LogicalTypeRoot#VARCHAR} type.
+ */
+class CharacterFamilyTrimmingAndPaddingCastRule
+ extends AbstractNullAwareCodeGeneratorCastRule<StringData, StringData>
{
+
+ static final CharacterFamilyTrimmingAndPaddingCastRule INSTANCE =
+ new CharacterFamilyTrimmingAndPaddingCastRule();
+
+ private CharacterFamilyTrimmingAndPaddingCastRule() {
+ super(
+ CastRulePredicate.builder()
+ .predicate(
+ (inputType, targetType) ->
+
targetType.is(LogicalTypeFamily.CHARACTER_STRING)
+ && !isStringType(targetType))
+ .build());
+ }
+
+ /* Example generated code for STRING() -> CHAR(6) cast
+
+ isNull$0 = _myInputIsNull;
+ if (!isNull$0) {
+ if (_myInput.numChars() > 6) {
+ result$1 = _myInput.substring(0, 6);
+ } else {
+ if (_myInput.numChars() < 6) {
+ int padLength$1;
+ padLength$1 = 6 - _myInput.numChars();
+ org.apache.flink.table.data.binary.BinaryStringData
padString$2;
+ padString$2 =
org.apache.flink.table.data.binary.BinaryStringData.blankString(padLength$1);
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringDataUtil.concat(_myInput,
padString$2);
+ } else {
+ result$1 = _myInput;
+ }
+ }
+ isNull$0 = result$1 == null;
+ } else {
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8;
+ }
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final int precision = LogicalTypeChecks.getLength(targetLogicalType);
+ CastRule<?, ?> castRule =
+ CastRuleProvider.resolve(inputLogicalType,
VarCharType.STRING_TYPE);
+
+ // Only used for non-Constructed types - for constructed, the
trimming/padding is applied
Review comment:
this comment is not very precise, RAW is neither CONSTRUCTED not
PREDEFINED
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/BooleanToStringCastRule.java
##########
@@ -33,8 +34,10 @@
private BooleanToStringCastRule() {
super(
CastRulePredicate.builder()
- .input(LogicalTypeRoot.BOOLEAN)
- .target(LogicalTypeFamily.CHARACTER_STRING)
+ .predicate(
Review comment:
how about we introduce a `.target(LogicalType)` instead? It can still
translate to this predicate. The input can stay as it is.
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java
##########
@@ -78,6 +78,7 @@
.addRule(ArrayToArrayCastRule.INSTANCE)
.addRule(RowToRowCastRule.INSTANCE)
// Special rules
+ .addRule(CharacterFamilyTrimmingAndPaddingCastRule.INSTANCE)
Review comment:
nit: maybe shorter: `CharVarCharTrimPadCastRule`?
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
+ *
+ * <p>This rule is used for casting from any of the {@link
LogicalTypeFamily#PREDEFINED} types to
+ * {@link LogicalTypeRoot#CHAR} or {@link LogicalTypeRoot#VARCHAR}. It calls
the underlying concrete
+ * matching rule, i.e.: {@link NumericToStringCastRule} to do the actual
conversion and then
+ * performs any necessary trimming or padding so that the length of the result
string value matches
+ * the one specified by the precision of the target {@link
LogicalTypeRoot#CHAR} or {@link
+ * LogicalTypeRoot#VARCHAR} type.
+ */
+class CharacterFamilyTrimmingAndPaddingCastRule
+ extends AbstractNullAwareCodeGeneratorCastRule<StringData, StringData>
{
+
+ static final CharacterFamilyTrimmingAndPaddingCastRule INSTANCE =
+ new CharacterFamilyTrimmingAndPaddingCastRule();
+
+ private CharacterFamilyTrimmingAndPaddingCastRule() {
+ super(
+ CastRulePredicate.builder()
+ .predicate(
+ (inputType, targetType) ->
+
targetType.is(LogicalTypeFamily.CHARACTER_STRING)
+ && !isStringType(targetType))
+ .build());
+ }
+
+ /* Example generated code for STRING() -> CHAR(6) cast
+
+ isNull$0 = _myInputIsNull;
+ if (!isNull$0) {
+ if (_myInput.numChars() > 6) {
+ result$1 = _myInput.substring(0, 6);
+ } else {
+ if (_myInput.numChars() < 6) {
+ int padLength$1;
+ padLength$1 = 6 - _myInput.numChars();
+ org.apache.flink.table.data.binary.BinaryStringData
padString$2;
+ padString$2 =
org.apache.flink.table.data.binary.BinaryStringData.blankString(padLength$1);
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringDataUtil.concat(_myInput,
padString$2);
+ } else {
+ result$1 = _myInput;
+ }
+ }
+ isNull$0 = result$1 == null;
+ } else {
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8;
+ }
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final int precision = LogicalTypeChecks.getLength(targetLogicalType);
+ CastRule<?, ?> castRule =
+ CastRuleProvider.resolve(inputLogicalType,
VarCharType.STRING_TYPE);
+
+ // Only used for non-Constructed types - for constructed, the
trimming/padding is applied
+ // on each individual rule, i.e.: ArrayToStringCastRule
+ if (castRule instanceof ExpressionCodeGeneratorCastRule) {
+ @SuppressWarnings("rawtypes")
+ final String stringExpr =
+ ((ExpressionCodeGeneratorCastRule) castRule)
+ .generateExpression(
+ context, inputTerm, inputLogicalType,
targetLogicalType);
+
+ CastRuleUtils.CodeWriter writer = new CastRuleUtils.CodeWriter();
Review comment:
nit: `final` otherwise it looks like the variable is mutated somewhere
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/ArrayToStringCastRule.java
##########
@@ -91,74 +121,94 @@ protected String generateCodeBlockInternal(
context.declareClassField(
className(StringBuilder.class), builderTerm,
constructorCall(StringBuilder.class));
- return new CastRuleUtils.CodeWriter()
- .stmt(methodCall(builderTerm, "setLength", 0))
- .stmt(methodCall(builderTerm, "append", strLiteral("[")))
- .forStmt(
- methodCall(inputTerm, "size"),
- (indexTerm, loopBodyWriter) -> {
- String elementTerm = newName("element");
- String elementIsNullTerm =
newName("elementIsNull");
+ final String resultStringTerm = newName("resultString");
+ final int precision = LogicalTypeChecks.getLength(targetLogicalType);
Review comment:
nit: `precision` -> `length` here and at other locations
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
+ *
+ * <p>This rule is used for casting from any of the {@link
LogicalTypeFamily#PREDEFINED} types to
+ * {@link LogicalTypeRoot#CHAR} or {@link LogicalTypeRoot#VARCHAR}. It calls
the underlying concrete
+ * matching rule, i.e.: {@link NumericToStringCastRule} to do the actual
conversion and then
+ * performs any necessary trimming or padding so that the length of the result
string value matches
+ * the one specified by the precision of the target {@link
LogicalTypeRoot#CHAR} or {@link
+ * LogicalTypeRoot#VARCHAR} type.
+ */
+class CharacterFamilyTrimmingAndPaddingCastRule
+ extends AbstractNullAwareCodeGeneratorCastRule<StringData, StringData>
{
Review comment:
input generic should be `Object`?
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
+ *
+ * <p>This rule is used for casting from any of the {@link
LogicalTypeFamily#PREDEFINED} types to
+ * {@link LogicalTypeRoot#CHAR} or {@link LogicalTypeRoot#VARCHAR}. It calls
the underlying concrete
+ * matching rule, i.e.: {@link NumericToStringCastRule} to do the actual
conversion and then
+ * performs any necessary trimming or padding so that the length of the result
string value matches
+ * the one specified by the precision of the target {@link
LogicalTypeRoot#CHAR} or {@link
+ * LogicalTypeRoot#VARCHAR} type.
+ */
+class CharacterFamilyTrimmingAndPaddingCastRule
+ extends AbstractNullAwareCodeGeneratorCastRule<StringData, StringData>
{
+
+ static final CharacterFamilyTrimmingAndPaddingCastRule INSTANCE =
+ new CharacterFamilyTrimmingAndPaddingCastRule();
+
+ private CharacterFamilyTrimmingAndPaddingCastRule() {
+ super(
+ CastRulePredicate.builder()
+ .predicate(
+ (inputType, targetType) ->
+
targetType.is(LogicalTypeFamily.CHARACTER_STRING)
+ && !isStringType(targetType))
+ .build());
+ }
+
+ /* Example generated code for STRING() -> CHAR(6) cast
+
+ isNull$0 = _myInputIsNull;
+ if (!isNull$0) {
+ if (_myInput.numChars() > 6) {
+ result$1 = _myInput.substring(0, 6);
+ } else {
+ if (_myInput.numChars() < 6) {
+ int padLength$1;
+ padLength$1 = 6 - _myInput.numChars();
+ org.apache.flink.table.data.binary.BinaryStringData
padString$2;
+ padString$2 =
org.apache.flink.table.data.binary.BinaryStringData.blankString(padLength$1);
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringDataUtil.concat(_myInput,
padString$2);
+ } else {
+ result$1 = _myInput;
+ }
+ }
+ isNull$0 = result$1 == null;
+ } else {
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8;
+ }
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final int precision = LogicalTypeChecks.getLength(targetLogicalType);
+ CastRule<?, ?> castRule =
+ CastRuleProvider.resolve(inputLogicalType,
VarCharType.STRING_TYPE);
+
+ // Only used for non-Constructed types - for constructed, the
trimming/padding is applied
+ // on each individual rule, i.e.: ArrayToStringCastRule
+ if (castRule instanceof ExpressionCodeGeneratorCastRule) {
+ @SuppressWarnings("rawtypes")
+ final String stringExpr =
+ ((ExpressionCodeGeneratorCastRule) castRule)
+ .generateExpression(
+ context, inputTerm, inputLogicalType,
targetLogicalType);
+
+ CastRuleUtils.CodeWriter writer = new CastRuleUtils.CodeWriter();
+ if (context.legacyBehaviour()
+ || !(shouldPossiblyTrim(precision)
+ || shouldPossiblyPad(targetLogicalType,
precision))) {
+ return writer.assignStmt(returnVariable,
stringExpr).toString();
+ }
+ return writer.ifStmt(
+ methodCall(stringExpr, "numChars") + " > " +
precision,
+ thenWriter ->
+ thenWriter.assignStmt(
+ returnVariable,
+ methodCall(stringExpr,
"substring", 0, precision)),
+ elseWriter -> {
+ if (shouldPossiblyPad(targetLogicalType,
precision)) {
+ final String padLength =
newName("padLength");
+ final String padString =
newName("padString");
+ elseWriter.ifStmt(
+ methodCall(stringExpr, "numChars")
+ " < " + precision,
+ thenInnerWriter ->
+ thenInnerWriter
+
.declStmt(int.class, padLength)
+ .assignStmt(
+ padLength,
+ precision
+ +
" - "
+ +
methodCall(
+
stringExpr,
+
"numChars"))
+ .declStmt(
+
BinaryStringData.class,
+ padString)
+ .assignStmt(
+ padString,
+ staticCall(
+
BinaryStringData.class,
+
"blankString",
+
padLength))
+ .assignStmt(
+
returnVariable,
+ staticCall(
+
BinaryStringDataUtil
+
.class,
+
"concat",
+
stringExpr,
+
padString)),
+ elseInnerWriter ->
+ elseInnerWriter.assignStmt(
+ returnVariable,
stringExpr));
+ } else {
+ elseWriter.assignStmt(returnVariable,
stringExpr).toString();
+ }
+ })
+ .toString();
+ } else {
+ throw new IllegalStateException("This is a bug. Please file an
issue.");
+ }
+ }
+
+ static String lengthExceedsPrecision(String strTerm, int precision) {
+ return methodCall(strTerm, "length") + " > " + precision;
+ }
+
+ static String precisionExceedsLength(String strTerm, int precision) {
+ return methodCall(strTerm, "length") + " < " + precision;
+ }
+
+ static boolean shouldPossiblyTrim(int precision) {
+ return precision < VarCharType.MAX_LENGTH;
+ }
+
+ static boolean shouldPossiblyPad(LogicalType targetType, int precision) {
+ return targetType.is(LogicalTypeRoot.CHAR) && precision <
VarCharType.MAX_LENGTH;
+ }
+
+ public static CastRuleUtils.CodeWriter appendAndTrimStringIfNeeded(
Review comment:
`appendAndTrimStringIfNeeded` -> `padAndTrimStringIfNeeded`
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
+ *
+ * <p>This rule is used for casting from any of the {@link
LogicalTypeFamily#PREDEFINED} types to
Review comment:
Only `PREDEFINED` or all types?
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CharacterFamilyTrimmingAndPaddingCastRule.java
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.casting;
+
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.data.binary.BinaryStringDataUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isStringType;
+
+/**
+ * Any source type to {@link LogicalTypeFamily#BINARY_STRING} cast rule.
+ *
+ * <p>This rule is used for casting from any of the {@link
LogicalTypeFamily#PREDEFINED} types to
+ * {@link LogicalTypeRoot#CHAR} or {@link LogicalTypeRoot#VARCHAR}. It calls
the underlying concrete
+ * matching rule, i.e.: {@link NumericToStringCastRule} to do the actual
conversion and then
+ * performs any necessary trimming or padding so that the length of the result
string value matches
+ * the one specified by the precision of the target {@link
LogicalTypeRoot#CHAR} or {@link
+ * LogicalTypeRoot#VARCHAR} type.
+ */
+class CharacterFamilyTrimmingAndPaddingCastRule
+ extends AbstractNullAwareCodeGeneratorCastRule<StringData, StringData>
{
+
+ static final CharacterFamilyTrimmingAndPaddingCastRule INSTANCE =
+ new CharacterFamilyTrimmingAndPaddingCastRule();
+
+ private CharacterFamilyTrimmingAndPaddingCastRule() {
+ super(
+ CastRulePredicate.builder()
+ .predicate(
+ (inputType, targetType) ->
+
targetType.is(LogicalTypeFamily.CHARACTER_STRING)
+ && !isStringType(targetType))
+ .build());
+ }
+
+ /* Example generated code for STRING() -> CHAR(6) cast
+
+ isNull$0 = _myInputIsNull;
+ if (!isNull$0) {
+ if (_myInput.numChars() > 6) {
+ result$1 = _myInput.substring(0, 6);
+ } else {
+ if (_myInput.numChars() < 6) {
+ int padLength$1;
+ padLength$1 = 6 - _myInput.numChars();
+ org.apache.flink.table.data.binary.BinaryStringData
padString$2;
+ padString$2 =
org.apache.flink.table.data.binary.BinaryStringData.blankString(padLength$1);
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringDataUtil.concat(_myInput,
padString$2);
+ } else {
+ result$1 = _myInput;
+ }
+ }
+ isNull$0 = result$1 == null;
+ } else {
+ result$1 =
org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8;
+ }
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final int precision = LogicalTypeChecks.getLength(targetLogicalType);
+ CastRule<?, ?> castRule =
+ CastRuleProvider.resolve(inputLogicalType,
VarCharType.STRING_TYPE);
+
+ // Only used for non-Constructed types - for constructed, the
trimming/padding is applied
+ // on each individual rule, i.e.: ArrayToStringCastRule
+ if (castRule instanceof ExpressionCodeGeneratorCastRule) {
+ @SuppressWarnings("rawtypes")
+ final String stringExpr =
+ ((ExpressionCodeGeneratorCastRule) castRule)
+ .generateExpression(
+ context, inputTerm, inputLogicalType,
targetLogicalType);
+
+ CastRuleUtils.CodeWriter writer = new CastRuleUtils.CodeWriter();
+ if (context.legacyBehaviour()
+ || !(shouldPossiblyTrim(precision)
+ || shouldPossiblyPad(targetLogicalType,
precision))) {
+ return writer.assignStmt(returnVariable,
stringExpr).toString();
+ }
+ return writer.ifStmt(
+ methodCall(stringExpr, "numChars") + " > " +
precision,
+ thenWriter ->
+ thenWriter.assignStmt(
+ returnVariable,
+ methodCall(stringExpr,
"substring", 0, precision)),
+ elseWriter -> {
+ if (shouldPossiblyPad(targetLogicalType,
precision)) {
+ final String padLength =
newName("padLength");
+ final String padString =
newName("padString");
+ elseWriter.ifStmt(
+ methodCall(stringExpr, "numChars")
+ " < " + precision,
+ thenInnerWriter ->
+ thenInnerWriter
+
.declStmt(int.class, padLength)
+ .assignStmt(
+ padLength,
+ precision
+ +
" - "
+ +
methodCall(
+
stringExpr,
+
"numChars"))
+ .declStmt(
+
BinaryStringData.class,
+ padString)
+ .assignStmt(
+ padString,
+ staticCall(
+
BinaryStringData.class,
+
"blankString",
+
padLength))
+ .assignStmt(
+
returnVariable,
+ staticCall(
+
BinaryStringDataUtil
+
.class,
+
"concat",
+
stringExpr,
+
padString)),
+ elseInnerWriter ->
+ elseInnerWriter.assignStmt(
+ returnVariable,
stringExpr));
+ } else {
+ elseWriter.assignStmt(returnVariable,
stringExpr).toString();
+ }
+ })
+ .toString();
+ } else {
+ throw new IllegalStateException("This is a bug. Please file an
issue.");
+ }
+ }
+
+ static String lengthExceedsPrecision(String strTerm, int precision) {
+ return methodCall(strTerm, "length") + " > " + precision;
+ }
+
+ static String precisionExceedsLength(String strTerm, int precision) {
+ return methodCall(strTerm, "length") + " < " + precision;
+ }
+
+ static boolean shouldPossiblyTrim(int precision) {
Review comment:
nit: `shouldPossiblyTrim` -> `shouldPossiblyTrim` "Possibly" is
redundant
##########
File path:
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RawToStringCastRule.java
##########
@@ -57,11 +94,19 @@ protected String generateCodeBlockInternal(
.ifStmt(
deserializedObjTerm + " != null",
thenWriter ->
- thenWriter.assignStmt(
- returnVariable,
- CastRuleUtils.staticCall(
-
BINARY_STRING_DATA_FROM_STRING(),
-
methodCall(deserializedObjTerm, "toString"))),
+ CharacterFamilyTrimmingAndPaddingCastRule
Review comment:
I think we can keep this rule unmodifed and let
`CharacterFamilyTrimmingAndPaddingCastRule` fire.
--
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]
