fix(warehouses): Opralog: Use Entries.LogicallyDeleted column to remove duplicates

infra/local/README.md

@@ -26,6 +26,7 @@ To test ELT scripts based on the Python package `dlt`, create and add the follow
 ```toml
 [destination.pyiceberg]
 bucket_url = "s3://playground"
+
 [destination.pyiceberg.credentials]
 uri = "http://localhost:50080/iceberg/catalog"
 project_id = "c4fcd44f-7ce7-4446-9f7c-dcc7ba76dd22"
@@ -36,6 +37,26 @@ client_secret = "s3cr3t"
 scope = "lakekeeper"
 ```
 
+To test dbt transform projects create and add the following to `$HOME/.dbt/profiles`:
+
+```yaml
+local_catalog:
+  target: local_trino
+  outputs:
+    local_trino:
+      type: trino
+      method: ldap
+      user: machine-infra
+      password: s3cr3t
+      host: localhost
+      port: 58443
+      http_scheme: https
+      cert: false
+      database: playground
+      schema: analytics
+      threads: 8
+```
+
 The values are defined in [`env-local`](./env-local).
 
 ## Start services

infra/local/trino/etc/catalog/playground.properties

@@ -5,7 +5,7 @@ iceberg.rest-catalog.uri=http://${ENV:ROUTER_HOSTNAME_INTERNAL}:${ENV:ROUTER_POR
 iceberg.rest-catalog.vended-credentials-enabled=false
 iceberg.rest-catalog.security=OAUTH2
 iceberg.rest-catalog.oauth2.server-uri=${ENV:KEYCLOAK_REALM_INTERNAL}/protocol/openid-connect/token
-iceberg.rest-catalog.oauth2.credential=trino:s3cr3t
+iceberg.rest-catalog.oauth2.credential=machine-infra:s3cr3t
 iceberg.rest-catalog.oauth2.scope=lakekeeper
 
 # Our S3 implementation does not have STS enabled so we need to provide

warehouses/accelerator/transform/accelerator/README.md

@@ -2,11 +2,20 @@
 
 [dbt](https://docs.getdbt.com/) projects for defining models within the `accelerator` catalog.
 
-_Prerequisities_:
+## One-time setup
 
-- Python requirements defined in `./requirements/requirements.txt`
+Install the Python requirements into a virtual environment:
 
-Try running the following commands:
+```bash
+>uv pip install -r ./requirements/requirements.txt
+```
 
-- `dbt run`
-- `dbt test`
+For running against the local, docker-based setup see [infra/local/README](../../../../infra/local/README.md#one-time-setup)
+to configure a `local_catalog` dbt profile. Ensure the docker services are running following the
+instructions in that file.
+
+Run the models against the local catalog:
+
+```bash
+dbt --profiles-dir ~/.dbt --profile local_catalog run
+```

warehouses/accelerator/transform/accelerator/models/staging/opralogweb/base/base_opralogweb__entries.sql

@@ -11,9 +11,13 @@ renamed as (
   select
 
       entry_id,
-      entry_timestamp as fault_occurred_at,
+      with_timezone(entry_timestamp, 'UTC') as fault_occurred_at,
       cast({{ adapter.quote('entry_timestamp') }} as date) as fault_date,
-      trim(additional_comment) as fault_description
+      trim(additional_comment) as fault_description,
+      case
+          when logically_deleted = 'Y' then true
+          else false
+      end as logically_deleted
 
   from source
 

warehouses/accelerator/transform/accelerator/models/staging/opralogweb/stg_opralogweb__mcr_equipment_downtime.sql

@@ -36,14 +36,15 @@ denormalized as (
 
   where
 
-    staging_logbooks.logbook_name = {{ MCR_LOGBOOK }}
+    staging_entries.logically_deleted = false
+    and staging_entries.fault_date >= from_iso8601_date({{ OPRALOG_EPOCH }})
+    and staging_logbooks.logbook_name = {{ MCR_LOGBOOK }}
     and staging_chapter_entry.logbook_id = staging_chapter_entry.principal_logbook
     and staging_additional_columns.column_title in ('Equipment', 'Group', 'Lost Time', 'Group Leader comments')
     and (
       staging_more_entry_columns.string_data is not null
       or staging_more_entry_columns.number_data is not null
     )
-    and staging_entries.fault_date >= from_iso8601_date({{ OPRALOG_EPOCH }})
 ),
 
 mcr_equipment_downtime as (

warehouses/accelerator/transform/accelerator/profiles.yml

@@ -5,11 +5,11 @@ trino_catalog:
     trino:
       type: trino
       method: ldap
-      user: "{{ env_var('DBT_TRINO_USER', 'trino') }}"
-      password: "{{ env_var('DBT_TRINO_PASSWORD', '') }}"
-      http_scheme: "{{ env_var('DBT_TRINO_HTTP_SCHEME', 'http') }}"
-      host: "{{ env_var('DBT_TRINO_HOST', 'analytics.localhost') }}"
-      port: "{{ env_var('DBT_TRINO_PORT', '59088') | int }}"
-      database: "{{ env_var('DBT_TRINO_CATALOG', 'accelerator') }}"
-      schema: "{{ env_var('DBT_TRINO_CATALOG_SCHEMA_PREFIX', 'dev_') }}analytics"
+      user: "{{ env_var('DBT_TRINO_USER') }}"
+      password: "{{ env_var('DBT_TRINO_PASSWORD') }}"
+      http_scheme: "{{ env_var('DBT_TRINO_HTTP_SCHEME') }}"
+      host: "{{ env_var('DBT_TRINO_HOST') }}"
+      port: "{{ env_var('DBT_TRINO_PORT') | int }}"
+      database: "{{ env_var('DBT_TRINO_CATALOG') }}"
+      schema: "{{ env_var('DBT_TRINO_CATALOG_SCHEMA_PREFIX') }}analytics"
       threads: "{{ env_var('DBT_TRINO_THREADS', '8') | int }}"