Implement California Consumer Privacy Act technical requirements. Covers data discovery, consumer request automation, data deletion pipelines, opt-out mechanisms, privacy signals, and the patterns that make CCPA compliance a technical capability.
CCPA Data Engineering ensures that data flows through systems comply with the California Consumer Privacy Act, particularly around data discovery, classification, access control, and consumer rights fulfillment. It matters when a company’s data pipeline must answer: *Which data belongs to which consumer? How is it processed? What is the source, retention policy, and purpose? And can we deliver a consumer’s data in a standardized, auditable format within 45 days of request?* The engineering effort lies not in policy documentation or legal review, but in building and maintaining a data infrastructure that automatically answers these questions across the entire data lifecycle.
To meet CCPA’s requirement that businesses maintain a detailed inventory of personal information, data engineers must implement a system that continuously discovers and catalogs data assets across sources.
A centralized data inventory is maintained in a relational database. Each data asset is represented as a row in a data_assets table:
CREATE TABLE data_assets (
id UUID PRIMARY KEY,
name TEXT NOT NULL,
source_system TEXT NOT NULL,
source_path TEXT NOT NULL,
schema_json JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
tags JSONB DEFAULT '[]'::JSONB
);A daily cron job runs a discover-data.sh script that scans all active data sources. The script uses psql to extract metadata from a sources table:
#!/bin/bash
set -euo pipefail
# Fetch all active sources
sources=$(psql -t -A -F',' -c "
SELECT id, type, connection_string, path
FROM sources
WHERE enabled = true
AND last_discovered_at < NOW() - INTERVAL '7 days'
")
while IFS=',' read -r source_id source_type connection_string path; do
case "$source_type" in
"s3")
# Use AWS CLI to list objects and extract metadata
aws s3api list-objects-v2 \
--bucket "$connection_string" \
--prefix "$path" \
--query 'Contents[].{Key: Key, Size: Size, LastModified: LastModified}' \
--output json \
--max-items 1000 \
> /tmp/s3-objects-$source_id.json
;;
"postgres")
# Use psql to extract table structure and sample data
psql "$connection_string" -c "
SELECT
table_name,
(SELECT json_agg(json_build_object(
'column', column_name,
'type', data_type,
'nullable', is_nullable
))
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = t.table_name
) AS schema,
(SELECT json_agg(row_to_json(t)) FROM (SELECT * FROM public."$path" LIMIT 100) t) AS sample_data
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = '$path'
" -t -A -F',' -o /tmp/pg-tables-$source_id.csv
;;
*)
echo "Unknown source type: $source_type"
exit 1
;;
esac
done <<< "$sources"
# Ingest new and changed assets
psql -c "
INSERT INTO data_assets (id, name, source_system, source_path, schema_json, tags)
SELECT
gen_random_uuid(),
s.table_name,
s.source_system,
s.path,
json_build_object(
'columns', s.schema,
'sample_data', s.sample_data,
'discovered_at', NOW()
),
json_build_array('discovered', 'ccpa', 'auto')
FROM (
SELECT
source_id AS source_system,
path,
schema,
sample_data
FROM
json_populate_recordset(NULL::record, pg_read_file('/tmp/pg-tables-*.csv'))
) s
ON CONFLICT (source_system, source_path)
DO UPDATE SET
schema_json = EXCLUDED.schema_json,
updated_at = NOW(),
tags = array_cat(data_assets.tags, EXCLUDED.tags)
WHERE data_assets.updated_at < EXCLUDED.updated_at;
"When a psql command fails, it returns a non-zero exit code and writes error details to stderr. A common failure is a missing schema column in sources table; the script fails silently if source_type is not properly indexed, and json_populate_recordset fails with:
ERROR: column "schema" does not exist
CONTEXT: SQL function "json_populate_recordset" during executionCCPA requires that personal information be classified into categories such as "Identifiers," "Sensitive Personal Information," and "Commercial Information." Data engineers use a set of rules to label data assets based on schema, content, and metadata.
Classification rules are stored in a classification_rules table:
CREATE TABLE classification_rules (
id UUID PRIMARY KEY,
name TEXT NOT NULL,
condition JSONB NOT NULL,
category TEXT NOT NULL,
priority INTEGER NOT NULL DEFAULT 1
);Each rule is a JSON object defining criteria. For example:
{
"name": "Email Field",
"condition": {
"schema": {
"column": "email",
"type": "text",
"pattern": "^.+@.+\\..+$"
},
"tags": ["contact", "customer"]
},
"category": "Identifiers",
"priority": 1
}A classify-data.sh script runs weekly and applies rules to all data assets:
#!/bin/bash
# classify-data.sh
set -euo pipefail
# Extract all data assets and classify them
psql -t -A -F',' -c "
SELECT id, name, schema_json, tags
FROM data_assets
WHERE classification_status = 'pending'
ORDER BY created_at DESC
" > /tmp/data-assets.csv
while IFS=',' read -r asset_id name schema_json tags; do
# Apply classification rules
for rule in $(psql -t -A -c "
SELECT id, category, priority
FROM classification_rules
WHERE
(condition->'schema') ? 'column'
AND (condition->'schema'->'column') = '$name'
ORDER BY priority
"); do
# Parse rule output: id, category, priority
rule_id=$(echo "$rule" | cut -d',' -f1)
category=$(echo "$rule" | cut -d',' -f2)
priority=$(echo "$rule" | cut -d',' -f3)
# Check if rule condition matches
if echo "$schema_json" | jq -e "
.columns | any(.column == \"$name\" and .type == \"text\" and .pattern) and
.sample_data | any(.email != null)
"; then
# Update asset with classification
psql -c "
INSERT INTO data_asset_classifications (asset_id, rule_id, category, applied_at)
VALUES ('$asset_id', '$rule_id', '$category', NOW())
ON CONFLICT (asset_id, rule_id)
DO UPDATE SET applied_at = NOW()
"
fi
done
done <<< "$(cat /tmp/data-assets.csv)"The jq command fails silently if schema_json is malformed. A common error is a missing sample_data field: jq: error: Cannot read field sample_data — which occurs when json_populate_recordset omits sample_data due to a mismatched column name or a pg_read_file path that doesn’t exist.
When a consumer submits a request (e.g., via a web form or API), the system must track and fulfill it within 45 days. The pipeline starts with a request table:
CREATE TABLE request (
id UUID PRIMARY KEY,
consumer_id TEXT NOT NULL,
type TEXT NOT NULL, -- 'access', 'deletion', 'opt-out'
status TEXT NOT NULL DEFAULT 'received',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
completed_at TIMESTAMPTZ,
metadata JSONB
);A fulfill-request.sh script processes all requests daily and executes the corresponding action:
#!/bin/bash
# fulfill-request.sh
set -euo pipefail
# Get all requests with status 'received'
requests=$(psql -t -A -F',' -c "
SELECT id, consumer_id, type, metadata
FROM request
WHERE status = 'received'
ORDER BY created_at
")
while IFS=',' read -r request_id consumer_id request_type metadata; do
case "$request_type" in
"access")
# Generate a PDF or JSON report of all data for the consumer
psql -c "
SELECT json_build_object(
'consumer', json_build_object(
'id', '$consumer_id',
'name', (SELECT name FROM customers WHERE id = '$consumer_id')
),
'data', (
SELECT json_agg(json_build_object(
'source', da.name,
'data', da.schema_json,
'matched', (
SELECT json_agg(c)
FROM (
SELECT * FROM data_asset_classifications
WHERE asset_id IN (
SELECT id FROM data_assets
WHERE schema_json ? 'email'
AND schema_json->'sample_data' ? 'email'
AND schema_json->'sample_data'->'email' = '$consumer_id'
)
) c
)
)
FROM data_assets da
WHERE da.schema_json->'columns' ? 'email'
AND da.schema_json->'columns'->'email'->>'type' = 'text'
)
) AS report
INTO TEMP TABLE report_table
FROM data_assets da
WHERE da.schema_json->'columns'->'email' ? 'pattern'
AND da.schema_json->'columns'->'email'->'pattern' = '^.+@.+\\..+$'
AND da.schema_json->'sample_data' ? 'email'
AND da.schema_json->'sample_data'->'email' = '$consumer_id'
"
# Export report to file
psql -c "
COPY (SELECT report FROM report_table) TO '/exports/access-report-$request_id.json' WITH (FORMAT json, FORCE_QUOTE all)
"
# Update request status
psql -c "
UPDATE request
SET status = 'processed',
metadata = jsonb_set(metadata, '{report_path}', '"'/exports/access-report-$request_id.json'"', true),
completed_at = NOW()
WHERE id = '$request_id'
"
;;
"deletion")
# Mark all data assets for deletion and enqueue a job
psql -c "
INSERT INTO deletion_jobs (request_id, asset_ids)
SELECT
'$request_id',
array_agg(da.id)
FROM data_assets da
WHERE
da.schema_json->'sample_data' ? 'email'
AND da.schema_json->'sample_data'->'email' = '$consumer_id'
GROUP BY da.id
"
psql -c "
UPDATE request
SET status = 'queued-for-deletion'
WHERE id = '$request_id'
"
;;
"opt-out")
# Update opt-out status in all relevant tables
psql -c "
UPDATE customers
SET opted_out = true,
opt_out_timestamp = NOW()
WHERE id = '$consumer_id'
"
psql -c "
INSERT INTO customer_opt_out_logs (consumer_id, request_id, timestamp)
VALUES ('$consumer_id', '$request_id', NOW())
"
psql -c "
UPDATE request
SET status = 'completed',
completed_at = NOW()
WHERE id = '$request_id'
"
;;
esac
done <<< "$requests"The COPY ... TO ... WITH (FORMAT json, FORCE_QUOTE all) command is essential: without FORCE_QUOTE all, nested objects in jsonb values are not properly quoted, leading to malformed JSON. A failure occurs when the jsonb column metadata contains a null value but the script expects jsonb, resulting in:
ERROR: column "metadata" is of type jsonb but expression is of type jsonCCPA mandates that personal information be retained for specified durations. Data engineers must track retention policies per data asset and enforce purging.
Retention policies are defined in a retention_policies table:
CREATE TABLE retention_policies (
id UUID PRIMARY KEY,
data_asset_id UUID REFERENCES data_assets(id),
category TEXT NOT NULL,
retention_period INTERVAL NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);A purge-older-data.sh script runs nightly and removes data older than the retention period:
#!/bin/bash
# purge-older-data.sh
set -euo pipefail
# Get all data assets with retention policies
psql -t -A -F',' -c "
SELECT da.id, da.source_system, da.source_path, rp.retention_period
FROM data_assets da
JOIN retention_policies rp ON da.id = rp.data_asset_id
WHERE da.last_purged_at < NOW() - rp.retention_period
" > /tmp/purge-targets.csv
while IFS=',' read -r asset_id source_system source_path retention_period; do
case "$source_system" in
"s3")
# Delete objects older than retention period
aws s3api list-objects-v2 \
--bucket "$source_system" \
--prefix "$source_path" \
--query 'Contents[?LastModified < `'$(( $(date -d "$retention_period" +%s) ))'`].Key' \
--output json \
> /tmp/objects-to-delete-$asset_id.json
# Delete objects
aws s3api delete-objects \
--bucket "$source_system" \
--delete '{
"Objects": [
$(jq -c -r '.[] | {Key: .}' /tmp/objects-to-delete-$asset_id.json | paste -sd,)
]
}'
;;
"postgres")
# Truncate or delete from table based on age
psql "$source_system" -c "
DELETE FROM $source_path
WHERE created_at < NOW() - '$retention_period'::interval
"
;;
esac
# Update last_purged_at
psql -c "
UPDATE data_assets
SET last_purged_at = NOW()
WHERE id = '$asset_id'
"
done <<< "$(cat /tmp/purge-targets.csv)"A silent failure occurs when date -d "$retention_period" expects a format like 1 year 2 months, but retention_period is stored as P1Y2M (ISO 8601 duration). The script runs but deletes data for a different period than expected, because date -d fails to parse P1Y2M. The fix is to convert P1Y2M to 1 year 2 months using sed or a CASE statement.
Every data operation is tracked in an audit_log table. The data_assets table is updated by triggers that write to this log:
CREATE TABLE audit_log (
id UUID PRIMARY KEY,
action TEXT NOT NULL, -- 'discovered', 'classified', 'purged', 'fulfilled'
entity_type TEXT NOT NULL,
entity_id UUID NOT NULL,
details JSONB NOT NULL,
performed_by TEXT,
performed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);A CREATE TRIGGER ensures that all updates to data_assets are logged:
CREATE OR REPLACE FUNCTION log_data_asset_change()
RETURNS TRIGGER AS $$
BEGIN
INSERT INTO audit_log (action, entity_type, entity_id, details, performed_by)
VALUES (
CASE
WHEN TG_OP = 'INSERT' THEN 'created'
WHEN TG_OP = 'UPDATE' THEN 'updated'
WHEN TG_OP = 'DELETE' THEN 'deleted'
END,
'data_asset',
NEW.id,
jsonb_build_object(
'old_values', OLD,
'new_values', NEW,
'changed_fields', (
SELECT array_agg(key)
FROM jsonb_each(NEW.schema_json)
EXCEPT
SELECT array_agg(key)
FROM jsonb_each(OLD.schema_json)
)
),
'system'
);
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER data_asset_audit_trigger
AFTER INSERT OR UPDATE OR DELETE
ON data_assets
FOR EACH ROW
EXECUTE FUNCTION log_data_asset_change();The jsonb_build_object function is critical: if schema_json is not properly jsonb, the details column contains a json value, leading to a mismatch in the audit log. The EXCEPT clause in the changed_fields query fails silently when jsonb_each returns null for a missing key, causing an incorrect set of changed fields.
CCPA Data Engineering is not about compliance documentation; it is about making the data itself the compliance record. The key insight is that every data asset must be discoverable, classifiable, trackable, and accountable. The pipeline from discovery to audit is driven by a few core principles: use psql for data processing, jq for JSON manipulation, aws s3api for cloud storage, and COPY and INSERT for data ingestion. Errors are caught early and are often silent: missing columns, malformed JSON, unindexed tables, and misconfigured date commands. The real edge comes from understanding that the pipeline fails not when a command returns an error, but when the data it produces does not match the expectations of the next step — a data asset with schema_json but no sample_data, or a retention policy stored as P1Y2M but expected as 1 year 2 months. The system works when data engineers stop thinking about "CCPA" and start thinking about "data that speaks for itself."
This page was rewritten on 18 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy