sorry for a little error, the program code should be:
package poc.flink.table;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.math.BigDecimal;
import java.sql.Timestamp;
import static org.apache.flink.table.api.Expressions.$;
public class PocLastDecimalJob {
private static final Logger LOGGER =
LoggerFactory.getLogger(PocLastDecimalJob.class);
public static class LastDecimalAccumulator extends Tuple1<BigDecimal> {
public LastDecimalAccumulator(BigDecimal f0) {
super(f0);
}
}
public static class LastDecimalAggFunction extends
AggregateFunction<BigDecimal, LastDecimalAccumulator> {
@Override
public BigDecimal getValue(LastDecimalAccumulator accumulator) {
return accumulator.f0;
}
@Override
public LastDecimalAccumulator createAccumulator() {
return new LastDecimalAccumulator(null);
}
public void accumulate(LastDecimalAccumulator accumulator,
@DataTypeHint("DECIMAL(38, 18)")
BigDecimal value) {
if (value != null) {
accumulator.f0 = value;
}
}
public void merge(LastDecimalAccumulator accumulator,
Iterable<BigDecimal> iterable) {
if (iterable != null) {
for (BigDecimal item : iterable) {
accumulator.f0 = item;
}
}
}
}
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
RowTypeInfo rowTypeInfo = new RowTypeInfo(
new TypeInformation[] {Types.SQL_TIMESTAMP, Types.INT,
Types.BIG_DEC},
new String[] {"rowtime", "id", "val"});
DataStream<Row> dataStream = env.fromElements(
Row.of(new Timestamp(1_100L), 1, new BigDecimal("1")),
Row.of(new Timestamp(2_200L), 1, new BigDecimal("2")),
Row.of(new Timestamp(3_300L), 2, new BigDecimal("3"))
).assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()).returns(rowTypeInfo);
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
tableEnv.createTemporarySystemFunction("LAST_DECIMAL", new
LastDecimalAggFunction());
tableEnv.createTemporaryView("InputTable", dataStream,
$("rowtime").rowtime(), $("id"), $("val"));
Table resultTable = tableEnv.sqlQuery("" +
"SELECT LAST_TIMESTAMP(rowtime), id, LAST_DECIMAL(bd) " +
"FROM InputTable " +
"GROUP BY HOP(rowtime, INTERVAL '1' SECOND, INTERVAL
'1' SECOND), id");
DataStream<Row> resultStream = tableEnv
.toRetractStream(resultTable, new
RowTypeInfo(Types.SQL_TIMESTAMP, Types.INT, Types.BIG_DEC))
.map((MapFunction<Tuple2<Boolean, Row>, Row>) value ->
value.f1);
resultStream.addSink(new SinkFunction<Row>() {
@Override
public void invoke(Row value, Context context) {
LOGGER.info("SinkFunction.invoke(): value={}", value);
}
});
env.execute();
}
}
LIU Xiao <liuxiaogen...@gmail.com> 于2021年7月30日周五 上午11:04写道：
> I'm currently converting our old code (based on Flink 1.6) to Flink 1.13
> and encountered a strange problem about the user-defined aggregate function
> which takes BigDecimal as the parameter and output:
>
> Exception in thread "main" org.apache.flink.table.api.ValidationException:
>> SQL validation failed. An error occurred in the type inference logic of
>> function 'LAST_DECIMAL'.
>> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
>> $apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:157)
>> at
>> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:110)
>> at
>> org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:201)
>> at
>> org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:101)
>> at
>> org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:704)
>> at poc.flink.table.PocTimestampUdf.main(PocTimestampUdf.java:101)
>> Caused by: org.apache.flink.table.api.ValidationException: An error
>> occurred in the type inference logic of function 'LAST_DECIMAL'.
>> at
>> org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.convertToBridgingSqlFunction(FunctionCatalogOperatorTable.java:163)
>> at
>> org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.convertToSqlFunction(FunctionCatalogOperatorTable.java:146)
>> at
>> org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lambda$lookupOperatorOverloads$0(FunctionCatalogOperatorTable.java:100)
>> at java.base/java.util.Optional.flatMap(Optional.java:294)
>> at
>> org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:98)
>> at
>> org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:67)
>> at
>> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1183)
>> at
>> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1200)
>> at
>> org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1169)
>> at
>> org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:945)
>> at
>> org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:704)
>> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
>> $apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:152)
>> ... 5 more
>> Caused by: org.apache.flink.table.api.ValidationException: Could not
>> extract a valid type inference for function class
>> 'poc.flink.table.LastDecimalAggFunction'. Please check for implementation
>> mistakes and/or provide a corresponding hint.
>> at
>> org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:361)
>> at
>> org.apache.flink.table.types.extraction.TypeInferenceExtractor.extractTypeInference(TypeInferenceExtractor.java:150)
>> at
>> org.apache.flink.table.types.extraction.TypeInferenceExtractor.forAggregateFunction(TypeInferenceExtractor.java:98)
>> at
>> org.apache.flink.table.functions.AggregateFunction.getTypeInference(AggregateFunction.java:212)
>> at
>> org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.convertToBridgingSqlFunction(FunctionCatalogOperatorTable.java:160)
>> ... 16 more
>> Caused by: org.apache.flink.table.api.ValidationException: Error in
>> extracting a signature to output mapping.
>> at
>> org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:361)
>> at
>> org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractOutputMapping(FunctionMappingExtractor.java:117)
>> at
>> org.apache.flink.table.types.extraction.TypeInferenceExtractor.extractTypeInferenceOrError(TypeInferenceExtractor.java:161)
>> at
>> org.apache.flink.table.types.extraction.TypeInferenceExtractor.extractTypeInference(TypeInferenceExtractor.java:148)
>> ... 19 more
>> Caused by: org.apache.flink.table.api.ValidationException: Unable to
>> extract a type inference from method:
>> public void
>> poc.flink.table.LastDecimalAggFunction.accumulate(poc.flink.table.LastDecimalAccumulator,java.math.BigDecimal)
>> at
>> org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:361)
>> at
>> org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractResultMappings(FunctionMappingExtractor.java:183)
>> at
>> org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractOutputMapping(FunctionMappingExtractor.java:114)
>> ... 21 more
>> Caused by: org.apache.flink.table.api.ValidationException: Could not
>> extract a data type from 'class java.math.BigDecimal' in generic class
>> 'org.apache.flink.table.functions.AggregateFunction' in class
>> poc.flink.table.LastDecimalAggFunction. Please pass the required data type
>> manually or allow RAW types.
>> at
>> org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:361)
>> at
>> org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrRawWithTemplate(DataTypeExtractor.java:241)
>> at
>> org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrRaw(DataTypeExtractor.java:219)
>> at
>> org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeWithClassContext(DataTypeExtractor.java:195)
>> at
>> org.apache.flink.table.types.extraction.DataTypeExtractor.extractFromGeneric(DataTypeExtractor.java:125)
>> at
>> org.apache.flink.table.types.extraction.FunctionMappingExtractor.lambda$createGenericResultExtraction$13(FunctionMappingExtractor.java:478)
>> at
>> org.apache.flink.table.types.extraction.FunctionMappingExtractor.putExtractedResultMappings(FunctionMappingExtractor.java:319)
>> at
>> org.apache.flink.table.types.extraction.FunctionMappingExtractor.collectMethodMappings(FunctionMappingExtractor.java:269)
>> at
>> org.apache.flink.table.types.extraction.FunctionMappingExtractor.extractResultMappings(FunctionMappingExtractor.java:169)
>> ... 22 more
>> Caused by: org.apache.flink.table.api.ValidationException: Values of
>> 'java.math.BigDecimal' need fixed precision and scale.
>> at
>> org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:361)
>> at
>> org.apache.flink.table.types.extraction.ExtractionUtils.extractionError(ExtractionUtils.java:356)
>> at
>> org.apache.flink.table.types.extraction.DataTypeExtractor.extractPredefinedType(DataTypeExtractor.java:398)
>> at
>> org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrError(DataTypeExtractor.java:276)
>> at
>> org.apache.flink.table.types.extraction.DataTypeExtractor.extractDataTypeOrRawWithTemplate(DataTypeExtractor.java:233)
>> ... 29 more
>>
>
> The program source code:
>
> package poc.flink.table;
>
> import org.apache.flink.api.common.eventtime.WatermarkStrategy;
> import org.apache.flink.api.common.functions.MapFunction;
> import org.apache.flink.api.common.typeinfo.TypeInformation;
> import org.apache.flink.api.common.typeinfo.Types;
> import org.apache.flink.api.java.tuple.Tuple1;
> import org.apache.flink.api.java.tuple.Tuple2;
> import org.apache.flink.api.java.typeutils.RowTypeInfo;
> import org.apache.flink.streaming.api.datastream.DataStream;
> import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> import org.apache.flink.streaming.api.functions.sink.SinkFunction;
> import org.apache.flink.table.annotation.DataTypeHint;
> import org.apache.flink.table.api.Table;
> import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
> import org.apache.flink.table.functions.AggregateFunction;
> import org.apache.flink.types.Row;
> import org.slf4j.Logger;
> import org.slf4j.LoggerFactory;
>
> import java.math.BigDecimal;
> import java.sql.Timestamp;
>
> import static org.apache.flink.table.api.Expressions.$;
>
> public class PocLastDecimalJob {
> private static final Logger LOGGER =
> LoggerFactory.getLogger(PocLastDecimalJob.class);
>
> public static class LastDecimalAccumulator extends Tuple1<BigDecimal> {
> public LastDecimalAccumulator(BigDecimal f0) {
> super(f0);
> }
> }
>
> public static class LastDecimalAggFunction extends
> AggregateFunction<BigDecimal, LastDecimalAccumulator> {
>
> @Override
> public BigDecimal getValue(LastDecimalAccumulator accumulator) {
> return accumulator.f0;
> }
>
> @Override
> public LastDecimalAccumulator createAccumulator() {
> return new LastDecimalAccumulator(null);
> }
>
> public void accumulate(LastDecimalAccumulator accumulator,
> @DataTypeHint("DECIMAL(38, 18)") BigDecimal
> value) {
> if (value != null) {
> accumulator.f0 = value;
> }
> }
>
> public void merge(LastDecimalAccumulator accumulator,
> Iterable<BigDecimal> iterable) {
> if (iterable != null) {
> for (BigDecimal item : iterable) {
> accumulator.f0 = item;
> }
> }
> }
> }
>
> public static void main(String[] args) throws Exception {
>
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
>
> RowTypeInfo rowTypeInfo = new RowTypeInfo(
> new TypeInformation[] {Types.SQL_TIMESTAMP, Types.INT,
> Types.BIG_DEC},
> new String[] {"rowtime", "id", "val"});
>
> DataStream<Row> dataStream = env.fromElements(
> Row.of(new Timestamp(1_100L), 1, new BigDecimal("1")),
> Row.of(new Timestamp(2_200L), 1, new BigDecimal("2")),
> Row.of(new Timestamp(3_300L), 2, new BigDecimal("3"))
>
> ).assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()).returns(rowTypeInfo);
>
> StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
> tableEnv.createTemporarySystemFunction("LAST_DECIMAL", new
> LastDecimalAggFunction());
>
> tableEnv.createTemporaryView("InputTable", dataStream,
> $("rowtime.rowtime"), $("id"), $("val"));
>
> Table resultTable = tableEnv.sqlQuery("" +
> "SELECT LAST_TIMESTAMP(rowtime), id, LAST_DECIMAL(bd) " +
> "FROM InputTable " +
> "GROUP BY HOP(rowtime, INTERVAL '1' SECOND, INTERVAL '1'
> SECOND), id");
>
> DataStream<Row> resultStream = tableEnv
> .toRetractStream(resultTable, new
> RowTypeInfo(Types.SQL_TIMESTAMP, Types.INT, Types.BIG_DEC))
> .map((MapFunction<Tuple2<Boolean, Row>, Row>) value ->
> value.f1);
>
> resultStream.addSink(new SinkFunction<Row>() {
> @Override
> public void invoke(Row value, Context context) {
> LOGGER.info("SinkFunction.invoke(): value={}", value);
> }
> });
>
> env.execute();
> }
> }
>
>
