{ "edges": [ { "to": "7835d9d6-b483-496e-a8f1-e4861d29fed2", "from": "dbe44b7a-936f-44cd-a376-888c71f5e242", "toPort": "0", "fromPort": "0" }, { "to": "8f05dfe2-f1b7-4cf4-8645-887cb7c9bb12", "from": "fecc8343-1930-4730-9fc0-8760aaf6097b", "toPort": "0", "fromPort": "0" }, { "to": "03b96358-29c8-497a-a1da-8bd9d4723c76", "from": "bd9b5a43-277f-4b70-940f-ea546ab079cd", "toPort": "0", "fromPort": "0" }, { "to": "dbe44b7a-936f-44cd-a376-888c71f5e242", "from": "42867364-5006-44b2-ab56-3f7c0897fcb8", "toPort": "0", "fromPort": "0" }, { "to": "42867364-5006-44b2-ab56-3f7c0897fcb8", "from": "b3eebfd6-88bb-435f-80d7-411fb423f546", "toPort": "0", "fromPort": "0" }, { "to": "7ac59653-ed75-41ac-b19d-02e2e2beeee3", "from": "40d7ac18-bef8-47f5-964a-0bbe59107125", "toPort": "0", "fromPort": "0" }, { "to": "d2790027-d863-4126-9ce8-c2998809ae3a", "from": "40d7ac18-bef8-47f5-964a-0bbe59107125", "toPort": "0", "fromPort": "0" }, { "to": "dbe44b7a-936f-44cd-a376-888c71f5e242", "from": "7ac59653-ed75-41ac-b19d-02e2e2beeee3", "toPort": "0", "fromPort": "0" }, { "to": "b3eebfd6-88bb-435f-80d7-411fb423f546", "from": "bad30ec6-11fa-4d9c-b66f-f685546fb82d", "toPort": "0", "fromPort": "0" }, { "to": "40d7ac18-bef8-47f5-964a-0bbe59107125", "from": "bad30ec6-11fa-4d9c-b66f-f685546fb82d", "toPort": "0", "fromPort": "0" }, { "to": "5a61e58d-fd77-4bc0-aef0-eabc3fc410a9", "from": "bad30ec6-11fa-4d9c-b66f-f685546fb82d", "toPort": "0", "fromPort": "0" }, { "to": "bad30ec6-11fa-4d9c-b66f-f685546fb82d", "from": "b2fcf63e-b2a0-4d0a-8df1-6e06e2b3dc1e", "toPort": "0", "fromPort": "0" }, { "to": "bd9b5a43-277f-4b70-940f-ea546ab079cd", "from": "8f05dfe2-f1b7-4cf4-8645-887cb7c9bb12", "toPort": "0", "fromPort": "0" }, { "to": "fecc8343-1930-4730-9fc0-8760aaf6097b", "from": "ba0f352a-1a7c-4c62-9c2b-bc3a9c42d730", "toPort": "0", "fromPort": "0" }, { "to": "ba0f352a-1a7c-4c62-9c2b-bc3a9c42d730", "from": "7835d9d6-b483-496e-a8f1-e4861d29fed2", "toPort": "0", "fromPort": "0" }, { "to": "dbe44b7a-936f-44cd-a376-888c71f5e242", "from": "d2790027-d863-4126-9ce8-c2998809ae3a", "toPort": "0", "fromPort": "0" } ], "nodes": [ { "id": "b2fcf63e-b2a0-4d0a-8df1-6e06e2b3dc1e", "name": "Schedule DB Quality Scan", "type": "trigger", "params": { "kind": "n8n-nodes-base.scheduleTrigger", "rule": { "interval": [ { "field": "hours", "hoursInterval": 6 } ] } }, "position": [ -672, 224 ] }, { "id": "bad30ec6-11fa-4d9c-b66f-f685546fb82d", "name": "Workflow Configuration", "type": "set", "params": { "options": {}, "assignments": { "assignments": [ { "id": "id-1", "name": "confidenceThreshold", "type": "number", "value": 0.85 }, { "id": "id-2", "name": "maxNullPercentage", "type": "number", "value": 0.15 }, { "id": "id-3", "name": "outlierStdDevThreshold", "type": "number", "value": 3 }, { "id": "id-4", "name": "auditTableName", "type": "string", "value": "data_quality_audit" }, { "id": "id-5", "name": "baselineTableName", "type": "string", "value": "data_quality_baselines" } ] }, "includeOtherFields": true }, "position": [ -464, 224 ] }, { "id": "b3eebfd6-88bb-435f-80d7-411fb423f546", "name": "Get Schema Metadata", "type": "integration", "params": { "query": "SELECT table_name, column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema = <__PLACEHOLDER_VALUE__target schema name__> ORDER BY table_name, ordinal_position", "options": {}, "service": "postgres", "operation": "executeQuery" }, "position": [ -128, -48 ] }, { "id": "40d7ac18-bef8-47f5-964a-0bbe59107125", "name": "Get Table Statistics", "type": "integration", "params": { "query": "SELECT schemaname, tablename, n_live_tup as row_count, n_dead_tup as dead_rows FROM pg_stat_user_tables WHERE schemaname = <__PLACEHOLDER_VALUE__target schema name__>", "options": {}, "service": "postgres", "operation": "executeQuery" }, "position": [ -128, 224 ] }, { "id": "5a61e58d-fd77-4bc0-aef0-eabc3fc410a9", "name": "Get Historical Baselines", "type": "integration", "params": { "query": "{{ 'SELECT * FROM ' + $('Workflow Configuration').first().json.baselineTableName + ' ORDER BY recorded_at DESC LIMIT 1' }}", "options": {}, "service": "postgres", "operation": "executeQuery" }, "position": [ -128, 416 ] }, { "id": "42867364-5006-44b2-ab56-3f7c0897fcb8", "name": "Detect Schema Drift", "type": "code", "params": { "language": "pythonNative", "pythonCode": "from datetime import datetime\nimport json\n\n# Get current schema metadata from Get Schema Metadata node\ncurrent_schema = _input.all()[0].get('json', {})\n\n# Get historical baselines from Get Historical Baselines node\nhistorical_data = _input.all()[1].get('json', {})\n\nissues = []\n\n# Parse current schema and historical baselines\nif isinstance(current_schema, list):\n current_tables = {item.get('table_name'): item for item in current_schema}\nelse:\n current_tables = {current_schema.get('table_name'): current_schema}\n\nif isinstance(historical_data, list):\n historical_tables = {item.get('table_name'): item for item in historical_data}\nelse:\n historical_tables = {historical_data.get('table_name'): historical_data}\n\ntimestamp = datetime.utcnow().isoformat()\n\n# Check for schema drift\nfor table_name, current_table in current_tables.items():\n if table_name not in historical_tables:\n # New table detected\n issues.append({\n 'type': 'schema_drift',\n 'severity': 'medium',\n 'table_name': table_name,\n 'column_name': None,\n 'change_description': f'New table {table_name} detected',\n 'detected_at': timestamp\n })\n continue\n \n historical_table = historical_tables[table_name]\n current_columns = current_table.get('columns', {})\n historical_columns = historical_table.get('columns', {})\n \n # Check for added columns\n for col_name, col_info in current_columns.items():\n if col_name not in historical_columns:\n issues.append({\n 'type': 'schema_drift',\n 'severity': 'low',\n 'table_name': table_name,\n 'column_name': col_name,\n 'change_description': f'New column {col_name} added to {table_name}',\n 'detected_at': timestamp\n })\n \n # Check for removed columns\n for col_name in historical_columns:\n if col_name not in current_columns:\n issues.append({\n 'type': 'schema_drift',\n 'severity': 'high',\n 'table_name': table_name,\n 'column_name': col_name,\n 'change_description': f'Column {col_name} removed from {table_name}',\n 'detected_at': timestamp\n })\n \n # Check for data type changes\n for col_name, col_info in current_columns.items():\n if col_name in historical_columns:\n current_type = col_info.get('data_type')\n historical_type = historical_columns[col_name].get('data_type')\n \n if current_type != historical_type:\n issues.append({\n 'type': 'schema_drift',\n 'severity': 'high',\n 'table_name': table_name,\n 'column_name': col_name,\n 'change_description': f'Data type changed from {historical_type} to {current_type} for {col_name} in {table_name}',\n 'detected_at': timestamp\n })\n \n # Check for nullability changes\n current_nullable = col_info.get('is_nullable', True)\n historical_nullable = historical_columns[col_name].get('is_nullable', True)\n \n if current_nullable != historical_nullable:\n nullable_change = 'nullable' if current_nullable else 'not nullable'\n issues.append({\n 'type': 'schema_drift',\n 'severity': 'medium',\n 'table_name': table_name,\n 'column_name': col_name,\n 'change_description': f'Nullability changed to {nullable_change} for {col_name} in {table_name}',\n 'detected_at': timestamp\n })\n\n# Check for removed tables\nfor table_name in historical_tables:\n if table_name not in current_tables:\n issues.append({\n 'type': 'schema_drift',\n 'severity': 'critical',\n 'table_name': table_name,\n 'column_name': None,\n 'change_description': f'Table {table_name} has been removed',\n 'detected_at': timestamp\n })\n\nreturn issues" }, "position": [ 208, -48 ] }, { "id": "7ac59653-ed75-41ac-b19d-02e2e2beeee3", "name": "Detect Null Explosions", "type": "code", "params": { "language": "pythonNative", "pythonCode": "from datetime import datetime\nimport json\n\n# Get configuration from Workflow Configuration node\nconfig = _input.first().json\ntables_to_monitor = config.get('tablesToMonitor', [])\nmax_null_percentage = config.get('maxNullPercentage', 5.0)\n\n# Get table statistics from Get Table Statistics node\ntable_stats = _input.all()\n\nissues = []\n\nfor stat in table_stats:\n table_name = stat.json.get('table_name')\n columns_data = stat.json.get('columns', [])\n \n # Check if this table should be monitored\n if table_name not in tables_to_monitor:\n continue\n \n # Analyze each column for null explosions\n for column in columns_data:\n column_name = column.get('column_name')\n null_count = column.get('null_count', 0)\n total_count = column.get('total_count', 0)\n \n if total_count == 0:\n continue\n \n null_percentage = (null_count / total_count) * 100\n \n # Check if null percentage exceeds threshold\n if null_percentage > max_null_percentage:\n severity = 'critical' if null_percentage > max_null_percentage * 2 else 'high'\n \n issues.append({\n 'type': 'null_explosion',\n 'severity': severity,\n 'table_name': table_name,\n 'column_name': column_name,\n 'null_percentage': round(null_percentage, 2),\n 'threshold': max_null_percentage,\n 'detected_at': datetime.utcnow().isoformat()\n })\n\nreturn issues" }, "position": [ 208, 160 ] }, { "id": "d2790027-d863-4126-9ce8-c2998809ae3a", "name": "Detect Outlier Distributions", "type": "code", "params": { "language": "pythonNative", "pythonCode": "from datetime import datetime\nimport math\n\n# Get configuration from Workflow Configuration node\nconfig = _input.first().get('json', {})\noutlier_threshold = config.get('outlierStdDevThreshold', 3)\n\n# Get table statistics from Get Table Statistics node\ntable_stats = []\nfor item in _input.all():\n if 'table_name' in item.get('json', {}):\n table_stats.append(item['json'])\n\n# Get historical baselines from Get Historical Baselines node\nbaselines = {}\nfor item in _input.all():\n baseline = item.get('json', {})\n if 'table_name' in baseline and 'column_name' in baseline:\n key = f\"{baseline['table_name']}.{baseline['column_name']}\"\n baselines[key] = baseline\n\nissues = []\n\n# Analyze each table's statistics\nfor stat in table_stats:\n table_name = stat.get('table_name')\n column_name = stat.get('column_name')\n column_type = stat.get('column_type', '')\n \n # Only analyze numeric columns\n if not any(t in column_type.lower() for t in ['int', 'float', 'numeric', 'decimal', 'double']):\n continue\n \n current_mean = stat.get('mean')\n current_stddev = stat.get('stddev')\n current_count = stat.get('row_count', 0)\n \n if current_mean is None or current_stddev is None:\n continue\n \n # Get historical baseline\n baseline_key = f\"{table_name}.{column_name}\"\n baseline = baselines.get(baseline_key, {})\n baseline_mean = baseline.get('baseline_mean')\n baseline_stddev = baseline.get('baseline_stddev')\n \n if baseline_mean is None or baseline_stddev is None or baseline_stddev == 0:\n continue\n \n # Calculate z-score for the current mean compared to baseline\n z_score = abs((current_mean - baseline_mean) / baseline_stddev)\n \n # Detect outliers based on threshold\n if z_score > outlier_threshold:\n # Estimate outlier count (simplified approach)\n outlier_count = int(current_count * (z_score / 10)) # Rough estimate\n \n # Determine severity based on z-score\n if z_score > outlier_threshold * 2:\n severity = 'critical'\n elif z_score > outlier_threshold * 1.5:\n severity = 'high'\n else:\n severity = 'medium'\n \n issues.append({\n 'type': 'outlier_distribution',\n 'severity': severity,\n 'table_name': table_name,\n 'column_name': column_name,\n 'outlier_count': outlier_count,\n 'z_score': round(z_score, 2),\n 'current_mean': round(current_mean, 2),\n 'baseline_mean': round(baseline_mean, 2),\n 'current_stddev': round(current_stddev, 2),\n 'detected_at': datetime.utcnow().isoformat()\n })\n\nreturn issues" }, "position": [ 208, 384 ] }, { "id": "dbe44b7a-936f-44cd-a376-888c71f5e242", "name": "Combine All Issues", "type": "passthrough", "params": { "mode": "aggregate", "options": {}, "aggregate": "aggregateAllItemData", "destinationFieldName": "issues" }, "position": [ 464, 128 ] }, { "id": "7835d9d6-b483-496e-a8f1-e4861d29fed2", "name": "Calculate Confidence Scores", "type": "code", "params": { "language": "pythonNative", "pythonCode": "# Calculate confidence scores for each data quality issue\n# Score based on: severity, historical frequency, data volume affected, and consistency\n\nimport json\n\n# Get all items from previous node\nitems = _items('all')\n\n# Process each issue and calculate confidence score\nfor item in items:\n issue = item['json']\n \n # Initialize score components (0-1 scale)\n severity_score = 0\n frequency_score = 0\n volume_score = 0\n consistency_score = 0\n \n # 1. Severity Score (based on issue type and impact)\n severity = issue.get('severity', 'medium').lower()\n if severity == 'critical':\n severity_score = 1.0\n elif severity == 'high':\n severity_score = 0.8\n elif severity == 'medium':\n severity_score = 0.5\n else:\n severity_score = 0.3\n \n # 2. Historical Frequency Score (how often this issue occurs)\n historical_count = issue.get('historical_count', 0)\n if historical_count >= 10:\n frequency_score = 1.0\n elif historical_count >= 5:\n frequency_score = 0.7\n elif historical_count >= 2:\n frequency_score = 0.5\n else:\n frequency_score = 0.2\n \n # 3. Data Volume Score (percentage of data affected)\n affected_rows = issue.get('affected_rows', 0)\n total_rows = issue.get('total_rows', 1)\n affected_percentage = (affected_rows / total_rows) * 100 if total_rows > 0 else 0\n \n if affected_percentage >= 50:\n volume_score = 1.0\n elif affected_percentage >= 25:\n volume_score = 0.8\n elif affected_percentage >= 10:\n volume_score = 0.6\n elif affected_percentage >= 1:\n volume_score = 0.4\n else:\n volume_score = 0.2\n \n # 4. Consistency Score (detected across multiple checks)\n detection_count = issue.get('detection_count', 1)\n if detection_count >= 3:\n consistency_score = 1.0\n elif detection_count >= 2:\n consistency_score = 0.7\n else:\n consistency_score = 0.4\n \n # Calculate weighted confidence score\n # Weights: severity (40%), volume (30%), frequency (20%), consistency (10%)\n confidence_score = (\n severity_score * 0.4 +\n volume_score * 0.3 +\n frequency_score * 0.2 +\n consistency_score * 0.1\n )\n \n # Round to 3 decimal places\n confidence_score = round(confidence_score, 3)\n \n # Add confidence score to the issue\n issue['confidence_score'] = confidence_score\n issue['confidence_breakdown'] = {\n 'severity': severity_score,\n 'volume': volume_score,\n 'frequency': frequency_score,\n 'consistency': consistency_score\n }\n\n# Return all items with confidence scores\nreturn items" }, "position": [ 640, 128 ] }, { "id": "ba0f352a-1a7c-4c62-9c2b-bc3a9c42d730", "name": "Check Confidence Threshold", "type": "if", "params": { "options": {}, "conditions": { "options": { "version": 3, "leftValue": "", "caseSensitive": true, "typeValidation": "strict" }, "combinator": "and", "conditions": [ { "id": "id-1", "operator": { "type": "array", "operation": "notEmpty" }, "leftValue": "={{ $json.issues }}" }, { "id": "id-2", "operator": { "type": "boolean", "operation": "true" }, "leftValue": "={{ $json.issues.some(issue => issue.confidence_score >= $('Workflow Configuration').first().json.confidenceThreshold) }}" } ] } }, "position": [ 800, 128 ] }, { "id": "fecc8343-1930-4730-9fc0-8760aaf6097b", "name": "Generate SQL Fixes", "type": "code", "params": { "language": "pythonNative", "pythonCode": "# Generate SQL fix suggestions for each high-confidence issue\nimport json\n\n# Get all items from previous node\nitems = _items('all')\n\nresults = []\n\nfor item in items:\n issue = item['json']\n issue_type = issue.get('issue_type', '')\n table_name = issue.get('table_name', '')\n column_name = issue.get('column_name', '')\n \n sql_fix = ''\n rollback_sql = ''\n \n if issue_type == 'schema_drift':\n # Generate ALTER TABLE statements for schema changes\n drift_details = issue.get('details', {})\n change_type = drift_details.get('change_type', '')\n \n if change_type == 'column_added':\n # Suggest documenting or validating the new column\n sql_fix = f\"-- New column detected: {column_name}\\nCOMMENT ON COLUMN {table_name}.{column_name} IS 'Added on {drift_details.get(\\\"detected_at\\\", \\\"unknown\\\")}';\\n\"\n rollback_sql = f\"ALTER TABLE {table_name} DROP COLUMN IF EXISTS {column_name};\"\n elif change_type == 'column_removed':\n # Suggest adding back the column if needed\n sql_fix = f\"-- Column removed: {column_name}\\n-- Review if this was intentional\\nALTER TABLE {table_name} ADD COLUMN {column_name} {drift_details.get('data_type', 'TEXT')};\"\n rollback_sql = f\"ALTER TABLE {table_name} DROP COLUMN IF EXISTS {column_name};\"\n elif change_type == 'type_changed':\n old_type = drift_details.get('old_type', 'TEXT')\n new_type = drift_details.get('new_type', 'TEXT')\n sql_fix = f\"-- Type change detected: {old_type} -> {new_type}\\nALTER TABLE {table_name} ALTER COLUMN {column_name} TYPE {old_type} USING {column_name}::{old_type};\"\n rollback_sql = f\"ALTER TABLE {table_name} ALTER COLUMN {column_name} TYPE {new_type} USING {column_name}::{new_type};\"\n \n elif issue_type == 'null_explosion':\n # Generate UPDATE or constraint additions for null explosions\n null_percentage = issue.get('null_percentage', 0)\n \n if null_percentage > 50:\n # High null rate - suggest adding NOT NULL constraint or default value\n sql_fix = f\"-- High null rate detected ({null_percentage}%)\\n-- Option 1: Add default value\\nUPDATE {table_name} SET {column_name} = '' WHERE {column_name} IS NULL;\\nALTER TABLE {table_name} ALTER COLUMN {column_name} SET DEFAULT '';\\n\\n-- Option 2: Add NOT NULL constraint (after fixing nulls)\\n-- ALTER TABLE {table_name} ALTER COLUMN {column_name} SET NOT NULL;\"\n rollback_sql = f\"ALTER TABLE {table_name} ALTER COLUMN {column_name} DROP NOT NULL;\\nALTER TABLE {table_name} ALTER COLUMN {column_name} DROP DEFAULT;\"\n else:\n # Moderate null rate - suggest investigation\n sql_fix = f\"-- Investigate null values\\nSELECT * FROM {table_name} WHERE {column_name} IS NULL LIMIT 100;\\n\\n-- If appropriate, update nulls:\\n-- UPDATE {table_name} SET {column_name} = '' WHERE {column_name} IS NULL;\"\n rollback_sql = f\"-- Manual rollback required based on UPDATE operation\"\n \n elif issue_type == 'outlier_distribution':\n # Generate data cleanup queries for outliers\n outlier_details = issue.get('details', {})\n outlier_count = outlier_details.get('outlier_count', 0)\n \n sql_fix = f\"-- Outlier distribution detected\\n-- Review outlier values:\\nSELECT {column_name}, COUNT(*) as frequency\\nFROM {table_name}\\nGROUP BY {column_name}\\nORDER BY frequency DESC;\\n\\n-- Option 1: Cap extreme values\\n-- UPDATE {table_name} SET {column_name} = WHERE {column_name} > ;\\n-- UPDATE {table_name} SET {column_name} = WHERE {column_name} < ;\\n\\n-- Option 2: Remove outliers (use with caution)\\n-- DELETE FROM {table_name} WHERE {column_name} > OR {column_name} < ;\"\n rollback_sql = f\"-- Backup data before cleanup:\\n-- CREATE TABLE {table_name}_backup AS SELECT * FROM {table_name};\\n-- Restore: INSERT INTO {table_name} SELECT * FROM {table_name}_backup;\"\n \n # Add SQL fix fields to the issue\n issue['sql_fix'] = sql_fix\n issue['rollback_sql'] = rollback_sql\n \n results.append({'json': issue})\n\nreturn results" }, "position": [ 1088, 112 ] }, { "id": "8f05dfe2-f1b7-4cf4-8645-887cb7c9bb12", "name": "Store Issue in Audit Log", "type": "integration", "params": { "table": { "__rl": true, "mode": "name", "value": "={{ $('Workflow Configuration').first().json.auditTableName }}" }, "schema": { "__rl": true, "mode": "list", "value": "public" }, "columns": { "value": { "sql_fix": "={{ $json.sql_fix }}", "severity": "={{ $json.severity }}", "issue_type": "={{ $json.type }}", "table_name": "={{ $json.table_name }}", "column_name": "={{ $json.column_name }}", "description": "={{ $json.change_description }}", "detected_at": "={{ $json.detected_at }}", "confidence_score": "={{ $json.confidence_score }}" }, "schema": [ { "id": "issue_type", "required": false, "displayName": "issue_type", "defaultMatch": true, "canBeUsedToMatch": true }, { "id": "table_name", "required": false, "displayName": "table_name", "defaultMatch": true, "canBeUsedToMatch": true }, { "id": "column_name", "required": false, "displayName": "column_name", "defaultMatch": true, "canBeUsedToMatch": true }, { "id": "severity", "required": false, "displayName": "severity", "defaultMatch": true, "canBeUsedToMatch": true }, { "id": "confidence_score", "required": false, "displayName": "confidence_score", "defaultMatch": true, "canBeUsedToMatch": true }, { "id": "description", "required": false, "displayName": "description", "defaultMatch": true, "canBeUsedToMatch": true }, { "id": "sql_fix", "required": false, "displayName": "sql_fix", "defaultMatch": true, "canBeUsedToMatch": true }, { "id": "detected_at", "required": false, "displayName": "detected_at", "defaultMatch": true, "canBeUsedToMatch": true } ], "mappingMode": "defineBelow", "matchingColumns": [ "issue_type", "table_name", "column_name", "severity", "confidence_score", "description", "sql_fix", "detected_at" ] }, "options": {}, "service": "postgres", "operation": "read" }, "position": [ 1344, 112 ] }, { "id": "bd9b5a43-277f-4b70-940f-ea546ab079cd", "name": "Send Alert to Team", "type": "integration", "params": { "text": "=🚨 **Data Quality Alert**\n\n**Severity:** {{ $json.severity }}\n**Confidence Score:** {{ $json.confidenceScore }}%\n\n**Affected Resources:**\n• Table: {{ $json.table }}\n• Column: {{ $json.column }}\n\n**Issue Details:**\n{{ $json.issueDescription }}\n\n**Suggested SQL Fix:**\n```sql\n{{ $json.sqlFix }}\n```\n\n**Detection Time:** {{ $json.detectedAt }}\n**Issue ID:** {{ $json.issueId }}", "select": "channel", "service": "slack", "channelId": { "__rl": true, "mode": "id", "value": "<__PLACEHOLDER_VALUE__Slack channel ID or name__>" }, "operation": "read", "otherOptions": { "includeLinkToWorkflow": true } }, "position": [ 1552, 112 ] }, { "id": "03b96358-29c8-497a-a1da-8bd9d4723c76", "name": "Update Baselines", "type": "integration", "params": { "table": { "__rl": true, "mode": "name", "value": "={{ $('Workflow Configuration').first().json.baselineTableName }}" }, "schema": { "__rl": true, "mode": "list", "value": "public" }, "columns": { "value": { "avg_value": "={{ $json.avg_value }}", "data_type": "={{ $json.data_type }}", "max_value": "={{ $json.max_value }}", "min_value": "={{ $json.min_value }}", "null_count": "={{ $json.null_count }}", "table_name": "={{ $json.table_name }}", "total_rows": "={{ $json.total_rows }}", "column_name": "={{ $json.column_name }}", "schema_name": "={{ $json.schema_name }}", "distinct_count": "={{ $json.distinct_count }}", "baseline_timestamp": "={{ $now }}" }, "mappingMode": "defineBelow" }, "options": {}, "service": "postgres", "operation": "read" }, "position": [ 1728, 112 ] } ] }

Monitor PostgreSQL data quality and generate remediation alerts with Slack