core/src/main/java/org/apache/calcite/sql/SqlWindowTableFunction.java - calcite - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to you under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.calcite.sql;

 import org.apache.calcite.linq4j.Ord;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.sql.type.ReturnTypes;
 import org.apache.calcite.sql.type.SqlOperandCountRanges;
 import org.apache.calcite.sql.type.SqlOperandMetadata;
 import org.apache.calcite.sql.type.SqlReturnTypeInference;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.type.SqlTypeUtil;
 import org.apache.calcite.sql.validate.SqlNameMatcher;
 import org.apache.calcite.sql.validate.SqlValidator;

 import com.google.common.collect.ImmutableList;

 import org.checkerframework.checker.nullness.qual.Nullable;

 import java.util.Collections;
 import java.util.List;

 import static com.google.common.base.Preconditions.checkArgument;

 import static org.apache.calcite.util.Static.RESOURCE;

 /**
  * Base class for a table-valued function that computes windows. Examples
  * include {@code TUMBLE}, {@code HOP} and {@code SESSION}.
  */
 public class SqlWindowTableFunction extends SqlFunction
     implements SqlTableFunction {

   /** The data source which the table function computes with. */
   protected static final String PARAM_DATA = "DATA";

   /** The time attribute column. Also known as the event time. */
   protected static final String PARAM_TIMECOL = "TIMECOL";

   /** The window duration INTERVAL. */
   protected static final String PARAM_SIZE = "SIZE";

   /** The optional align offset for each window. */
   protected static final String PARAM_OFFSET = "OFFSET";

   /** The session key(s), only used for SESSION window. */
   protected static final String PARAM_KEY = "KEY";

   /** The slide interval, only used for HOP window. */
   protected static final String PARAM_SLIDE = "SLIDE";

   /**
    * Type-inference strategy whereby the row type of a table function call is a
    * ROW, which is combined from the row type of operand #0 (which is a TABLE)
    * and two additional fields. The fields are as follows:
    *
    * <ol>
    * <li>{@code window_start}: TIMESTAMP type to indicate a window's start
    * <li>{@code window_end}: TIMESTAMP type to indicate a window's end
    * </ol>
    */
   public static final SqlReturnTypeInference ARG0_TABLE_FUNCTION_WINDOWING =
       SqlWindowTableFunction::inferRowType;

   /** Creates a window table function with a given name. */
   public SqlWindowTableFunction(String name, SqlOperandMetadata operandMetadata) {
     super(name, SqlKind.OTHER_FUNCTION, ReturnTypes.CURSOR, null,
         operandMetadata, SqlFunctionCategory.SYSTEM);
   }

   @Override public @Nullable SqlOperandMetadata getOperandTypeChecker() {
     return (@Nullable SqlOperandMetadata) super.getOperandTypeChecker();
   }

   @Override public SqlReturnTypeInference getRowTypeInference() {
     return ARG0_TABLE_FUNCTION_WINDOWING;
   }

   /**
    * {@inheritDoc}
    *
    * <p>Overrides because the first parameter of
    * table-value function windowing is an explicit TABLE parameter,
    * which is not scalar.
    */
   @Override public boolean argumentMustBeScalar(int ordinal) {
     return ordinal != 0;
   }

   /** Helper for {@link #ARG0_TABLE_FUNCTION_WINDOWING}. */
   private static RelDataType inferRowType(SqlOperatorBinding opBinding) {
     final RelDataType inputRowType = opBinding.getOperandType(0);
     final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
     return typeFactory.builder()
         .kind(inputRowType.getStructKind())
         .addAll(inputRowType.getFieldList())
         .add("window_start", SqlTypeName.TIMESTAMP, 3)
         .add("window_end", SqlTypeName.TIMESTAMP, 3)
         .build();
   }

   /** Partial implementation of operand type checker. */
   protected abstract static class AbstractOperandMetadata
       implements SqlOperandMetadata {
     final List<String> paramNames;
     final int mandatoryParamCount;

     AbstractOperandMetadata(List<String> paramNames,
         int mandatoryParamCount) {
       this.paramNames = ImmutableList.copyOf(paramNames);
       this.mandatoryParamCount = mandatoryParamCount;
       checkArgument(mandatoryParamCount >= 0
           && mandatoryParamCount <= paramNames.size());
     }

     @Override public SqlOperandCountRange getOperandCountRange() {
       return SqlOperandCountRanges.between(mandatoryParamCount,
           paramNames.size());
     }

     @Override public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
       return Collections.nCopies(paramNames.size(),
           typeFactory.createSqlType(SqlTypeName.ANY));
     }

     @Override public List<String> paramNames() {
       return paramNames;
     }

     @Override public boolean isOptional(int i) {
       return i > getOperandCountRange().getMin()
           && i <= getOperandCountRange().getMax();
     }

     boolean throwValidationSignatureErrorOrReturnFalse(SqlCallBinding callBinding,
         boolean throwOnFailure) {
       if (throwOnFailure) {
         throw callBinding.newValidationSignatureError();
       } else {
         return false;
       }
     }

     /**
      * Checks whether the heading operands are in the form
      * {@code (ROW, DESCRIPTOR, DESCRIPTOR ..., other params)},
      * returning whether successful, and throwing if any columns are not found.
      *
      * @param callBinding The call binding
      * @param descriptorCount The number of descriptors following the first
      * operand (e.g. the table)
      *
      * @return true if validation passes; throws if any columns are not found
      */
     boolean checkTableAndDescriptorOperands(SqlCallBinding callBinding,
         int descriptorCount) {
       final SqlNode operand0 = callBinding.operand(0);
       final SqlValidator validator = callBinding.getValidator();
       final RelDataType type = validator.getValidatedNodeType(operand0);
       if (type.getSqlTypeName() != SqlTypeName.ROW) {
         return false;
       }
       for (int i = 1; i < descriptorCount + 1; i++) {
         final SqlNode operand = callBinding.operand(i);
         if (operand.getKind() != SqlKind.DESCRIPTOR) {
           return false;
         }
         validateColumnNames(validator, type.getFieldNames(),
             ((SqlCall) operand).getOperandList());
       }
       return true;
     }

     /**
      * Checks whether the type that the operand of time col descriptor refers to is valid.
      *
      * @param callBinding The call binding
      * @param pos The position of the descriptor at the operands of the call
      * @return true if validation passes, false otherwise
      */
     boolean checkTimeColumnDescriptorOperand(SqlCallBinding callBinding, int pos) {
       SqlValidator validator = callBinding.getValidator();
       SqlNode operand0 = callBinding.operand(0);
       RelDataType type = validator.getValidatedNodeType(operand0);
       List<SqlNode> operands = ((SqlCall) callBinding.operand(pos)).getOperandList();
       SqlIdentifier identifier = (SqlIdentifier) operands.get(0);
       String columnName = identifier.getSimple();
       SqlNameMatcher matcher = validator.getCatalogReader().nameMatcher();
       for (RelDataTypeField field : type.getFieldList()) {
         if (matcher.matches(field.getName(), columnName)) {
           return SqlTypeUtil.isTimestamp(field.getType());
         }
       }
       return false;
     }

     /**
      * Checks whether the operands starting from position {@code startPos} are
      * all of type {@code INTERVAL}, returning whether successful.
      *
      * @param callBinding The call binding
      * @param startPos    The start position to validate (starting index is 0)
      *
      * @return true if validation passes
      */
     boolean checkIntervalOperands(SqlCallBinding callBinding, int startPos) {
       final SqlValidator validator = callBinding.getValidator();
       for (int i = startPos; i < callBinding.getOperandCount(); i++) {
         final RelDataType type = validator.getValidatedNodeType(callBinding.operand(i));
         if (!SqlTypeUtil.isInterval(type)) {
           return false;
         }
       }
       return true;
     }

     void validateColumnNames(SqlValidator validator,
         List<String> fieldNames, List<SqlNode> columnNames) {
       final SqlNameMatcher matcher = validator.getCatalogReader().nameMatcher();
       Ord.forEach(SqlIdentifier.simpleNames(columnNames), (name, i) -> {
         if (matcher.indexOf(fieldNames, name) < 0) {
           final SqlIdentifier columnName = (SqlIdentifier) columnNames.get(i);
           throw SqlUtil.newContextException(columnName.getParserPosition(),
               RESOURCE.unknownIdentifier(name));
         }
       });
     }
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to you under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.calcite.sql;

	import org.apache.calcite.linq4j.Ord;
	import org.apache.calcite.rel.type.RelDataType;
	import org.apache.calcite.rel.type.RelDataTypeFactory;
	import org.apache.calcite.rel.type.RelDataTypeField;
	import org.apache.calcite.sql.type.ReturnTypes;
	import org.apache.calcite.sql.type.SqlOperandCountRanges;
	import org.apache.calcite.sql.type.SqlOperandMetadata;
	import org.apache.calcite.sql.type.SqlReturnTypeInference;
	import org.apache.calcite.sql.type.SqlTypeName;
	import org.apache.calcite.sql.type.SqlTypeUtil;
	import org.apache.calcite.sql.validate.SqlNameMatcher;
	import org.apache.calcite.sql.validate.SqlValidator;

	import com.google.common.collect.ImmutableList;

	import org.checkerframework.checker.nullness.qual.Nullable;

	import java.util.Collections;
	import java.util.List;

	import static com.google.common.base.Preconditions.checkArgument;

	import static org.apache.calcite.util.Static.RESOURCE;

	/**
	* Base class for a table-valued function that computes windows. Examples
	* include {@code TUMBLE}, {@code HOP} and {@code SESSION}.
	*/
	public class SqlWindowTableFunction extends SqlFunction
	implements SqlTableFunction {

	/** The data source which the table function computes with. */
	protected static final String PARAM_DATA = "DATA";

	/** The time attribute column. Also known as the event time. */
	protected static final String PARAM_TIMECOL = "TIMECOL";

	/** The window duration INTERVAL. */
	protected static final String PARAM_SIZE = "SIZE";

	/** The optional align offset for each window. */
	protected static final String PARAM_OFFSET = "OFFSET";

	/** The session key(s), only used for SESSION window. */
	protected static final String PARAM_KEY = "KEY";

	/** The slide interval, only used for HOP window. */
	protected static final String PARAM_SLIDE = "SLIDE";

	/**
	* Type-inference strategy whereby the row type of a table function call is a
	* ROW, which is combined from the row type of operand #0 (which is a TABLE)
	* and two additional fields. The fields are as follows:
	*
	* <ol>
	* <li>{@code window_start}: TIMESTAMP type to indicate a window's start
	* <li>{@code window_end}: TIMESTAMP type to indicate a window's end
	* </ol>
	*/
	public static final SqlReturnTypeInference ARG0_TABLE_FUNCTION_WINDOWING =
	SqlWindowTableFunction::inferRowType;

	/** Creates a window table function with a given name. */
	public SqlWindowTableFunction(String name, SqlOperandMetadata operandMetadata) {
	super(name, SqlKind.OTHER_FUNCTION, ReturnTypes.CURSOR, null,
	operandMetadata, SqlFunctionCategory.SYSTEM);
	}

	@Override public @Nullable SqlOperandMetadata getOperandTypeChecker() {
	return (@Nullable SqlOperandMetadata) super.getOperandTypeChecker();
	}

	@Override public SqlReturnTypeInference getRowTypeInference() {
	return ARG0_TABLE_FUNCTION_WINDOWING;
	}

	/**
	* {@inheritDoc}
	*
	* <p>Overrides because the first parameter of
	* table-value function windowing is an explicit TABLE parameter,
	* which is not scalar.
	*/
	@Override public boolean argumentMustBeScalar(int ordinal) {
	return ordinal != 0;
	}

	/** Helper for {@link #ARG0_TABLE_FUNCTION_WINDOWING}. */
	private static RelDataType inferRowType(SqlOperatorBinding opBinding) {
	final RelDataType inputRowType = opBinding.getOperandType(0);
	final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
	return typeFactory.builder()
	.kind(inputRowType.getStructKind())
	.addAll(inputRowType.getFieldList())
	.add("window_start", SqlTypeName.TIMESTAMP, 3)
	.add("window_end", SqlTypeName.TIMESTAMP, 3)
	.build();
	}

	/** Partial implementation of operand type checker. */
	protected abstract static class AbstractOperandMetadata
	implements SqlOperandMetadata {
	final List<String> paramNames;
	final int mandatoryParamCount;

	AbstractOperandMetadata(List<String> paramNames,
	int mandatoryParamCount) {
	this.paramNames = ImmutableList.copyOf(paramNames);
	this.mandatoryParamCount = mandatoryParamCount;
	checkArgument(mandatoryParamCount >= 0
	&& mandatoryParamCount <= paramNames.size());
	}

	@Override public SqlOperandCountRange getOperandCountRange() {
	return SqlOperandCountRanges.between(mandatoryParamCount,
	paramNames.size());
	}

	@Override public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
	return Collections.nCopies(paramNames.size(),
	typeFactory.createSqlType(SqlTypeName.ANY));
	}

	@Override public List<String> paramNames() {
	return paramNames;
	}

	@Override public boolean isOptional(int i) {
	return i > getOperandCountRange().getMin()
	&& i <= getOperandCountRange().getMax();
	}

	boolean throwValidationSignatureErrorOrReturnFalse(SqlCallBinding callBinding,
	boolean throwOnFailure) {
	if (throwOnFailure) {
	throw callBinding.newValidationSignatureError();
	} else {
	return false;
	}
	}

	/**
	* Checks whether the heading operands are in the form
	* {@code (ROW, DESCRIPTOR, DESCRIPTOR ..., other params)},
	* returning whether successful, and throwing if any columns are not found.
	*
	* @param callBinding The call binding
	* @param descriptorCount The number of descriptors following the first
	* operand (e.g. the table)
	*
	* @return true if validation passes; throws if any columns are not found
	*/
	boolean checkTableAndDescriptorOperands(SqlCallBinding callBinding,
	int descriptorCount) {
	final SqlNode operand0 = callBinding.operand(0);
	final SqlValidator validator = callBinding.getValidator();
	final RelDataType type = validator.getValidatedNodeType(operand0);
	if (type.getSqlTypeName() != SqlTypeName.ROW) {
	return false;
	}
	for (int i = 1; i < descriptorCount + 1; i++) {
	final SqlNode operand = callBinding.operand(i);
	if (operand.getKind() != SqlKind.DESCRIPTOR) {
	return false;
	}
	validateColumnNames(validator, type.getFieldNames(),
	((SqlCall) operand).getOperandList());
	}
	return true;
	}

	/**
	* Checks whether the type that the operand of time col descriptor refers to is valid.
	*
	* @param callBinding The call binding
	* @param pos The position of the descriptor at the operands of the call
	* @return true if validation passes, false otherwise
	*/
	boolean checkTimeColumnDescriptorOperand(SqlCallBinding callBinding, int pos) {
	SqlValidator validator = callBinding.getValidator();
	SqlNode operand0 = callBinding.operand(0);
	RelDataType type = validator.getValidatedNodeType(operand0);
	List<SqlNode> operands = ((SqlCall) callBinding.operand(pos)).getOperandList();
	SqlIdentifier identifier = (SqlIdentifier) operands.get(0);
	String columnName = identifier.getSimple();
	SqlNameMatcher matcher = validator.getCatalogReader().nameMatcher();
	for (RelDataTypeField field : type.getFieldList()) {
	if (matcher.matches(field.getName(), columnName)) {
	return SqlTypeUtil.isTimestamp(field.getType());
	}
	}
	return false;
	}

	/**
	* Checks whether the operands starting from position {@code startPos} are
	* all of type {@code INTERVAL}, returning whether successful.
	*
	* @param callBinding The call binding
	* @param startPos The start position to validate (starting index is 0)
	*
	* @return true if validation passes
	*/
	boolean checkIntervalOperands(SqlCallBinding callBinding, int startPos) {
	final SqlValidator validator = callBinding.getValidator();
	for (int i = startPos; i < callBinding.getOperandCount(); i++) {
	final RelDataType type = validator.getValidatedNodeType(callBinding.operand(i));
	if (!SqlTypeUtil.isInterval(type)) {
	return false;
	}
	}
	return true;
	}

	void validateColumnNames(SqlValidator validator,
	List<String> fieldNames, List<SqlNode> columnNames) {
	final SqlNameMatcher matcher = validator.getCatalogReader().nameMatcher();
	Ord.forEach(SqlIdentifier.simpleNames(columnNames), (name, i) -> {
	if (matcher.indexOf(fieldNames, name) < 0) {
	final SqlIdentifier columnName = (SqlIdentifier) columnNames.get(i);
	throw SqlUtil.newContextException(columnName.getParserPosition(),
	RESOURCE.unknownIdentifier(name));
	}
	});
	}
	}
	}