session.add(DatasetModel(uri=dataset4.uri, is_orphaned=True))
            session.flush()
            dag1 = DAG(dag_id="datasets-1", start_date=DEFAULT_DATE, schedule=[dataset1])
            BashOperator(dag=dag1, task_id="task", bash_command="echo 1", outlets=[dataset3])
            DAG.bulk_write_to_db([dag1], session=session)
            
            non_orphaned_datasets = [
                dataset.uri
                for dataset in session.query(DatasetModel.uri)
                .filter(~DatasetModel.is_orphaned)
                .order_by(DatasetModel.uri)
            ]
            assert non_orphaned_datasets == ["ds1", "ds3"]
            orphaned_datasets = [
                dataset.uri
                for dataset in session.query(DatasetModel.uri)
                .filter(DatasetModel.is_orphaned)
                .order_by(DatasetModel.uri)
            ]
            assert orphaned_datasets == ["ds2", "ds4"]
            
            dag1 = DAG(dag_id="datasets-1", start_date=DEFAULT_DATE, schedule=[dataset1, dataset2])
            BashOperator(dag=dag1, task_id="task", bash_command="echo 1", outlets=[dataset3, dataset4])
            DAG.bulk_write_to_db([dag1], session=session)
            
            non_orphaned_dataset_count = session.query(DatasetModel).filter(~DatasetModel.is_orphaned).count()
            assert non_orphaned_dataset_count == 4
            orphaned_dataset_count = session.query(DatasetModel).filter(DatasetModel.is_orphaned).count()
            assert orphaned_dataset_count == 0
    def test_sync_to_db(self):