Disable CompiledProgram with data parallel in StandaloneExecutor (#44264)

* Disable CompiledProgram with data parallel in StandaloneExecutor

* Fix typos

* Fix typos

python/paddle/fluid/executor.py

@@ -409,18 +409,6 @@ def _is_enable_standalone_executor():
     return flag
 
 
-def _is_standalone_executor_enable_compiled_program():
-    """
-    Whether to use experimental executor `StandaloneExecutor` in CompiledProgram.
-    Convert Graph to Program.
-    """
-    flag = False
-    env_val = os.environ.get('FLAGS_CONVERT_GRAPH_TO_PROGRAM', None)
-    if env_val in [1, '1', True, 'True', 'true']:
-        flag = True
-    return flag
-
-
 def _prepare_fleet_executor():
     from ..distributed.fleet.proto import fleet_executor_desc_pb2
     trainer_endpoints_str = os.getenv("PADDLE_TRAINER_ENDPOINTS", "")
@@ -1405,18 +1393,22 @@ class Executor(object):
                 return False
 
             compiled = isinstance(program, compiler.CompiledProgram)
-            # print("compiled is : {}".format(compiled))
-            # NOTE(zhiqiu): do not support compiled program now
             if compiled:
-                if program._program is not None and _is_standalone_executor_enable_compiled_program(
-                ):
-                    return True
+                # Unsupported case 1 : the CompiledProgram is constructed by Graph
+                if program._program is None:
+                    return False
+
+                # Unsupported case 2 : disabled by FLAGS_CONVERT_GRAPH_TO_PROGRAM
+                if os.environ.get('FLAGS_CONVERT_GRAPH_TO_PROGRAM',
+                                  None) not in [1, '1', True, 'True', 'true']:
                     return False
-                # if program._is_data_parallel and len(
-                #         program._get_places(place, program._places)) == 1:
-                #     return True
-                # else:
-                #     return False
+
+                # Unsupported case 3: data parallel
+                if program._is_data_parallel == True and len(
+                        program._get_places(place, program._places)) != 1:
+                    return False
+
+                return True
             else:
                 if isinstance(program._graph, compiler.CompiledProgram):
                     return False
@@ -1447,11 +1439,10 @@ class Executor(object):
                 # a little bit tricy here, use inner_program before _add_feed_fetch_ops to get key
                 # while use program to geet _StandaloneExecutor
                 if key not in self._executor_cache._cached_executors:
+                    # To apply IR pass, compile the Program to IrGraph and convert it back to Program
                     if isinstance(program, compiler.CompiledProgram):
                         program._compile(scope, self.place)
-                        compiled_graph = program._graph
-                        ir_graph = framework.IrGraph(compiled_graph,
-                                                     for_test=True)
+                        ir_graph = framework.IrGraph(program._graph)
                         inner_program = ir_graph.to_program()
                     program = self._add_feed_fetch_ops(
                         program=inner_program,