Do the CompoundSelect check for number of columns in the compile phase

Starting to go forward with the general idea of moving more
of Core / ORM construction into the compile phase.   Bigger
initiatives like the refactor of Query will follow onto this.

Change-Id: I0f364d3182e21e32ed85ef34cfd11fd9d11cf653

2 files changed, 33 insertions, 27 deletions

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index a28ce465a..4ec3b93ea 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -2149,7 +2149,9 @@ class SQLCompiler(Compiled):
         if not populate_result_map and "add_to_result_map" in kwargs:
             del kwargs["add_to_result_map"]
 
-        froms = self._setup_select_stack(select, entry, asfrom, lateral)
+        froms = self._setup_select_stack(
+            select, entry, asfrom, lateral, compound_index
+        )
 
         column_clause_args = kwargs.copy()
         column_clause_args.update(
@@ -2254,10 +2256,33 @@ class SQLCompiler(Compiled):
         hint_text = self.get_select_hint_text(byfrom)
         return hint_text, byfrom
 
-    def _setup_select_stack(self, select, entry, asfrom, lateral):
+    def _setup_select_stack(
+        self, select, entry, asfrom, lateral, compound_index
+    ):
         correlate_froms = entry["correlate_froms"]
         asfrom_froms = entry["asfrom_froms"]
 
+        if compound_index > 0:
+            # note this is cached
+            select_0 = entry["selectable"].selects[0]
+            if select_0._is_select_container:
+                select_0 = select_0.element
+            numcols = len(select_0.selected_columns)
+            # numcols = len(select_0._columns_plus_names)
+            if len(select._columns_plus_names) != numcols:
+                raise exc.CompileError(
+                    "All selectables passed to "
+                    "CompoundSelect must have identical numbers of "
+                    "columns; select #%d has %d columns, select "
+                    "#%d has %d"
+                    % (
+                        1,
+                        numcols,
+                        compound_index + 1,
+                        len(select.selected_columns),
+                    )
+                )
+
         if asfrom and not lateral:
             froms = select._get_display_froms(
                 explicit_correlate_froms=correlate_froms.difference(
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 5f609f8fd..bece0b3c5 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -2782,31 +2782,12 @@ class CompoundSelect(GenerativeSelect):
     def __init__(self, keyword, *selects, **kwargs):
         self._auto_correlate = kwargs.pop("correlate", False)
         self.keyword = keyword
-        self.selects = []
-
-        numcols = None
-
-        # some DBs do not like ORDER BY in the inner queries of a UNION, etc.
-        for n, s in enumerate(selects):
-            s = coercions.expect(roles.CompoundElementRole, s)
-
-            if not numcols:
-                numcols = len(s.selected_columns)
-            elif len(s.selected_columns) != numcols:
-                raise exc.ArgumentError(
-                    "All selectables passed to "
-                    "CompoundSelect must have identical numbers of "
-                    "columns; select #%d has %d columns, select "
-                    "#%d has %d"
-                    % (
-                        1,
-                        len(self.selects[0].selected_columns),
-                        n + 1,
-                        len(s.selected_columns),
-                    )
-                )
-
-            self.selects.append(s.self_group(against=self))
+        self.selects = [
+            coercions.expect(roles.CompoundElementRole, s).self_group(
+                against=self
+            )
+            for s in selects
+        ]
 
         GenerativeSelect.__init__(self, **kwargs)