class EvalRegressionBatchOp extends BatchOperator<EvalRegressionBatchOp> implements EvalRegressionParams<EvalRegressionBatchOp>, EvaluationMetricsCollector<RegressionMetrics, EvalRegressionBatchOp> {

    @Override
    public EvalRegressionBatchOp linkFrom(BatchOperator<?>... inputs) {
        BatchOperator in = checkAndGetFirst(inputs);

        TableUtil.findColIndexWithAssertAndHint(in.getColNames(), this.getLabelCol());
        TableUtil.findColIndexWithAssertAndHint(in.getColNames(), this.getPredictionCol());

        TableUtil.assertNumericalCols(in.getSchema(), this.getLabelCol(), this.getPredictionCol());
        DataSet<Row> filter = in.select(new String[]{this.getLabelCol(), this.getPredictionCol()})
            .getDataSet()
            .filter(new FilterFunction<Row>() {
                @Override
                public boolean filter(Row value) throws Exception {
                    return EvaluationUtil.checkRowFieldNotNull(value);
                }
            });

        DataSet<Row> out = filter
            .mapPartition(new CalcLocal())
            .withBroadcastSet(DataSetUtil.count(filter), "count")
            .reduce(new EvaluationUtil.ReduceBaseMetrics())
            .flatMap(new EvaluationUtil.SaveDataAsParams());

        this.setOutputTable(DataSetConversionUtil.toTable(getMLEnvironmentId(),
            out, new TableSchema(new String[] {"regression_eval_result"}, new TypeInformation[] {Types.STRING})
        ));
        return this;
    }

    public  EvalRegressionBatchOp();
    public  EvalRegressionBatchOp(Params params);

}