You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2021/08/09 11:47:35 UTC
[flink] branch master updated: [FLINK-23479][table-planner] Fix the
unstable test cases about json plan
This is an automated email from the ASF dual-hosted git repository.
godfrey 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 785c72a [FLINK-23479][table-planner] Fix the unstable test cases about json plan
785c72a is described below
commit 785c72a5e9dc317584be92e9f174b652650ebe72
Author: godfreyhe <go...@163.com>
AuthorDate: Mon Jul 26 21:46:55 2021 +0800
[FLINK-23479][table-planner] Fix the unstable test cases about json plan
The reason about the unstable test cases is the inner serializer instance in RawType is serialized directly as String to json which is mutable. While for a RawType representing DataView, the inner logical type is stable and accessible, we should serialize the inner logical type to make the json plan stable. For other case, we just keep the previous behavior.
This closes #16599
---
.../exec/serde/LogicalTypeJsonDeserializer.java | 55 +++++
.../exec/serde/LogicalTypeJsonSerializer.java | 146 ++++++++++---
.../table/planner/typeutils/DataViewUtils.java | 42 +++-
.../nodes/exec/serde/LogicalTypeSerdeTest.java | 82 ++++++-
.../exec/stream/GroupAggregateJsonPlanTest.java | 12 +-
.../jsonplan/GroupAggregateJsonPlanITCase.java | 48 +++-
...tDistinctAggCalls[isMiniBatchEnabled=false].out | 124 +++++++----
...stDistinctAggCalls[isMiniBatchEnabled=true].out | 242 ++++++++++++++++-----
.../testIncrementalAggregate.out | 28 ++-
.../testDistinctSplitEnabled.out | 28 ++-
.../testEventTimeTumbleWindow.out | 76 ++++++-
.../testEventTimeTumbleWindowWithOffset.out | 76 ++++++-
.../testEventTimeTumbleWindow.out | 56 ++++-
.../runtime/typeutils/ExternalSerializer.java | 8 +
14 files changed, 863 insertions(+), 160 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonDeserializer.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonDeserializer.java
index a6173d2..a2a040f 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonDeserializer.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonDeserializer.java
@@ -20,7 +20,11 @@ package org.apache.flink.table.planner.plan.nodes.exec.serde;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.dataview.ListView;
+import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.planner.typeutils.DataViewUtils;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BinaryType;
import org.apache.flink.table.types.logical.CharType;
@@ -30,6 +34,7 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RawType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.table.types.logical.SymbolType;
@@ -40,6 +45,8 @@ import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.ZonedTimestampType;
import org.apache.flink.table.types.logical.utils.LogicalTypeParser;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+import org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter;
import org.apache.flink.table.utils.EncodingUtils;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
@@ -55,6 +62,7 @@ import java.util.List;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_ATTRIBUTES;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_COMPARISON;
+import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_DATA_VIEW_CLASS;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_DESCRIPTION;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_ELEMENT_TYPE;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_FIELDS;
@@ -62,6 +70,7 @@ import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJs
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_IDENTIFIER;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_IMPLEMENTATION_CLASS;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_INSTANTIABLE;
+import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_IS_INTERNAL_TYPE;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_KEY_TYPE;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_LENGTH;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.LogicalTypeJsonSerializer.FIELD_NAME_LOGICAL_TYPE;
@@ -130,6 +139,11 @@ public class LogicalTypeJsonDeserializer extends StdDeserializer<LogicalType> {
return deserializeArrayType(logicalTypeNode, serdeCtx);
case MULTISET:
return deserializeMultisetType(logicalTypeNode, serdeCtx);
+ case RAW:
+ // This method only deserializes the LogicalType with `type='RAW'`,
+ // otherwise it will fallback to
+ // `LogicalTypeParser.parse(logicalTypeNode.asText())`
+ return deserializeRawType(logicalTypeNode, serdeCtx);
default:
throw new TableException("Unsupported type name:" + typeName);
}
@@ -355,4 +369,45 @@ public class LogicalTypeJsonDeserializer extends StdDeserializer<LogicalType> {
TimestampKind.valueOf(logicalTypeNode.get(FIELD_NAME_TIMESTAMP_KIND).asText());
return new LocalZonedTimestampType(nullable, timestampKind, precision);
}
+
+ private RawType<?> deserializeRawType(JsonNode logicalTypeNode, SerdeContext serdeCtx) {
+ if (!logicalTypeNode.has(FIELD_NAME_DATA_VIEW_CLASS)) {
+ throw new TableException("Only RowType for DataView class is supported now");
+ }
+ boolean nullable = logicalTypeNode.get(FIELD_NAME_NULLABLE).asBoolean();
+ String dataViewClass = logicalTypeNode.get(FIELD_NAME_DATA_VIEW_CLASS).asText();
+ final DataType dataViewDataType;
+ if (MapView.class.getName().equals(dataViewClass)) {
+ DataType keyDataType =
+ deserializeDataTypeForDataView(logicalTypeNode, FIELD_NAME_KEY_TYPE, serdeCtx);
+ DataType valueDataType =
+ deserializeDataTypeForDataView(
+ logicalTypeNode, FIELD_NAME_VALUE_TYPE, serdeCtx);
+ dataViewDataType = MapView.newMapViewDataType(keyDataType, valueDataType);
+ } else if (ListView.class.getName().equals(dataViewClass)) {
+ DataType elementDataType =
+ deserializeDataTypeForDataView(
+ logicalTypeNode, FIELD_NAME_ELEMENT_TYPE, serdeCtx);
+ dataViewDataType = ListView.newListViewDataType(elementDataType);
+ } else {
+ throw new TableException("Only MapView and ListView are supported now");
+ }
+ // we only Use DataViewUtils.adjustDataViews method to create the DataType for DataView,
+ // so the hasStateBackedDataViews is always false
+ DataType dataType = DataViewUtils.adjustDataViews(dataViewDataType, false);
+ RawType<?> rawType = (RawType<?>) LogicalTypeDataTypeConverter.toLogicalType(dataType);
+ return (RawType<?>) rawType.copy(nullable);
+ }
+
+ private DataType deserializeDataTypeForDataView(
+ JsonNode logicalTypeNode, String key, SerdeContext serdeCtx) {
+ JsonNode jsonNode = logicalTypeNode.get(key);
+ LogicalType logicalType = deserialize(jsonNode.get(FIELD_NAME_TYPE_NAME), serdeCtx);
+ DataType dataType = LogicalTypeDataTypeConverter.toDataType(logicalType);
+ boolean isInternalType = jsonNode.get(FIELD_NAME_IS_INTERNAL_TYPE).asBoolean();
+ if (isInternalType) {
+ dataType = DataTypeUtils.toInternalDataType(dataType);
+ }
+ return dataType;
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java
index 6e48e38..e86332d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeJsonSerializer.java
@@ -18,7 +18,14 @@
package org.apache.flink.table.planner.plan.nodes.exec.serde;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.dataview.ListView;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.planner.typeutils.DataViewUtils;
+import org.apache.flink.table.runtime.typeutils.ExternalSerializer;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BinaryType;
import org.apache.flink.table.types.logical.CharType;
@@ -28,6 +35,7 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RawType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.table.types.logical.SymbolType;
@@ -37,6 +45,8 @@ import org.apache.flink.table.types.logical.UnresolvedUserDefinedType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.ZonedTimestampType;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+import org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter;
import org.apache.flink.table.utils.EncodingUtils;
import org.apache.flink.util.Preconditions;
@@ -85,6 +95,9 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
public static final String FIELD_NAME_VALUE_TYPE = "valueType";
// ArrayType/MultiSetType
public static final String FIELD_NAME_ELEMENT_TYPE = "elementType";
+ // RawType for DataView
+ public static final String FIELD_NAME_DATA_VIEW_CLASS = "dataViewClass";
+ public static final String FIELD_NAME_IS_INTERNAL_TYPE = "isInternal";
public LogicalTypeJsonSerializer() {
super(LogicalType.class);
@@ -98,37 +111,37 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
throws IOException {
if (logicalType instanceof CharType) {
// Zero-length character strings have no serializable string representation.
- serialize((CharType) logicalType, jsonGenerator);
+ serializeRowType((CharType) logicalType, jsonGenerator);
} else if (logicalType instanceof VarCharType) {
// Zero-length character strings have no serializable string representation.
- serialize((VarCharType) logicalType, jsonGenerator);
+ serializeVarCharType((VarCharType) logicalType, jsonGenerator);
} else if (logicalType instanceof BinaryType) {
// Zero-length binary strings have no serializable string representation.
- serialize((BinaryType) logicalType, jsonGenerator);
+ serializeBinaryType((BinaryType) logicalType, jsonGenerator);
} else if (logicalType instanceof VarBinaryType) {
// Zero-length binary strings have no serializable string representation.
- serialize((VarBinaryType) logicalType, jsonGenerator);
+ serializeVarBinaryType((VarBinaryType) logicalType, jsonGenerator);
} else if (logicalType instanceof SymbolType) {
// SymbolType does not support `asSerializableString`
- serialize((SymbolType<?>) logicalType, jsonGenerator);
+ serializeSymbolType((SymbolType<?>) logicalType, jsonGenerator);
} else if (logicalType instanceof TypeInformationRawType) {
// TypeInformationRawType does not support `asSerializableString`
- serialize((TypeInformationRawType<?>) logicalType, jsonGenerator);
+ serializeTypeInformationRawType((TypeInformationRawType<?>) logicalType, jsonGenerator);
} else if (logicalType instanceof StructuredType) {
// StructuredType does not full support `asSerializableString`
- serialize((StructuredType) logicalType, jsonGenerator);
+ serializeStructuredType((StructuredType) logicalType, jsonGenerator);
} else if (logicalType instanceof DistinctType) {
// DistinctType does not full support `asSerializableString`
- serialize((DistinctType) logicalType, jsonGenerator);
+ serializeDistinctType((DistinctType) logicalType, jsonGenerator);
} else if (logicalType instanceof TimestampType) {
// TimestampType does not consider `TimestampKind`
- serialize((TimestampType) logicalType, jsonGenerator);
+ serializeTimestampType((TimestampType) logicalType, jsonGenerator);
} else if (logicalType instanceof ZonedTimestampType) {
// ZonedTimestampType does not consider `TimestampKind`
- serialize((ZonedTimestampType) logicalType, jsonGenerator);
+ serializeZonedTimestampType((ZonedTimestampType) logicalType, jsonGenerator);
} else if (logicalType instanceof LocalZonedTimestampType) {
// LocalZonedTimestampType does not consider `TimestampKind`
- serialize((LocalZonedTimestampType) logicalType, jsonGenerator);
+ serializeLocalZonedTimestampType((LocalZonedTimestampType) logicalType, jsonGenerator);
} else if (logicalType instanceof RowType) {
serializeRowType((RowType) logicalType, jsonGenerator, serializerProvider);
} else if (logicalType instanceof MapType) {
@@ -137,6 +150,8 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
serializeArrayType((ArrayType) logicalType, jsonGenerator, serializerProvider);
} else if (logicalType instanceof MultisetType) {
serializeMultisetType((MultisetType) logicalType, jsonGenerator, serializerProvider);
+ } else if (logicalType instanceof RawType) {
+ serializeRawType((RawType<?>) logicalType, jsonGenerator, serializerProvider);
} else if (logicalType instanceof UnresolvedUserDefinedType) {
throw new TableException(
"Can not serialize an UnresolvedUserDefinedType instance. \n"
@@ -205,47 +220,49 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeEndObject();
}
- private void serialize(CharType charType, JsonGenerator jsonGenerator) throws IOException {
+ private void serializeRowType(CharType charType, JsonGenerator jsonGenerator)
+ throws IOException {
// Zero-length character strings have no serializable string representation.
if (charType.getLength() == CharType.EMPTY_LITERAL_LENGTH) {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, charType.getTypeRoot().name());
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, charType.isNullable());
- jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, charType.getLength());
+ jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, 0);
jsonGenerator.writeEndObject();
} else {
jsonGenerator.writeObject(charType.asSerializableString());
}
}
- private void serialize(VarCharType varCharType, JsonGenerator jsonGenerator)
+ private void serializeVarCharType(VarCharType varCharType, JsonGenerator jsonGenerator)
throws IOException {
// Zero-length character strings have no serializable string representation.
if (varCharType.getLength() == VarCharType.EMPTY_LITERAL_LENGTH) {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, varCharType.getTypeRoot().name());
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, varCharType.isNullable());
- jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, varCharType.getLength());
+ jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, 0);
jsonGenerator.writeEndObject();
} else {
jsonGenerator.writeObject(varCharType.asSerializableString());
}
}
- private void serialize(BinaryType binaryType, JsonGenerator jsonGenerator) throws IOException {
+ private void serializeBinaryType(BinaryType binaryType, JsonGenerator jsonGenerator)
+ throws IOException {
// Zero-length binary strings have no serializable string representation.
if (binaryType.getLength() == BinaryType.EMPTY_LITERAL_LENGTH) {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, binaryType.getTypeRoot().name());
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, binaryType.isNullable());
- jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, binaryType.getLength());
+ jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, 0);
jsonGenerator.writeEndObject();
} else {
jsonGenerator.writeObject(binaryType.asSerializableString());
}
}
- private void serialize(VarBinaryType varBinaryType, JsonGenerator jsonGenerator)
+ private void serializeVarBinaryType(VarBinaryType varBinaryType, JsonGenerator jsonGenerator)
throws IOException {
// Zero-length binary strings have no serializable string representation.
if (varBinaryType.getLength() == VarBinaryType.EMPTY_LITERAL_LENGTH) {
@@ -253,14 +270,14 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeStringField(
FIELD_NAME_TYPE_NAME, varBinaryType.getTypeRoot().name());
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, varBinaryType.isNullable());
- jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, varBinaryType.getLength());
+ jsonGenerator.writeNumberField(FIELD_NAME_LENGTH, 0);
jsonGenerator.writeEndObject();
} else {
jsonGenerator.writeObject(varBinaryType.asSerializableString());
}
}
- private void serialize(SymbolType<?> symbolType, JsonGenerator jsonGenerator)
+ private void serializeSymbolType(SymbolType<?> symbolType, JsonGenerator jsonGenerator)
throws IOException {
jsonGenerator.writeStartObject();
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, symbolType.isNullable());
@@ -269,8 +286,8 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeEndObject();
}
- private void serialize(TypeInformationRawType<?> rawType, JsonGenerator jsonGenerator)
- throws IOException {
+ private void serializeTypeInformationRawType(
+ TypeInformationRawType<?> rawType, JsonGenerator jsonGenerator) throws IOException {
jsonGenerator.writeStartObject();
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, rawType.isNullable());
jsonGenerator.writeStringField(
@@ -279,7 +296,7 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeEndObject();
}
- private void serialize(StructuredType structuredType, JsonGenerator jsonGenerator)
+ private void serializeStructuredType(StructuredType structuredType, JsonGenerator jsonGenerator)
throws IOException {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(
@@ -322,7 +339,7 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeEndObject();
}
- private void serialize(DistinctType distinctType, JsonGenerator jsonGenerator)
+ private void serializeDistinctType(DistinctType distinctType, JsonGenerator jsonGenerator)
throws IOException {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, LogicalTypeRoot.DISTINCT_TYPE.name());
@@ -337,7 +354,7 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeEndObject();
}
- private void serialize(TimestampType timestampType, JsonGenerator jsonGenerator)
+ private void serializeTimestampType(TimestampType timestampType, JsonGenerator jsonGenerator)
throws IOException {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, timestampType.getTypeRoot().name());
@@ -347,8 +364,8 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeEndObject();
}
- private void serialize(ZonedTimestampType timestampType, JsonGenerator jsonGenerator)
- throws IOException {
+ private void serializeZonedTimestampType(
+ ZonedTimestampType timestampType, JsonGenerator jsonGenerator) throws IOException {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, timestampType.getTypeRoot().name());
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, timestampType.isNullable());
@@ -357,8 +374,8 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeEndObject();
}
- private void serialize(LocalZonedTimestampType timestampType, JsonGenerator jsonGenerator)
- throws IOException {
+ private void serializeLocalZonedTimestampType(
+ LocalZonedTimestampType timestampType, JsonGenerator jsonGenerator) throws IOException {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, timestampType.getTypeRoot().name());
jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, timestampType.isNullable());
@@ -366,4 +383,75 @@ public class LogicalTypeJsonSerializer extends StdSerializer<LogicalType> {
jsonGenerator.writeObjectField(FIELD_NAME_TIMESTAMP_KIND, timestampType.getKind());
jsonGenerator.writeEndObject();
}
+
+ @SuppressWarnings("rawtypes")
+ private void serializeRawType(
+ RawType<?> rawType, JsonGenerator jsonGenerator, SerializerProvider serializerProvider)
+ throws IOException {
+ TypeSerializer<?> typeSer = rawType.getTypeSerializer();
+ if (typeSer instanceof ExternalSerializer) {
+ ExternalSerializer externalSer = (ExternalSerializer) typeSer;
+ // Currently, ExternalSerializer with `isInternalInput=false` will be serialized,
+ // Once `isInternalInput=true` needs to be serialized, we can add individual field in
+ // the json to support it, and the new json plan is compatible with the previous one.
+ if (externalSer.isInternalInput()) {
+ throw new TableException(
+ "ExternalSerializer with `isInternalInput=true` is not supported.");
+ }
+ DataType dataType = externalSer.getDataType();
+ boolean isMapView = DataViewUtils.isMapViewDataType(dataType);
+ boolean isListView = DataViewUtils.isListViewDataType(dataType);
+ if (isMapView || isListView) {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeStringField(FIELD_NAME_TYPE_NAME, LogicalTypeRoot.RAW.name());
+ jsonGenerator.writeBooleanField(FIELD_NAME_NULLABLE, rawType.isNullable());
+ if (isMapView) {
+ jsonGenerator.writeStringField(
+ FIELD_NAME_DATA_VIEW_CLASS, MapView.class.getName());
+ KeyValueDataType keyValueDataType =
+ DataViewUtils.extractKeyValueDataTypeForMapView(dataType);
+ serializeDataTypeForDataView(
+ FIELD_NAME_KEY_TYPE,
+ keyValueDataType.getKeyDataType(),
+ jsonGenerator,
+ serializerProvider);
+ serializeDataTypeForDataView(
+ FIELD_NAME_VALUE_TYPE,
+ keyValueDataType.getValueDataType(),
+ jsonGenerator,
+ serializerProvider);
+ } else {
+ jsonGenerator.writeStringField(
+ FIELD_NAME_DATA_VIEW_CLASS, ListView.class.getName());
+ DataType elementType =
+ DataViewUtils.extractElementDataTypeForListView(dataType);
+ serializeDataTypeForDataView(
+ FIELD_NAME_ELEMENT_TYPE,
+ elementType,
+ jsonGenerator,
+ serializerProvider);
+ }
+ jsonGenerator.writeEndObject();
+ return;
+ }
+ }
+
+ jsonGenerator.writeObject(rawType.asSerializableString());
+ }
+
+ private void serializeDataTypeForDataView(
+ String key,
+ DataType dataType,
+ JsonGenerator jsonGenerator,
+ SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeFieldName(key);
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeBooleanField(
+ FIELD_NAME_IS_INTERNAL_TYPE, DataTypeUtils.isInternal(dataType));
+ jsonGenerator.writeFieldName(FIELD_NAME_TYPE_NAME);
+ LogicalType logicalType = LogicalTypeDataTypeConverter.toLogicalType(dataType);
+ serialize(logicalType, jsonGenerator, serializerProvider);
+ jsonGenerator.writeEndObject();
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/DataViewUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/DataViewUtils.java
index 648af47..52c7bd9 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/DataViewUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/DataViewUtils.java
@@ -30,6 +30,7 @@ import org.apache.flink.table.dataview.NullSerializer;
import org.apache.flink.table.runtime.typeutils.ExternalSerializer;
import org.apache.flink.table.types.CollectionDataType;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.FieldsDataType;
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.table.types.inference.TypeTransformation;
import org.apache.flink.table.types.logical.LogicalType;
@@ -138,7 +139,7 @@ public final class DataViewUtils {
// --------------------------------------------------------------------------------------------
- private static boolean isDataView(LogicalType t, Class<? extends DataView> viewClass) {
+ public static boolean isDataView(LogicalType t, Class<? extends DataView> viewClass) {
return hasRoot(t, LogicalTypeRoot.STRUCTURED_TYPE)
&& ((StructuredType) t)
.getImplementationClass()
@@ -146,6 +147,45 @@ public final class DataViewUtils {
.orElse(false);
}
+ /**
+ * Check if the given data type represents a {@link ListView}. This method must be in sync with
+ * {@link ListView#newListViewDataType(DataType)}.
+ */
+ public static boolean isListViewDataType(DataType dataType) {
+ return dataType.getConversionClass().equals(ListView.class)
+ && dataType instanceof FieldsDataType
+ && dataType.getChildren().size() == 1
+ && dataType.getChildren().get(0) instanceof CollectionDataType;
+ }
+
+ public static DataType extractElementDataTypeForListView(DataType dataType) {
+ if (!isListViewDataType(dataType)) {
+ throw new TableException(
+ "The given type: " + dataType + " is not the expected type for ListView.");
+ }
+ CollectionDataType collectionDataType = (CollectionDataType) dataType.getChildren().get(0);
+ return collectionDataType.getElementDataType();
+ }
+
+ /**
+ * Check if the given data type represents a {@link MapView}. This method must be in sync with
+ * {@link MapView#newMapViewDataType(DataType, DataType)}.
+ */
+ public static boolean isMapViewDataType(DataType dataType) {
+ return dataType.getConversionClass().equals(MapView.class)
+ && dataType instanceof FieldsDataType
+ && dataType.getChildren().size() == 1
+ && dataType.getChildren().get(0) instanceof KeyValueDataType;
+ }
+
+ public static KeyValueDataType extractKeyValueDataTypeForMapView(DataType dataType) {
+ if (!isMapViewDataType(dataType)) {
+ throw new TableException(
+ "The given type: " + dataType + " is not the expected type for MapView.");
+ }
+ return (KeyValueDataType) dataType.getChildren().get(0);
+ }
+
private static String createStateId(int fieldIndex, String fieldName) {
return "agg" + fieldIndex + "$" + fieldName;
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeSerdeTest.java
index 37f1f86..4306b45 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeSerdeTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeSerdeTest.java
@@ -22,11 +22,15 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.dataview.ListView;
+import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.expressions.TimeIntervalUnit;
import org.apache.flink.table.planner.calcite.FlinkContextImpl;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable;
+import org.apache.flink.table.planner.typeutils.DataViewUtils;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BinaryType;
@@ -60,6 +64,7 @@ import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.YearMonthIntervalType;
import org.apache.flink.table.types.logical.ZonedTimestampType;
+import org.apache.flink.table.types.utils.DataTypeUtils;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
@@ -76,6 +81,8 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import static org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter.toDataType;
+import static org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter.toLogicalType;
import static org.junit.Assert.assertEquals;
/** Tests for {@link LogicalType} serialization and deserialization. */
@@ -242,15 +249,78 @@ public class LogicalTypeSerdeTest {
new RawType<>(
PojoClass.class,
new KryoSerializer<>(PojoClass.class, new ExecutionConfig())));
- List<LogicalType> newTypes = new ArrayList<>();
+ List<LogicalType> mutableTypes = new ArrayList<>(types);
+ // RawType for MapView
+ addRawTypesForMapView(mutableTypes, new VarCharType(100), new VarCharType(100));
+ addRawTypesForMapView(mutableTypes, new VarCharType(100), new BigIntType());
+ addRawTypesForMapView(mutableTypes, new BigIntType(), new VarCharType(100));
+ addRawTypesForMapView(mutableTypes, new BigIntType(), new BigIntType());
+ // RawType for ListView
+ addRawTypesForListView(mutableTypes, new VarCharType(100));
+ addRawTypesForListView(mutableTypes, new BigIntType());
+
+ List<LogicalType> finalTypes = new ArrayList<>();
// consider nullable
- for (LogicalType type : types) {
- newTypes.add(type.copy(true));
- newTypes.add(type.copy(false));
+ for (LogicalType type : mutableTypes) {
+ finalTypes.add(type.copy(true));
+ finalTypes.add(type.copy(false));
}
// ignore nullable for NullType
- newTypes.add(new NullType());
- return newTypes;
+ finalTypes.add(new NullType());
+ return finalTypes;
+ }
+
+ private static void addRawTypesForMapView(
+ List<LogicalType> types, LogicalType keyType, LogicalType valueType) {
+ for (boolean hasStateBackedDataViews : Arrays.asList(true, false)) {
+ for (boolean keyNullable : Arrays.asList(true, false)) {
+ for (boolean isInternalKeyType : Arrays.asList(true, false)) {
+ for (boolean valueNullable : Arrays.asList(true, false)) {
+ for (boolean isInternalValueType : Arrays.asList(true, false)) {
+ types.add(
+ toLogicalType(
+ DataViewUtils.adjustDataViews(
+ MapView.newMapViewDataType(
+ convertToInternalTypeIfNeeded(
+ toDataType(
+ keyType.copy(
+ keyNullable)),
+ isInternalKeyType),
+ convertToInternalTypeIfNeeded(
+ toDataType(
+ valueType.copy(
+ valueNullable)),
+ isInternalValueType)),
+ hasStateBackedDataViews)));
+ }
+ }
+ }
+ }
+ }
+ }
+
+ private static void addRawTypesForListView(List<LogicalType> types, LogicalType elementType) {
+ for (boolean hasStateBackedDataViews : Arrays.asList(true, false)) {
+ for (boolean elementNullable : Arrays.asList(true, false)) {
+ for (boolean isInternalType : Arrays.asList(true, false)) {
+ types.add(
+ toLogicalType(
+ DataViewUtils.adjustDataViews(
+ ListView.newListViewDataType(
+ convertToInternalTypeIfNeeded(
+ toDataType(
+ elementType.copy(
+ elementNullable)),
+ isInternalType)),
+ hasStateBackedDataViews)));
+ }
+ }
+ }
+ }
+
+ private static DataType convertToInternalTypeIfNeeded(
+ DataType dataType, boolean isInternalType) {
+ return isInternalType ? DataTypeUtils.toInternalDataType(dataType) : dataType;
}
/** Testing class. */
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest.java
index aaff6e5..86238e0 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest.java
@@ -134,25 +134,27 @@ public class GroupAggregateJsonPlanTest extends TableTestBase {
public void testDistinctAggCalls() {
String sinkTableDdl =
"CREATE TABLE MySink (\n"
- + " c varchar,\n"
+ + " d bigint,\n"
+ " cnt_a1 bigint,\n"
+ " cnt_a2 bigint,\n"
+ " sum_a bigint,\n"
+ " sum_b int,\n"
- + " avg_b double\n"
+ + " avg_b double,\n"
+ + " cnt_c bigint\n"
+ ") with (\n"
+ " 'connector' = 'values',\n"
+ " 'sink-insert-only' = 'false',\n"
+ " 'table-sink-class' = 'DEFAULT')";
tEnv.executeSql(sinkTableDdl);
util.verifyJsonPlan(
- "insert into MySink select c, "
+ "insert into MySink select d, "
+ "count(distinct a) filter (where b > 10) as cnt_a1, "
+ "count(distinct a) as cnt_a2, "
+ "sum(distinct a) as sum_a, "
+ "sum(distinct b) as sum_b, "
- + "avg(b) as avg_b "
- + "from MyTable group by c");
+ + "avg(b) as avg_b, "
+ + "count(distinct c) as cnt_d "
+ + "from MyTable group by d");
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/GroupAggregateJsonPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/GroupAggregateJsonPlanITCase.java
index 09751f9..8356729 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/GroupAggregateJsonPlanITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/GroupAggregateJsonPlanITCase.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.runtime.stream.jsonplan;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions;
import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.VarSum1AggFunction;
import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.VarSum2AggFunction;
import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.WeightedAvg;
@@ -115,6 +116,7 @@ public class GroupAggregateJsonPlanITCase extends JsonPlanTestBase {
"sum_a bigint",
"sum_b bigint",
"avg_b double",
+ "cnt_d bigint",
"primary key (e) not enforced");
String jsonPlan =
tableEnv.getJsonPlan(
@@ -123,21 +125,22 @@ public class GroupAggregateJsonPlanITCase extends JsonPlanTestBase {
+ "count(distinct a) as cnt_a2, "
+ "sum(distinct a) as sum_a, "
+ "sum(distinct b) as sum_b, "
- + "avg(b) as avg_b "
+ + "avg(b) as avg_b, "
+ + "count(distinct d) as concat_d "
+ "from MyTable group by e");
tableEnv.executeJsonPlan(jsonPlan).await();
List<String> result = TestValuesTableFactory.getResults("MySink");
assertResult(
Arrays.asList(
- "+I[1, 1, 4, 12, 32, 6.0]",
- "+I[2, 1, 4, 14, 57, 8.0]",
- "+I[3, 1, 2, 8, 31, 10.0]"),
+ "+I[1, 1, 4, 12, 32, 6.0, 5]",
+ "+I[2, 1, 4, 14, 57, 8.0, 7]",
+ "+I[3, 1, 2, 8, 31, 10.0, 3]"),
result);
}
@Test
- public void testUserDefinedAggCalls() throws Exception {
+ public void testUserDefinedAggCallsWithoutMerge() throws Exception {
tableEnv.createTemporaryFunction("my_sum1", new VarSum1AggFunction());
tableEnv.createFunction("my_avg", WeightedAvg.class);
tableEnv.createTemporarySystemFunction("my_sum2", VarSum2AggFunction.class);
@@ -172,4 +175,39 @@ public class GroupAggregateJsonPlanITCase extends JsonPlanTestBase {
assertResult(
Arrays.asList("+I[1, 77, 0, 1]", "+I[2, 120, 0, 2]", "+I[3, 58, 0, 3]"), result);
}
+
+ @Test
+ public void testUserDefinedAggCallsWithMerge() throws Exception {
+ tableEnv.createFunction("my_avg", JavaUserDefinedAggFunctions.WeightedAvgWithMerge.class);
+ tableEnv.createTemporarySystemFunction(
+ "my_concat_agg", JavaUserDefinedAggFunctions.ConcatDistinctAggFunction.class);
+
+ createTestValuesSourceTable(
+ "MyTable",
+ JavaScalaConversionUtil.toJava(TestData.data2()),
+ "a int",
+ "b bigint",
+ "c int",
+ "d varchar",
+ "e bigint");
+ createTestNonInsertOnlyValuesSinkTable(
+ "MySink", "d bigint", "s1 bigint", "c1 varchar", "primary key (d) not enforced");
+
+ String jsonPlan =
+ tableEnv.getJsonPlan(
+ "insert into MySink select "
+ + "e, "
+ + "my_avg(e, a) as s1, "
+ + "my_concat_agg(d) as c1 "
+ + "from MyTable group by e");
+ tableEnv.executeJsonPlan(jsonPlan).await();
+
+ List<String> result = TestValuesTableFactory.getResults("MySink");
+ assertResult(
+ Arrays.asList(
+ "+I[1, 1, Hallo Welt wie|Hallo|GHI|EFG|DEF]",
+ "+I[2, 2, Hallo Welt wie gehts?|Hallo Welt|ABC|FGH|CDE|JKL|KLM]",
+ "+I[3, 3, HIJ|IJK|BCD]"),
+ result);
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out
index f79817a..0971eda 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out
@@ -19,50 +19,36 @@
"bounded" : "false",
"schema.0.name" : "a",
"schema.1.data-type" : "INT NOT NULL"
- },
- "sourceAbilitySpecs" : [ {
- "type" : "ProjectPushDown",
- "projectedFields" : [ [ 2 ], [ 0 ], [ 1 ] ],
- "producedType" : {
- "type" : "ROW",
- "nullable" : false,
- "fields" : [ {
- "c" : "VARCHAR(2147483647)"
- }, {
- "a" : "BIGINT"
- }, {
- "b" : "INT NOT NULL"
- } ]
- }
- } ]
+ }
},
"id" : 1,
"outputType" : {
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
- }, {
"a" : "BIGINT"
}, {
"b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ }, {
+ "d" : "BIGINT"
} ]
},
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[c, a, b]]], fields=[c, a, b])",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])",
"inputProperties" : [ ]
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 0,
+ "inputIndex" : 3,
"type" : {
- "typeName" : "VARCHAR",
- "nullable" : true,
- "precision" : 2147483647
+ "typeName" : "BIGINT",
+ "nullable" : true
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 1,
+ "inputIndex" : 0,
"type" : {
"typeName" : "BIGINT",
"nullable" : true
@@ -76,7 +62,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "INTEGER",
"nullable" : false
@@ -95,11 +81,19 @@
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "INTEGER",
"nullable" : false
}
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : {
+ "typeName" : "VARCHAR",
+ "nullable" : true,
+ "precision" : 2147483647
+ }
} ],
"condition" : null,
"id" : 2,
@@ -114,16 +108,18 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"a" : "BIGINT"
}, {
"$f2" : "BOOLEAN NOT NULL"
}, {
"b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
} ]
},
- "description" : "Calc(select=[c, a, (b > 10) AS $f2, b])"
+ "description" : "Calc(select=[d, a, (b > 10) AS $f2, b, c])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
"id" : 3,
@@ -139,16 +135,18 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"a" : "BIGINT"
}, {
"$f2" : "BOOLEAN NOT NULL"
}, {
"b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
} ]
},
- "description" : "Exchange(distribution=[hash[c]])"
+ "description" : "Exchange(distribution=[hash[d]])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecGroupAggregate",
"grouping" : [ 0 ],
@@ -232,8 +230,24 @@
"typeName" : "INTEGER",
"nullable" : false
}
+ }, {
+ "name" : "cnt_d",
+ "aggFunction" : {
+ "name" : "COUNT",
+ "kind" : "COUNT",
+ "syntax" : "FUNCTION_STAR"
+ },
+ "argList" : [ 4 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : false
+ }
} ],
- "aggCallNeedRetractions" : [ false, false, false, false, false ],
+ "aggCallNeedRetractions" : [ false, false, false, false, false, false ],
"generateUpdateBefore" : true,
"needRetraction" : false,
"id" : 4,
@@ -248,7 +262,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"cnt_a1" : "BIGINT NOT NULL"
}, {
@@ -259,18 +273,19 @@
"sum_b" : "INT NOT NULL"
}, {
"avg_b" : "INT NOT NULL"
+ }, {
+ "cnt_d" : "BIGINT NOT NULL"
} ]
},
- "description" : "GroupAggregate(groupBy=[c], select=[c, COUNT(DISTINCT a) FILTER $f2 AS cnt_a1, COUNT(DISTINCT a) AS cnt_a2, SUM(DISTINCT a) AS sum_a, SUM(DISTINCT b) AS sum_b, AVG(b) AS avg_b])"
+ "description" : "GroupAggregate(groupBy=[d], select=[d, COUNT(DISTINCT a) FILTER $f2 AS cnt_a1, COUNT(DISTINCT a) AS cnt_a2, SUM(DISTINCT a) AS sum_a, SUM(DISTINCT b) AS sum_b, AVG(b) AS avg_b, COUNT(DISTINCT c) AS cnt_d])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
"inputIndex" : 0,
"type" : {
- "typeName" : "VARCHAR",
- "nullable" : true,
- "precision" : 2147483647
+ "typeName" : "BIGINT",
+ "nullable" : true
}
}, {
"kind" : "REX_CALL",
@@ -355,6 +370,25 @@
"typeName" : "DOUBLE",
"nullable" : true
}
+ }, {
+ "kind" : "REX_CALL",
+ "operator" : {
+ "name" : "CAST",
+ "kind" : "CAST",
+ "syntax" : "SPECIAL"
+ },
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : false
+ }
+ } ],
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
} ],
"condition" : null,
"id" : 5,
@@ -369,7 +403,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"cnt_a1" : "BIGINT"
}, {
@@ -380,9 +414,11 @@
"sum_b" : "INT"
}, {
"avg_b" : "DOUBLE"
+ }, {
+ "cnt_c" : "BIGINT"
} ]
},
- "description" : "Calc(select=[c, CAST(cnt_a1) AS cnt_a1, CAST(cnt_a2) AS cnt_a2, sum_a, CAST(sum_b) AS sum_b, CAST(avg_b) AS avg_b])"
+ "description" : "Calc(select=[d, CAST(cnt_a1) AS cnt_a1, CAST(cnt_a2) AS cnt_a2, sum_a, CAST(sum_b) AS sum_b, CAST(avg_b) AS avg_b, CAST(cnt_d) AS cnt_c])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink",
"dynamicTableSink" : {
@@ -394,9 +430,10 @@
"catalogTable" : {
"schema.5.name" : "avg_b",
"sink-insert-only" : "false",
- "schema.0.data-type" : "VARCHAR(2147483647)",
+ "schema.0.data-type" : "BIGINT",
"schema.2.name" : "cnt_a2",
"schema.1.name" : "cnt_a1",
+ "schema.6.data-type" : "BIGINT",
"schema.4.name" : "sum_b",
"schema.1.data-type" : "BIGINT",
"schema.3.data-type" : "BIGINT",
@@ -404,9 +441,10 @@
"schema.2.data-type" : "BIGINT",
"schema.3.name" : "sum_a",
"connector" : "values",
+ "schema.6.name" : "cnt_c",
"schema.5.data-type" : "DOUBLE",
"schema.4.data-type" : "INT",
- "schema.0.name" : "c"
+ "schema.0.name" : "d"
}
},
"inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER" ],
@@ -422,7 +460,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"cnt_a1" : "BIGINT"
}, {
@@ -433,9 +471,11 @@
"sum_b" : "INT"
}, {
"avg_b" : "DOUBLE"
+ }, {
+ "cnt_c" : "BIGINT"
} ]
},
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[c, cnt_a1, cnt_a2, sum_a, sum_b, avg_b])"
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[d, cnt_a1, cnt_a2, sum_a, sum_b, avg_b, cnt_c])"
} ],
"edges" : [ {
"source" : 1,
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out
index c2eb568..f98d378 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out
@@ -19,36 +19,23 @@
"bounded" : "false",
"schema.0.name" : "a",
"schema.1.data-type" : "INT NOT NULL"
- },
- "sourceAbilitySpecs" : [ {
- "type" : "ProjectPushDown",
- "projectedFields" : [ [ 2 ], [ 0 ], [ 1 ] ],
- "producedType" : {
- "type" : "ROW",
- "nullable" : false,
- "fields" : [ {
- "c" : "VARCHAR(2147483647)"
- }, {
- "a" : "BIGINT"
- }, {
- "b" : "INT NOT NULL"
- } ]
- }
- } ]
+ }
},
"id" : 1,
"outputType" : {
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
- }, {
"a" : "BIGINT"
}, {
"b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ }, {
+ "d" : "BIGINT"
} ]
},
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[c, a, b]]], fields=[c, a, b])",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])",
"inputProperties" : [ ]
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecMiniBatchAssigner",
@@ -68,11 +55,13 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
- }, {
"a" : "BIGINT"
}, {
"b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ }, {
+ "d" : "BIGINT"
} ]
},
"description" : "MiniBatchAssigner(interval=[10000ms], mode=[ProcTime])"
@@ -80,15 +69,14 @@
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 0,
+ "inputIndex" : 3,
"type" : {
- "typeName" : "VARCHAR",
- "nullable" : true,
- "precision" : 2147483647
+ "typeName" : "BIGINT",
+ "nullable" : true
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 1,
+ "inputIndex" : 0,
"type" : {
"typeName" : "BIGINT",
"nullable" : true
@@ -102,7 +90,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "INTEGER",
"nullable" : false
@@ -121,11 +109,19 @@
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "INTEGER",
"nullable" : false
}
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : {
+ "typeName" : "VARCHAR",
+ "nullable" : true,
+ "precision" : 2147483647
+ }
} ],
"condition" : null,
"id" : 3,
@@ -140,16 +136,18 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"a" : "BIGINT"
}, {
"$f2" : "BOOLEAN NOT NULL"
}, {
"b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
} ]
},
- "description" : "Calc(select=[c, a, (b > 10) AS $f2, b])"
+ "description" : "Calc(select=[d, a, (b > 10) AS $f2, b, c])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLocalGroupAggregate",
"grouping" : [ 0 ],
@@ -233,8 +231,24 @@
"typeName" : "INTEGER",
"nullable" : false
}
+ }, {
+ "name" : "cnt_d",
+ "aggFunction" : {
+ "name" : "COUNT",
+ "kind" : "COUNT",
+ "syntax" : "FUNCTION_STAR"
+ },
+ "argList" : [ 4 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : false
+ }
} ],
- "aggCallNeedRetractions" : [ false, false, false, false, false ],
+ "aggCallNeedRetractions" : [ false, false, false, false, false, false ],
"needRetraction" : false,
"id" : 4,
"inputProperties" : [ {
@@ -248,7 +262,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"count$0" : "BIGINT"
}, {
@@ -262,12 +276,52 @@
}, {
"count$5" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "count$6" : "BIGINT"
+ }, {
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "BIGINT"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
- "distinct$1" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$1" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "INT NOT NULL"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
+ }, {
+ "distinct$2" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
} ]
},
- "description" : "LocalGroupAggregate(groupBy=[c], select=[c, COUNT(distinct$0 a) FILTER $f2 AS count$0, COUNT(distinct$0 a) AS count$1, SUM(distinct$0 a) AS sum$2, SUM(distinct$1 b) AS sum$3, AVG(b) AS (sum$4, count$5), DISTINCT(a) AS distinct$0, DISTINCT(b) AS distinct$1])"
+ "description" : "LocalGroupAggregate(groupBy=[d], select=[d, COUNT(distinct$0 a) FILTER $f2 AS count$0, COUNT(distinct$0 a) AS count$1, SUM(distinct$0 a) AS sum$2, SUM(distinct$1 b) AS sum$3, AVG(b) AS (sum$4, count$5), COUNT(distinct$2 c) AS count$6, DISTINCT(a) AS distinct$0, DISTINCT(b) AS distinct$1, DISTINCT(c) AS distinct$2])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
"id" : 5,
@@ -283,7 +337,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"count$0" : "BIGINT"
}, {
@@ -297,12 +351,52 @@
}, {
"count$5" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "count$6" : "BIGINT"
+ }, {
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "BIGINT"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
- "distinct$1" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$1" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "INT NOT NULL"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
+ }, {
+ "distinct$2" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
} ]
},
- "description" : "Exchange(distribution=[hash[c]])"
+ "description" : "Exchange(distribution=[hash[d]])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecGlobalGroupAggregate",
"grouping" : [ 0 ],
@@ -386,19 +480,37 @@
"typeName" : "INTEGER",
"nullable" : false
}
+ }, {
+ "name" : "cnt_d",
+ "aggFunction" : {
+ "name" : "COUNT",
+ "kind" : "COUNT",
+ "syntax" : "FUNCTION_STAR"
+ },
+ "argList" : [ 4 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : false
+ }
} ],
- "aggCallNeedRetractions" : [ false, false, false, false, false ],
+ "aggCallNeedRetractions" : [ false, false, false, false, false, false ],
"localAggInputRowType" : {
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"a" : "BIGINT"
}, {
"$f2" : "BOOLEAN NOT NULL"
}, {
"b" : "INT NOT NULL"
+ }, {
+ "c" : "VARCHAR(2147483647)"
} ]
},
"generateUpdateBefore" : true,
@@ -415,7 +527,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"cnt_a1" : "BIGINT NOT NULL"
}, {
@@ -426,18 +538,19 @@
"sum_b" : "INT NOT NULL"
}, {
"avg_b" : "INT NOT NULL"
+ }, {
+ "cnt_d" : "BIGINT NOT NULL"
} ]
},
- "description" : "GlobalGroupAggregate(groupBy=[c], select=[c, COUNT(distinct$0 count$0) AS cnt_a1, COUNT(distinct$0 count$1) AS cnt_a2, SUM(distinct$0 sum$2) AS sum_a, SUM(distinct$1 sum$3) AS sum_b, AVG((sum$4, count$5)) AS avg_b])"
+ "description" : "GlobalGroupAggregate(groupBy=[d], select=[d, COUNT(distinct$0 count$0) AS cnt_a1, COUNT(distinct$0 count$1) AS cnt_a2, SUM(distinct$0 sum$2) AS sum_a, SUM(distinct$1 sum$3) AS sum_b, AVG((sum$4, count$5)) AS avg_b, COUNT(distinct$2 count$6) AS cnt_d])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
"inputIndex" : 0,
"type" : {
- "typeName" : "VARCHAR",
- "nullable" : true,
- "precision" : 2147483647
+ "typeName" : "BIGINT",
+ "nullable" : true
}
}, {
"kind" : "REX_CALL",
@@ -522,6 +635,25 @@
"typeName" : "DOUBLE",
"nullable" : true
}
+ }, {
+ "kind" : "REX_CALL",
+ "operator" : {
+ "name" : "CAST",
+ "kind" : "CAST",
+ "syntax" : "SPECIAL"
+ },
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : false
+ }
+ } ],
+ "type" : {
+ "typeName" : "BIGINT",
+ "nullable" : true
+ }
} ],
"condition" : null,
"id" : 7,
@@ -536,7 +668,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"cnt_a1" : "BIGINT"
}, {
@@ -547,9 +679,11 @@
"sum_b" : "INT"
}, {
"avg_b" : "DOUBLE"
+ }, {
+ "cnt_c" : "BIGINT"
} ]
},
- "description" : "Calc(select=[c, CAST(cnt_a1) AS cnt_a1, CAST(cnt_a2) AS cnt_a2, sum_a, CAST(sum_b) AS sum_b, CAST(avg_b) AS avg_b])"
+ "description" : "Calc(select=[d, CAST(cnt_a1) AS cnt_a1, CAST(cnt_a2) AS cnt_a2, sum_a, CAST(sum_b) AS sum_b, CAST(avg_b) AS avg_b, CAST(cnt_d) AS cnt_c])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink",
"dynamicTableSink" : {
@@ -561,9 +695,10 @@
"catalogTable" : {
"schema.5.name" : "avg_b",
"sink-insert-only" : "false",
- "schema.0.data-type" : "VARCHAR(2147483647)",
+ "schema.0.data-type" : "BIGINT",
"schema.2.name" : "cnt_a2",
"schema.1.name" : "cnt_a1",
+ "schema.6.data-type" : "BIGINT",
"schema.4.name" : "sum_b",
"schema.1.data-type" : "BIGINT",
"schema.3.data-type" : "BIGINT",
@@ -571,9 +706,10 @@
"schema.2.data-type" : "BIGINT",
"schema.3.name" : "sum_a",
"connector" : "values",
+ "schema.6.name" : "cnt_c",
"schema.5.data-type" : "DOUBLE",
"schema.4.data-type" : "INT",
- "schema.0.name" : "c"
+ "schema.0.name" : "d"
}
},
"inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER" ],
@@ -589,7 +725,7 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "c" : "VARCHAR(2147483647)"
+ "d" : "BIGINT"
}, {
"cnt_a1" : "BIGINT"
}, {
@@ -600,9 +736,11 @@
"sum_b" : "INT"
}, {
"avg_b" : "DOUBLE"
+ }, {
+ "cnt_c" : "BIGINT"
} ]
},
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[c, cnt_a1, cnt_a2, sum_a, sum_b, avg_b])"
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[d, cnt_a1, cnt_a2, sum_a, sum_b, avg_b, cnt_c])"
} ],
"edges" : [ {
"source" : 1,
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out
index 64d9ba3..5f84036 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out
@@ -188,7 +188,19 @@
}, {
"count$0" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
} ]
},
"description" : "LocalGroupAggregate(groupBy=[a, $f2], partialFinalType=[PARTIAL], select=[a, $f2, COUNT(distinct$0 c) AS count$0, DISTINCT(c) AS distinct$0])"
@@ -213,7 +225,19 @@
}, {
"count$0" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
} ]
},
"description" : "Exchange(distribution=[hash[a, $f2]])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out
index ec272bb..7e131f5 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out
@@ -395,7 +395,19 @@
}, {
"count$2" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
@@ -426,7 +438,19 @@
}, {
"count$2" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
index 0e3ae6c..011f243 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
@@ -377,17 +377,49 @@
"implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$ConcatAcc",
"attributes" : [ {
"name" : "list",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.ListView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3 [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.ListView",
+ "elementType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ }
+ }
}, {
"name" : "map",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3J [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BOOLEAN"
+ }
+ }
} ],
"final" : true,
"instantiable" : true,
"comparison" : "NONE"
}
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
@@ -422,17 +454,49 @@
"implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$ConcatAcc",
"attributes" : [ {
"name" : "list",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.ListView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3 [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.ListView",
+ "elementType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ }
+ }
}, {
"name" : "map",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3J [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BOOLEAN"
+ }
+ }
} ],
"final" : true,
"instantiable" : true,
"comparison" : "NONE"
}
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out
index 81cf74d..da5f735 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out
@@ -378,17 +378,49 @@
"implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$ConcatAcc",
"attributes" : [ {
"name" : "list",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.ListView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3 [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.ListView",
+ "elementType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ }
+ }
}, {
"name" : "map",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3J [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BOOLEAN"
+ }
+ }
} ],
"final" : true,
"instantiable" : true,
"comparison" : "NONE"
}
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
@@ -423,17 +455,49 @@
"implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$ConcatAcc",
"attributes" : [ {
"name" : "list",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.ListView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3 [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.ListView",
+ "elementType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ }
+ }
}, {
"name" : "map",
- "logicalType" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3J [...]
+ "logicalType" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : false,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BOOLEAN"
+ }
+ }
} ],
"final" : true,
"instantiable" : true,
"comparison" : "NONE"
}
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
index 9ad8ac3..f432747 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
@@ -324,7 +324,19 @@
}, {
"count$1" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
@@ -351,7 +363,19 @@
}, {
"count$1" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
@@ -944,7 +968,19 @@
}, {
"count$1" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
@@ -971,7 +1007,19 @@
}, {
"count$1" : "BIGINT"
}, {
- "distinct$0" : "RAW('org.apache.flink.table.api.dataview.MapView', 'AFZvcmcuYXBhY2hlLmZsaW5rLnRhYmxlLnJ1bnRpbWUudHlwZXV0aWxzLkV4dGVybmFsU2VyaWFsaXplciRFeHRlcm5hbFNlcmlhbGl6ZXJTbmFwc2hvdAAAAAMADecEAAAAAaztAAVzcgArb3JnLmFwYWNoZS5mbGluay50YWJsZS50eXBlcy5GaWVsZHNEYXRhVHlwZfSwrBytgZ9fAgABTAAOZmllbGREYXRhVHlwZXN0ABBMamF2YS91dGlsL0xpc3Q7eHIAJW9yZy5hcGFjaGUuZmxpbmsudGFibGUudHlwZXMuRGF0YVR5cGV5y2rIj5/EeAIAAkwAD2NvbnZlcnNpb25DbGFzc3QAEUxqYXZhL2xhbmcvQ2xhc3M7TAALbG9naWNhbFR5cGV0ADJMb3JnL2Fw [...]
+ "distinct$0" : {
+ "type" : "RAW",
+ "nullable" : true,
+ "dataViewClass" : "org.apache.flink.table.api.dataview.MapView",
+ "keyType" : {
+ "isInternal" : true,
+ "type" : "VARCHAR(2147483647)"
+ },
+ "valueType" : {
+ "isInternal" : true,
+ "type" : "BIGINT NOT NULL"
+ }
+ }
}, {
"$slice_end" : "BIGINT"
} ]
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ExternalSerializer.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ExternalSerializer.java
index a277659..d9e7caa 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ExternalSerializer.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ExternalSerializer.java
@@ -93,6 +93,14 @@ public final class ExternalSerializer<I, E> extends TypeSerializer<E> {
dataType, InternalSerializers.create(dataType.getLogicalType()), isInternalInput);
}
+ public DataType getDataType() {
+ return dataType;
+ }
+
+ public boolean isInternalInput() {
+ return isInternalInput;
+ }
+
@Override
public boolean isImmutableType() {
return internalSerializer.isImmutableType();