You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by as...@apache.org on 2025/01/31 11:19:09 UTC
(airflow) 01/01: Revert "Mostly use relative fileloc in dag processor (#46290)"
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch revert-46290-mostly-use-relative-loc-in-dag-processor
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7e944f987a5e39546a44457801d650b04b1483c7
Author: Ash Berlin-Taylor <as...@firemirror.com>
AuthorDate: Fri Jan 31 11:18:58 2025 +0000
Revert "Mostly use relative fileloc in dag processor (#46290)"
This reverts commit 000796a1d555701d72544e925f1838a5ddafa7c8.
---
airflow/dag_processing/collection.py | 1 -
airflow/dag_processing/manager.py | 74 +++++++++++++--------------------
airflow/models/dag.py | 14 +++----
airflow/models/dagbag.py | 2 +-
airflow/models/errors.py | 2 +-
tests/dag_processing/test_manager.py | 80 +++++++++++-------------------------
tests/models/test_dag.py | 8 ++--
7 files changed, 67 insertions(+), 114 deletions(-)
diff --git a/airflow/dag_processing/collection.py b/airflow/dag_processing/collection.py
index c8bac5cef96..879d2349644 100644
--- a/airflow/dag_processing/collection.py
+++ b/airflow/dag_processing/collection.py
@@ -210,7 +210,6 @@ def _serialize_dag_capturing_errors(
except Exception:
log.exception("Failed to write serialized DAG dag_id=%s fileloc=%s", dag.dag_id, dag.fileloc)
dagbag_import_error_traceback_depth = conf.getint("core", "dagbag_import_error_traceback_depth")
- # todo AIP-66: this needs to use bundle name / rel fileloc instead
return [(dag.fileloc, traceback.format_exc(limit=-dagbag_import_error_traceback_depth))]
diff --git a/airflow/dag_processing/manager.py b/airflow/dag_processing/manager.py
index fa68cb4fab1..a966f0a29c8 100644
--- a/airflow/dag_processing/manager.py
+++ b/airflow/dag_processing/manager.py
@@ -102,16 +102,10 @@ log = logging.getLogger("airflow.processor_manager")
class DagFileInfo:
"""Information about a DAG file."""
- rel_path: Path
+ path: str # absolute path of the file
bundle_name: str
bundle_path: Path | None = field(compare=False, default=None)
- @property
- def absolute_path(self) -> Path:
- if not self.bundle_path:
- raise ValueError("bundle_path not set")
- return self.bundle_path / self.rel_path
-
def _config_int_factory(section: str, key: str):
return functools.partial(conf.getint, section, key)
@@ -247,9 +241,7 @@ class DagFileProcessorManager:
elapsed_time_since_refresh = now - self._last_deactivate_stale_dags_time
if elapsed_time_since_refresh > self.parsing_cleanup_interval:
last_parsed = {
- file_info: stat.last_finish_time
- for file_info, stat in self._file_stats.items()
- if stat.last_finish_time
+ fp: stat.last_finish_time for fp, stat in self._file_stats.items() if stat.last_finish_time
}
self.deactivate_stale_dags(last_parsed=last_parsed)
self._last_deactivate_stale_dags_time = time.monotonic()
@@ -277,11 +269,14 @@ class DagFileProcessorManager:
# last_parsed_time is the processor_timeout. Longer than that indicates that the DAG is
# no longer present in the file. We have a stale_dag_threshold configured to prevent a
# significant delay in deactivation of stale dags when a large timeout is configured
- file_info = DagFileInfo(rel_path=Path(dag.relative_fileloc), bundle_name=dag.bundle_name)
- if last_finish_time := last_parsed.get(file_info, None):
- if dag.last_parsed_time + timedelta(seconds=self.stale_dag_threshold) < last_finish_time:
- self.log.info("DAG %s is missing and will be deactivated.", dag.dag_id)
- to_deactivate.add(dag.dag_id)
+ dag_file_path = DagFileInfo(path=dag.fileloc, bundle_name=dag.bundle_name)
+ if (
+ dag_file_path in last_parsed
+ and (dag.last_parsed_time + timedelta(seconds=self.stale_dag_threshold))
+ < last_parsed[dag_file_path]
+ ):
+ self.log.info("DAG %s is missing and will be deactivated.", dag.dag_id)
+ to_deactivate.add(dag.dag_id)
if to_deactivate:
deactivated_dagmodel = session.execute(
@@ -488,30 +483,30 @@ class DagFileProcessorManager:
"Version changed for %s, new version: %s", bundle.name, version_after_refresh
)
- found_file_infos = [
- DagFileInfo(rel_path=p, bundle_name=bundle.name, bundle_path=bundle.path)
- for p in self._find_files_in_bundle(bundle)
- ]
+ bundle_file_paths = self._find_files_in_bundle(bundle)
new_file_paths = [f for f in self._file_paths if f.bundle_name != bundle.name]
- new_file_paths.extend(found_file_infos)
+ new_file_paths.extend(
+ DagFileInfo(path=path, bundle_path=bundle.path, bundle_name=bundle.name)
+ for path in bundle_file_paths
+ )
self.set_file_paths(new_file_paths)
- self.deactivate_deleted_dags(active_files=found_file_infos)
+ self.deactivate_deleted_dags(bundle_file_paths)
self.clear_nonexistent_import_errors()
self._bundle_versions[bundle.name] = bundle.get_current_version()
- def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
- """Get relative file paths from bundle dir."""
+ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[str]:
+ """Refresh file paths from bundle dir."""
# Build up a list of Python files that could contain DAGs
self.log.info("Searching for files in %s at %s", bundle.name, bundle.path)
- file_paths = [Path(x).relative_to(bundle.path) for x in list_py_file_paths(bundle.path)]
+ file_paths = list_py_file_paths(bundle.path)
self.log.info("Found %s files for bundle %s", len(file_paths), bundle.name)
return file_paths
- def deactivate_deleted_dags(self, active_files: list[DagFileInfo]) -> None:
+ def deactivate_deleted_dags(self, file_paths: set[str]) -> None:
"""Deactivate DAGs that come from files that are no longer present."""
def _iter_dag_filelocs(fileloc: str) -> Iterator[str]:
@@ -531,20 +526,10 @@ class DagFileProcessorManager:
except zipfile.BadZipFile:
self.log.exception("There was an error accessing ZIP file %s %s", fileloc)
- active_subpaths: set[tuple[str, str]] = set()
- """
- 'subpath' here means bundle + modified rel path. What does modified rel path mean?
- Well, '_iter_dag_filelocs' walks through zip files and may return a "path" that is,
- rel path to the zip, plus the rel path within the zip. So, since this is is a bit different
- from most uses of the word "rel path", I wanted to call it something different.
- A set is used presumably since many dags can be in one file.
- """
-
- for info in active_files:
- for path in _iter_dag_filelocs(str(info.absolute_path)):
- active_subpaths.add((info.bundle_name, path))
+ dag_filelocs = {full_loc for path in file_paths for full_loc in _iter_dag_filelocs(path)}
- DagModel.deactivate_deleted_dags(active_subpaths)
+ # TODO: AIP-66: make bundle aware, as fileloc won't be unique long term.
+ DagModel.deactivate_deleted_dags(dag_filelocs)
def _print_stat(self):
"""Occasionally print out stats about how fast the files are getting processed."""
@@ -568,8 +553,7 @@ class DagFileProcessorManager:
if self._file_paths:
query = query.where(
tuple_(ParseImportError.filename, ParseImportError.bundle_name).notin_(
- # todo AIP-66: ParseImportError should have rel fileloce + bundle name
- [(str(f.absolute_path), f.bundle_name) for f in self._file_paths]
+ [(f.path, f.bundle_name) for f in self._file_paths]
),
)
@@ -614,7 +598,7 @@ class DagFileProcessorManager:
proc = self._processors.get(file_path)
num_dags = stat.num_dags
num_errors = stat.import_errors
- file_name = Path(file_path.rel_path).stem
+ file_name = Path(file_path.path).stem
processor_pid = proc.pid if proc else None
processor_start_time = proc.start_time if proc else None
runtime = (now - processor_start_time) if processor_start_time else None
@@ -766,7 +750,7 @@ class DagFileProcessorManager:
self._latest_log_symlink_date = datetime.today()
bundle = next(b for b in self._dag_bundles if b.name == dag_file.bundle_name)
- relative_path = Path(dag_file.rel_path)
+ relative_path = Path(dag_file.path).relative_to(bundle.path)
return os.path.join(self._get_log_dir(), bundle.name, f"{relative_path}.log")
def _get_logger_for_dag_file(self, dag_file: DagFileInfo):
@@ -784,7 +768,7 @@ class DagFileProcessorManager:
return DagFileProcessorProcess.start(
id=id,
- path=dag_file.absolute_path,
+ path=dag_file.path,
bundle_path=cast(Path, dag_file.bundle_path),
callbacks=callback_to_execute_for_file,
selector=self.selector,
@@ -836,7 +820,7 @@ class DagFileProcessorManager:
for file_path in self._file_paths:
if is_mtime_mode:
try:
- files_with_mtime[file_path] = os.path.getmtime(file_path.absolute_path)
+ files_with_mtime[file_path] = os.path.getmtime(file_path.path)
except FileNotFoundError:
self.log.warning("Skipping processing of missing file: %s", file_path)
self._file_stats.pop(file_path, None)
@@ -862,7 +846,7 @@ class DagFileProcessorManager:
if is_mtime_mode:
file_paths = sorted(files_with_mtime, key=files_with_mtime.get, reverse=True)
elif list_mode == "alphabetical":
- file_paths.sort(key=lambda f: f.rel_path)
+ file_paths.sort(key=lambda f: f.path)
elif list_mode == "random_seeded_by_host":
# Shuffle the list seeded by hostname so multiple DAG processors can work on different
# set of files. Since we set the seed, the sort order will remain same per host
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index a0f6e901dc3..a22452a748c 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -24,7 +24,7 @@ import logging
import sys
import time
from collections import defaultdict
-from collections.abc import Collection, Generator, Iterable, Sequence
+from collections.abc import Collection, Container, Generator, Iterable, Sequence
from contextlib import ExitStack
from datetime import datetime, timedelta
from functools import cache
@@ -2241,25 +2241,25 @@ class DagModel(Base):
@provide_session
def deactivate_deleted_dags(
cls,
- active: set[tuple[str, str]],
+ alive_dag_filelocs: Container[str],
session: Session = NEW_SESSION,
) -> None:
"""
Set ``is_active=False`` on the DAGs for which the DAG files have been removed.
- :param active_paths: file paths of alive DAGs
+ :param alive_dag_filelocs: file paths of alive DAGs
:param session: ORM Session
"""
log.debug("Deactivating DAGs (for which DAG files are deleted) from %s table ", cls.__tablename__)
dag_models = session.scalars(
select(cls).where(
- cls.relative_fileloc.is_not(None),
+ cls.fileloc.is_not(None),
)
)
- for dm in dag_models:
- if (dm.bundle_name, dm.relative_fileloc) not in active:
- dm.is_active = False
+ for dag_model in dag_models:
+ if dag_model.fileloc not in alive_dag_filelocs:
+ dag_model.is_active = False
@classmethod
def dags_needing_dagruns(cls, session: Session) -> tuple[Query, dict[str, tuple[datetime, datetime]]]:
diff --git a/airflow/models/dagbag.py b/airflow/models/dagbag.py
index a18b133f8e6..96eceb4c525 100644
--- a/airflow/models/dagbag.py
+++ b/airflow/models/dagbag.py
@@ -119,7 +119,7 @@ class DagBag(LoggingMixin):
def __init__(
self,
- dag_folder: str | Path | None = None, # todo AIP-66: rename this to path
+ dag_folder: str | Path | None = None,
include_examples: bool | ArgNotSet = NOTSET,
safe_mode: bool | ArgNotSet = NOTSET,
read_dags_from_db: bool = False,
diff --git a/airflow/models/errors.py b/airflow/models/errors.py
index 748d56c46b4..21c2236e2c1 100644
--- a/airflow/models/errors.py
+++ b/airflow/models/errors.py
@@ -29,6 +29,6 @@ class ParseImportError(Base):
__tablename__ = "import_error"
id = Column(Integer, primary_key=True)
timestamp = Column(UtcDateTime)
- filename = Column(String(1024)) # todo AIP-66: make this bundle and relative fileloc
+ filename = Column(String(1024))
bundle_name = Column(StringID())
stacktrace = Column(Text)
diff --git a/tests/dag_processing/test_manager.py b/tests/dag_processing/test_manager.py
index a68749f831e..a33884e6d4e 100644
--- a/tests/dag_processing/test_manager.py
+++ b/tests/dag_processing/test_manager.py
@@ -75,8 +75,8 @@ TEST_DAG_FOLDER = Path(__file__).parents[1].resolve() / "dags"
DEFAULT_DATE = timezone.datetime(2016, 1, 1)
-def _get_dag_file_paths(files: list[str | Path]) -> list[DagFileInfo]:
- return [DagFileInfo(bundle_name="testing", bundle_path=TEST_DAGS_FOLDER, rel_path=Path(f)) for f in files]
+def _get_dag_file_paths(files: list[str]) -> list[DagFileInfo]:
+ return [DagFileInfo(bundle_name="testing", bundle_path=TEST_DAGS_FOLDER, path=f) for f in files]
class TestDagFileProcessorManager:
@@ -164,9 +164,9 @@ class TestDagFileProcessorManager:
"""
manager = DagFileProcessorManager(max_runs=1)
- file_1 = DagFileInfo(bundle_name="testing", rel_path=Path("file_1.py"), bundle_path=TEST_DAGS_FOLDER)
- file_2 = DagFileInfo(bundle_name="testing", rel_path=Path("file_2.py"), bundle_path=TEST_DAGS_FOLDER)
- file_3 = DagFileInfo(bundle_name="testing", rel_path=Path("file_3.py"), bundle_path=TEST_DAGS_FOLDER)
+ file_1 = DagFileInfo(bundle_name="testing", path="file_1.py", bundle_path=TEST_DAGS_FOLDER)
+ file_2 = DagFileInfo(bundle_name="testing", path="file_2.py", bundle_path=TEST_DAGS_FOLDER)
+ file_3 = DagFileInfo(bundle_name="testing", path="file_3.py", bundle_path=TEST_DAGS_FOLDER)
manager._file_path_queue = deque([file_1, file_2, file_3])
# Mock that only one processor exists. This processor runs with 'file_1'
@@ -188,9 +188,7 @@ class TestDagFileProcessorManager:
def test_set_file_paths_when_processor_file_path_not_in_new_file_paths(self):
"""Ensure processors and file stats are removed when the file path is not in the new file paths"""
manager = DagFileProcessorManager(max_runs=1)
- file = DagFileInfo(
- bundle_name="testing", rel_path=Path("missing_file.txt"), bundle_path=TEST_DAGS_FOLDER
- )
+ file = DagFileInfo(bundle_name="testing", path="missing_file.txt", bundle_path=TEST_DAGS_FOLDER)
manager._processors[file] = MagicMock()
manager._file_stats[file] = DagFileStat()
@@ -201,7 +199,7 @@ class TestDagFileProcessorManager:
def test_set_file_paths_when_processor_file_path_is_in_new_file_paths(self):
manager = DagFileProcessorManager(max_runs=1)
- file = DagFileInfo(bundle_name="testing", rel_path=Path("abc.txt"), bundle_path=TEST_DAGS_FOLDER)
+ file = DagFileInfo(bundle_name="testing", path="abc.txt", bundle_path=TEST_DAGS_FOLDER)
mock_processor = MagicMock()
manager._processors[file] = mock_processor
@@ -289,10 +287,7 @@ class TestDagFileProcessorManager:
assert manager._file_path_queue == deque(ordered_files)
manager.set_file_paths(
- [
- *dag_files,
- DagFileInfo(bundle_name="testing", rel_path=Path("file_4.py"), bundle_path=TEST_DAGS_FOLDER),
- ]
+ [*dag_files, DagFileInfo(bundle_name="testing", path="file_4.py", bundle_path=TEST_DAGS_FOLDER)]
)
manager.add_new_file_path_to_queue()
ordered_files = _get_dag_file_paths(["file_4.py", "file_3.py", "file_2.py", "file_1.py"])
@@ -306,9 +301,7 @@ class TestDagFileProcessorManager:
"""
freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
initial_file_1_mtime = (freezed_base_time - timedelta(minutes=5)).timestamp()
- dag_file = DagFileInfo(
- bundle_name="testing", rel_path=Path("file_1.py"), bundle_path=TEST_DAGS_FOLDER
- )
+ dag_file = DagFileInfo(bundle_name="testing", path="file_1.py", bundle_path=TEST_DAGS_FOLDER)
dag_files = [dag_file]
mock_getmtime.side_effect = [initial_file_1_mtime]
@@ -378,15 +371,10 @@ class TestDagFileProcessorManager:
test_dag_path = DagFileInfo(
bundle_name="testing",
- rel_path=Path("test_example_bash_operator.py"),
+ path=str(TEST_DAG_FOLDER / "test_example_bash_operator.py"),
bundle_path=TEST_DAGS_FOLDER,
)
- dagbag = DagBag(
- test_dag_path.absolute_path,
- read_dags_from_db=False,
- include_examples=False,
- bundle_path=test_dag_path.bundle_path,
- )
+ dagbag = DagBag(test_dag_path.path, read_dags_from_db=False, include_examples=False)
with create_session() as session:
# Add stale DAG to the DB
@@ -409,11 +397,7 @@ class TestDagFileProcessorManager:
active_dag_count = (
session.query(func.count(DagModel.dag_id))
- .filter(
- DagModel.is_active,
- DagModel.relative_fileloc == str(test_dag_path.rel_path),
- DagModel.bundle_name == test_dag_path.bundle_name,
- )
+ .filter(DagModel.is_active, DagModel.fileloc == test_dag_path.path)
.scalar()
)
assert active_dag_count == 1
@@ -422,11 +406,7 @@ class TestDagFileProcessorManager:
active_dag_count = (
session.query(func.count(DagModel.dag_id))
- .filter(
- DagModel.is_active,
- DagModel.relative_fileloc == str(test_dag_path.rel_path),
- DagModel.bundle_name == test_dag_path.bundle_name,
- )
+ .filter(DagModel.is_active, DagModel.fileloc == test_dag_path.path)
.scalar()
)
assert active_dag_count == 0
@@ -446,9 +426,7 @@ class TestDagFileProcessorManager:
processor = self.mock_processor()
processor._process.create_time.return_value = timezone.make_aware(datetime.min).timestamp()
manager._processors = {
- DagFileInfo(
- bundle_name="testing", rel_path=Path("abc.txt"), bundle_path=TEST_DAGS_FOLDER
- ): processor
+ DagFileInfo(bundle_name="testing", path="abc.txt", bundle_path=TEST_DAGS_FOLDER): processor
}
with mock.patch.object(type(processor), "kill") as mock_kill:
manager._kill_timed_out_processors()
@@ -464,9 +442,7 @@ class TestDagFileProcessorManager:
processor = self.mock_processor()
processor._process.create_time.return_value = timezone.make_aware(datetime.max).timestamp()
manager._processors = {
- DagFileInfo(
- bundle_name="testing", rel_path=Path("abc.txt"), bundle_path=TEST_DAGS_FOLDER
- ): processor
+ DagFileInfo(bundle_name="testing", path="abc.txt", bundle_path=TEST_DAGS_FOLDER): processor
}
with mock.patch.object(type(processor), "kill") as mock_kill:
manager._kill_timed_out_processors()
@@ -684,11 +660,6 @@ class TestDagFileProcessorManager:
@pytest.mark.skip("AIP-66: callbacks are not implemented yet")
def test_callback_queue(self, tmp_path):
- """
- This test has gotten a bit out of sync with the codebase.
-
- I am just updating it to be consistent with the changes in DagFileInfo
- """
# given
manager = DagFileProcessorManager(
max_runs=1,
@@ -696,17 +667,17 @@ class TestDagFileProcessorManager:
)
dag1_path = DagFileInfo(
- bundle_name="testing", rel_path=Path("green_eggs/ham/file1.py"), bundle_path=TEST_DAGS_FOLDER
+ bundle_name="testing", path="/green_eggs/ham/file1.py", bundle_path=TEST_DAGS_FOLDER
)
dag1_req1 = DagCallbackRequest(
- full_filepath=TEST_DAGS_FOLDER / "green_eggs/ham/file1.py",
+ full_filepath="/green_eggs/ham/file1.py",
dag_id="dag1",
run_id="run1",
is_failure_callback=False,
msg=None,
)
dag1_req2 = DagCallbackRequest(
- full_filepath=TEST_DAGS_FOLDER / "green_eggs/ham/file1.py",
+ full_filepath="/green_eggs/ham/file1.py",
dag_id="dag1",
run_id="run1",
is_failure_callback=False,
@@ -714,10 +685,10 @@ class TestDagFileProcessorManager:
)
dag2_path = DagFileInfo(
- bundle_name="testing", rel_path=Path("green_eggs/ham/file2.py"), bundle_path=TEST_DAGS_FOLDER
+ bundle_name="testing", path="/green_eggs/ham/file2.py", bundle_path=TEST_DAGS_FOLDER
)
dag2_req1 = DagCallbackRequest(
- full_filepath=TEST_DAGS_FOLDER / "green_eggs/ham/file2.py",
+ full_filepath="/green_eggs/ham/file2.py",
dag_id="dag2",
run_id="run1",
is_failure_callback=False,
@@ -800,12 +771,10 @@ class TestDagFileProcessorManager:
bundleone = MagicMock()
bundleone.name = "bundleone"
- bundleone.path = "/dev/null"
bundleone.refresh_interval = 0
bundleone.get_current_version.return_value = None
bundletwo = MagicMock()
bundletwo.name = "bundletwo"
- bundletwo.path = "/dev/null"
bundletwo.refresh_interval = 300
bundletwo.get_current_version.return_value = None
@@ -847,7 +816,7 @@ class TestDagFileProcessorManager:
manager.run()
assert bundletwo.refresh.call_count == 2
- def test_bundles_versions_are_stored(self, session):
+ def test_bundles_versions_are_stored(self):
config = [
{
"name": "mybundle",
@@ -857,8 +826,7 @@ class TestDagFileProcessorManager:
]
mybundle = MagicMock()
- mybundle.name = "mybundle"
- mybundle.path = "/dev/null"
+ mybundle.name = "bundleone"
mybundle.refresh_interval = 0
mybundle.supports_versioning = True
mybundle.get_current_version.return_value = "123"
@@ -868,11 +836,11 @@ class TestDagFileProcessorManager:
with mock.patch(
"airflow.dag_processing.bundles.manager.DagBundlesManager"
) as mock_bundle_manager:
- mock_bundle_manager.return_value._bundle_config = {"mybundle": None}
+ mock_bundle_manager.return_value._bundle_config = {"bundleone": None}
mock_bundle_manager.return_value.get_all_dag_bundles.return_value = [mybundle]
manager = DagFileProcessorManager(max_runs=1)
manager.run()
with create_session() as session:
- model = session.get(DagBundleModel, "mybundle")
+ model = session.get(DagBundleModel, "bundleone")
assert model.version == "123"
diff --git a/tests/models/test_dag.py b/tests/models/test_dag.py
index 78fffd45fd0..8766a89a810 100644
--- a/tests/models/test_dag.py
+++ b/tests/models/test_dag.py
@@ -1081,16 +1081,18 @@ class TestDag:
dag.clear()
self._clean_up(dag_id)
- def test_dag_is_deactivated_upon_dagfile_deletion(self, dag_maker):
+ def test_dag_is_deactivated_upon_dagfile_deletion(self):
dag_id = "old_existing_dag"
- with dag_maker(dag_id, schedule=None, is_paused_upon_creation=True) as dag:
- ...
+ dag_fileloc = "/usr/local/airflow/dags/non_existing_path.py"
+ dag = DAG(dag_id, schedule=None, is_paused_upon_creation=True)
+ dag.fileloc = dag_fileloc
session = settings.Session()
dag.sync_to_db(session=session)
orm_dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).one()
assert orm_dag.is_active
+ assert orm_dag.fileloc == dag_fileloc
DagModel.deactivate_deleted_dags(list_py_file_paths(settings.DAGS_FOLDER))