[flink] branch release-1.15 updated: [FLINK-26467][table] Compile RowDataToStringConverter lazily

[twalthr]

This is an automated email from the ASF dual-hosted git repository.

twalthr pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.15 by this push:
     new e4f0b539643 [FLINK-26467][table] Compile RowDataToStringConverter 
lazily
e4f0b539643 is described below

commit e4f0b539643f302f09fcaeac65bbdbe138a4fc40
Author: slinkydeveloper <francescogu...@gmail.com>
AuthorDate: Mon Mar 14 17:12:28 2022 +0100

    [FLINK-26467][table] Compile RowDataToStringConverter lazily
    
    This closes #19087.
---
 .../casting/RowDataToStringConverterImpl.java      | 53 +++++++++++-----------
 1 file changed, 27 insertions(+), 26 deletions(-)

diff --git 
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowDataToStringConverterImpl.java
 
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowDataToStringConverterImpl.java
index 3698e7f553a..fcbd429ebdf 100644
--- 
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowDataToStringConverterImpl.java
+++ 
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowDataToStringConverterImpl.java
@@ -25,22 +25,23 @@ import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.utils.CastExecutor;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VarCharType;
 import org.apache.flink.table.utils.DateTimeUtils;
 import org.apache.flink.table.utils.print.PrintStyle;
 import org.apache.flink.table.utils.print.RowDataToStringConverter;
 
 import java.time.ZoneId;
 import java.util.List;
-import java.util.Objects;
 import java.util.function.Function;
 
-import static org.apache.flink.table.api.DataTypes.STRING;
-
 /** {@link RowData} to {@link String} converter using {@link CastRule}. */
 @Internal
 public final class RowDataToStringConverterImpl implements 
RowDataToStringConverter {
 
-    private final Function<RowData, String>[] columnConverters;
+    private final DataType dataType;
+    private final CastRule.Context castRuleContext;
+
+    private Function<RowData, String>[] columnConverters;
 
     @VisibleForTesting
     public RowDataToStringConverterImpl(DataType dataType) {
@@ -51,9 +52,14 @@ public final class RowDataToStringConverterImpl implements 
RowDataToStringConver
                 false);
     }
 
-    @SuppressWarnings("unchecked")
     public RowDataToStringConverterImpl(
             DataType dataType, ZoneId zoneId, ClassLoader classLoader, boolean 
legacyBehaviour) {
+        this.dataType = dataType;
+        this.castRuleContext = CastRule.Context.create(legacyBehaviour, 
zoneId, classLoader);
+    }
+
+    @SuppressWarnings("unchecked")
+    private void init() {
         List<DataType> rowDataTypes = DataType.getFieldDataTypes(dataType);
         this.columnConverters = new Function[rowDataTypes.size()];
 
@@ -64,34 +70,29 @@ public final class RowDataToStringConverterImpl implements 
RowDataToStringConver
             CastExecutor<Object, StringData> castExecutor =
                     (CastExecutor<Object, StringData>)
                             CastRuleProvider.create(
-                                    CastRule.Context.create(legacyBehaviour, 
zoneId, classLoader),
-                                    fieldType,
-                                    STRING().getLogicalType());
+                                    castRuleContext, fieldType, 
VarCharType.STRING_TYPE);
             if (castExecutor == null) {
-                // Fallback in case no casting rule is defined, for example 
for MULTISET and
-                // STRUCTURED
-                // Links to https://issues.apache.org/jira/browse/FLINK-24403
-                this.columnConverters[index] =
-                        row -> {
-                            if (row.isNullAt(index)) {
-                                return PrintStyle.NULL_VALUE;
-                            }
-                            return 
Objects.toString(getter.getFieldOrNull(row));
-                        };
-            } else {
-                this.columnConverters[index] =
-                        row -> {
-                            if (row.isNullAt(index)) {
-                                return PrintStyle.NULL_VALUE;
-                            }
-                            return 
castExecutor.cast(getter.getFieldOrNull(row)).toString();
-                        };
+                throw new IllegalStateException(
+                        "Cannot create a cast executor for converting "
+                                + fieldType
+                                + " to string. This is a bug, please open an 
issue.");
             }
+            this.columnConverters[index] =
+                    row -> {
+                        if (row.isNullAt(index)) {
+                            return PrintStyle.NULL_VALUE;
+                        }
+                        return 
castExecutor.cast(getter.getFieldOrNull(row)).toString();
+                    };
         }
     }
 
     @Override
     public String[] convert(RowData rowData) {
+        if (this.columnConverters == null) {
+            init();
+        }
+
         String[] result = new String[rowData.getArity()];
         for (int i = 0; i < result.length; i++) {
             result[i] = this.columnConverters[i].apply(rowData);