class BeamSqlFieldAccessExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, BeamSqlExpressionEnvironment env) {
    BeamSqlPrimitive targetObject = referenceExpression.evaluate(inputRow, window, env);
    SqlTypeName targetFieldType = targetObject.getOutputType();

    Object targetFieldValue;

    if (SqlTypeName.ARRAY.equals(targetFieldType)) {
      targetFieldValue = ((List) targetObject.getValue()).get(nestedFieldIndex);
    } else if (SqlTypeName.ROW.equals(targetFieldType)) {
      targetFieldValue = ((Row) targetObject.getValue()).getValue(nestedFieldIndex);
    } else {
      throw new IllegalArgumentException(
          "Attempt to access field of unsupported type "
              + targetFieldType.getDeclaringClass().getSimpleName()
              + ". Field access operator is only supported for arrays or rows");
    }

    return BeamSqlPrimitive.of(outputType, targetFieldValue);
  }

  public  BeamSqlFieldAccessExpression(
      BeamSqlExpression referenceExpression, int nestedFieldIndex, SqlTypeName nestedFieldType);

}