code: public static BaseRowKeySelector getBaseRowSelector(int[] keyFields, BaseRowTypeInfo rowType) {
  		if (keyFields.length > 0) {
  			InternalType[] inputFieldTypes = rowType.getInternalTypes();
  			String[] inputFieldNames = rowType.getFieldNames();
  			InternalType[] keyFieldTypes = new InternalType[keyFields.length];
  			String[] keyFieldNames = new String[keyFields.length];
  			for (int i = 0; i < keyFields.length; ++i) {
  				keyFieldTypes[i] = inputFieldTypes[keyFields[i]];
  				keyFieldNames[i] = inputFieldNames[keyFields[i]];
  			}
  			RowType returnType = new RowType(keyFieldTypes, keyFieldNames);
  			RowType inputType = new RowType(inputFieldTypes, rowType.getFieldNames());
  			GeneratedProjection generatedProjection = ProjectionCodeGenerator.generateProjection(
  				CodeGeneratorContext.apply(new TableConfig()),
  				"KeyProjection",
  				inputType,
  				returnType, keyFields);
  			BaseRowTypeInfo keyRowType = returnType.toTypeInfo();
  			// check if type implements proper equals/hashCode
  			TypeCheckUtils.validateEqualsHashCode("grouping", keyRowType);
  			return new BinaryRowKeySelector(keyRowType, generatedProjection);
  		} else {
  			return NullBinaryRowKeySelector.INSTANCE;
  		}
  	}
docs: Create a BaseRowKeySelector to extract keys from DataStream which type is BaseRowTypeInfo.
  
  @param keyFields key fields
  @param rowType type of DataStream to extract keys
  @return the BaseRowKeySelector to extract keys from DataStream which type is BaseRowTypeInfo.