aljoscha commented on a change in pull request #8685: [FLINK-12799][table]
Improve expression based TableSchema extraction from DataStream/DataSet
URL: https://github.com/apache/flink/pull/8685#discussion_r292471431
##########
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java
##########
@@ -381,105 +397,225 @@ public String getFieldName() {
public int getIndex() {
return index;
}
+
+ public DataType getType() {
+ return type;
+ }
}
- private static class IndexedExprToFieldInfo extends
ApiExpressionDefaultVisitor<Optional<FieldInfo>> {
+ private static class IndexedExprToFieldInfo extends
ApiExpressionDefaultVisitor<FieldInfo> {
+ private final CompositeType<?> inputType;
private final int index;
- private IndexedExprToFieldInfo(int index) {
+ private IndexedExprToFieldInfo(CompositeType<?> inputType, int
index) {
+ this.inputType = inputType;
this.index = index;
}
@Override
- public Optional<FieldInfo>
visitUnresolvedReference(UnresolvedReferenceExpression unresolvedReference) {
+ public FieldInfo
visitUnresolvedReference(UnresolvedReferenceExpression unresolvedReference) {
String fieldName = unresolvedReference.getName();
- return Optional.of(new FieldInfo(fieldName, index));
+ return new FieldInfo(fieldName, index,
fromLegacyInfoToDataType(getTypeAt(unresolvedReference)));
}
@Override
- public Optional<FieldInfo> visitCall(CallExpression call) {
+ public FieldInfo visitCall(CallExpression call) {
if (call.getFunctionDefinition() ==
BuiltInFunctionDefinitions.AS) {
- List<Expression> children = call.getChildren();
- Expression origExpr = children.get(0);
- String newName =
ExpressionUtils.extractValue(children.get(1), String.class)
- .orElseThrow(() ->
- new TableException("Alias
expects string literal as new name. Got: " + children.get(1)));
-
- if (origExpr instanceof
UnresolvedReferenceExpression) {
- throw new TableException(
- format("Alias '%s' is not
allowed if other fields are referenced by position.", newName));
- } else if (isTimeAttribute(origExpr)) {
- return Optional.empty();
- }
- } else if (isTimeAttribute(call)) {
- return Optional.empty();
+ return visitAlias(call);
+ } else if (isRowTimeExpression(call)) {
+ validateRowtimeReplacesCompatibleType(call);
+ return
createTimeAttributeField(getChildAsReference(call), TimestampKind.ROWTIME,
null);
+ } else if (isProcTimeExpression(call)) {
+ validateProcTimeAttributeAppended(call);
+ return
createTimeAttributeField(getChildAsReference(call), TimestampKind.PROCTIME,
null);
}
return defaultMethod(call);
}
+ private FieldInfo visitAlias(CallExpression call) {
+ List<Expression> children = call.getChildren();
+ String newName =
ExpressionUtils.extractValue(children.get(1), String.class).orElseThrow(() ->
+ new ValidationException("Alias expects
string literal as new name. Got: " + children.get(1)));
+
+ Expression child = call.getChildren().get(0);
+ if (isProcTimeExpression(child)) {
+ validateProcTimeAttributeAppended(call);
+ return
createTimeAttributeField(getChildAsReference(child), TimestampKind.PROCTIME,
newName);
+ } else {
+ throw new ValidationException(
+ format("Alias '%s' is not allowed if
other fields are referenced by position.", newName));
+ }
+ }
+
+ private void
validateRowtimeReplacesCompatibleType(CallExpression call) {
+ if (index < inputType.getArity()) {
+ checkRowtimeType(getTypeAt(call));
+ }
+ }
+
+ private void validateProcTimeAttributeAppended(CallExpression
call) {
+ if (index < inputType.getArity()) {
+ throw new
ValidationException(String.format("The proctime attribute can only be appended
to the" +
+ " table schema and not replace an
existing field. Please move '%s' to the end of the" +
+ " schema.", call));
+ }
+ }
+
+ private TypeInformation<Object> getTypeAt(Expression expr) {
+ if (index >= inputType.getArity()) {
+ throw new ValidationException(String.format(
+ "Number of expressions does not match
number of input fields.\n" +
+ "Available fields: %s\n" +
+ "Could not map: %s",
+
Arrays.toString(inputType.getFieldNames()),
+ expr));
+ }
+ return inputType.getTypeAt(index);
+ }
+
@Override
- protected Optional<FieldInfo> defaultMethod(Expression
expression) {
- throw new TableException("Field reference expression or
alias on field expression expected.");
+ protected FieldInfo defaultMethod(Expression expression) {
+ throw new ValidationException("Field reference
expression or alias on field expression expected.");
}
}
- private static class ExprToFieldInfo extends
ApiExpressionDefaultVisitor<Optional<FieldInfo>> {
+ private static class ExprToFieldInfo extends
ApiExpressionDefaultVisitor<FieldInfo> {
private final CompositeType ct;
private ExprToFieldInfo(CompositeType ct) {
this.ct = ct;
}
+ private ValidationException noFieldFound(String name) {
Review comment:
nitpick: my sense would say `fieldNotFound` but this is perfectly fine. 👌
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
