Virtualized EDW

Federated queries across the legacy EMR, the MariaDB EDW and the lakehouse through one SQL endpoint. Data is read in place; filters and aggregations are pushed down to each source where the connector supports it.

Encounters live in the legacy EMR (SQL Server), claims and department mappings in the MariaDB EDW, and the enterprise patient identifier in the lakehouse. One federated query, no copies.

SQL

SELECT dd.service_line,
       c.payer,
       count(DISTINCT sp.enterprise_patient_id) AS patients,
       count(*) AS claims,
       sum(c.billed_amount) AS billed,
       sum(c.paid_amount) AS paid
FROM mssql.dbo.encounter e
JOIN mariadb.edw.claim c ON c.encounter_id = e.encounter_id
JOIN mariadb.edw.dim_department dd ON dd.department = e.department
JOIN lakehouse.silver.patient sp ON sp.mrn = e.mrn AND sp.is_current
WHERE c.claim_dt >= DATE '2025-01-01'
GROUP BY dd.service_line, c.payer
ORDER BY paid DESC
LIMIT 20

Sources touched and pushdown

Catalog	Source table	Operator	Pushed down to source
mariadb	`edw.claim`	ScanFilter	filter on claim_dt, 4 columns projected
mssql	`legacy_emr.dbo.encounter`	ScanFilter	3 columns projected
mariadb	`edw.dim_department`	TableScan	2 columns projected
lakehouse	`lakehouse.silver.patient`	TableScan	filter on is_current, 2 columns projected

3 source catalogs, 4 scans. Full distributed plan below (170 lines).

Result (20 rows in 321 ms)

service_line	payer	patients	claims	billed	paid
Diagnostics	Medicare	506	593	1,767,566.00	622,241.03
Surgery	Medicare	37	37	1,555,524.50	591,551.76
Diagnostics	Excellus BCBS	284	333	1,050,095.50	520,538.66
Surgery	Excellus BCBS	28	28	948,366.75	466,531.02
Surgery	Independent Health	18	18	631,965.50	327,187.39
Diagnostics	Independent Health	201	243	709,423.00	313,698.57
Medical Oncology	Medicare	658	2,453	773,008.20	273,046.09
Infusion Services	Medicare	105	510	623,157.00	215,785.57
Medical Oncology	Excellus BCBS	374	1,368	432,886.76	213,621.12
Diagnostics	Aetna	108	132	420,685.50	201,690.50
Infusion Services	Excellus BCBS	62	329	403,208.25	198,835.09
Diagnostics	Univera	99	117	366,168.25	144,251.40
Medical Oncology	Independent Health	257	963	300,152.91	137,381.07
Surgery	Univera	7	7	306,618.75	113,500.88
Surgery	Aetna	10	10	283,890.75	110,933.43
Infusion Services	Independent Health	39	188	226,329.90	106,605.49
Diagnostics	Medicaid NY	93	107	352,168.00	93,793.06
Radiation Oncology	Medicare	24	124	258,353.75	90,289.81
Inpatient Services	Excellus BCBS	10	11	153,678.00	82,113.75
Infusion Services	Aetna	24	130	157,956.00	80,854.23

Query plan

Trino version: 484
Fragment 0 [SINGLE]
    Output layout: [service_line_3, payer, count, count_5, sum, sum_6]
    Output partitioning: SINGLE []
    Output[columnNames = [service_line, payer, patients, claims, billed, paid]]
         Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, count_5:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
         Estimates: {rows: 20 (3.16kB), cpu: 0, memory: 0B, network: 0B}
         service_line := service_line_3
         patients := count
         claims := count_5
         billed := sum
         paid := sum_6
       Project[]
            Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, count_5:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
            Estimates: {rows: 20 (3.16kB), cpu: 3.16k, memory: 0B, network: 0B}
            count_5 := COALESCE(coalesce_expr, bigint '0')
          TopN[count = 20, orderBy = [sum_6 DESC NULLS LAST]]
               Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
               Estimates: {rows: 20 (3.16kB), cpu: ?, memory: ?, network: ?}
             LocalExchange[partitioning = SINGLE]
                  Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
                  Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B}
                RemoteSource[sourceFragmentIds = [1]]
                    Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]

Fragment 1 [HASH]
    Output layout: [service_line_3, payer, count, coalesce_expr, sum, sum_6]
    Output partitioning: SINGLE []
    TopNPartial[count = 20, orderBy = [sum_6 DESC NULLS LAST]]
         Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
       LocalExchange[partitioning = SINGLE]
            Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
            Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B}
          TopNPartial[count = 20, orderBy = [sum_6 DESC NULLS LAST]]
               Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
             LocalExchange[partitioning = ROUND_ROBIN]
                  Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
                  Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
                TopNPartial[count = 20, orderBy = [sum_6 DESC NULLS LAST]]
                     Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
                   Aggregate[type = FINAL, keys = [service_line_3, payer]]
                        Layout: [service_line_3:varchar(40), payer:varchar(40), count:bigint, coalesce_expr:bigint, sum:decimal(38,2), sum_6:decimal(38,2)]
                        Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}
                        count := count(count_9)
                        coalesce_expr := any_value(any_value)
                        sum := any_value(any_value_10)
                        sum_6 := any_value(any_value_11)
                      LocalExchange[partitioning = HASH, arguments = [service_line_3::varchar(40), payer::varchar(40)]]
                           Layout: [service_line_3:varchar(40), payer:varchar(40), count_9:bigint, any_value:bigint, any_value_10:decimal(38,2), any_value_11:decimal(38,2)]
                           Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
                         RemoteSource[sourceFragmentIds = [2]]
                             Layout: [service_line_3:varchar(40), payer:varchar(40), count_9:bigint, any_value:bigint, any_value_10:decimal(38,2), any_value_11:decimal(38,2)]

Fragment 2 [HASH]
    Output layout: [service_line_3, payer, count_9, any_value, any_value_10, any_value_11]
    Output partitioning: HASH [service_line_3, payer]
    Aggregate[type = PARTIAL, keys = [service_line_3, payer]]
         Layout: [service_line_3:varchar(40), payer:varchar(40), count_9:bigint, any_value:bigint, any_value_10:decimal(38,2), any_value_11:decimal(38,2)]
         count_9 := count(enterprise_patient_id) (mask = gid_filter)
         any_value := any_value(inner) (mask = non_distinct_gid_filter)
         any_value_10 := any_value(inner_7) (mask = non_distinct_gid_filter)
         any_value_11 := any_value(inner_8) (mask = non_distinct_gid_filter)
       Project[]
            Layout: [non_distinct_gid_filter:boolean, gid_filter:boolean, service_line_3:varchar(40), payer:varchar(40), enterprise_patient_id:uuid, inner:bigint, inner_7:decimal(38,2), inner_8:decimal(38,2)]
            Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
            non_distinct_gid_filter := (group = bigint '0')
            gid_filter := (group = bigint '1')
          Aggregate[type = FINAL, keys = [service_line_3, payer, enterprise_patient_id, group]]
               Layout: [service_line_3:varchar(40), payer:varchar(40), enterprise_patient_id:uuid, group:bigint, inner:bigint, inner_7:decimal(38,2), inner_8:decimal(38,2)]
               Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}
               inner := count(count_12)
               inner_7 := sum(sum_13)
               inner_8 := sum(sum_14)
             LocalExchange[partitioning = HASH, arguments = [service_line_3::varchar(40), payer::varchar(40), enterprise_patient_id::uuid, group::bigint]]
                  Layout: [service_line_3:varchar(40), payer:varchar(40), enterprise_patient_id:uuid, group:bigint, count_12:bigint, sum_13:varbinary, sum_14:varbinary]
                  Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
                RemoteSource[sourceFragmentIds = [3]]
                    Layout: [service_line_3:varchar(40), payer:varchar(40), enterprise_patient_id:uuid, group:bigint, count_12:bigint, sum_13:varbinary, sum_14:varbinary]

Fragment 3 [HASH]
    Output layout: [service_line_3, payer, enterprise_patient_id, group, count_12, sum_13, sum_14]
    Output partitioning: HASH [service_line_3, payer, enterprise_patient_id, group]
    Aggregate[type = PARTIAL, keys = [service_line_3, payer, enterprise_patient_id, group]]
         Layout: [service_line_3:varchar(40), payer:varchar(40), enterprise_patient_id:uuid, group:bigint, count_12:bigint, sum_13:varbinary, sum_14:varbinary]
         count_12 := count(*)
         sum_13 := sum(billed_amount)
         sum_14 := sum(paid_amount)
       GroupId[symbols = [[service_line_3, payer, billed_amount, paid_amount], [service_line_3, payer, enterprise_patient_id]]]
            Layout: [service_line_3:varchar(40), payer:varchar(40), billed_amount:decimal(12,2), paid_amount:decimal(12,2), enterprise_patient_id:uuid, group:bigint]
            service_line_3 := service_line_3
            payer := payer
            billed_amount := billed_amount
            paid_amount := paid_amount
            enterprise_patient_id := enterprise_patient_id
          InnerJoin[criteria = (expr = mrn_4), distribution = PARTITIONED]
               Layout: [payer:varchar(40), billed_amount:decimal(12,2), paid_amount:decimal(12,2), service_line_3:varchar(40), enterprise_patient_id:uuid]
               Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}
               Distribution: PARTITIONED
               dynamicFilterAssignments = {mrn_4 -> #df_838}
             RemoteSource[sourceFragmentIds = [4]]
                  Layout: [payer:varchar(40), billed_amount:decimal(12,2), paid_amount:decimal(12,2), service_line_3:varchar(40), expr:varchar]
             LocalExchange[partitioning = HASH, arguments = [mrn_4::varchar]]
                  Layout: [enterprise_patient_id:uuid, mrn_4:varchar]
                  Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
                RemoteSource[sourceFragmentIds = [7]]
                    Layout: [enterprise_patient_id:uuid, mrn_4:varchar]

Fragment 4 [SOURCE]
    Output layout: [payer, billed_amount, paid_amount, service_line_3, expr]
    Output partitioning: HASH [expr]
    Project[]
         Layout: [payer:varchar(40), billed_amount:decimal(12,2), paid_amount:decimal(12,2), service_line_3:varchar(40), expr:varchar]
         Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
         expr := CAST(mrn AS varchar)
       InnerJoin[criteria = (department = department_2), distribution = REPLICATED]
            Layout: [payer:varchar(40), billed_amount:decimal(12,2), paid_amount:decimal(12,2), mrn:varchar(12), service_line_3:varchar(40)]
            Estimates: {rows: ? (?), cpu: ?, memory: 1.61kB, network: 0B}
            Distribution: REPLICATED
            dynamicFilterAssignments = {department_2 -> #df_839}
          InnerJoin[criteria = (encounter_id_1 = encounter_id), distribution = REPLICATED]
                Layout: [payer:varchar(40), billed_amount:decimal(12,2), paid_amount:decimal(12,2), mrn:varchar(12), department:varchar(60)]
                Estimates: {rows: ? (?), cpu: ?, memory: 3.39MB, network: 0B}
                Distribution: REPLICATED
                dynamicFilterAssignments = {encounter_id -> #df_840}
              ScanFilter[table = mariadb:edw.claim edw.claim constraint on [claim_dt] columns=[encounter_id:integer:INT, payer:varchar(40):VARCHAR, billed_amount:decimal(12,2):DECIMAL, paid_amount:decimal(12,2):DECIMAL], dynamicFilters = {encounter_id_1 = #df_840}]
                   Layout: [encounter_id_1:integer, payer:varchar(40), billed_amount:decimal(12,2), paid_amount:decimal(12,2)]
                   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}
                   payer := payer:varchar(40):VARCHAR
                   encounter_id_1 := encounter_id:integer:INT
                   billed_amount := billed_amount:decimal(12,2):DECIMAL
                   paid_amount := paid_amount:decimal(12,2):DECIMAL
              LocalExchange[partitioning = SINGLE]
                   Layout: [encounter_id:integer, mrn:varchar(12), department:varchar(60)]
                   Estimates: {rows: 80785 (3.39MB), cpu: 0, memory: 0B, network: 0B}
                 RemoteSource[sourceFragmentIds = [5]]
                     Layout: [encounter_id:integer, mrn:varchar(12), department:varchar(60)]
          LocalExchange[partitioning = SINGLE]
               Layout: [department_2:varchar(60), service_line_3:varchar(40)]
               Estimates: {rows: 15 (1.61kB), cpu: 0, memory: 0B, network: 0B}
             RemoteSource[sourceFragmentIds = [6]]