Do only map known columns

dask_sql/datacontainer.py

@@ -156,6 +156,7 @@ def assign(self) -> dd.DataFrame:
             **{
                 col_from: self.df[col_to]
                 for col_from, col_to in self.column_container.mapping()
+                if col_from in self.column_container.columns
             }
         )
         return df[self.column_container.columns]

tests/integration/test_complex.py

@@ -0,0 +1,42 @@
+from unittest import TestCase
+
+from dask.datasets import timeseries
+
+from dask_sql import Context
+
+
+class TimeSeriesTestCase(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.c = Context()
+
+        df = timeseries().persist()
+        cls.c.register_dask_table(df, "timeseries")
+
+    def test_complex_query(self):
+        result = self.c.sql(
+            """
+            SELECT
+                lhs.name,
+                lhs.id,
+                lhs.x
+            FROM
+                timeseries AS lhs
+            JOIN
+                (
+                    SELECT
+                        name AS max_name,
+                        MAX(x) AS max_x
+                    FROM timeseries
+                    GROUP BY name
+                ) AS rhs
+            ON
+                lhs.name = rhs.max_name AND
+                lhs.x = rhs.max_x
+        """
+        )
+
+        # should not fail
+        df = result.compute()
+
+        self.assertGreater(len(df), 0)