Address comments

Author: JingsongLi

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LagAggFunction.java

@@ -100,6 +100,10 @@ public void accumulate(LagAcc<T> acc, T value) throws Exception {
     }
 
     public void accumulate(LagAcc<T> acc, T value, int offset) throws Exception {
+        if (offset < 0) {
+            throw new TableException(String.format("Offset(%d) should be positive.", offset));
+        }
+
         acc.offset = offset;
         accumulate(acc, value);
     }

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala

@@ -45,9 +45,10 @@ import scala.collection.JavaConversions._
  * as subclasses of [[SqlAggFunction]] in Calcite but not as [[BridgingSqlAggFunction]]. The factory
  * returns [[DeclarativeAggregateFunction]] or [[BuiltInAggregateFunction]].
  *
- * @param inputType the input rel data type
- * @param orderKeyIdx the indexes of order key (null when is not over agg)
- * @param needRetraction true if need retraction
+ * @param inputRowType the input row type
+ * @param orderKeyIndexes the indexes of order key (null when is not over agg)
+ * @param aggCallNeedRetractions true if need retraction
+ * @param isBounded true if the source is bounded source
  */
 class AggFunctionFactory(
     inputRowType: RowType,