Make Dag Serialization a hard requirement

UPDATING.md

@@ -244,6 +244,14 @@ The Old and New provider configuration keys that have changed are as follows
 
 For more information, visit https://flask-appbuilder.readthedocs.io/en/latest/security.html#authentication-oauth
 
+### DAG Serialization will be strictly required
+
+Until Airflow 2.0, DAG Serialization was disabled by default. However from Airflow 2.0, DAG Serialization
+will be enabled by default. From Airflow 2.0, Scheduler will use Serialized DAGs to make scheduling decisions
+and so DAG Serialization can not be turned off.
+
+The previous setting `[core] store_serialized_dags` will be ignored.
+
 ### Changes to the KubernetesExecutor
 
 #### The KubernetesExecutor Will No Longer Read from the airflow.cfg for Base Pod Configurations

airflow/api/common/experimental/__init__.py

@@ -19,7 +19,6 @@
 from datetime import datetime
 from typing import Optional
 
-from airflow.configuration import conf
 from airflow.exceptions import DagNotFound, DagRunNotFound, TaskNotFound
 from airflow.models import DagBag, DagModel, DagRun
 
@@ -32,10 +31,10 @@ def check_and_get_dag(dag_id: str, task_id: Optional[str] = None) -> DagModel:
 
     dagbag = DagBag(
         dag_folder=dag_model.fileloc,
-        read_dags_from_db=conf.getboolean('core', 'store_serialized_dags')
+        read_dags_from_db=True
     )
-    dag = dagbag.get_dag(dag_id)  # prefetch dag if it is stored serialized
-    if dag_id not in dagbag.dags:
+    dag = dagbag.get_dag(dag_id)
+    if not dag:
         error_message = "Dag id {} not found".format(dag_id)
         raise DagNotFound(error_message)
     if task_id and not dag.has_task(task_id):

airflow/api/common/experimental/delete_dag.py

@@ -24,7 +24,6 @@
 from airflow.exceptions import DagNotFound
 from airflow.models import DagModel, TaskFail
 from airflow.models.serialized_dag import SerializedDagModel
-from airflow.settings import STORE_SERIALIZED_DAGS
 from airflow.utils.session import provide_session
 
 log = logging.getLogger(__name__)
@@ -47,7 +46,7 @@ def delete_dag(dag_id: str, keep_records_in_log: bool = True, session=None) -> i
 
     # Scheduler removes DAGs without files from serialized_dag table every dag_dir_list_interval.
     # There may be a lag, so explicitly removes serialized DAG here.
-    if STORE_SERIALIZED_DAGS and SerializedDagModel.has_dag(dag_id=dag_id, session=session):
+    if SerializedDagModel.has_dag(dag_id=dag_id, session=session):
         SerializedDagModel.remove_dag(dag_id=dag_id, session=session)
 
     count = 0

airflow/api/common/experimental/mark_tasks.py

@@ -61,15 +61,16 @@ def _create_dagruns(dag, execution_dates, state, run_type):
 
 @provide_session
 def set_state(
-        tasks: Iterable[BaseOperator],
-        execution_date: datetime.datetime,
-        upstream: bool = False,
-        downstream: bool = False,
-        future: bool = False,
-        past: bool = False,
-        state: str = State.SUCCESS,
-        commit: bool = False,
-        session=None):  # pylint: disable=too-many-arguments,too-many-locals
+    tasks: Iterable[BaseOperator],
+    execution_date: datetime.datetime,
+    upstream: bool = False,
+    downstream: bool = False,
+    future: bool = False,
+    past: bool = False,
+    state: str = State.SUCCESS,
+    commit: bool = False,
+    session=None
+):  # pylint: disable=too-many-arguments,too-many-locals
     """
     Set the state of a task instance and if needed its relatives. Can set state
     for future tasks (calculated from execution_date) and retroactively
@@ -181,7 +182,7 @@ def get_subdag_runs(dag, session, state, task_ids, commit, confirmed_dates):
                 continue
 
             current_task = current_dag.get_task(task_id)
-            if isinstance(current_task, SubDagOperator):
+            if isinstance(current_task, SubDagOperator) or current_task.task_type == "SubDagOperator":
                 # this works as a kind of integrity check
                 # it creates missing dag runs for subdag operators,
                 # maybe this should be moved to dagrun.verify_integrity

airflow/api/common/experimental/trigger_dag.py

@@ -114,13 +114,7 @@ def trigger_dag(
     if dag_model is None:
         raise DagNotFound("Dag id {} not found in DagModel".format(dag_id))
 
-    def read_store_serialized_dags():
-        from airflow.configuration import conf
-        return conf.getboolean('core', 'store_serialized_dags')
-    dagbag = DagBag(
-        dag_folder=dag_model.fileloc,
-        read_dags_from_db=read_store_serialized_dags()
-    )
+    dagbag = DagBag(dag_folder=dag_model.fileloc, read_dags_from_db=True)
     triggers = _trigger_dag(
         dag_id=dag_id,
         dag_bag=dagbag,

airflow/api_connexion/endpoints/task_instance_endpoint.py

@@ -34,6 +34,7 @@
     TaskInstanceReferenceCollection,
     set_task_instance_state_form,
 )
+from airflow.exceptions import SerializedDagNotFound
 from airflow.models.dagrun import DagRun as DR
 from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
 from airflow.models import SlaMiss
@@ -286,9 +287,13 @@ def post_set_task_instances_state(dag_id, session):
     except ValidationError as err:
         raise BadRequest(detail=str(err.messages))
 
-    dag = current_app.dag_bag.get_dag(dag_id)
-    if not dag:
-        error_message = "Dag ID {} not found".format(dag_id)
+    error_message = "Dag ID {} not found".format(dag_id)
+    try:
+        dag = current_app.dag_bag.get_dag(dag_id)
+        if not dag:
+            raise NotFound(error_message)
+    except SerializedDagNotFound:
+        # If DAG is not found in serialized_dag table
         raise NotFound(error_message)
 
     task_id = data['task_id']

airflow/cli/commands/sync_perm_command.py

@@ -16,7 +16,6 @@
 # specific language governing permissions and limitations
 # under the License.
 """Sync permission command"""
-from airflow import settings
 from airflow.models import DagBag
 from airflow.utils import cli as cli_utils
 from airflow.www.app import cached_app
@@ -29,7 +28,7 @@ def sync_perm(args):
     print('Updating permission, view-menu for all existing roles')
     appbuilder.sm.sync_roles()
     print('Updating permission on all DAG views')
-    dags = DagBag(store_serialized_dags=settings.STORE_SERIALIZED_DAGS).dags.values()
+    dags = DagBag(read_dags_from_db=True).dags.values()
     for dag in dags:
         appbuilder.sm.sync_perm_for_dag(
             dag.dag_id,

airflow/exceptions.py

@@ -126,6 +126,10 @@ class DuplicateTaskIdFound(AirflowException):
     """Raise when a Task with duplicate task_id is defined in the same DAG"""
 
 
+class SerializedDagNotFound(DagNotFound):
+    """Raise when DAG is not found in the serialized_dags table in DB"""
+
+
 class TaskNotFound(AirflowNotFoundException):
     """Raise when a Task is not available in the system"""
 

airflow/models/dagbag.py

@@ -36,7 +36,7 @@
 from airflow import settings
 from airflow.configuration import conf
 from airflow.dag.base_dag import BaseDagBag
-from airflow.exceptions import AirflowClusterPolicyViolation, AirflowDagCycleException
+from airflow.exceptions import AirflowClusterPolicyViolation, AirflowDagCycleException, SerializedDagNotFound
 from airflow.plugins_manager import integrate_dag_plugins
 from airflow.stats import Stats
 from airflow.utils import timezone
@@ -75,10 +75,8 @@ class DagBag(BaseDagBag, LoggingMixin):
     :param include_smart_sensor: whether to include the smart sensor native
         DAGs that create the smart sensor operators for whole cluster
     :type include_smart_sensor: bool
-    :param read_dags_from_db: Read DAGs from DB if store_serialized_dags is ``True``.
-        If ``False`` DAGs are read from python files. This property is not used when
-        determining whether or not to write Serialized DAGs, that is done by checking
-        the config ``store_serialized_dags``.
+    :param read_dags_from_db: Read DAGs from DB if ``True`` is passed.
+        If ``False`` DAGs are read from python files.
     :type read_dags_from_db: bool
     """
 
@@ -214,7 +212,7 @@ def _add_dag_from_db(self, dag_id: str, session: Session):
         from airflow.models.serialized_dag import SerializedDagModel
         row = SerializedDagModel.get(dag_id, session)
         if not row:
-            raise ValueError(f"DAG '{dag_id}' not found in serialized_dag table")
+            raise SerializedDagNotFound(f"DAG '{dag_id}' not found in serialized_dag table")
 
         dag = row.dag
         for subdag in dag.subdags:
@@ -527,8 +525,6 @@ def sync_to_db(self, session: Optional[Session] = None):
         from airflow.models.serialized_dag import SerializedDagModel
         self.log.debug("Calling the DAG.bulk_sync_to_db method")
         DAG.bulk_write_to_db(self.dags.values(), session=session)
-        # Write Serialized DAGs to DB if DAG Serialization is turned on
-        # Even though self.read_dags_from_db is False
-        if settings.STORE_SERIALIZED_DAGS or self.read_dags_from_db:
-            self.log.debug("Calling the SerializedDagModel.bulk_sync_to_db method")
-            SerializedDagModel.bulk_sync_to_db(self.dags.values(), session=session)
+        # Write Serialized DAGs to DB
+        self.log.debug("Calling the SerializedDagModel.bulk_sync_to_db method")
+        SerializedDagModel.bulk_sync_to_db(self.dags.values(), session=session)

airflow/models/dagcode.py

@@ -39,7 +39,6 @@ class DagCode(Base):
     dag_code table contains code of DAG files synchronized by scheduler.
     This feature is controlled by:
 
-    * ``[core] store_serialized_dags = True``: enable this feature
     * ``[core] store_dag_code = True``: enable this feature
 
     For details on dag serialization see SerializedDagModel

airflow/models/serialized_dag.py

@@ -47,16 +47,15 @@ class SerializedDagModel(Base):
     serialized_dag table is a snapshot of DAG files synchronized by scheduler.
     This feature is controlled by:
 
-    * ``[core] store_serialized_dags = True``: enable this feature
     * ``[core] min_serialized_dag_update_interval = 30`` (s):
       serialized DAGs are updated in DB when a file gets processed by scheduler,
       to reduce DB write rate, there is a minimal interval of updating serialized DAGs.
     * ``[scheduler] dag_dir_list_interval = 300`` (s):
       interval of deleting serialized DAGs in DB when the files are deleted, suggest
       to use a smaller interval such as 60
 
-    It is used by webserver to load dags when ``store_serialized_dags=True``.
-    Because reading from database is lightweight compared to importing from files,
+    It is used by webserver to load dags
+    because reading from database is lightweight compared to importing from files,
     it solves the webserver scalability issue.
     """
 

airflow/models/taskinstance.py

@@ -51,7 +51,6 @@
 from airflow.models.variable import Variable
 from airflow.models.xcom import XCOM_RETURN_KEY, XCom
 from airflow.sentry import Sentry
-from airflow.settings import STORE_SERIALIZED_DAGS
 from airflow.stats import Stats
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.dependencies_deps import REQUEUEABLE_DEPS, RUNNING_DEPS
@@ -1160,9 +1159,8 @@ def signal_handler(signum, frame):  # pylint: disable=unused-argument
         start_time = time.time()
 
         self.render_templates(context=context)
-        if STORE_SERIALIZED_DAGS:
-            RenderedTaskInstanceFields.write(RenderedTaskInstanceFields(ti=self, render_templates=False))
-            RenderedTaskInstanceFields.delete_old_records(self.task_id, self.dag_id)
+        RenderedTaskInstanceFields.write(RenderedTaskInstanceFields(ti=self, render_templates=False))
+        RenderedTaskInstanceFields.delete_old_records(self.task_id, self.dag_id)
 
         # Export context to make it available for operators to use.
         airflow_context_vars = context_to_airflow_vars(context, in_env_var_format=True)
@@ -1576,28 +1574,22 @@ def get(
         }
 
     def get_rendered_template_fields(self):
-        """
-        Fetch rendered template fields from DB if Serialization is enabled.
-        Else just render the templates
-        """
+        """Fetch rendered template fields from DB"""
         from airflow.models.renderedtifields import RenderedTaskInstanceFields
-        if STORE_SERIALIZED_DAGS:
-            rendered_task_instance_fields = RenderedTaskInstanceFields.get_templated_fields(self)
-            if rendered_task_instance_fields:
-                for field_name, rendered_value in rendered_task_instance_fields.items():
-                    setattr(self.task, field_name, rendered_value)
-            else:
-                try:
-                    self.render_templates()
-                except (TemplateAssertionError, UndefinedError) as e:
-                    raise AirflowException(
-                        "Webserver does not have access to User-defined Macros or Filters "
-                        "when Dag Serialization is enabled. Hence for the task that have not yet "
-                        "started running, please use 'airflow tasks render' for debugging the "
-                        "rendering of template_fields."
-                    ) from e
+        rendered_task_instance_fields = RenderedTaskInstanceFields.get_templated_fields(self)
+        if rendered_task_instance_fields:
+            for field_name, rendered_value in rendered_task_instance_fields.items():
+                setattr(self.task, field_name, rendered_value)
         else:
-            self.render_templates()
+            try:
+                self.render_templates()
+            except (TemplateAssertionError, UndefinedError) as e:
+                raise AirflowException(
+                    "Webserver does not have access to User-defined Macros or Filters "
+                    "when Dag Serialization is enabled. Hence for the task that have not yet "
+                    "started running, please use 'airflow tasks render' for debugging the "
+                    "rendering of template_fields."
+                ) from e
 
     def overwrite_params_with_dag_run_conf(self, params, dag_run):
         """Overwrite Task Params with DagRun.conf"""

airflow/operators/dagrun_operator.py

@@ -20,7 +20,6 @@
 from typing import Dict, Optional, Union
 from urllib.parse import quote
 
-from airflow import settings
 from airflow.api.common.experimental.trigger_dag import trigger_dag
 from airflow.exceptions import DagNotFound, DagRunAlreadyExists
 from airflow.models import BaseOperator, BaseOperatorLink, DagBag, DagModel, DagRun
@@ -122,7 +121,7 @@ def execute(self, context: Dict):
 
                 dag_bag = DagBag(
                     dag_folder=dag_model.fileloc,
-                    store_serialized_dags=settings.STORE_SERIALIZED_DAGS
+                    read_dags_from_db=True
                 )
 
                 dag = dag_bag.get_dag(self.trigger_dag_id)

airflow/settings.py

@@ -344,9 +344,6 @@ def initialize():
 WEB_COLORS = {'LIGHTBLUE': '#4d9de0',
               'LIGHTORANGE': '#FF9933'}
 
-# If store_serialized_dags is True, scheduler writes serialized DAGs to DB, and webserver
-# reads DAGs from DB instead of importing from files.
-STORE_SERIALIZED_DAGS = conf.getboolean('core', 'store_serialized_dags', fallback=False)
 
 # Updating serialized DAG can not be faster than a minimum interval to reduce database
 # write rate.
@@ -360,8 +357,7 @@ def initialize():
 
 # Whether to persist DAG files code in DB. If set to True, Webserver reads file contents
 # from DB instead of trying to access files in a DAG folder.
-# Defaults to same as the store_serialized_dags setting.
-STORE_DAG_CODE = conf.getboolean("core", "store_dag_code", fallback=STORE_SERIALIZED_DAGS)
+STORE_DAG_CODE = conf.getboolean("core", "store_dag_code", fallback=True)
 
 # If donot_modify_handlers=True, we do not modify logging handlers in task_run command
 # If the flag is set to False, we remove all handlers from the root logger

airflow/utils/dag_processing.py

@@ -38,9 +38,10 @@
 
 import airflow.models
 from airflow.configuration import conf
-from airflow.models import errors
+from airflow.models import DagModel, errors
+from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.taskinstance import SimpleTaskInstance
-from airflow.settings import STORE_DAG_CODE, STORE_SERIALIZED_DAGS
+from airflow.settings import STORE_DAG_CODE
 from airflow.stats import Stats
 from airflow.utils import timezone
 from airflow.utils.callback_requests import CallbackRequest, SlaCallbackRequest, TaskCallbackRequest
@@ -734,11 +735,8 @@ def _refresh_dag_dir(self):
             except Exception:  # noqa pylint: disable=broad-except
                 self.log.exception("Error removing old import errors")
 
-            if STORE_SERIALIZED_DAGS:
-                from airflow.models.dag import DagModel
-                from airflow.models.serialized_dag import SerializedDagModel
-                SerializedDagModel.remove_deleted_dags(self._file_paths)
-                DagModel.deactivate_deleted_dags(self._file_paths)
+            SerializedDagModel.remove_deleted_dags(self._file_paths)
+            DagModel.deactivate_deleted_dags(self._file_paths)
 
             if self.store_dag_code:
                 from airflow.models.dagcode import DagCode

airflow/www/extensions/init_dagbag.py

@@ -18,7 +18,7 @@
 import os
 
 from airflow.models import DagBag
-from airflow.settings import DAGS_FOLDER, STORE_SERIALIZED_DAGS
+from airflow.settings import DAGS_FOLDER
 
 
 def init_dagbag(app):
@@ -29,4 +29,4 @@ def init_dagbag(app):
     if os.environ.get('SKIP_DAGS_PARSING') == 'True':
         app.dag_bag = DagBag(os.devnull, include_examples=False)
     else:
-        app.dag_bag = DagBag(DAGS_FOLDER, read_dags_from_db=STORE_SERIALIZED_DAGS)
+        app.dag_bag = DagBag(DAGS_FOLDER, read_dags_from_db=True)

airflow/www/views.py

@@ -1867,7 +1867,12 @@ def duration(self, session=None):  # pylint: disable=too-many-locals
         """Get Dag as duration graph."""
         default_dag_run = conf.getint('webserver', 'default_dag_run_display_number')
         dag_id = request.args.get('dag_id')
-        dag = current_app.dag_bag.get_dag(dag_id)
+
+        try:
+            dag = current_app.dag_bag.get_dag(dag_id)
+        except airflow.exceptions.SerializedDagNotFound:
+            dag = None
+
         base_date = request.args.get('base_date')
         num_runs = request.args.get('num_runs')
         num_runs = int(num_runs) if num_runs else default_dag_run
@@ -2159,10 +2164,7 @@ def refresh(self, session=None):
     @action_logging
     def refresh_all(self):
         """Refresh everything"""
-        if settings.STORE_SERIALIZED_DAGS:
-            current_app.dag_bag.collect_dags_from_db()
-        else:
-            current_app.dag_bag.collect_dags(only_if_updated=False)
+        current_app.dag_bag.collect_dags_from_db()
 
         # sync permissions for all dags
         for dag_id, dag in current_app.dag_bag.dags.items():