SQL Dialects and Built-in Functions Questions

Approach (Postgres): use LAG() to get previous subscription's end_date per customer ordered by start_date, compute days between, and flag gaps > 1 day and overlaps (start_date <= prev_end_date). Treat NULL end_date (active) carefully — consider them as very recent (no gap after) or exclude depending on business rule.

SQL (Postgres):

WITH ordered AS (
  SELECT
    customer_id,
    subscription_id,
    start_date::date AS start_date,
    end_date::date AS end_date,
    LAG(end_date) OVER (PARTITION BY customer_id ORDER BY start_date, end_date) AS prev_end_date
  FROM subscriptions
)
SELECT
  customer_id,
  subscription_id,
  prev_end_date,
  start_date,
  end_date,
  -- days difference between start and previous end
  (start_date - prev_end_date) AS days_between,
  CASE
    WHEN prev_end_date IS NULL THEN NULL
    WHEN start_date <= prev_end_date THEN 'overlap'
    WHEN (start_date - prev_end_date) > 1 THEN 'gap'
    ELSE 'contiguous'
  END AS status
FROM ordered
WHERE prev_end_date IS NOT NULL
  AND (
    start_date <= prev_end_date         -- overlap
    OR (start_date - prev_end_date) > 1 -- gap greater than 1 day
  )
ORDER BY customer_id, start_date;

Explanation / reasoning:

• LAG gives the immediate prior end_date per customer.
• start_date <= prev_end_date indicates overlapping periods (includes same-day overlap).
• (start_date - prev_end_date) returns integer days difference; >1 filters gaps longer than one day (so a start exactly prev_end_date+1 is contiguous).
• Excluding prev_end_date IS NULL avoids flagging a customer's first record.

Testing with sample data and edge cases:
Insert rows to cover:

• Normal contiguous: prev_end = 2025-01-10, start = 2025-01-11 → contiguous (no row returned).
• One-day gap: prev_end = 2025-01-10, start = 2025-01-12 → gap (days_between = 2, flagged).
• Overlap: prev_end = 2025-01-10, start = 2025-01-09 → overlap.
• Same-day overlap: prev_end = 2025-01-10, start = 2025-01-10 → overlap.
• Null end_date (active): prev_end = 2025-03-01, next start = 2025-03-05 — decide policy: if NULL present, adjust CTE to COALESCE(prev_end_date, 'infinite') or treat NULL as current and skip.
• Duplicate rows and out-of-order timestamps: test ordering by start_date then end_date to ensure deterministic prev selection.
• Boundary: end_date after start_date (bad data) — include validations to flag invalid ranges (end_date < start_date).

Example validation query for bad rows:

SELECT * FROM subscriptions WHERE end_date IS NOT NULL AND end_date < start_date;

Notes / alternatives:

• For SQL Server use DATEDIFF(day, prev_end_date, start_date).
• If business treats adjacency differently (e.g., same-day contiguous), adjust condition accordingly.
• For long-running active subscriptions with NULL end_date, you may want to treat NULL as CURRENT_DATE for comparison or exclude those sequences.

This query surfaces customers and subscription IDs with either gaps >1 day or overlaps; combine with reporting (counts, percent of customers affected) to prioritize data-quality fixes.

To get orders from the last 7 days inclusive of today (i.e., today and previous 6 days), use a date boundary that is clear about inclusivity. For a timestamptz column:

SELECT *
FROM orders
WHERE order_ts >= (CURRENT_DATE - INTERVAL '6 days')       -- start of 7-day window
  AND order_ts < (CURRENT_DATE + INTERVAL '1 day');       -- exclusive next-day boundary

Why use CURRENT_DATE vs now()::date:

• CURRENT_DATE returns the current date in the session time zone (midnight reference). now()::date is equivalent in most cases, but now() returns a timestamptz (current timestamp) then cast to date. CURRENT_DATE is slightly clearer/intentional when comparing dates.
• For comparisons against timestamptz, using date arithmetic with CURRENT_DATE and intervals yields deterministic midnight boundaries.

If order_ts is timestamp without time zone (timestamp):

• Interpret it in a known timezone before comparing, e.g. assume UTC:

WHERE order_ts AT TIME ZONE 'UTC' >= (CURRENT_DATE - INTERVAL '6 days')
  AND order_ts AT TIME ZONE 'UTC' < (CURRENT_DATE + INTERVAL '1 day')

Or, if you want the window in a specific TZ (e.g., 'America/Los_Angeles'):

WHERE (order_ts AT TIME ZONE 'America/Los_Angeles') >= (CURRENT_DATE - INTERVAL '6 days')
  AND (order_ts AT TIME ZONE 'America/Los_Angeles') < (CURRENT_DATE + INTERVAL '1 day')

Notes:

• Use half-open interval [start, next_day) to avoid issues with fractional seconds and inclusivity.
• If you prefer inclusive end: use <= (CURRENT_DATE + INTERVAL '1 day' - INTERVAL '1 microsecond') but half-open is safer.

Requirements:

• Provide daily aggregates for the last 365 days (one row per date + metrics).
• Refresh once per day (low latency) but remain correct when late-arriving or updated source rows arrive.
• Keep refresh time low and predictable.

Design overview:

• Use a summary table daily_summary(date, metric1, metric2..., updated_at).
• Source data is partitioned by event_date in the raw table (or ingestion date if event_date missing).
• Maintain a lightweight change-tracking mechanism: CDC table or daily staging (staging_changes(date, op, row_id, payload, source_ts, arrived_at)).

Incremental refresh strategy:

1. Daily recompute window:

   • Always recompute today (current_date) and yesterday (current_date - 1) because late arrivals often affect recent days.
   • Recompute partitions where CDC indicates changes (any distinct event_date present in staging_changes since last refresh).
   • Recompute any dates in a configurable “late-arrival buffer” (e.g., last 7 days) proactively if business needs demand higher correctness for recent history.

2. Implementation steps (SQL-like pseudocode for a data warehouse supporting MERGE, e.g., BigQuery/Snowflake/Redshift):

• Identify affected_dates:
  sql

  affected_dates AS (
    SELECT DISTINCT event_date
    FROM staging_changes
    WHERE arrived_at >= last_refresh_time
       OR source_ts >= last_refresh_time
  )
  UNION (current_date, current_date - 1)
  UNION (dates BETWEEN current_date - 6 AND current_date)  -- optional buffer
  

• Recompute aggregates only for affected_dates into a temp table:
  sql

  CREATE OR REPLACE TABLE tmp_daily AS
  SELECT event_date,
         COUNT(*) AS cnt,
         SUM(amount) AS total_amount,
         MAX(latest_change_ts) AS updated_at
  FROM raw_events
  WHERE event_date IN (SELECT event_date FROM affected_dates)
    AND event_date >= current_date - 365
  GROUP BY event_date;
  

• Merge/upsert into daily_summary:
  sql

  MERGE INTO daily_summary t
  USING tmp_daily s
  ON t.date = s.event_date
  WHEN MATCHED THEN UPDATE SET t.cnt = s.cnt, t.total_amount = s.total_amount, t.updated_at = s.updated_at
  WHEN NOT MATCHED THEN INSERT (date, cnt, total_amount, updated_at) VALUES (...);
  

• Delete rows older than 365 days:
  sql

  DELETE FROM daily_summary WHERE date < current_date - 365;
  

Handling late-arriving and backdated updates:

• Track source_ts or event_ts in staging_changes and raw_events. If a late row arrives for date D:
  • It will appear in staging_changes; D becomes part of affected_dates and will be recomputed on next refresh.
  • For corrections (update/delete), ensure staging captures operation type (upsert/delete) so recompute uses the authoritative raw_events snapshot.
• For very late data that affects dates beyond the 365-day window: if business wants historic correction, allow manual backfill jobs; otherwise ignore or log.

Keeping refresh time low:

• Narrow recompute to affected_dates only; avoid full-table aggregates.
• Ensure raw_events is clustered/partitioned by event_date so queries for specific dates are fast.
• Use pre-aggregations in upstream ETL if possible (e.g., hourly pre-agg per date) and merge those.
• Use incremental materialized view features where available (BigQuery materialized views, Snowflake tasks+streams) to offload merging logic.
• Parallelize per-date aggregation: compute multiple dates concurrently if the warehouse supports it.
• Maintain small indexes or sort keys on date to speed MERGE.

Correctness and auditing:

• Make refresh idempotent (recomputing same date yields same result).
• Keep updated_at and source_watermark in daily_summary to know last source time considered.
• Maintain an audit table logging refresh runs, affected_dates, row counts, duration, and any errors.
• Add tests: reconcile sample totals against a full re-aggregate weekly.

Trade-offs:

• Larger proactive buffer improves apparent accuracy but increases compute. Prefer targeted recompute driven by CDC.
• Full daily recompute is simplest but costly at scale.

This approach balances performance and correctness by recomputing only changed partitions (days), using CDC/staging to detect affected dates, clustering by event_date for fast reads, and keeping strong auditing for trust.

To assign session IDs with correct timezone-aware times and a 30-minute inactivity threshold, parse event timestamps using each row's timezone, sort per user, compute gaps, and increment session when gap > 30 minutes (handles cross-midnight and DST because times are timezone-aware). Here's a vectorized pandas approach and notes on scaling.

import pandas as pd
import pytz

def assign_sessions(df, ts_col='event_ts', tz_col='user_tz', user_col='user_id', inactivity_min=30):
    # Parse naive strings to timezone-aware timestamps per-row
    # Assume event_ts is ISO-like without tz info, e.g. "2025-03-09 01:30:00"
    df = df.copy()
    df['ts'] = pd.to_datetime(df[ts_col])
    # localize each timestamp using the user's tz, then convert to UTC for safe diffing
    df['tzinfo'] = df[tz_col].apply(pytz.timezone)
    df['ts_local'] = df.apply(lambda r: r['tzinfo'].localize(r['ts'], is_dst=None), axis=1)
    df['ts_utc'] = df['ts_local'].dt.tz_convert('UTC')
    # sort and compute differences per user
    df = df.sort_values([user_col, 'ts_utc'])
    df['prev_ts'] = df.groupby(user_col)['ts_utc'].shift(1)
    df['delta_min'] = (df['ts_utc'] - df['prev_ts']).dt.total_seconds() / 60
    # start new session if first event for user or gap > threshold
    df['new_session'] = (df['delta_min'].isna()) | (df['delta_min'] > inactivity_min)
    df['session_id'] = df.groupby(user_col)['new_session'].cumsum().astype(int)
    # optional: produce global session UUID
    df['session_uuid'] = df[user_col].astype(str) + '_' + df['session_id'].astype(str)
    return df.drop(columns=['tzinfo','ts','ts_local','ts_utc','prev_ts','delta_min','new_session'])

Key points:

• Using per-row localization handles DST transitions and cross-midnight correctly because localize accounts for ambiguous/nonexistent times.
• Converting to UTC gives monotonic times for diffing across timezones.

Time/space:

• Complexity O(N log N) due to sort; O(N) otherwise. Memory: O(N).

Edge cases:

• Ambiguous DST times: localize(..., is_dst=None) raises — handle with try/except to choose policy (prefer earlier/later).
• Missing/invalid tz strings: validate and fallback (UTC).
• Out-of-order ingestion: sort fixes.

Scalability:

• For tens of millions of rows:
  • Use chunked processing by user-key partitioning (hash by user_id) so each partition fits memory; preserve ordering within partitions.
  • Use Dask DataFrame with same code pattern (map_partitions) ensuring partitioning by user_id and sorting inside partitions.
  • Use Spark for distributed processing: parse timestamps with to_utc_timestamp and window/lag per user.
  • For streaming: maintain last event timestamp per user in a key-value store (Redis) and emit session starts when gap > threshold.
• Persist session state and use incremental processing to avoid reprocessing historical data.

Approach: identify rows where transaction_ts > now(), group by source system to prioritize high-impact sources, and capture metrics (count, earliest/latest future ts, percent of source). Store offenders in a quarantine/audit table for manual review or automated remediation.

SQL (Postgres-compatible example):

-- Find future-dated transactions per source system
WITH totals AS (
  SELECT source_system, COUNT(*) AS total_count
  FROM transactions
  GROUP BY source_system
),
future AS (
  SELECT
    source_system,
    COUNT(*) AS future_count,
    MIN(transaction_ts) AS earliest_future_ts,
    MAX(transaction_ts) AS latest_future_ts,
    MAX(transaction_ts) - now() AS max_skew
  FROM transactions
  WHERE transaction_ts > now()
  GROUP BY source_system
)
SELECT
  f.source_system,
  f.future_count,
  t.total_count,
  ROUND(100.0 * f.future_count / t.total_count, 4) AS pct_future,
  f.earliest_future_ts,
  f.latest_future_ts,
  f.max_skew
FROM future f
JOIN totals t USING (source_system)
ORDER BY pct_future DESC, future_count DESC;

Automated detection rules:

• Rule 1 (hard): Any single transaction_ts > now() by > 24 hours -> immediate alert.
• Rule 2 (rate): If pct_future for a source_system > 0.1% in a 24-hour window -> investigate.
• Rule 3 (absolute volume): If future_count > 100 in last hour -> escalate.
• Rule 4 (rolling baseline): Compare current rate to 7-day rolling mean; trigger if > 5σ above baseline.

Suggested remediation workflow:

1. Quarantine: copy offending rows to transactions_quarantine with metadata (detected_at, rule_id, original_ts, detected_by).
   sql

   INSERT INTO transactions_quarantine (select *, now() as detected_at, 'future_ts_rule' as rule_id
     from transactions WHERE transaction_ts > now() AND <your_rule_filter>);
   

2. Stop propagation: prevent quarantined rows from downstream ETL/analytics until resolved (flag in ETL).
3. Classify cause automatically where possible:
   • If skew is consistent per-source (e.g., median skew ≈ +3600s) assume clock skew; mark for auto-adjust.
   • If timestamps parse to improbable year (e.g., 1970/2038) mark as parsing error.
4. Auto-adjust with audit trail (only for low-risk cases): create corrected_ts and original_ts columns and apply a shift derived from median skew per-source:
   sql

   UPDATE transactions_quarantine
   SET corrected_ts = transaction_ts - interval '1 second' * ROUND(median_skew_seconds)
   WHERE rule_id = 'future_ts_rule' AND <auto_adjust_allowed>;
   

   Record operator and rationale.
5. Notify source owners with payload summary (sample rows, skew stats, suggested fix) and ticket link. Require confirmation for bulk corrections.
6. Re-ingest / release: after QA, move corrected rows back to production table with provenance fields.
7. Monitor: create dashboards tracking future-rate over time and source ownership.

Causes and reasoning:

• Clock skew: source machine/system clock ahead of canonical time (common in VMs, IoT). If skew is consistent, a fixed shift can correct timestamps.
• Timezone misinterpretation: storing local times without timezone metadata or double-applying offsets can create apparent future times (e.g., interpreting UTC as local).
• Parsing/ETL bugs: wrong format strings, milliseconds vs seconds, epoch misinterpretation (ms vs s) can produce large future offsets.
• Upstream batching delays: very old events stamped in future due to incorrect event timestamps.

Precautions:

• Never blind auto-fix everything—log original values, require approvals for high-volume changes.
• Prefer preventing issues: enforce schema with timestamptz, validate timestamp parsing during ingestion, use NTP on sources, add per-source validation and sampling.
• Track SLA for source-owner response and automated rollbacks if corrections introduce regressions.