%0 = tf_executor.graph {
    %1:3 = tf_executor.NextIteration.Source : tensor<*xf32>
    "tf_executor.NextIteration.Sink"(%1#1, %arg1) : (!tf_executor.token, i1) -> ()
// expected-error@-1 {{'tf_executor.NextIteration.Sink' op input type 'i1' mismatch the tf_executor.NextIteration.Source output type: 'tensor<*xf32>'}}
    tf_executor.fetch %1#0 : tensor<*xf32>
  }

defined as a non-dead input (selected in a non-defined way if multiple inputs
are non-dead). If all inputs are dead, the output is also dead.

### NextIteration: `tf_executor.NextIteration.Source` and `tf_executor.NextIteration.Sink` Operation

The TensorFlow
[`NextIteration`](https://www.tensorflow.org/api_docs/cc/class/tensorflow/ops/next-iteration)
op is modeled using these two paired operations. Since _NextIteration_ is

    result[i] = outputs[i].node();
  }
  return result;
}

// Manually generates the name of the `loop_var_idx`-th NextIteration node of a
// loop being constructed with `scope`. This is used to define the backedge
// before the NextIteration node is created.
string NextIterationName(const Scope& scope, int loop_var_idx) {
  string result;
  const string& prefix = scope.impl()->name();

}

// CHECK-LABEL: func @nextiteration(%{{.*}}: tensor<*xf32>, %{{.*}}: i1) -> tensor<*xf32> {
func.func @nextiteration(%arg0: tensor<*xf32>, %arg1: i1) -> tensor<*xf32> {
  %0 = tf_executor.graph {
    %1:3 = tf_executor.NextIteration.Source : tensor<*xf32>
    tf_executor.NextIteration.Sink[%1#1] %1#0 : tensor<*xf32>
// CHECK: tf_executor.NextIteration.Source : tensor<*xf32>

    return (!lhs_device_attr && !rhs_device_attr) ||
           (lhs_device_attr && rhs_device_attr &&
            lhs_device_attr.getValue() == rhs_device_attr.getValue());
  };

  // Check if tf_executor.NextIteration.Source/tf_executor.NextIteration.Sink
  // pair has matching devices or no devices.
  if (auto source = llvm::dyn_cast<tf_executor::NextIterationSourceOp>(op)) {
    return ops_have_same_device(source, source.GetSink());

//               |           |
//               v           |
//           NextIteration --+
//
// we won't put (Read, Write) in the returned set.  This is fine if
// auto-clustering can only cluster the Read->Write edge, but it is a problem if
// it clusters the Write->NextIteration->Merge->Read edges instead.  So we rely
// on auto-clustering to not cluster NextIteration->Merge edges.  The same

        scope.WithOpName("while/add/y").WithControlDependencies(identity), 1);
    auto add = ops::Add(scope.WithOpName("while/add"), identity_reshaped, one);
    auto next_iteration =
        ops::NextIteration(scope.WithOpName("while/NextIteration"), add);

    auto sink = ops::Identity(scope.WithOpName("sink"), exit);

    // Remove the dummy node and add the loop backedge.
    scope.graph()->RemoveNode(dummy.node());

    }
  }
  return absl::OkStatus();
}

// Returns true if the executor/control dialect op should map to Ref node in
// TensorFlow Graph. For control dialect NextIteration it uses the 1st operand
// type. For executor dialect NextIteration it uses the 2nd operand type. For
// all others (Enter/Exit/Merge/Switch), if the output type is ref, they
// correspond to the Ref equivalent op in TF Graph.

        });
  }

  VisitOpOperands(graph, op, reachable_ops, ops_to_visit);

  // If op is a `tf_executor.NextIteration.Source`, visit its associated
  // `tf_executor.NextIteration.Sink` op.
  if (auto source_op = llvm::dyn_cast<NextIterationSourceOp>(op)) {
    Operation* sink_op = source_op.GetSink().getOperation();

    %1:2 = tf_executor.Enter %0#0 frame "while/while_context" : (tensor<i32>) -> (tensor<*xi32>, !tf_executor.control)
    %2 = tf_executor.island wraps "tf.opB"() : () -> ()
    %3:3 = tf_executor.NextIteration.Source : tensor<*xi32>
    %4:3 = tf_executor.Merge %3#0, %1#0 : tensor<*xi32>
    %5:2 = tf_executor.island(%4#2) wraps "tf.opC"() : () -> tensor<i32>