Dev testing

.gitignore

@@ -176,3 +176,6 @@ metastore_db/
 
 # Kubernetes/Helm
 secrets.yaml
+
+# GigaSpatial data (Google/MS buildings, GHSL, WorldPop)
+dagster/bronze/

azure/templates/create-config.yaml

@@ -86,6 +86,8 @@ jobs:
                     CATALOG_BASE: "$(catalogBase)"
                     DATABASE_ID: "$(databaseId)"
                     SLACK_WORKFLOW_WEBHOOK: "$(slackWorkflowWebhook)"
+                    GIGAMETER_API_BASE_URL: "$(gigameterApiBaseUrl)"
+                    GIGAMETER_API_TOKEN: "$(gigameterApiToken)"
 
             - task: Kubernetes@1
               displayName: Create authproxy secrets

azure/templates/variables.yaml

@@ -52,4 +52,6 @@ variables:
   catalogBase: $(CATALOG_BASE)
   databaseId: $(DATABASE_ID)
   slackWorkflowWebhook: $(SLACK_WORKFLOW_WEBHOOK)
+  gigameterApiBaseUrl: $(GIGAMETER_API_BASE_URL)
+  gigameterApiToken: $(GIGAMETER_API_TOKEN)
   system.debug: true

dagster/models/file_upload.py

@@ -15,6 +15,7 @@ class DQStatusEnum(Enum):
     ERROR = "ERROR"
     TIMEOUT = "TIMEOUT"
     SKIPPED = "SKIPPED"
+    FILE_CHECKED = "FILE_CHECKED"
 
 
 class FileUpload(BaseModel):

dagster/models/mappings.py

@@ -5,6 +5,7 @@
 VALID_PRIMITIVES = [
     "string",
     "integer",
+    "int",
     "long",
     "float",
     "double",
@@ -22,6 +23,7 @@ class TypeMapping(BaseModel):
 class TypeMappings(BaseModel):
     string: TypeMapping
     integer: TypeMapping
+    int: TypeMapping  # Alias for integer
     long: TypeMapping
     float: TypeMapping
     double: TypeMapping

dagster/pyproject.toml

@@ -9,7 +9,7 @@ readme = "../docs/README.md"
 python = ">=3.11,<3.13"
 scipy = ">=1.10.0"
 cryptography = ">=42.0.4"
-sentry-sdk = "0.10.2"
+sentry-sdk = ">=1.33.1"
 openpyxl = "^3.1.2"
 loguru = "^0.7.2"
 icecream = "^2.1.3"
@@ -34,18 +34,19 @@ azure-storage-file-datalake = "^12.13.1"
 pytest-cov = "4.0.0"
 azure-identity = ">=1.16.1"
 geopy = "^2.4.0"
-h3 = "^3.7.6"
+h3 = "^4.2"
 fastparquet = "^2023.8.0"
 grpcio-status = "^1.59.0"
-acryl-datahub = { extras = ["azure-ad"], version = "0.14.0.2" }
+acryl-datahub = { extras = ["azure-ad"], version = "^1.3.1" }
 shapely = "^2.0.2"
-geopandas = "^0.14.1"
+geopandas = "^1.0"
 thefuzz = "^0.20.0"
 country-converter = "^1.1.1"
 sqlalchemy = "^2.0.28"
 psycopg2-binary = "^2.9.9"
 cuid2 = "^2.0.0"
-pydantic = { extras = ["email"], version = "<2.0.0" }
+pydantic = { extras = ["email"], version = "^2.10" }
+pydantic-settings = "^2.0"
 msgraph-sdk = "^1.2.0"
 unidecode = "^1.3.8"
 aiohttp = ">=3.10.2"
@@ -60,13 +61,16 @@ urllib3 = ">=2.2.2"
 tornado = ">=6.4.1"
 xlrd = "^2.0.1"
 chardet = "^5.2.0"
+giga-spatial = "==0.9.2"
+numpy = "^2.4.4"
+pandas = "^2.1"
+networkx = "^3.2"
+earthengine-api = ">=0.1.400"
 
 [tool.poetry.group.spark.dependencies]
 pyspark = "3.5.0"
 delta-spark = "3.0.0"
-deltalake = "0.12.0"
-pyarrow = "11.0.0"
-pyarrow-hotfix = "^0.5"
+pyarrow = "^16.0"
 roapi = "0.9.0"
 
 [tool.poetry.group.notebook.dependencies]
@@ -124,7 +128,7 @@ combine-as-imports = true
 force-wrap-aliases = true
 
 [tool.ruff.lint.pep8-naming]
-classmethod-decorators = ["pydantic.validator"]
+classmethod-decorators = ["pydantic.field_validator"]
 
 [tool.bandit]
 tests = [

dagster/src/assets/adhoc/master_csv_to_gold.py

@@ -42,7 +42,6 @@
     check_table_exists,
     create_delta_table,
     create_schema,
-    sync_schema,
 )
 from src.utils.logger import ContextLoggerWithLoguruFallback
 from src.utils.metadata import get_output_metadata, get_table_preview
@@ -608,38 +607,6 @@ def adhoc__publish_master_to_gold(
     )
     gold = compute_row_hash(gold)
 
-    table_exists = check_table_exists(
-        spark=spark.spark_session,
-        schema_name="school_master",
-        table_name=config.country_code.lower(),
-        data_tier=DataTier.GOLD,
-    )
-
-    if table_exists:
-        table_name = f"{config.metastore_schema}.{config.country_code}"
-        updated_schema = StructType(
-            get_schema_columns(
-                spark=spark.spark_session, schema_name=config.metastore_schema
-            )
-        )
-
-        context.log.info(f"Existing table name: {table_name}")
-
-        spark.spark_session.catalog.refreshTable(table_name)
-        existing_df = DeltaTable.forName(
-            sparkSession=spark.spark_session, tableOrViewName=table_name
-        ).toDF()
-
-        existing_schema = existing_df.schema
-
-        sync_schema(
-            table_name=table_name,
-            existing_schema=existing_schema,
-            updated_schema=updated_schema,
-            spark=spark.spark_session,
-            context=context,
-        )
-
     schema_reference = get_schema_columns_datahub(
         spark.spark_session,
         config.metastore_schema,
@@ -690,38 +657,6 @@ def adhoc__publish_reference_to_gold(
     )
     gold = compute_row_hash(gold)
 
-    table_exists = check_table_exists(
-        spark=spark.spark_session,
-        schema_name="school_reference",
-        table_name=config.country_code.lower(),
-        data_tier=DataTier.GOLD,
-    )
-
-    if table_exists:
-        table_name = f"{config.metastore_schema}.{config.country_code}"
-        updated_schema = StructType(
-            get_schema_columns(
-                spark=spark.spark_session, schema_name=config.metastore_schema
-            )
-        )
-
-        context.log.info(f"Existing table name: {table_name}")
-
-        spark.spark_session.catalog.refreshTable(table_name)
-        existing_df = DeltaTable.forName(
-            sparkSession=spark.spark_session,
-            tableOrViewName=table_name,
-        ).toDF()
-        existing_schema = existing_df.schema
-
-        sync_schema(
-            table_name=table_name,
-            existing_schema=existing_schema,
-            updated_schema=updated_schema,
-            spark=spark.spark_session,
-            context=context,
-        )
-
     schema_reference = get_schema_columns_datahub(
         spark.spark_session,
         config.metastore_schema,

dagster/src/assets/admin/assets.py

@@ -1,6 +1,8 @@
+from typing import Annotated
+
 from dagster_pyspark import PySparkResource
 from delta import DeltaTable
-from pydantic import conint
+from pydantic import Field
 from pyspark.sql import SparkSession
 from src.constants.constants_class import constants
 from src.settings import settings
@@ -47,7 +49,7 @@ def admin__terminate_all_runs(context: OpExecutionContext):
 class RollbackTableConfig(Config):
     schema_name: str
     table_name: str
-    version: conint(ge=0)
+    version: Annotated[int, Field(ge=0)]
 
 
 @asset

dagster/src/assets/common/assets.py

@@ -630,7 +630,7 @@ def reset_staging_table(
             raise
 
 
-def _handle_null_columns(schema_columns, primary_key):
+def handle_null_columns(schema_columns, primary_key):
     """Handle null columns by providing default values based on data type.
 
     If the column value is NULL, add a placeholder value if the following
@@ -790,7 +790,7 @@ def master(
     else:
         new_master = silver
 
-    column_actions = _handle_null_columns(schema_columns, primary_key)
+    column_actions = handle_null_columns(schema_columns, primary_key)
     new_master = new_master.withColumns(column_actions)
     new_master = compute_row_hash(new_master)
 
@@ -862,7 +862,7 @@ def reference(
     else:
         new_reference = silver
 
-    column_actions = _handle_null_columns(schema_columns, primary_key)
+    column_actions = handle_null_columns(schema_columns, primary_key)
     new_reference = new_reference.withColumns(column_actions)
     new_reference = compute_row_hash(new_reference)
 

dagster/src/assets/debug/assets.py

@@ -226,10 +226,10 @@ async def debug__send_test_email(
         res = await client.post(
             "/api/email/send-email",
             headers={"Authorization": f"Bearer {settings.EMAIL_RENDERER_BEARER_TOKEN}"},
-            json=config.dict(),
+            json=config.model_dump(),
         )
         if res.is_error:
             context.log.error(res.json())
             res.raise_for_status()
 
-    return Output(None, metadata=config.dict())
+    return Output(None, metadata=config.model_dump())

dagster/src/assets/migrations/core.py

@@ -4,7 +4,11 @@
 from models import VALID_PRIMITIVES, Schema
 from pyspark import sql
 from pyspark.sql.functions import col, when
-from src.utils.delta import execute_query_with_error_handler
+from src.utils.delta import (
+    create_delta_table,
+    persist_column_id_map,
+    sync_schema,
+)
 
 from dagster import OpExecutionContext
 
@@ -42,17 +46,32 @@ def save_schema_delta_table(context: OpExecutionContext, df: sql.DataFrame):
     full_table_name = f"{schema_name}.{table_name}"
 
     columns = Schema.fields
-    query = (
-        DeltaTable.createOrReplace(spark).tableName(full_table_name).addColumns(columns)
+    create_delta_table(
+        spark,
+        schema_name,
+        table_name,
+        columns,
+        context,
+        if_not_exists=True,
+    )
+    sync_schema(
+        table_name=full_table_name,
+        existing_schema=spark.table(full_table_name).schema,
+        updated_schema=Schema.schema,
+        spark=spark,
+        context=context,
     )
-    execute_query_with_error_handler(spark, query, schema_name, table_name, context)
 
     spark.catalog.refreshTable(full_table_name)
     (
         DeltaTable.forName(spark, full_table_name)
         .alias("master")
-        .merge(df.alias("updates"), "master.name = updates.name")
+        .merge(df.alias("updates"), "master.id = updates.id")
         .whenMatchedUpdateAll()
         .whenNotMatchedInsertAll()
+        .whenNotMatchedBySourceDelete()
         .execute()
     )
+
+    # Persist column-ID mapping after merge succeeds
+    persist_column_id_map(spark, full_table_name, table_name)

dagster/src/assets/school_geolocation/assets.py

@@ -15,6 +15,7 @@
 from pyspark.sql.types import StringType, StructField, StructType
 from sqlalchemy import select
 from src.constants import DataTier
+from src.data_quality_checks.dq_context import DQContext, DQMode
 from src.data_quality_checks.utils import (
     aggregate_report_json,
     aggregate_report_spark_df,
@@ -128,10 +129,19 @@ def geolocation_metadata(
     context.log.info("Create spark dataframe")
     metadata_df = s.createDataFrame(metadata_df)
 
-    table_columns = get_schema_columns(s, "school_geolocation_metadata")
     table_name = "school_geolocation_metadata"
     table_schema_name = "pipeline_tables"
 
+    context.log.info("Get schema columns for metadata table")
+    try:
+        table_columns = get_schema_columns(s, "school_geolocation_metadata")
+    except Exception:
+        context.log.warning(
+            "Schema table schemas.school_geolocation_metadata not found; "
+            "using DataFrame schema for metadata table creation"
+        )
+        table_columns = list(metadata_df.schema.fields)
+
     context.log.info("Create the schema and table if they do not exist")
     metadata_df = add_missing_columns(metadata_df, table_columns)
     metadata_df = metadata_df.select(*StructType(table_columns).fieldNames())
@@ -228,7 +238,7 @@ def geolocation_bronze(
     df = df.withColumn("school_id_govt", f.col("school_id_govt").cast(StringType()))
 
     df = create_bronze_layer_columns_updated(
-        df, mode, uploaded_columns, country_code, s
+        df, mode, uploaded_columns, country_code, file_upload.source, s
     )
 
     t2 = time.time()
@@ -301,9 +311,13 @@ def geolocation_data_quality_results(
     dq_results = row_level_checks(
         df=renamed_bronze,
         silver=casted_silver,
-        dataset_type=dataset_type,
-        _country_code_iso3=country_code,
-        mode=config.metadata["mode"],
+        dq_context=DQContext(
+            dq_mode=DQMode(config.metadata.get("dq_mode", "master")),
+            dataset_type=dataset_type,
+            country_code_iso3=country_code,
+            upload_id=id,
+            upload_mode=config.metadata.get("mode"),
+        ),
         context=context,
     )
 
@@ -327,7 +341,9 @@ def geolocation_data_quality_results(
     )
 
     dq_results_schema_name = f"{schema_name}_dq_results"
-    table_name = f"{id}_{country_code}_{current_timestamp}"
+    # Replace hyphens with underscores so the identifier is valid in Spark SQL
+    safe_id = id.replace("-", "_")
+    table_name = f"{safe_id}_{country_code}_{current_timestamp}"
 
     schema_columns = [
         StructField(field.name, field.dataType, nullable=True)
@@ -696,6 +712,10 @@ def geolocation_staging(
     spark: PySparkResource,
     config: FileConfig,
 ) -> Output[None]:
+    if config.metadata.get("dq_mode") == "uploaded":
+        context.log.info("Skipping staging as dq_mode is 'uploaded'")
+        return Output(None)
+
     if geolocation_dq_passed_rows.isEmpty():
         context.log.warning("Skipping staging as there are no rows passing DQ checks")
         return Output(None)