switches redshift database name

.github/workflows/test_dbt_runner.yml

@@ -11,7 +11,7 @@ on:
 env:
   # all credentials must be present to be passed to dbt runner
   DESTINATION__POSTGRES__CREDENTIALS: postgresql://[email protected]:5432/dlt_data
-  DESTINATION__REDSHIFT__CREDENTIALS: postgresql://[email protected]:5439/chat_analytics_rasa_ci
+  DESTINATION__REDSHIFT__CREDENTIALS: postgresql://[email protected]:5439/dlt_ci
   DESTINATION__CREDENTIALS__PASSWORD: ${{ secrets.PG_PASSWORD }}
 
   DESTINATION__CREDENTIALS__PROJECT_ID: chat-analytics-rasa-ci

.github/workflows/test_destinations.yml

@@ -10,7 +10,7 @@ on:
 
 env:
   DESTINATION__POSTGRES__CREDENTIALS: postgresql://[email protected]:5432/dlt_data
-  DESTINATION__REDSHIFT__CREDENTIALS: postgresql://[email protected]:5439/chat_analytics_rasa_ci
+  DESTINATION__REDSHIFT__CREDENTIALS: postgresql://[email protected]:5439/dlt_ci
   DESTINATION__DUCKDB__CREDENTIALS: duckdb:///_storage/test_quack.duckdb
   # password is the same so it will be shared
   CREDENTIALS__PASSWORD: ${{ secrets.PG_PASSWORD }}

dlt/destinations/redshift/README.md

@@ -11,9 +11,9 @@ https://www.intermix.io/blog/top-14-performance-tuning-techniques-for-amazon-red
 
 # loader account setup
 
-1. Create new database `CREATE DATABASE chat_analytics_rasa_ci`
+1. Create new database `CREATE DATABASE dlt_ci`
 2. Create new user, set password
-3. Set as database owner (we could set lower permission) `ALTER DATABASE chat_analytics_rasa_ci OWNER TO loader`
+3. Set as database owner (we could set lower permission) `ALTER DATABASE dlt_ci OWNER TO loader`
 
 # Public access setup for Serverless
 Follow https://docs.aws.amazon.com/redshift/latest/mgmt/serverless-connecting.html `Connecting from the public subnet to the Amazon Redshift Serverless endpoint using Network Load Balancer`

dlt/extract/pipe.py

@@ -567,7 +567,10 @@ def _resolve_futures(self) -> ResolvablePipeItem:
             return self._resolve_futures()
 
         if future.exception():
-            raise future.exception()
+            ex = future.exception()
+            if isinstance(ex, (PipelineException, ExtractorException, DltSourceException, PipeException)):
+                raise ex
+            raise ResourceExtractionError(pipe.name, future, str(ex), "future") from ex
 
         item = future.result()
         if isinstance(item, DataItemWithMeta):

tests/extract/test_extract_pipe.py

@@ -8,7 +8,7 @@
 import dlt
 from dlt.common import sleep
 from dlt.common.typing import TDataItems
-from dlt.extract.exceptions import CreatePipeException
+from dlt.extract.exceptions import CreatePipeException, ResourceExtractionError
 from dlt.extract.typing import DataItemWithMeta, FilterItem, MapItem, YieldMapItem
 from dlt.extract.pipe import ManagedPipeIterator, Pipe, PipeItem, PipeIterator
 
@@ -552,8 +552,9 @@ def assert_pipes_closed(raise_gen, long_gen) -> None:
 
     pit: PipeIterator = None
     with PipeIterator.from_pipe(Pipe.from_data("failing", raise_gen, parent=Pipe.from_data("endless", long_gen()))) as pit:
-        with pytest.raises(RuntimeError):
+        with pytest.raises(ResourceExtractionError) as py_ex:
             list(pit)
+        assert isinstance(py_ex.value.__cause__, RuntimeError)
     # it got closed
     assert pit._sources == []
     assert close_pipe_got_exit is True
@@ -563,8 +564,9 @@ def assert_pipes_closed(raise_gen, long_gen) -> None:
     close_pipe_got_exit = False
     close_pipe_yielding = False
     pit = ManagedPipeIterator.from_pipe(Pipe.from_data("failing", raise_gen, parent=Pipe.from_data("endless", long_gen())))
-    with pytest.raises(RuntimeError):
+    with pytest.raises(ResourceExtractionError) as py_ex:
         list(pit)
+    assert isinstance(py_ex.value.__cause__, RuntimeError)
     assert pit._sources == []
     assert close_pipe_got_exit is True
     # while long gen was still yielding

tests/load/pipeline/test_dbt_helper.py

@@ -30,6 +30,8 @@ def test_run_jaffle_package(destination_name: str, dbt_venv: Venv) -> None:
     dbt = dlt.dbt.package(pipeline, "https://github.com/dbt-labs/jaffle_shop.git", venv=dbt_venv)
     # no default schema
     assert pipeline.default_schema_name is None
+    # inject default schema otherwise dataset is not deleted
+    pipeline._inject_schema(Schema("default"))
     # run the package (clone/pull repo, deps, seed, source tests, run)
     models = dbt.run_all()
     # all models were executed