Add comprehensive BigQuery/Postgres integration tests with symmetric aggregation

- Add 7 new BigQuery SemanticLayer tests: ORDER BY, LIMIT, symmetric aggs, 3-way joins
- Add 10 new Postgres SemanticLayer tests: filters, ORDER BY, LIMIT, symmetric aggs, 3-way joins, compile
- Implement dialect-aware symmetric aggregation for BigQuery and Postgres to prevent fan-out double-counting
- Fix fanout detection: trigger symmetric agg on any one-to-many join, not just multiple
- Fix BigQuery CI: use docker run with args instead of service container
- Update test expectations to match correct symmetric agg behavior

Integration tests: 27 passed, 2 skipped
Regular tests: 570 passed

Author: nicosuave

.github/workflows/integration.yml

@@ -48,20 +48,19 @@ jobs:
   bigquery-integration:
     runs-on: ubuntu-latest
 
-    services:
-      bigquery:
-        image: ghcr.io/goccy/bigquery-emulator:latest
-        ports:
-          - 9050:9050
-        options: >-
-          --health-cmd "grpc_health_probe -addr=:9050"
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
-
     steps:
       - uses: actions/checkout@v4
 
+      - name: Start BigQuery emulator
+        run: |
+          docker run -d --name bigquery-emulator \
+            -p 9050:9050 \
+            ghcr.io/goccy/bigquery-emulator:latest \
+            --project=test-project --dataset=test_dataset
+
+          # Wait for emulator to be ready
+          sleep 5
+
       - name: Install uv
         uses: astral-sh/setup-uv@v5
         with:

sidemantic/core/symmetric_aggregate.py

@@ -17,6 +17,7 @@ def build_symmetric_aggregate_sql(
     primary_key: str,
     agg_type: Literal["sum", "avg", "count", "count_distinct"],
     model_alias: str | None = None,
+    dialect: str = "duckdb",
 ) -> str:
     """Build SQL for symmetric aggregate to prevent double-counting in fan-out joins.
 
@@ -25,30 +26,56 @@ def build_symmetric_aggregate_sql(
         primary_key: The primary key field to use for deduplication
         agg_type: Type of aggregation (sum, avg, count, count_distinct)
         model_alias: Optional table/CTE alias to prefix columns
+        dialect: SQL dialect (duckdb, bigquery, postgres)
 
     Returns:
         SQL expression using symmetric aggregates
 
     Examples:
         >>> build_symmetric_aggregate_sql("amount", "order_id", "sum")
-        '(SUM(DISTINCT HASH(order_id) * 1e15 + amount) - SUM(DISTINCT HASH(order_id) * 1e15))'
+        '(SUM(DISTINCT (HASH(order_id)::HUGEINT * (1::HUGEINT << 20)) + amount) - SUM(DISTINCT (HASH(order_id)::HUGEINT * (1::HUGEINT << 20))))'
 
         >>> build_symmetric_aggregate_sql("amount", "order_id", "avg", "orders_cte")
-        '(SUM(DISTINCT HASH(orders_cte.order_id) * 1e15 + orders_cte.amount) - SUM(DISTINCT HASH(orders_cte.order_id) * 1e15)) / NULLIF(COUNT(DISTINCT orders_cte.order_id), 0)'
+        '(SUM(DISTINCT (HASH(orders_cte.order_id)::HUGEINT * (1::HUGEINT << 20)) + orders_cte.amount) - SUM(DISTINCT (HASH(orders_cte.order_id)::HUGEINT * (1::HUGEINT << 20)))) / NULLIF(COUNT(DISTINCT orders_cte.order_id), 0)'
     """
     # Add table prefix if provided
     pk_col = f"{model_alias}.{primary_key}" if model_alias else primary_key
     measure_col = f"{model_alias}.{measure_expr}" if model_alias else measure_expr
 
+    # Dialect-specific hash and multiplier functions
+    if dialect == "bigquery":
+
+        def hash_func(col):
+            return f"FARM_FINGERPRINT(CAST({col} AS STRING))"
+
+        multiplier = "1048576"  # 2^20 as literal
+    elif dialect in ("postgres", "postgresql"):
+        # Use hashtext which returns int4, then cast to bigint and multiply
+        # Use smaller multiplier (2^10 = 1024) to avoid overflow
+        def hash_func(col):
+            return f"hashtext({col}::text)::bigint"
+
+        multiplier = "1024"  # 2^10 as literal (smaller to avoid overflow)
+    else:  # duckdb
+
+        def hash_func(col):
+            return f"HASH({col})::HUGEINT"
+
+        multiplier = "(1::HUGEINT << 20)"
+
     if agg_type == "sum":
-        # SUM(DISTINCT HASH(pk) * power_of_2 + value) - SUM(DISTINCT HASH(pk) * power_of_2)
-        # Use 2^20 (~1 million) for the multiplier - enough headroom for typical values
-        # Use HUGEINT (128-bit) to avoid overflow
-        return f"(SUM(DISTINCT (HASH({pk_col})::HUGEINT * (1::HUGEINT << 20)) + {measure_col}) - SUM(DISTINCT (HASH({pk_col})::HUGEINT * (1::HUGEINT << 20))))"
+        # SUM(DISTINCT HASH(pk) * multiplier + value) - SUM(DISTINCT HASH(pk) * multiplier)
+        hash_expr = hash_func(pk_col)
+        return (
+            f"(SUM(DISTINCT ({hash_expr} * {multiplier}) + {measure_col}) - SUM(DISTINCT ({hash_expr} * {multiplier})))"
+        )
 
     elif agg_type == "avg":
         # Sum divided by distinct count
-        sum_expr = f"(SUM(DISTINCT (HASH({pk_col})::HUGEINT * (1::HUGEINT << 20)) + {measure_col}) - SUM(DISTINCT (HASH({pk_col})::HUGEINT * (1::HUGEINT << 20))))"
+        hash_expr = hash_func(pk_col)
+        sum_expr = (
+            f"(SUM(DISTINCT ({hash_expr} * {multiplier}) + {measure_col}) - SUM(DISTINCT ({hash_expr} * {multiplier})))"
+        )
         count_expr = f"COUNT(DISTINCT {pk_col})"
         return f"{sum_expr} / NULLIF({count_expr}, 0)"
 

sidemantic/sql/generator.py

@@ -680,7 +680,7 @@ def collect_measures_from_metric(metric_ref: str):
     def _has_fanout_joins(self, base_model_name: str, other_models: list[str]) -> dict[str, bool]:
         """Determine which models need symmetric aggregates due to fan-out.
 
-        When multiple one-to-many joins exist from the base model, measures from
+        When one-to-many joins exist from the base model, measures from
         the base model need symmetric aggregates to prevent double-counting.
 
         Args:
@@ -692,24 +692,35 @@ def _has_fanout_joins(self, base_model_name: str, other_models: list[str]) -> di
         """
         needs_symmetric = {}
 
-        # Check if there are multiple one-to-many relationships
+        # Check if there are any one-to-many relationships
         one_to_many_count = 0
+        many_to_one_models = []
 
         for other_model in other_models:
             try:
                 join_path = self.graph.find_relationship_path(base_model_name, other_model)
                 # Check if first hop is one-to-many
                 if join_path and join_path[0].relationship == "one_to_many":
                     one_to_many_count += 1
+                elif join_path and join_path[0].relationship == "many_to_one":
+                    # Track models with many-to-one from base perspective
+                    many_to_one_models.append(other_model)
             except (ValueError, KeyError):
                 pass
 
-        # If we have multiple one-to-many joins, the base model needs symmetric aggregates
-        needs_symmetric[base_model_name] = one_to_many_count > 1
+        # Base model needs symmetric aggregates if there are any one-to-many joins
+        needs_symmetric[base_model_name] = one_to_many_count > 0
 
-        # Other models generally don't need it (they're on the "many" side)
+        # Models on the "many" side of a many-to-one relationship also need symmetric
+        # aggregation if they're being joined (because from their perspective,
+        # they're creating fan-out for the "one" side)
         for other_model in other_models:
-            needs_symmetric[other_model] = False
+            if other_model in many_to_one_models:
+                # Check if the "one" side (base) has metrics - if so, it needs symmetric agg
+                # But we're checking from the perspective of this model, so mark False
+                needs_symmetric[other_model] = False
+            else:
+                needs_symmetric[other_model] = False
 
         return needs_symmetric
 
@@ -841,6 +852,7 @@ def _build_main_select(
                                 primary_key=pk,
                                 agg_type=measure.agg,
                                 model_alias=f"{model_name}_cte",
+                                dialect=self.dialect,
                             )
                         else:
                             # Regular aggregation

tests/db/test_bigquery_integration.py

@@ -207,3 +207,173 @@ def test_semantic_layer_sql_generation(bigquery_layer):
     assert "SELECT" in sql.upper()
     assert "SUM" in sql.upper()
     assert bigquery_layer.dialect == "bigquery"
+
+
+def test_semantic_layer_order_by(bigquery_layer):
+    """Test ORDER BY with SemanticLayer."""
+    scores = Model(
+        name="scores",
+        table="""(
+            SELECT 'Alice' as name, 85 as score UNION ALL
+            SELECT 'Bob', 92 UNION ALL
+            SELECT 'Charlie', 78 UNION ALL
+            SELECT 'Diana', 95
+        )""",
+        primary_key="name",
+        dimensions=[Dimension(name="name", type="categorical")],
+        metrics=[Metric(name="avg_score", agg="avg", sql="score")],
+    )
+    bigquery_layer.add_model(scores)
+
+    # Order by dimension
+    result = bigquery_layer.query(
+        dimensions=["scores.name"], metrics=["scores.avg_score"], order_by=["scores.avg_score DESC"]
+    )
+    rows = result.fetchall()
+    # First row should have highest score
+    assert rows[0][1] == 95  # Diana
+    assert rows[-1][1] == 78  # Charlie
+
+
+def test_semantic_layer_limit(bigquery_layer):
+    """Test LIMIT with SemanticLayer."""
+    items = Model(
+        name="items",
+        table="""(
+            SELECT 1 as id, 'A' as category UNION ALL
+            SELECT 2, 'B' UNION ALL
+            SELECT 3, 'A' UNION ALL
+            SELECT 4, 'C' UNION ALL
+            SELECT 5, 'B'
+        )""",
+        primary_key="id",
+        dimensions=[Dimension(name="category", type="categorical")],
+        metrics=[Metric(name="count", agg="count", sql="id")],
+    )
+    bigquery_layer.add_model(items)
+
+    result = bigquery_layer.query(dimensions=["items.category"], metrics=["items.count"], limit=2)
+    rows = result.fetchall()
+    assert len(rows) == 2
+
+
+@pytest.mark.skip(reason="FORMAT_DATE appears to hang with BigQuery emulator")
+def test_semantic_layer_date_functions(bigquery_layer):
+    """Test date/time functions in metrics."""
+    events = Model(
+        name="events",
+        table="""(
+            SELECT DATE('2024-01-15') as event_date, 1 as event_id UNION ALL
+            SELECT DATE('2024-01-20'), 2 UNION ALL
+            SELECT DATE('2024-02-10'), 3 UNION ALL
+            SELECT DATE('2024-02-15'), 4
+        )""",
+        primary_key="event_id",
+        dimensions=[
+            Dimension(name="month", type="time", sql="FORMAT_DATE('%Y-%m', event_date)", granularity="month"),
+            Dimension(name="year", type="time", sql="FORMAT_DATE('%Y', event_date)", granularity="year"),
+        ],
+        metrics=[Metric(name="event_count", agg="count", sql="event_id")],
+    )
+    bigquery_layer.add_model(events)
+
+    result = bigquery_layer.query(dimensions=["events.month"], metrics=["events.event_count"])
+    rows = result.fetchall()
+    results_dict = {row[0]: row[1] for row in rows}
+
+    assert results_dict["2024-01"] == 2
+    assert results_dict["2024-02"] == 2
+
+
+def test_semantic_layer_symmetric_aggregates(bigquery_layer):
+    """Test symmetric aggregates handle fan-out joins correctly."""
+    # Create a fan-out scenario: order has multiple line_items
+    orders_sym = Model(
+        name="orders_sym",
+        table="""(
+            SELECT 1 as order_id, 100 as subtotal UNION ALL
+            SELECT 2, 200
+        )""",
+        primary_key="order_id",
+        metrics=[Metric(name="total_subtotal", agg="sum", sql="subtotal")],
+    )
+
+    line_items_sym = Model(
+        name="line_items_sym",
+        table="""(
+            SELECT 1 as item_id, 1 as order_id, 50 as price UNION ALL
+            SELECT 2, 1, 30 UNION ALL
+            SELECT 3, 1, 20 UNION ALL
+            SELECT 4, 2, 100 UNION ALL
+            SELECT 5, 2, 100
+        )""",
+        primary_key="item_id",
+        metrics=[Metric(name="total_price", agg="sum", sql="price")],
+        relationships=[Relationship(name="orders_sym", type="many_to_one", foreign_key="order_id")],
+    )
+
+    bigquery_layer.add_model(orders_sym)
+    bigquery_layer.add_model(line_items_sym)
+
+    # Query both metrics - should use symmetric aggregation to avoid fan-out
+    result = bigquery_layer.query(metrics=["orders_sym.total_subtotal", "line_items_sym.total_price"])
+    row = result.fetchone()
+    cols = [desc[0] for desc in result.description]
+    row_dict = dict(zip(cols, row))
+
+    # Without symmetric aggregation, total_subtotal would be inflated
+    assert row_dict["total_subtotal"] == 300  # 100 + 200, not inflated
+    assert row_dict["total_price"] == 300  # 50+30+20+100+100
+
+
+def test_semantic_layer_multiple_joins(bigquery_layer):
+    """Test joining 3+ models together."""
+    users_multi = Model(
+        name="users_multi",
+        table="""(
+            SELECT 1 as user_id, 'Alice' as name UNION ALL
+            SELECT 2, 'Bob'
+        )""",
+        primary_key="user_id",
+        dimensions=[Dimension(name="name", type="categorical")],
+    )
+
+    orders_multi = Model(
+        name="orders_multi",
+        table="""(
+            SELECT 1 as order_id, 1 as user_id, 1 as product_id, 100 as amount UNION ALL
+            SELECT 2, 1, 2, 150 UNION ALL
+            SELECT 3, 2, 1, 200
+        )""",
+        primary_key="order_id",
+        metrics=[Metric(name="total", agg="sum", sql="amount")],
+        relationships=[
+            Relationship(name="users_multi", type="many_to_one", foreign_key="user_id"),
+            Relationship(name="products_multi", type="many_to_one", foreign_key="product_id"),
+        ],
+    )
+
+    products_multi = Model(
+        name="products_multi",
+        table="""(
+            SELECT 1 as product_id, 'Widget' as product_name UNION ALL
+            SELECT 2, 'Gadget'
+        )""",
+        primary_key="product_id",
+        dimensions=[Dimension(name="product_name", type="categorical")],
+    )
+
+    bigquery_layer.add_model(users_multi)
+    bigquery_layer.add_model(products_multi)
+    bigquery_layer.add_model(orders_multi)
+
+    # Query across 3 models
+    result = bigquery_layer.query(
+        metrics=["orders_multi.total"], dimensions=["users_multi.name", "products_multi.product_name"]
+    )
+    rows = result.fetchall()
+    results_dict = {(row[0], row[1]): row[2] for row in rows}
+
+    assert results_dict[("Alice", "Widget")] == 100
+    assert results_dict[("Alice", "Gadget")] == 150
+    assert results_dict[("Bob", "Widget")] == 200