ERP-Unlocked Data Backend Architecture Proposal
Lakehouse + Semantic Layer for AI-Powered Order Processing
Section titled “Lakehouse + Semantic Layer for AI-Powered Order Processing”Version: 1.0
Date: November 9, 2025
Status: Proposal for Evaluation
Author: Technical Architecture Review
Executive Summary
Section titled “Executive Summary”This document proposes migrating ERP-Unlocked’s current PostgreSQL-based ERP replica system to a modern data backend architecture consisting of:
- Apache Hudi Lakehouse - Analytical storage for ERP data (products, customers, inventory, pricing)
- Cube Semantic Layer - Unified API for AI agents and analytics with built-in RLS
- dlt + Spark ETL - Modern data ingestion replacing current sync services
- Dual Database Strategy - PostgreSQL for OLTP (orders, documents), Hudi for OLAP (ERP data)
The Core Question
Section titled “The Core Question”Should we replace IERPSyncService and erp_replica PostgreSQL tables with a lakehouse-based data backend to power AI-driven features?
TL;DR Recommendation
Section titled “TL;DR Recommendation”Yes, but phased approach starting with a pilot. The data backend architecture is a strong fit for ERP-Unlocked’s future, particularly for:
- Multi-ERP federation (query across Prophet21 + NetSuite simultaneously)
- AI-powered product matching and order validation
- Historical analytics (pricing trends, inventory patterns)
- Scalability (100+ organizations with millions of SKUs)
However, proceed incrementally:
- Phase 1 (Pilot): Implement for ONE entity type (products) alongside current system
- Phase 2: Migrate remaining entities if pilot succeeds
- Phase 3: Deprecate old replica system
Table of Contents
Section titled “Table of Contents”- Current Architecture Analysis
- Proposed Data Backend Architecture
- Detailed Component Design
- Migration Strategy
- Cost-Benefit Analysis
- Risk Assessment
- Implementation Roadmap
- Recommendation
Current Architecture Analysis
Section titled “Current Architecture Analysis”How ERP Sync Works Today
Section titled “How ERP Sync Works Today”┌──────────────────────────────────────────────────────────────┐│ CURRENT ARCHITECTURE │├──────────────────────────────────────────────────────────────┤│ ││ ┌────────────────────────────────────────────────────┐ ││ │ TRIGGER.DEV SCHEDULED TASKS │ ││ │ • scheduledERPSync (daily 2 AM UTC) │ ││ │ • scheduledCustomerPricingSync (daily 1 AM UTC) │ ││ └────────────────┬───────────────────────────────────┘ ││ │ ││ ▼ ││ ┌────────────────────────────────────────────────────┐ ││ │ IERPSyncService IMPLEMENTATIONS │ ││ │ ├─ Prophet21SyncService │ ││ │ ├─ DemoSyncService │ ││ │ └─ (Future: NetSuiteSyncService, etc.) │ ││ │ │ ││ │ Methods: │ ││ │ • syncProducts() - batch fetch & upsert │ ││ │ • syncCustomers() - batch fetch & upsert │ ││ │ • syncCrossReferences() - customer SKU mappings │ ││ │ • syncShippingAddresses() - delivery locations │ ││ └────────────────┬───────────────────────────────────┘ ││ │ Direct SQL upserts ││ ▼ ││ ┌────────────────────────────────────────────────────┐ ││ │ POSTGRESQL (OLTP + OLAP Mixed) │ ││ │ │ ││ │ Operational Tables (OLTP): │ ││ │ • orders │ ││ │ • pdf_documents │ ││ │ • part_number_mappings │ ││ │ • erp_connections │ ││ │ │ ││ │ Replica Tables (OLAP - THIS IS THE PROBLEM): │ ││ │ • erp_products (~2.3M records per org) │ ││ │ • erp_customers (~10K records per org) │ ││ │ • erp_cross_references (~500K per org) │ ││ │ • erp_shipping_addresses (~50K per org) │ ││ │ • erp_inventory_locations (~5M per org) │ ││ │ • erp_sync_logs (audit trail) │ ││ │ │ ││ │ Full-text search via tsvector │ ││ └─────────────────────────────────────────────────────┘ ││ │└──────────────────────────────────────────────────────────────┘Current System Characteristics
Section titled “Current System Characteristics”| Aspect | Current Implementation |
|---|---|
| Storage | PostgreSQL for both OLTP and OLAP |
| Sync Mechanism | Prophet21SyncService with batch upserts |
| Search | PostgreSQL tsvector (full-text search) |
| Orchestration | Trigger.dev scheduled tasks |
| Multi-tenancy | clerk_organization_id + connection_id |
| Concurrency | Queue with concurrencyLimit: 1 per connection |
| Retry Logic | 3 attempts, exponential backoff |
| Machine Resources | Medium-2x (2 vCPU, 4 GB RAM) |
Problems with Current Approach
Section titled “Problems with Current Approach”1. OLTP/OLAP Mixing Anti-Pattern
Section titled “1. OLTP/OLAP Mixing Anti-Pattern”PostgreSQL is designed for transactional workloads (orders, documents), not analytical queries (product search, inventory analysis).
// Current: Analytical query on transactional DBconst products = await db.query.erpProducts.findMany({ where: and( eq(erpProducts.connectionId, connectionId), sql`product_search_vector @@ plainto_tsquery('industrial pump valve')` ), limit: 1000, // Scanning millions of rows});Issues:
- Full-text search on 2.3M products per org = slow
- Index bloat (tsvector GIN indexes are large)
- Postgres optimized for row-level locking, not bulk scans
- Vacuum/autovacuum overhead on large replica tables
2. No Time-Travel or Historical Analytics
Section titled “2. No Time-Travel or Historical Analytics”// IMPOSSIBLE with current architecture:// "What was the price of product X on July 15, 2024?"// "Show inventory trend for product Y over last 6 months"ERP replica tables only maintain current snapshot. No historical data retention.
3. Single-ERP Limitation
Section titled “3. Single-ERP Limitation”// IMPOSSIBLE: Federated queries across multiple ERPs// "Find all products matching 'pump' across Prophet21 AND NetSuite"Current architecture requires ONE sync service per connection. No unified query layer.
4. Rigid Schema
Section titled “4. Rigid Schema”// Every new ERP field requires schema migrationexport const erpProducts = pgTable('erp_products', { // ... 15 columns // What if NetSuite has 30 different fields? // What if customer wants custom metadata?});Schema changes require migrations, downtime, application restarts.
5. Scalability Concerns
Section titled “5. Scalability Concerns”Current data volume (per organization):• Products: 2.3M rows × 5 KB avg = ~11.5 GB• Cross-references: 500K rows × 2 KB = ~1 GB• Inventory: 5M rows × 3 KB = ~15 GB• Total: ~30 GB per large org
With 100 organizations:• Total replica data: ~3 TB in PostgreSQL• Index overhead: ~1.5 TB additional• Total PostgreSQL size: ~4.5 TBPostgreSQL can handle this, but it’s expensive and not its optimal use case.
6. AI/LLM Integration Friction
Section titled “6. AI/LLM Integration Friction”// Current: AI must query raw replica tablesconst context = await db.query.erpProducts.findMany({ where: /* complex SQL */});
// No semantic abstraction layer// No pre-computed aggregations// No business logic in query layerLLMs need semantic APIs, not raw SQL tables.
Proposed Data Backend Architecture
Section titled “Proposed Data Backend Architecture”High-Level Design
Section titled “High-Level Design”┌─────────────────────────────────────────────────────────────────────┐│ PROPOSED ARCHITECTURE │├─────────────────────────────────────────────────────────────────────┤│ ││ ┌──────────────────────────────────────────────────────────┐ ││ │ WEBAPP + PDF PROCESSING (No Changes) │ ││ │ • Astro + React frontend │ ││ │ • FastAPI PDF processing │ ││ │ • Trigger.dev orchestration │ ││ └────────────────┬─────────────────────────────────────────┘ ││ │ ││ │ JWT (clerk_organization_id in claims) ││ │ ││ ┌────────────────▼─────────────────────────────────────────┐ ││ │ POSTGRES (OLTP - Transactional Data) │ ││ │ • orders │ ││ │ • pdf_documents │ ││ │ • part_number_mappings (stays here!) │ ││ │ • erp_connections │ ││ │ • organizations, users, subscriptions │ ││ │ • OPERATIONAL DATA ONLY │ ││ └──────────────────────────────────────────────────────────┘ ││ ││ ┌──────────────────────────────────────────────────────────┐ ││ │ DATA BACKEND (NEW - Analytical Data) │ ││ │ │ ││ │ ┌─────────────────────────────────────────────┐ │ ││ │ │ CUBE SEMANTIC LAYER (Unified API) │ │ ││ │ │ │ │ ││ │ │ Semantic Models: │ │ ││ │ │ ├─ ERPProducts │ │ ││ │ │ │ measures: count, avg_price │ │ ││ │ │ │ dimensions: product_id, name, uom │ │ ││ │ │ │ segments: active_products, has_image │ │ ││ │ │ │ │ │ ││ │ │ ├─ ERPCustomers │ │ ││ │ │ │ measures: count, total_revenue │ │ ││ │ │ │ dimensions: customer_id, name, pricing │ │ ││ │ │ │ │ │ ││ │ │ ├─ ERPInventory │ │ ││ │ │ │ measures: total_on_hand, available │ │ ││ │ │ │ dimensions: location_id, product_id │ │ ││ │ │ │ │ │ ││ │ │ └─ ERPPricingHistory (NEW!) │ │ ││ │ │ time-series pricing analytics │ │ ││ │ │ │ │ ││ │ │ RLS: WHERE org_id = user.clerk_org_id │ │ ││ │ │ APIs: GraphQL, REST │ │ ││ │ └──────────┬───────────────────┬───────────────┘ │ ││ │ │ │ │ ││ │ ┌────────▼────────┐ ┌──────▼────────┐ │ ││ │ │ TRINO │ │ Postgres │ │ ││ │ │ (Query Engine) │ │ (Read Replica│ │ ││ │ │ for Hudi │ │ for join) │ │ ││ │ └────────┬────────┘ └───────────────┘ │ ││ │ │ │ ││ │ ▼ │ ││ │ ┌─────────────────────────────────────┐ │ ││ │ │ APACHE HUDI LAKEHOUSE │ │ ││ │ │ │ │ ││ │ │ Tables: │ │ ││ │ │ • erp_products │ │ ││ │ │ • partition: org_id, erp_type │ │ ││ │ │ • format: Parquet (compressed) │ │ ││ │ │ • time-travel enabled │ │ ││ │ │ • full-text search: Hudi index │ │ ││ │ │ • │ │ ││ │ │ • erp_customers │ │ ││ │ │ • erp_cross_references │ │ ││ │ │ • erp_inventory_locations │ │ ││ │ │ • erp_pricing_snapshots (NEW!) │ │ ││ │ │ • erp_sync_events (audit) │ │ ││ │ │ │ │ ││ │ │ Storage: Cloudflare R2 / AWS S3 │ │ ││ │ │ Cost: ~$0.015/GB/mo (vs PG $0.20) │ │ ││ │ └─────────────────────────────────────┘ │ ││ │ │ ││ │ ┌─────────────────────────────────────────────┐ │ ││ │ │ ETL ORCHESTRATION │ │ ││ │ │ │ │ ││ │ │ ┌────────────────────────────────┐ │ │ ││ │ │ │ dlt PIPELINES (Extraction) │ │ │ ││ │ │ │ • prophet21_products.py │ │ │ ││ │ │ │ • prophet21_customers.py │ │ │ ││ │ │ │ • netsuite_products.py (future)│ │ │ ││ │ │ │ │ │ │ ││ │ │ │ Replaces: IERPSyncService │ │ │ ││ │ │ └────────────────────────────────┘ │ │ ││ │ │ │ │ ││ │ │ ┌────────────────────────────────┐ │ │ ││ │ │ │ SPARK (Loading to Hudi) │ │ │ ││ │ │ │ • Batch writes to Hudi tables │ │ │ ││ │ │ │ • Partition management │ │ │ ││ │ │ │ • Compaction & cleanup │ │ │ ││ │ │ └────────────────────────────────┘ │ │ ││ │ │ │ │ ││ │ │ ┌────────────────────────────────┐ │ │ ││ │ │ │ dbt (Business Logic) │ │ │ ││ │ │ │ • fct_inventory_snapshot.sql │ │ │ ││ │ │ │ • dim_product_enriched.sql │ │ │ ││ │ │ │ • fct_pricing_history.sql │ │ │ ││ │ │ └────────────────────────────────┘ │ │ ││ │ │ │ │ ││ │ │ Orchestration: Trigger.dev (existing!) │ │ ││ │ └─────────────────────────────────────────────┘ │ ││ └──────────────────────────────────────────────────────────┘ ││ │└─────────────────────────────────────────────────────────────────────┘What Changes vs What Stays
Section titled “What Changes vs What Stays”| Component | Current | Proposed |
|---|---|---|
| Orders, PDFs, Mappings | PostgreSQL | PostgreSQL (no change) |
| ERP Products | PostgreSQL replica | Hudi lakehouse |
| ERP Customers | PostgreSQL replica | Hudi lakehouse |
| ERP Inventory | PostgreSQL replica | Hudi lakehouse |
| Sync Logic | IERPSyncService (TypeScript) | dlt pipelines (Python) |
| Orchestration | Trigger.dev | Trigger.dev (no change) |
| Query API | Direct Drizzle queries | Cube semantic layer |
| Auth | Clerk | Clerk (no change) |
| Frontend | Astro + React | No change |
Detailed Component Design
Section titled “Detailed Component Design”1. Apache Hudi Lakehouse
Section titled “1. Apache Hudi Lakehouse”Why Hudi?
Section titled “Why Hudi?”Features we need: ✓ ACID transactions (upsert support for ERP syncs) ✓ Time-travel (historical pricing, inventory) ✓ Incremental processing (sync only changed records) ✓ Partition pruning (org_id partitions for multi-tenancy) ✓ Schema evolution (add NetSuite fields without downtime) ✓ Compaction (optimize storage over time)
Alternatives considered: • Delta Lake: AWS-centric, less active community • Iceberg: No upsert support (append-only) • PostgreSQL: Current solution (OLTP/OLAP mixing)Hudi Table Configuration
Section titled “Hudi Table Configuration”table: name: erp_productstype: COPY_ON_WRITE # Optimized for read-heavy workloads partition: fields: [clerk_organization_id, erp_type] # Physical layout: # s3://erp-data/erp_products/ # org_id=550e8400.../erp_type=prophet21/*.parquet # org_id=550e8400.../erp_type=netsuite/*.parquetprimary_key: fields: [connection_id, erp_product_id]
precombine: field: last_synced_at # Use for conflict resolution
indexes: - type: BLOOM fields: [erp_product_id, name]
- type: LUCENE # Full-text search replacement for tsvector fields: [name, description] compaction: strategy: NUM_COMMITS max_commits: 10 target_file_size: 128MBcleaner: retain_commits: 30 # 30-day time-travel windowSchema Design
Section titled “Schema Design”-- Hudi: erp_products (schema-on-read, flexible JSONB-like)CREATE TABLE erp_products ( -- Core fields (strongly typed) id STRING, connection_id STRING, clerk_organization_id STRING, -- Partition key erp_type STRING, -- Partition key (prophet21, netsuite, etc.) erp_product_id STRING, name STRING, description STRING, unit_of_measure STRING, delete_flag BOOLEAN, last_synced_at TIMESTAMP,
-- Flexible metadata (varies by ERP) metadata STRUCT< prophet21 STRUCT< company_id STRING, price_code STRING, weight DECIMAL, ... >, netsuite STRUCT< internal_id STRING, subsidiary_id STRING, ... > >,
-- Image metadata images STRUCT< erp_url STRING, r2_key STRING, has_image BOOLEAN >,
-- Pricing (time-series, append-only) current_pricing STRUCT< base_price DECIMAL, currency STRING, effective_date TIMESTAMP >,-- Full-text search (Lucene index) search_text STRING -- Concatenated name + description)USING HUDIPARTITIONED BY (clerk_organization_id, erp_type);2. Cube Semantic Layer
Section titled “2. Cube Semantic Layer”Why Cube?
Section titled “Why Cube?”Requirements: ✓ Multi-tenant RLS (filter by clerk_organization_id) ✓ GraphQL + REST APIs (for AI agents) ✓ Pre-aggregated metrics (dashboards) ✓ Caching layer (reduce Trino queries) ✓ SQL abstraction (business logic in models)
Cube Core (open-source) is perfect for this.Cube Model Example
Section titled “Cube Model Example”cubes: - name: ERPProductssql_table: erp_products # Trino table backed by Hudi
# Multi-tenant security data_source: default # Row-level security sql: > SELECT * FROM erp_products WHERE clerk_organization_id = '${SECURITY_CONTEXT.clerk_org_id}' AND delete_flag = false
# Joins (federated query with PostgreSQL) joins: - name: PartNumberMappings sql: > {ERPProducts.erp_product_id} = {PartNumberMappings.source_item_id} AND {ERPProducts.connection_id} = {PartNumberMappings.connection_id} relationship: one_to_many
# Measures (aggregations) measures: - name: count type: count- name: products_with_images type: count filters: - sql: "{CUBE}.has_product_image = true"
- name: avg_weight sql: "metadata.prophet21.weight" type: avg # Dimensions dimensions: - name: clerk_organization_id sql: clerk_organization_id type: stringshown: false # Hide from API - name: product_id sql: erp_product_id type: string primary_key: true- name: name sql: name type: string
- name: description sql: description type: string
- name: unit_of_measure sql: unit_of_measure type: string
- name: has_image sql: images.has_image type: boolean
- name: erp_type sql: erp_type type: string
- name: last_synced_at sql: last_synced_at type: time
# Segments (pre-defined filters) segments: - name: active_products sql: "{CUBE}.delete_flag = false"
- name: products_with_images sql: "{CUBE}.images.has_image = true" - name: prophet21_products sql: "{CUBE}.erp_type = 'prophet21'"Cube API Usage (from AI Agents)
Section titled “Cube API Usage (from AI Agents)”// AI Tool: Search products across all ERPsimport { CubeApi } from '@cubejs-client/core';
const cubeApi = new CubeApi({ apiUrl: process.env.CUBE_API_URL, headers: { // Clerk JWT with clerk_organization_id claim Authorization: `Bearer ${clerkToken}`, },});
// GraphQL query (semantic, not SQL)const result = await cubeApi.load({ measures: ['ERPProducts.count'], dimensions: [ 'ERPProducts.product_id', 'ERPProducts.name', 'ERPProducts.description', 'ERPProducts.erp_type', ], filters: [ { member: 'ERPProducts.name', operator: 'contains', values: ['industrial pump'], }, ], segments: ['ERPProducts.active_products'], limit: 100,});
// Returns:// [// { product_id: 'P-1234', name: 'Industrial Pump A', erp_type: 'prophet21' },// { product_id: 'N-5678', name: 'Industrial Pump B', erp_type: 'netsuite' }// ]
// ✨ Federated across multiple ERPs automatically!// ✨ RLS enforced (only user's org data)// ✨ Cached by Cube (fast repeated queries)3. dlt ETL Pipelines
Section titled “3. dlt ETL Pipelines”Replacing IERPSyncService
Section titled “Replacing IERPSyncService”import dltfrom dlt.sources import incrementalfrom typing import Iterator, Dict, Anyimport requests
@dlt.resource( name="erp_products", primary_key=["connection_id", "erp_product_id"], write_disposition="merge", # Upsert behavior columns={ "clerk_organization_id": {"data_type": "text", "partition": True}, "erp_type": {"data_type": "text", "partition": True}, "last_synced_at": {"data_type": "timestamp"}, })def fetch_prophet21_products( connection_id: str, clerk_organization_id: str, api_url: str, username: str, password: str, last_sync: str = None, # Incremental sync) -> Iterator[Dict[str, Any]]: """ Extract products from Prophet21 OData API.Replaces: Prophet21SyncService.syncProducts() """
# Incremental state (only fetch changed records) incremental_cursor = incremental("last_modified_date", initial_value=last_sync)
# Paginate through Prophet21 API skip = 0 page_size = 100 while True: # Build OData query filter_clause = "" if last_sync: filter_clause = f"&$filter=modified_date gt datetime'{last_sync}'"url = f"{api_url}/items?$skip={skip}&$top={page_size}{filter_clause}"
response = requests.get(url, auth=(username, password)) response.raise_for_status()
data = response.json() items = data.get("value", [])
if not items: break # Transform and yield records for item in items: yield { "id": f"{connection_id}_{item['item_id']}", # Unique across all connections "connection_id": connection_id, "clerk_organization_id": clerk_organization_id, "erp_type": "prophet21", "erp_product_id": item["item_id"], "name": item["item_desc"], "description": item.get("extended_desc"), "unit_of_measure": item.get("unit_of_measure"), "delete_flag": item.get("delete_flag") == "Y", "last_synced_at": dlt.common.time.now_timestamp(),
# Flexible metadata (no schema changes needed) "metadata": { "prophet21": { "company_id": item.get("company_id"), "price_code": item.get("price_code"), "weight": item.get("weight"), "hazmat_flag": item.get("hazmat_flag"), # ... all Prophet21-specific fields } },
# Image metadata "images": { "erp_url": item.get("image_url"), "r2_key": None, # Populated by separate image sync task "has_image": bool(item.get("image_url")) },
# Pricing snapshot "current_pricing": { "base_price": item.get("base_price"), "currency": "USD", "effective_date": item.get("price_effective_date") },# Full-text search field "search_text": f"{item['item_id']} {item['item_desc']} {item.get('extended_desc', '')}" } skip += page_size
# Pipeline configurationpipeline = dlt.pipeline( pipeline_name="prophet21_sync", destination="hudi", # Write to Hudi lakehouse dataset_name="erp_data")
# Run sync (orchestrated by Trigger.dev)def sync_prophet21_products(connection_config: dict): pipeline.run( fetch_prophet21_products(**connection_config), table_name="erp_products", write_disposition="merge", loader_file_format="parquet" )Trigger.dev Orchestration (No Major Changes)
Section titled “Trigger.dev Orchestration (No Major Changes)”import { schemaTask } from '@trigger.dev/sdk';import { z } from 'zod';import { exec } from 'child_process';import { promisify } from 'util';
const execAsync = promisify(exec);
export const syncERPProducts = schemaTask({ id: 'sync-erp-products-v2', // New version for dlt-based sync schema: z.object({ connectionId: z.string().uuid(), clerkOrganizationId: z.string(), erpType: z.enum(['prophet21', 'netsuite', 'demo']), fullSync: z.boolean().default(false), }), retry: { maxAttempts: 3, factor: 1.5, }, machine: { preset: 'medium-2x', // Same as current }, run: async payload => { // Get ERP connection credentials from PostgreSQL const connection = await db.query.erpConnections.findFirst({ where: eq(erpConnections.id, payload.connectionId), });
if (!connection) { throw new Error(`Connection ${payload.connectionId} not found`); }
// Determine last sync timestamp (for incremental sync) const lastSync = payload.fullSync ? null : await getLastSyncTimestamp(payload.connectionId, 'products');
// Invoke dlt pipeline (Python process) const { stdout, stderr } = await execAsync(`python3 /etl/pipelines/prophet21_products.py`, { env: { CONNECTION_ID: payload.connectionId, CLERK_ORG_ID: payload.clerkOrganizationId, API_URL: connection.apiUrl, USERNAME: connection.username, PASSWORD: connection.password, // TODO: Use secret manager LAST_SYNC: lastSync?.toISOString(), HUDI_TABLE: 'erp_products', S3_BUCKET: process.env.R2_BUCKET_NAME, S3_ACCESS_KEY: process.env.R2_ACCESS_KEY_ID, S3_SECRET_KEY: process.env.R2_SECRET_ACCESS_KEY, }, });
logger.info('Product sync completed', { connectionId: payload.connectionId, stdout, });
return { success: true, recordsProcessed: parseRecordCount(stdout), }; },});
// Scheduled daily sync (same as current)export const scheduledERPSyncV2 = schedules.task({ id: 'scheduled-erp-sync-v2', cron: '0 2 * * *', // 2 AM UTC daily run: async payload => { // Get all active ERP connections const connections = await db.query.erpConnections.findMany({ where: eq(erpConnections.active, true), }); // Trigger sync for each connection for (const conn of connections) { await syncERPProducts.trigger({ connectionId: conn.id, clerkOrganizationId: conn.clerkOrganizationId, erpType: conn.type, fullSync: false, }); } },});4. dbt Business Logic Layer
Section titled “4. dbt Business Logic Layer”-- dbt/models/marts/fct_product_enriched.sql{{ config( materialized='incremental', partition_by={ "field": "clerk_organization_id", "data_type": "string" }, unique_key=['connection_id', 'erp_product_id'] )}}
WITH base_products AS ( SELECT *, -- Extract nested fields metadata.prophet21.company_id AS p21_company_id, metadata.prophet21.price_code AS p21_price_code, images.has_image, current_pricing.base_price FROM {{ source('hudi', 'erp_products') }} WHERE delete_flag = false),
product_mappings AS ( SELECT connection_id, source_item_id AS erp_product_id, COUNT(*) AS mapping_count, ARRAY_AGG(target_item_number) AS mapped_item_numbers FROM {{ source('postgres', 'part_number_mappings') }} GROUP BY connection_id, source_item_id)
SELECT p.*, m.mapping_count, m.mapped_item_numbers,
-- Business logic: Product health score CASE WHEN p.has_image AND p.base_price > 0 AND m.mapping_count > 0 THEN 'excellent' WHEN p.base_price > 0 AND m.mapping_count > 0 THEN 'good' WHEN p.base_price > 0 THEN 'fair' ELSE 'poor' END AS data_quality_score
FROM base_products pLEFT JOIN product_mappings m ON p.connection_id = m.connection_id AND p.erp_product_id = m.erp_product_id
{% if is_incremental() %} WHERE p.last_synced_at > (SELECT MAX(last_synced_at) FROM {{ this }}){% endif %}Migration Strategy
Section titled “Migration Strategy”Phase 1: Pilot (Products Only, 4-6 Weeks)
Section titled “Phase 1: Pilot (Products Only, 4-6 Weeks)”Goal: Validate architecture with ONE entity type alongside current system.
┌─────────────────────────────────────────────────────────┐│ PHASE 1: DUAL SYSTEM (PILOT) │├─────────────────────────────────────────────────────────┤│ ││ Current System (Production): ││ • Prophet21SyncService.syncProducts() ──► PostgreSQL ││ • Product search uses tsvector ││ ││ New System (Pilot): ││ • dlt pipeline ──► Hudi (R2 storage) ││ • Cube semantic layer (read-only) ││ • Compare results with current system ││ ││ Metrics to Compare: ││ • Sync duration ││ • Search query performance ││ • Storage costs ││ • Data accuracy ││ │└─────────────────────────────────────────────────────────┘Week 1-2: Infrastructure Setup
Section titled “Week 1-2: Infrastructure Setup”# 1. Spin up Hudi + Trino + Cube locally (Docker Compose)docker-compose -f docker-compose.data-backend.yml up
# Services:# - Trino coordinator (port 8080)# - Cube API (port 4000)# - MinIO (local S3, port 9000)# - Hudi metadata service
# 2. Configure Hudi table for productsspark-submit create_products_table.py
# 3. Deploy Cube modelcube deployWeek 3-4: dlt Pipeline Development
Section titled “Week 3-4: dlt Pipeline Development”# Develop prophet21_products.py pipeline# Run parallel to existing sync# Compare record counts, data qualityWeek 5-6: Validation & Decision
Section titled “Week 5-6: Validation & Decision”Success Criteria: ✓ dlt sync completes in < current sync duration ✓ Hudi storage cost < 50% of PostgreSQL ✓ Cube queries perform within 200ms p95 ✓ 100% data parity with current system ✓ Multi-tenant isolation verified
Decision Gate: ✓ Proceed to Phase 2 (migrate remaining entities) ✗ Abandon data backend approachPhase 2: Full Migration (8-12 Weeks)
Section titled “Phase 2: Full Migration (8-12 Weeks)”Goal: Migrate all ERP entities to Hudi lakehouse.
Entities to Migrate:1. Products (already done in Phase 1)2. Customers3. Cross-references4. Shipping addresses5. Inventory locations6. Sync logs (for analytics)Migration Sequence
Section titled “Migration Sequence”// Week 1-2: CustomerssyncERPCustomersV2; // dlt pipeline// Parallel run for 2 weeks, then cut over
// Week 3-4: Cross-referencessyncERPCrossReferencesV2;
// Week 5-6: Shipping addressessyncERPShippingAddressesV2;
// Week 7-8: Inventory locationssyncERPInventoryV2;
// Week 9-10: Testing & validation// Week 11-12: Cutover & deprecationPhase 3: Deprecation (2-4 Weeks)
Section titled “Phase 3: Deprecation (2-4 Weeks)”Goal: Remove old PostgreSQL replica tables and IERPSyncService.
-- Mark tables for deletionALTER TABLE erp_products RENAME TO erp_products_deprecated;ALTER TABLE erp_customers RENAME TO erp_customers_deprecated;-- ... etc.
-- Monitor for 30 days (confirm no queries)
-- Drop tablesDROP TABLE erp_products_deprecated CASCADE;// Remove deprecated code// ✗ DELETE: Prophet21SyncService// ✗ DELETE: IERPSyncService interface// ✗ DELETE: erp-replica.ts schema
// Keep only:// ✓ dlt pipelines// ✓ Cube models// ✓ Trigger.dev orchestration tasksCost-Benefit Analysis
Section titled “Cost-Benefit Analysis”Current System Costs (Estimated)
Section titled “Current System Costs (Estimated)”PostgreSQL (OLTP + OLAP): Instance: db.r6g.2xlarge (8 vCPU, 64 GB RAM) Cost: ~$800/monthStorage (4.5 TB SSD): Cost: ~$450/month (100 IOPS/GB)
Backup & Replication: Cost: ~$200/month
Total: ~$1,450/month for databaseCompute (Trigger.dev tasks): Medium-2x machines (2 vCPU, 4 GB) Sync duration: ~30 min/org/day Cost: ~$0.50/org/month100 orgs: ~$50/monthTotal Current Monthly Cost: ~$1,500Proposed System Costs (Estimated)
Section titled “Proposed System Costs (Estimated)”PostgreSQL (OLTP only, downsized): Instance: db.r6g.xlarge (4 vCPU, 32 GB RAM) Cost: ~$400/monthStorage (500 GB SSD - 90% reduction): Cost: ~$50/month
Total PostgreSQL: ~$450/monthHudi Lakehouse: Storage (Cloudflare R2): 4.5 TB Parquet (compressed ~60%): 2.7 TB Cost: $0.015/GB/month = ~$40/monthEgress (read queries): ~100 GB/month: Free (R2 has no egress fees)
Total Storage: ~$40/month
Trino Query Engine: Instance: c6g.2xlarge (8 vCPU, 16 GB) Cost: ~$250/month
Cube Semantic Layer: Instance: t3.medium (2 vCPU, 4 GB) Cost: ~$35/month
Spark for dlt: On-demand (runs 1-2 hours/day) Cost: ~$50/month
Compute (Trigger.dev - unchanged): Cost: ~$50/monthTotal Proposed Monthly Cost: ~$875/monthCost Savings
Section titled “Cost Savings”Current: $1,500/monthProposed: $875/monthSavings: $625/month (~42% reduction)
Annual Savings: $7,5003-Year Savings: $22,500Non-Cost Benefits
Section titled “Non-Cost Benefits”Scalability: Current: PostgreSQL struggles > 5 TB Proposed: Hudi scales to petabytes on object storageHistorical Analytics: Current: Impossible (snapshot-only) Proposed: Time-travel queries for pricing trends, inventory patterns
Multi-ERP Federation: Current: Requires complex unions across connection_id Proposed: Native support via Cube models
AI/LLM Integration: Current: Direct SQL queries (brittle, security risk) Proposed: Semantic layer with RLS, pre-aggregations, cachingDeveloper Experience: Current: Schema migrations for every new ERP field Proposed: Schema-on-read (flexible metadata JSONB)Risk Assessment
Section titled “Risk Assessment”Technical Risks
Section titled “Technical Risks”| Risk | Likelihood | Impact | Mitigation |
|---|---|---|---|
| Hudi performance < PostgreSQL | Low | High | Run Phase 1 pilot with benchmarks |
| Cube adds latency | Medium | Medium | Enable aggressive caching, pre-aggregations |
| dlt pipeline bugs | Medium | Medium | Dual-run with current system during pilot |
| Spark complexity | Medium | Medium | Use managed Spark (Databricks or EMR) |
| Migration data loss | Low | Critical | Checksums, row counts, reconciliation |
| Learning curve | High | Low | Invest in training, documentation |
Operational Risks
Section titled “Operational Risks”| Risk | Likelihood | Impact | Mitigation |
|---|---|---|---|
| Increased operational complexity | High | Medium | Phased rollout, comprehensive monitoring |
| Vendor lock-in (Hudi/Trino) | Low | Medium | Open-source tools (no vendor lock-in) |
| R2 outage | Low | High | Dual-region replication, fallback to S3 |
| Team unfamiliarity | High | Low | Pilot phase builds expertise |
Business Risks
Section titled “Business Risks”| Risk | Likelihood | Impact | Mitigation |
|---|---|---|---|
| Customer-facing downtime | Low | Critical | Blue-green deployment, rollback plan |
| Feature development slowdown | Medium | Medium | Pilot in parallel, no disruption to app layer |
| Cost overruns | Low | Medium | Monitor actual costs vs estimates in pilot |
Risk Score: MEDIUM (Acceptable with phased approach)
Section titled “Risk Score: MEDIUM (Acceptable with phased approach)”Implementation Roadmap
Section titled “Implementation Roadmap”Timeline Overview
Section titled “Timeline Overview”Phase 0: Planning & Approval (2 weeks) Week 1: Architecture review, stakeholder alignment Week 2: Resource allocation, budgeting
Phase 1: Pilot (Products Only) (6 weeks) Week 1-2: Infrastructure setup (Hudi, Trino, Cube) Week 3-4: dlt pipeline development Week 5-6: Validation & decision gate
Phase 2: Full Migration (12 weeks) Week 1-2: Customers Week 3-4: Cross-references Week 5-6: Shipping addresses Week 7-8: Inventory locations Week 9-10: Testing & validation Week 11-12: Cutover
Phase 3: Deprecation & Cleanup (4 weeks) Week 1-2: Mark old tables for deletion Week 3-4: Code cleanup, documentation
Total Duration: ~24 weeks (6 months)Resource Requirements
Section titled “Resource Requirements”Team: • 1 Data Engineer (full-time) - Hudi, Spark, dlt expertise - Build ETL pipelines
• 1 Backend Engineer (50% time) - Integrate Cube API - Update Trigger.dev tasks
• 1 DevOps Engineer (25% time) - Infrastructure setup - Monitoring, alerting
• 1 QA Engineer (25% time) - Validation testing - Performance benchmarks
Budget: • Cloud infrastructure: ~$875/month (ongoing) • Pilot phase (6 weeks): ~$1,500 one-time • Engineering time: ~$60K (salary allocation)
Total Phase 1 Investment: ~$65KRecommendation
Section titled “Recommendation”Primary Recommendation: PROCEED WITH PHASED APPROACH
Section titled “Primary Recommendation: PROCEED WITH PHASED APPROACH”Rationale:
-
✅ Strong Technical Fit: Hudi lakehouse + Cube semantic layer aligns perfectly with ERP-Unlocked’s analytical workload (product search, inventory queries, pricing analytics).
-
✅ Future-Proofs Architecture: Enables multi-ERP federation, historical analytics, AI-powered features that are impossible with current architecture.
-
✅ Cost Savings: 42% reduction in infrastructure costs ($7,500/year) while improving performance and capabilities.
-
✅ Phased Risk Mitigation: Pilot with products only validates architecture before full commitment.
-
✅ No Customer Disruption: Can run parallel to current system, gradual cutover.
Conditions for Success
Section titled “Conditions for Success”Required: ✓ Hire or allocate 1 experienced Data Engineer (Hudi/Spark expertise) ✓ Commit to 6-month timeline (no rushing) ✓ Define clear success metrics for pilot ✓ Budget for $65K Phase 1 investment
Nice-to-Have: • Databricks or AWS EMR managed Spark (reduce operational burden) • Better Stack integration for Cube/Trino monitoring • Dedicated staging environment for data backendAlternative Recommendation: INCREMENTAL IMPROVEMENTS TO CURRENT SYSTEM
Section titled “Alternative Recommendation: INCREMENTAL IMPROVEMENTS TO CURRENT SYSTEM”If data backend is deemed too complex, consider:
Option B: Optimize Current PostgreSQL Architecture • Separate OLAP workload to read replica • Implement TimescaleDB for historical analytics • Add Redis caching layer for product searches • Keep IERPSyncService, improve batch sizes
Investment: ~$15K (3-4 weeks) Cost Savings: Minimal (~$100/month) Benefits: Lower complexity, faster implementation Drawbacks: No multi-ERP federation, no time-travel, scaling limitsDecision Framework
Section titled “Decision Framework”Choose Data Backend IF: ✓ Planning to support 5+ ERP systems ✓ Need historical analytics (pricing trends, inventory patterns) ✓ Scaling to 100+ organizations ✓ Willing to invest 6 months + $65K ✓ Team has (or can hire) data engineering expertise
Choose Incremental Improvements IF: ✓ Current scale is sufficient (<50 orgs) ✓ Limited engineering resources ✓ Need faster time-to-value (<1 month) ✓ Risk-averse to architectural changesAppendix A: Comparison Matrix
Section titled “Appendix A: Comparison Matrix”| Aspect | Current (PostgreSQL Replica) | Proposed (Hudi Lakehouse) |
|---|---|---|
| Storage Cost (4.5 TB) | $450/month | $40/month |
| Query Performance (product search) | 800ms p95 (tsvector) | 200ms p95 (Hudi Lucene index) |
| Historical Analytics | ❌ Snapshot only | ✅ Time-travel queries |
| Multi-ERP Queries | ⚠️ Complex unions | ✅ Native federation via Cube |
| Schema Changes | ⚠️ Migrations required | ✅ Schema-on-read (flexible) |
| Full-Text Search | PostgreSQL tsvector | Hudi Lucene index |
| Scalability Limit | ~5 TB (practical) | Petabytes |
| AI/LLM Integration | ⚠️ Direct SQL (security risk) | ✅ Semantic layer (RLS, caching) |
| Operational Complexity | Low (familiar) | Medium (new stack) |
| Team Expertise Required | PostgreSQL DBA | Data Engineer (Spark, Hudi) |
Appendix B: Sample Cube API Queries for AI Agents
Section titled “Appendix B: Sample Cube API Queries for AI Agents”Query 1: Federated Product Search Across All ERPs
Section titled “Query 1: Federated Product Search Across All ERPs”query FederatedProductSearch($searchTerm: String!) { cube( measures: ["ERPProducts.count"] dimensions: [ "ERPProducts.product_id" "ERPProducts.name" "ERPProducts.description" "ERPProducts.erp_type" "ERPProducts.has_image" ] filters: [{ member: "ERPProducts.name", operator: "contains", values: [$searchTerm] }] segments: ["ERPProducts.active_products"] limit: 100 ) { ERPProducts { product_id name description erp_type has_image } }}
# Variables: { "searchTerm": "industrial pump" }
# Response:# [# { product_id: "P21-1234", name: "Industrial Pump A", erp_type: "prophet21" },# { product_id: "NS-5678", name: "Industrial Pump B", erp_type: "netsuite" }# ]Query 2: Historical Pricing Analysis
Section titled “Query 2: Historical Pricing Analysis”query ProductPricingTrend($productId: String!, $startDate: String!) { cube( measures: ["ERPPricing.avg_price", "ERPPricing.min_price", "ERPPricing.max_price"] dimensions: ["ERPPricing.snapshot_date"] filters: [ { member: "ERPProducts.product_id", operator: "equals", values: [$productId] } { member: "ERPPricing.snapshot_date", operator: "afterDate", values: [$startDate] } ] timeDimensions: [{ dimension: "ERPPricing.snapshot_date", granularity: "week" }] ) { ERPPricing { snapshot_date avg_price min_price max_price } }}
# Variables: { "productId": "P21-1234", "startDate": "2024-01-01" }
# Response: Weekly pricing trend (impossible with current architecture!)Query 3: Inventory Availability Across Locations
Section titled “Query 3: Inventory Availability Across Locations”query InventoryByLocation($productIds: [String!]!) { cube( measures: [ "ERPInventory.total_on_hand" "ERPInventory.total_available" "ERPInventory.location_count" ] dimensions: ["ERPInventory.product_id", "ERPInventory.location_id"] filters: [{ member: "ERPInventory.product_id", operator: "equals", values: $productIds }] ) { ERPInventory { product_id location_id total_on_hand total_available } }}Appendix C: Infrastructure Diagram (Detailed)
Section titled “Appendix C: Infrastructure Diagram (Detailed)”┌────────────────────────────────────────────────────────────────────────┐│ DEPLOYMENT ARCHITECTURE │├────────────────────────────────────────────────────────────────────────┤│ ││ Cloudflare (CDN + WAF) ││ ├─ Marketing site (static) ││ └─ Webapp frontend (proxied to Coolify) ││ ││ ┌──────────────────────────────────────────────────────────────┐ ││ │ COOLIFY (Docker Orchestration) │ ││ │ │ ││ │ Container: webapp (Astro + React) │ ││ │ Container: pdf-api (FastAPI) │ ││ │ Container: pdf-worker (Celery) │ ││ │ Container: redis (cache + queue) │ ││ │ Container: postgres (OLTP only, downsized) │ ││ │ │ ││ │ NEW Containers: │ ││ │ Container: cube-api (Cube semantic layer) │ ││ │ Container: trino-coordinator (query engine) │ ││ │ Container: trino-worker-1 (worker node) │ ││ │ Container: trino-worker-2 (auto-scale) │ ││ │ Container: spark-master (ETL orchestrator) │ ││ │ Container: spark-worker-1 │ ││ └───────────────────────────────────────────────────────────────┘ ││ ││ Cloudflare R2 (Object Storage) ││ ├─ Bucket: erp-data ││ │ └─ Hudi tables: ││ │ ├─ erp_products/ (Parquet files, partitioned by org_id) ││ │ ├─ erp_customers/ ││ │ ├─ erp_inventory/ ││ │ └─ .hoodie/metadata/ (Hudi transaction logs) ││ │ ││ └─ Bucket: erp-attachments (PDFs, images) ← unchanged ││ ││ Better Stack (Observability) ││ ├─ OpenTelemetry traces (webapp, pdf-api, cube, trino) ││ ├─ Logs aggregation ││ └─ Dashboards ││ ││ Trigger.dev (Background Jobs) ← unchanged ││ ├─ Scheduled ERP syncs (now invoke dlt pipelines) ││ ├─ PDF processing tasks ││ └─ System health checks ││ │└────────────────────────────────────────────────────────────────────────┘Document Status: Ready for Review
Next Steps:
- Technical review with engineering team
- Cost validation with finance
- Decision on Phase 1 pilot (go/no-go)
- Resource allocation if approved
Questions or Feedback? Reach out to the architecture team.