@Override
	public RexNode[] convertToRexNodes(String[] exprs) {
    //query语句是 SELECT ts - INTERVAL '0.001' SECOND FROM __temp_table__
		String query = String.format(QUERY_FORMAT, String.join(",", exprs));
    //通过 Flink Planner 解析成 SqlSelect
    //ts - INTERVAL '0.001' SECOND 是SqlBasicCall  ts和Interval '0.001' SECOND是operands
		SqlNode parsed = planner.parser().parse(query);
    // validate 来校验 运算对象operands是否合法存在
		SqlNode validated = planner.validate(parsed);
    // SqlNode转成 RelNode,LogicalProject,input为 LogicalTableScan
		RelNode rel = planner.rel(validated).rel;
		// The plan should in the following tree
		// LogicalProject
		// +- TableScan
    
    //再通过LogicalProject 获取每列的Row Expression 即RexNode
		if (rel instanceof LogicalProject
			&& rel.getInput(0) != null
			&& rel.getInput(0) instanceof TableScan) {
      // RexCall "-($0, 1:INTERVAL SECOND)",operands:0 RexInputRef"$0",operands:1 RexLiteral"1:INTERVAL SECOND"
			return ((LogicalProject) rel).getProjects().toArray(new RexNode[0]);
		} else {
			throw new IllegalStateException("The root RelNode should be LogicalProject, but is " + rel.toString());
		}
	}

      
      public final class WatermarkGenerator$0
          extends org.apache.flink.table.runtime.generated.WatermarkGenerator {
         
        
         
        public WatermarkGenerator$0(Object[] references) throws Exception {
          
        }
        
        @Override
        public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
          
        }
        @Override
        public Long currentWatermark(org.apache.flink.table.data.RowData row) throws Exception {
          
          org.apache.flink.table.data.TimestampData field$1;
          boolean isNull$1;
          boolean isNull$2;
          org.apache.flink.table.data.TimestampData result$3;
          isNull$1 = row.isNullAt(0);
          field$1 = null;
          if (!isNull$1) {
            field$1 = row.getTimestamp(0, 3);
          }
          
          
          
          isNull$2 = isNull$1 || false;
          result$3 = null;
          if (!isNull$2) {
            
            result$3 = org.apache.flink.table.data.TimestampData.fromEpochMillis(field$1.getMillisecond() - ((long) 1L), field$1.getNanoOfMillisecond());
            
          }
          
          if (isNull$2) {
            return null;
          } else {
            return result$3.getMillisecond();
          }
        }
       
        @Override
        public void close() throws Exception {
          
        }
      }
    

LogicalTableScan
-- LogicalProject
super(operand(LogicalProject.class,
      operand(LogicalTableScan.class, none())),
 /**
 match方法
 **/
 matches方法中进行条件匹配,条件符合可进行转换,
 只要Source支持SupportsProjectionPushDown即可
 tableSourceTable.tableSource() instanceof SupportsProjectionPushDown
 
 /**
 onMatch方法
 **/
 onMatch 中进行 transformTo
 
 LogicalTableScan newScan = new LogicalTableScan(
				scan.getCluster(), scan.getTraitSet(), scan.getHints(), newTableSourceTable);
		// rewrite input field in projections
		List<RexNode> newProjects = RexNodeRewriter.rewriteWithNewFieldInput(project.getProjects(), usedFields);
		LogicalProject newProject = project.copy(
				project.getTraitSet(),
				newScan,
				newProjects,
				project.getRowType());

		if (ProjectRemoveRule.isTrivial(newProject)) {
			// drop project if the transformed program merely returns its input
			call.transformTo(newScan);
		} else {
			call.transformTo(newProject);
		}
 

/**
 * Insertion operation.
 */
INSERT,

/**
 * Update operation.
 */
UPDATE,

/**
 * Deletion operation.
 */
DELETE

/**
    * Optimize max or min with retraction agg. MaxWithRetract can be optimized to Max if input is
    * update increasing.
    */
  def getNeedRetractions(
      groupCount: Int,
      needRetraction: Boolean,
      monotonicity: RelModifiedMonotonicity,
      aggCalls: Seq[AggregateCall]): Array[Boolean] = {
    val needRetractionArray = Array.fill(aggCalls.size)(needRetraction)
    if (monotonicity != null && needRetraction) {
      aggCalls.zipWithIndex.foreach { case (aggCall, idx) =>
        aggCall.getAggregation match {
          // if monotonicity is decreasing and aggCall is min with retract,
          // set needRetraction to false
          case a: SqlMinMaxAggFunction
            if a.getKind == SqlKind.MIN &&
              monotonicity.fieldMonotonicities(groupCount + idx) == SqlMonotonicity.DECREASING =>
            needRetractionArray(idx) = false
          // if monotonicity is increasing and aggCall is max with retract,
          // set needRetraction to false
          case a: SqlMinMaxAggFunction
            if a.getKind == SqlKind.MAX &&
              monotonicity.fieldMonotonicities(groupCount + idx) == SqlMonotonicity.INCREASING =>
            needRetractionArray(idx) = false
          case _ => // do nothing
        }
      }
    }