chore(ducklake): proper feature flag gating for DuckLake data-model copy (#42541)

Author: bill-ph

posthog/ducklake/README.md

@@ -22,6 +22,13 @@ For local dev the defaults are:
 - `DUCKLAKE_S3_ACCESS_KEY=object_storage_root_user`
 - `DUCKLAKE_S3_SECRET_KEY=object_storage_root_password`
 
+## Feature flag gating
+
+The modeling workflow launches the DuckLake copy child only when the
+`ducklake-data-modeling-copy-workflow` feature flag is enabled for the team (as evaluated
+via `feature_enabled`). Create or update that flag locally to target the team you are testing
+with—otherwise the copy workflow will be skipped even if the rest of the configuration is correct.
+
 ## Target bucket layout
 
 Every model copy is written to a deterministic prefix inside the DuckLake data bucket. Each workflow
@@ -50,12 +57,7 @@ For AWS S3, grant the worker role at least `s3:ListBucket`, `s3:GetObject`, `s3:
 Follow this checklist to exercise the DuckLake copy workflow on a local checkout without needing extra tribal knowledge:
 
 1. **Start the dev stack**  
-   Run `hogli start` (or `bin/start`) so Postgres, MinIO, Temporal, and all DuckLake defaults are up.
-
-   ```bash
-   export DUCKLAKE_DATA_MODELING_COPY_WORKFLOW_ENABLED=true
-   hogli start
-   ```
+   Run `hogli start` (or `bin/start`) so Postgres, MinIO, Temporal, and all DuckLake defaults are up. Make sure the `ducklake-data-modeling-copy-workflow` feature flag is enabled for the team you plan to use.
 
 2. **Trigger a model materialization from the app**  
    In the PostHog UI, open Data Warehouse → Views, pick (or create) a view, open the Materialization section, enable it if needed, and click **Sync now**. This schedules the `data-modeling-run` workflow for that team/view.

posthog/settings/base_variables.py

@@ -35,14 +35,6 @@
         "️WARNING! Environment variable E2E_TESTING is enabled. This is a security vulnerability unless you are running tests."
     )
 
-ducklake_data_modeling_env = get_from_env(
-    "DUCKLAKE_DATA_MODELING_COPY_WORKFLOW_ENABLED", optional=True, type_cast=str_to_bool
-)
-if ducklake_data_modeling_env is None:
-    ducklake_data_modeling_env = get_from_env("DUCKLAKE_COPY_WORKFLOW_ENABLED", False, type_cast=str_to_bool)
-
-DUCKLAKE_DATA_MODELING_COPY_WORKFLOW_ENABLED: bool = ducklake_data_modeling_env
-
 IS_COLLECT_STATIC = len(sys.argv) > 1 and sys.argv[1] == "collectstatic"
 SERVER_GATEWAY_INTERFACE = get_from_env("SERVER_GATEWAY_INTERFACE", "WSGI", type_cast=str)
 

posthog/temporal/data_modeling/__init__.py

@@ -1,6 +1,7 @@
 from posthog.temporal.data_modeling.ducklake_copy_workflow import (
     DuckLakeCopyDataModelingWorkflow,
     copy_data_modeling_model_to_ducklake_activity,
+    ducklake_copy_workflow_gate_activity,
     prepare_data_modeling_ducklake_metadata_activity,
 )
 from posthog.temporal.data_modeling.run_workflow import (
@@ -26,5 +27,6 @@
     create_job_model_activity,
     cleanup_running_jobs_activity,
     prepare_data_modeling_ducklake_metadata_activity,
+    ducklake_copy_workflow_gate_activity,
     copy_data_modeling_model_to_ducklake_activity,
 ]

posthog/temporal/data_modeling/ducklake_copy_workflow.py

@@ -7,6 +7,7 @@
 from django.conf import settings
 
 import duckdb
+import posthoganalytics
 from structlog.contextvars import bind_contextvars
 from temporalio import activity, workflow
 from temporalio.common import RetryPolicy
@@ -18,6 +19,8 @@
     get_config,
     normalize_endpoint,
 )
+from posthog.exceptions_capture import capture_exception
+from posthog.models import Team
 from posthog.sync import database_sync_to_async
 from posthog.temporal.common.base import PostHogWorkflow
 from posthog.temporal.common.logger import get_logger
@@ -49,6 +52,39 @@ class DuckLakeCopyActivityInputs:
     model: DuckLakeCopyModelMetadata
 
 
+@dataclasses.dataclass
+class DuckLakeCopyWorkflowGateInputs:
+    team_id: int
+
+
+@activity.defn
+async def ducklake_copy_workflow_gate_activity(inputs: DuckLakeCopyWorkflowGateInputs) -> bool:
+    """Evaluate whether the DuckLake copy workflow should run for a team."""
+    bind_contextvars(team_id=inputs.team_id)
+    logger = LOGGER.bind()
+
+    try:
+        team = await database_sync_to_async(Team.objects.only("uuid", "organization_id").get)(id=inputs.team_id)
+    except Team.DoesNotExist:
+        await logger.aerror("Team does not exist when evaluating DuckLake copy workflow gate")
+        return False
+
+    try:
+        return posthoganalytics.feature_enabled(
+            "ducklake-data-modeling-copy-workflow",
+            str(team.uuid),
+            groups={"organization": str(team.organization_id)},
+            only_evaluate_locally=True,
+        )
+    except Exception as error:
+        await logger.awarning(
+            "Failed to evaluate DuckLake copy workflow feature flag",
+            error=str(error),
+        )
+        capture_exception(error)
+        return False
+
+
 @activity.defn
 async def prepare_data_modeling_ducklake_metadata_activity(
     inputs: DataModelingDuckLakeCopyInputs,
@@ -137,6 +173,20 @@ async def run(self, inputs: DataModelingDuckLakeCopyInputs) -> None:
             workflow.logger.info("No models to copy - exiting early", **inputs.properties_to_log)
             return
 
+        should_copy = await workflow.execute_activity(
+            ducklake_copy_workflow_gate_activity,
+            DuckLakeCopyWorkflowGateInputs(team_id=inputs.team_id),
+            start_to_close_timeout=dt.timedelta(seconds=30),
+            retry_policy=RetryPolicy(maximum_attempts=1),
+        )
+
+        if not should_copy:
+            workflow.logger.info(
+                "DuckLake copy workflow disabled by feature flag",
+                **inputs.properties_to_log,
+            )
+            return
+
         metadata = await workflow.execute_activity(
             prepare_data_modeling_ducklake_metadata_activity,
             inputs,

posthog/temporal/data_modeling/run_workflow.py

@@ -1565,11 +1565,7 @@ async def run(self, inputs: RunWorkflowInputs) -> Results:
             ),
         )
 
-        if (
-            settings.DUCKLAKE_DATA_MODELING_COPY_WORKFLOW_ENABLED
-            and self.ducklake_copy_inputs
-            and self.ducklake_copy_inputs.models
-        ):
+        if self.ducklake_copy_inputs and self.ducklake_copy_inputs.models:
             temporalio.workflow.logger.info(
                 "Triggering DuckLake copy child workflow",
                 job_id=job_id,

posthog/temporal/tests/data_modeling/test_ducklake_copy_workflow.py

@@ -1,9 +1,14 @@
 import uuid
+import datetime as dt
 
 import pytest
 
 from django.test import override_settings
 
+import temporalio.worker
+from temporalio import activity as temporal_activity
+from temporalio.testing import WorkflowEnvironment
+
 from posthog.temporal.data_modeling import ducklake_copy_workflow as ducklake_module
 from posthog.temporal.data_modeling.ducklake_copy_workflow import (
     DuckLakeCopyActivityInputs,
@@ -126,3 +131,139 @@ def fake_ensure_bucket(config):
     assert "read_parquet('s3://source/table/**/*.parquet')" in table_calls[0]
     assert any("ATTACH" in statement for statement in fake_conn.sql_statements), "Expected DuckLake catalog attach"
     assert fake_conn.closed is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.django_db
+async def test_ducklake_copy_workflow_skips_when_feature_flag_disabled(monkeypatch, ateam):
+    call_counts = {"metadata": 0, "copy": 0}
+
+    @temporal_activity.defn
+    async def metadata_stub(inputs: DataModelingDuckLakeCopyInputs):
+        call_counts["metadata"] += 1
+        return [
+            DuckLakeCopyModelMetadata(
+                model_label="model",
+                saved_query_id=str(uuid.uuid4()),
+                saved_query_name="model",
+                normalized_name="model",
+                source_glob_uri="s3://source/table/**/*.parquet",
+                schema_name="data_modeling_team_1",
+                table_name="model",
+            )
+        ]
+
+    @temporal_activity.defn
+    async def copy_stub(inputs: DuckLakeCopyActivityInputs):
+        call_counts["copy"] += 1
+
+    monkeypatch.setattr(
+        "posthog.temporal.data_modeling.ducklake_copy_workflow.posthoganalytics.feature_enabled",
+        lambda *args, **kwargs: False,
+    )
+    monkeypatch.setattr(ducklake_module, "prepare_data_modeling_ducklake_metadata_activity", metadata_stub)
+    monkeypatch.setattr(ducklake_module, "copy_data_modeling_model_to_ducklake_activity", copy_stub)
+
+    inputs = DataModelingDuckLakeCopyInputs(
+        team_id=ateam.pk,
+        job_id="job",
+        models=[
+            DuckLakeCopyModelInput(
+                model_label="model",
+                saved_query_id=str(uuid.uuid4()),
+                table_uri="s3://source/table",
+                file_uris=["s3://source/table/part-0.parquet"],
+            )
+        ],
+    )
+
+    async with await WorkflowEnvironment.start_time_skipping() as env:
+        async with temporalio.worker.Worker(
+            env.client,
+            task_queue="ducklake-test",
+            workflows=[ducklake_module.DuckLakeCopyDataModelingWorkflow],
+            activities=[
+                ducklake_module.ducklake_copy_workflow_gate_activity,
+                ducklake_module.prepare_data_modeling_ducklake_metadata_activity,
+                ducklake_module.copy_data_modeling_model_to_ducklake_activity,
+            ],
+            workflow_runner=temporalio.worker.UnsandboxedWorkflowRunner(),
+        ):
+            await env.client.execute_workflow(
+                ducklake_module.DuckLakeCopyDataModelingWorkflow.run,
+                inputs,
+                id=str(uuid.uuid4()),
+                task_queue="ducklake-test",
+                execution_timeout=dt.timedelta(seconds=30),
+            )
+
+    assert call_counts["metadata"] == 0
+    assert call_counts["copy"] == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.django_db
+async def test_ducklake_copy_workflow_runs_when_feature_flag_enabled(monkeypatch, ateam):
+    call_counts = {"metadata": 0, "copy": 0}
+
+    @temporal_activity.defn
+    async def metadata_stub(inputs: DataModelingDuckLakeCopyInputs):
+        call_counts["metadata"] += 1
+        return [
+            DuckLakeCopyModelMetadata(
+                model_label="model",
+                saved_query_id=str(uuid.uuid4()),
+                saved_query_name="model",
+                normalized_name="model",
+                source_glob_uri="s3://source/table/**/*.parquet",
+                schema_name="data_modeling_team_1",
+                table_name="model",
+            )
+        ]
+
+    @temporal_activity.defn
+    async def copy_stub(inputs: DuckLakeCopyActivityInputs):
+        call_counts["copy"] += 1
+
+    monkeypatch.setattr(
+        "posthog.temporal.data_modeling.ducklake_copy_workflow.posthoganalytics.feature_enabled",
+        lambda *args, **kwargs: True,
+    )
+    monkeypatch.setattr(ducklake_module, "prepare_data_modeling_ducklake_metadata_activity", metadata_stub)
+    monkeypatch.setattr(ducklake_module, "copy_data_modeling_model_to_ducklake_activity", copy_stub)
+
+    inputs = DataModelingDuckLakeCopyInputs(
+        team_id=ateam.pk,
+        job_id="job",
+        models=[
+            DuckLakeCopyModelInput(
+                model_label="model",
+                saved_query_id=str(uuid.uuid4()),
+                table_uri="s3://source/table",
+                file_uris=["s3://source/table/part-0.parquet"],
+            )
+        ],
+    )
+
+    async with await WorkflowEnvironment.start_time_skipping() as env:
+        async with temporalio.worker.Worker(
+            env.client,
+            task_queue="ducklake-test",
+            workflows=[ducklake_module.DuckLakeCopyDataModelingWorkflow],
+            activities=[
+                ducklake_module.ducklake_copy_workflow_gate_activity,
+                ducklake_module.prepare_data_modeling_ducklake_metadata_activity,
+                ducklake_module.copy_data_modeling_model_to_ducklake_activity,
+            ],
+            workflow_runner=temporalio.worker.UnsandboxedWorkflowRunner(),
+        ):
+            await env.client.execute_workflow(
+                ducklake_module.DuckLakeCopyDataModelingWorkflow.run,
+                inputs,
+                id=str(uuid.uuid4()),
+                task_queue="ducklake-test",
+                execution_timeout=dt.timedelta(seconds=30),
+            )
+
+    assert call_counts["metadata"] == 1
+    assert call_counts["copy"] == 1

posthog/temporal/tests/data_modeling/test_run_workflow.py

@@ -1031,7 +1031,7 @@ async def fail_jobs_stub(inputs):
     monkeypatch.setattr(run_workflow_module, "finish_run_activity", finish_run_stub)
     monkeypatch.setattr(run_workflow_module, "fail_jobs_activity", fail_jobs_stub)
 
-    with override_settings(DUCKLAKE_DATA_MODELING_COPY_WORKFLOW_ENABLED=True, DATA_MODELING_TASK_QUEUE="ducklake-test"):
+    with override_settings(DATA_MODELING_TASK_QUEUE="ducklake-test"):
         child_ducklake_workflow_runs.clear()
         async with await WorkflowEnvironment.start_time_skipping() as env:
             async with temporalio.worker.Worker(