NodeWith(Op("Switch"), Inputs(Out(0, xla_compile), Out(1, xla_compile)));
  auto xla_run =
      NodeWith(Op("_XlaRun"), Inputs(Out(1, predicated_compilation_key)));
  auto tf_call =
      NodeWith(Op("StatefulPartitionedCall"),
               CtrlDeps(NodeWith(Op("Identity"),
                                 Inputs(Out(0, predicated_compilation_key)))));
  auto merge = NodeWith(Op("_XlaMerge"), Inputs(Out(tf_call), Out(xla_run)));

    key: "value"
    value {
      tensor {
        dtype: DT_FLOAT
        tensor_shape {
        }
        float_val: 0.0
      }
    }
  }
}
node {
  name: "test_fn_call"
  op: "StatefulPartitionedCall"
  input: "const"
  attr {
    key: "Tin"
    value {
      list {
        type: DT_FLOAT
      }
    }
  }
  attr {
    key: "Tout"
    value {
      list {
        type: DT_FLOAT

  (*f.mutable_attr())["T"].set_type(DT_FLOAT);
  auto results =
      PartitionedCall(scope_, std::initializer_list<Input>{x}, {DT_FLOAT}, f);
  RunTest(x, {}, results[0], {});

  auto stateful_results = StatefulPartitionedCall(
      scope_, std::initializer_list<Input>{x}, {DT_FLOAT}, f);
  RunTest(x, {}, stateful_results[0], {});
}

}  // namespace
}  // namespace ops

// -----

// Tests PartitionedCall/StatefulPartitionedCall.

// CHECK-LABEL: func @main
func.func @main(%arg0: tensor<i1>) -> () {
  %max_size = "tf.Const"() {value = dense<10> : tensor<i32>} : () -> tensor<i32>
  // CHECK-NOT: tf.Stack
  %stack = "tf.StackV2"(%max_size) {elem_type = f32, stack_name = "s"} : (tensor<i32>) -> tensor<!tf_type.resource>
  // CHECK: "tf.StatefulPartitionedCall"
  // CHECK-SAME: f = @callee_stack_decomposed

    %0 = tf_executor.graph {

  // CHECK: [[handle:%.*]] = "tf.VarHandleOp"()
  %handle = "tf.VarHandleOp"() {container = "", shared_name = "x"} : () -> tensor<!tf_type.resource<tensor<i32>>>
  // CHECK: "tf.StatefulPartitionedCall"([[handle]])
  %x = "tf.StatefulPartitionedCall"(%handle) {device = "/CPU:0", config = "", config_proto = "", executor_type = "", f = @some_func} : (tensor<!tf_type.resource<tensor<i32>>>) -> (tensor<i32>)

  // CHECK-NOT: tf.VarHandleOp
  // CHECK:  "tf._TfrtGetResource"()
  %handle = "tf.VarHandleOp"() {container = "", shared_name = "x"} : () -> tensor<!tf_type.resource<tensor<i32>>>
  // CHECK: tf.StatefulPartitionedCall
  %x = "tf.StatefulPartitionedCall"(%handle) {device = "/CPU:0", config = "", config_proto = "", executor_type = "", f = @some_func} : (tensor<!tf_type.resource<tensor<i32>>>) -> (tensor<i32>)

  // When add_edges_to_output_of_downstream_nodes is true, the output edges of
  // the xla_launch_node's immediate downstream nodes would be attached to the
  // generated xla node. For example, if the original graph is
  // StatefulPartitionedCall{_xla_compile_id=1} -> XlaClusterOutput -> NodeA
  // The output graph of this function would look like the following when
  // add_edges_to_output_of_downstream_nodes is true:
  // XlaLaunch -> NodeA

    tf_executor.graph {
      %control = tf_executor.island {
        "tf.StatefulPartitionedCall"(%arg0) <{config = "", config_proto = "", executor_type = "", f = @partitioned}> : (tensor<*x!tf_type.resource>) -> ()
        tf_executor.yield
      }
      tf_executor.fetch %control : !tf_executor.control
    }

// -----

// This test makes sure functions with tf._noinline=true is not inlined.

module {
  func.func @stateful_partitioned_call(%arg0: tensor<1x2x2x3xf32>) -> (tensor<1x2x2x3xf32>) {
    %0 = "tf.StatefulPartitionedCall"(%arg0) <{
      config = "", config_proto = "", executor_type = "", f = @some_func
    }> {
      _collective_manager_ids = [], device = ""
    } : (tensor<1x2x2x3xf32>) -> tensor<1x2x2x3xf32>