diff --git a/CHANGELOG.md b/CHANGELOG.md index d4315220..a8582f42 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ All notable changes to Data Hopper EDW (formerly hop-datavault) are documented i ## Unreleased +### Partition large BV SCD2 loads (issue #141) + +- SCD2 table option **Hash-key partitions** (None / 4 / 8 / 16) splits a Full rebuild so each satellite `ORDER BY` covers a first-byte slice of the parent hash key +- A generated workflow truncates the BV target once, then a driver pipeline runs the parameterized SCD2 pipeline for partition numbers `0 .. N-1` (`'${PARTITION_COUNT}'`, `'${PARTITION_NUMBER}'`) +- Truncate is always a SQL action; Table Output and Native bulk then append each partition +- Staging file writes `{pipeline}-${PARTITION_NUMBER}-${copy}.csv` per partition, then bulk-loads those files from the wrapper workflow +- Incremental build mode cannot be combined with hash-key partitions +- Business Vault Update runs partitioned SCD2 wrapper workflows before the free-pipeline orchestrator + ### Read-only existing Data Vault models - Data Vault configuration checkbox **Read-only existing vault** documents an already-built raw vault so Business Vault and dimensional models can sit on top diff --git a/docs/ai-file-schemas/models/hbv.md b/docs/ai-file-schemas/models/hbv.md index 542a452a..13d4e61d 100644 --- a/docs/ai-file-schemas/models/hbv.md +++ b/docs/ai-file-schemas/models/hbv.md @@ -41,7 +41,8 @@ Inspect real retail samples for exact nested tags (`satellite_config`, `field_ma - Parent DV hub (via references) - One or more **satellite configs** (`satelliteName`, source indicator, field mappings `sourceFieldName` → `targetFieldName`) -- Timeline fields from configuration (`validFromField`, `validToField`, open sentinels) +- Timeline fields from configuration (`validFromField`, `validToField`, open sentinels) +- Optional `hashKeyPartitionCount` (`NONE`, `4`, `8`, `16`) for large full rebuilds — SQL truncate once, then Table Output, Native bulk, or Staging file (one CSV set per partition) ## Anti-patterns diff --git a/docs/ai-file-schemas/samples/hbv-excerpt.xml b/docs/ai-file-schemas/samples/hbv-excerpt.xml index 32c98eb2..0e73a1c6 100644 --- a/docs/ai-file-schemas/samples/hbv-excerpt.xml +++ b/docs/ai-file-schemas/samples/hbv-excerpt.xml @@ -33,6 +33,7 @@ FULL_REBUILD + NONEx_load_ts diff --git a/docs/business-vault-scd2.adoc b/docs/business-vault-scd2.adoc index ff4f46c2..43521562 100644 --- a/docs/business-vault-scd2.adoc +++ b/docs/business-vault-scd2.adoc @@ -53,6 +53,25 @@ For each SCD2 table the plugin builds a Hop pipeline that: Use **Debug** or **Show build pipeline** on an SCD2 table in the `.hbv` editor to inspect the generated pipeline before running a workflow. +=== Hash-key partitions (large full rebuilds) + +When a satellite is too large to `ORDER BY` hash key and load date in one pass, set **Hash-key partitions** to 4, 8, or 16 on the SCD2 table. Full rebuild then: + +1. Truncates the BV target **once** with a workflow SQL action (`TRUNCATE TABLE` via the Hop database dialect). +2. Generates partition numbers `0 .. N-1` and runs the SCD2 pipeline once per part, passing `'${PARTITION_COUNT}'` and `'${PARTITION_NUMBER}'`. +3. Filters each satellite `TableInput` with a first-byte modulus of the parent hash key (BINARY: first octet; HEX: first two hex characters; STRING: first dash-separated token). Every version of a hub/link key stays in one partition. +4. Writes with `TableOutput` **truncate off**, so later partitions append. + +**Show build pipeline** opens the parameterized SCD2 pipeline, the partition driver, and the wrapper workflow. Running the inner pipeline alone with defaults loads partition 0 of N. + +Constraints: + +* Full rebuild only (Incremental already filters by watermark). +* Target load mode is honored: **Table Output** and **Native bulk** append after the SQL truncate; **Staging file** writes one CSV set per partition (`…-${PARTITION_NUMBER}-${Internal.Transform.CopyNr}.csv`) then bulk-loads those files in the wrapper workflow. +* Sequential partition execution (lower peak memory, not parallel wall-clock). + +Integration test: `integration-tests/tests/multi-satellite-bv/update-customer-360-partitioned.hwf` uses `customer-360-partitioned.hbv` (4 parts) and the same golden current-state dataset as the unpartitioned full rebuild. + === Incremental build mode Set **Build mode** to *Incremental* on the SCD2 table when you want to append new functional versions instead of truncating and reloading the BV table on every run. @@ -79,7 +98,7 @@ The watermark is always read from the **Business Vault target table** (`MAX(wate On the first incremental run against an empty BV table the sentinel (`1900-01-01 00:00:00`) applies, so behaviour matches a full rebuild. Later runs process only satellite deltas and close prior open rows in place. -Integration test: `integration-tests/tests/multi-satellite-bv/update-customer-360-incremental.hwf` loads four DV satellite waves, runs **Business Vault Update** once on `customer-360-incremental.hbv`, then validates current open rows against the same golden dataset as the full-rebuild fixture. Multi-run close-and-append behaviour is covered by unit tests in `BvScd2PipelineSupportTest`. +Integration test: `integration-tests/tests/multi-satellite-bv/update-customer-360-incremental.hwf` loads four DV satellite waves, runs **Business Vault Update** once on `customer-360-incremental.hbv`, then validates current open rows against the same golden dataset as the full-rebuild fixture. Multi-run close-and-append behaviour is covered by unit tests in `BvScd2PipelineSupportTest`. Hash-key partitioned full rebuild: `update-customer-360-partitioned.hwf` / `customer-360-partitioned.hbv` (4 parts, same golden). == Single-satellite SCD2 @@ -150,6 +169,9 @@ Open intervals use sentinels from configuration (default `1900-01-01 00:00:00` a |Build mode |`FULL_REBUILD` (default) truncates and reloads the BV table; `INCREMENTAL` filters satellite history and appends/closes versions. +|Hash-key partitions +|`None` (default), or `4` / `8` / `16` parts for a large full rebuild. See <>. + |Incremental watermark field |Optional BV column used for `MAX(...)` watermark reads and incremental write filtering. Defaults to the resolved functional timestamp field. diff --git a/docs/business-vault-update-action.adoc b/docs/business-vault-update-action.adoc index a2a743bd..49801c35 100644 --- a/docs/business-vault-update-action.adoc +++ b/docs/business-vault-update-action.adoc @@ -14,7 +14,8 @@ At runtime the action: * Optionally validates both models before any work is done * Optionally generates CREATE TABLE DDL for Business Vault tables on the BV target database * Generates **build pipelines** for each selected SCD2 and PIT table (and other supported types when implemented) -* Stages pipelines and runs them through a parallel orchestrator +* For SCD2 tables with **hash-key partitions**, generates a wrapper workflow (truncate once, then sequential partition loads; Staging file mode then bulk-loads each partition CSV) and runs it before the free-pipeline orchestrator +* Stages remaining pipelines and runs them through a parallel orchestrator * Optionally publishes BV target table layouts to the data catalog * Aggregates pipeline results into the workflow action result @@ -102,5 +103,6 @@ SCD2 and PIT read independent data paths (PIT does not consume SCD2 output). Ord Sample workflows: * `integration-tests/tests/multi-satellite-bv/update-customer-360.hwf` — multi-satellite SCD2 (full rebuild) +* `integration-tests/tests/multi-satellite-bv/update-customer-360-partitioned.hwf` — same golden as full rebuild, SCD2 hash-key partitions = 4 * `integration-tests/tests/multi-satellite-bv/update-customer-360-incremental.hwf` — multi-satellite SCD2 incremental build mode * `integration-tests/tests/basic/update-vault1.hwf` — SCD2 + PIT on the introductory vault1 model \ No newline at end of file diff --git a/docs/feature-overview.adoc b/docs/feature-overview.adoc index 53e90b7d..2a28d5a7 100644 --- a/docs/feature-overview.adoc +++ b/docs/feature-overview.adoc @@ -324,7 +324,7 @@ image::images/validate-resource-definitions-action-dialog.png[Validate resource === Business Vault (`.hbv`) -Linked to a `.hdv` model. Defines **SCD2** consumption tables (single- or multi-satellite merge), **PIT** helpers, and **SQL views/tables** (dbt-style `ref` / `source`, optional Jinja macros). **Import dbt models** on the canvas (and the **Import dbt project** workflow action) loads a dbt-core project as SQL business tables. **Business Vault Update** validates, optionally publishes target layouts to the catalog, generates SCD2 build pipelines, and orchestrates parallel execution. +Linked to a `.hdv` model. Defines **SCD2** consumption tables (single- or multi-satellite merge), **PIT** helpers, and **SQL views/tables** (dbt-style `ref` / `source`, optional Jinja macros). **Import dbt models** on the canvas (and the **Import dbt project** workflow action) loads a dbt-core project as SQL business tables. **Business Vault Update** validates, optionally publishes target layouts to the catalog, generates SCD2 build pipelines (hash-key partitioned full rebuilds run as truncate-then-load wrapper workflows), and orchestrates parallel execution. image::images/business-vault-dbt-import-models-dialog.png[Import dbt models dialog — project folder, model list, destination, and macro library,align="center"] diff --git a/docs/getting-started-integration-tests.adoc b/docs/getting-started-integration-tests.adoc index e70a1e96..1251798f 100644 --- a/docs/getting-started-integration-tests.adoc +++ b/docs/getting-started-integration-tests.adoc @@ -104,6 +104,8 @@ Run `integration-tests/tests/multi-satellite-bv/update-customer-360.hwf` for end For incremental SCD2, open `customer-360-incremental.hbv` (same mappings, `buildMode=INCREMENTAL`) and run `update-customer-360-incremental.hwf`. That workflow runs the same DV load waves, then **Business Vault Update** with the incremental pipeline, and validates against the same golden current-state dataset. +For a large-table Full rebuild split, open `customer-360-partitioned.hbv` (`hashKeyPartitionCount=4`) and run `update-customer-360-partitioned.hwf`. Same golden current-state dataset; the target is truncated once, then four hash-key slices are loaded. + == Chapter 7 — External read-only raw vault tables When dbt or another tool loads raw vault tables, Hop can still model them for Business Vault. diff --git a/docs/help/bv-scd2-table-dialog.adoc b/docs/help/bv-scd2-table-dialog.adoc index 84882f96..40cdc9bd 100644 --- a/docs/help/bv-scd2-table-dialog.adoc +++ b/docs/help/bv-scd2-table-dialog.adoc @@ -46,6 +46,9 @@ Raw vault satellites remain insert-only (technical load history). SCD2 rewrites |**Build mode** |`FULL_REBUILD` (truncate and reload) or `INCREMENTAL` (watermark, append, close open rows). +|**Hash-key partitions** +|`None` (default) or 4 / 8 / 16. Full rebuild only: truncate once, then load hash-key slices so each satellite `ORDER BY` is smaller. Works with Table Output, Native bulk, and Staging file (one CSV set per partition, then bulk-load). + |**Incremental watermark field** |Optional BV column for `MAX(...)` watermark reads; defaults to the functional timestamp. @@ -69,8 +72,9 @@ The functional timestamp drives interval boundaries — not the technical load d - **Full rebuild** — truncate and reload the BV table from satellite history. - **Incremental** — read satellite deltas above a watermark, close the prior open version, append new versions. Prefer after an initial full load when history is large. +- **Hash-key partitions** — optional 4 / 8 / 16 split on Full rebuild. A wrapper workflow truncates the target, then runs the SCD2 pipeline per partition with `'${PARTITION_COUNT}'` and `'${PARTITION_NUMBER}'`. Honors Table Output, Native bulk, and Staging file. Disabled for Incremental. -Use **Debug** or **Show build pipeline** on the canvas to inspect the generated Hop pipeline before workflow runs. +Use **Debug** or **Show build pipeline** on the canvas to inspect the generated Hop pipeline (and the partition workflow when enabled) before workflow runs. == Field mappings diff --git a/integration-tests/tests/multi-satellite-bv/customer-360-partitioned.hbv b/integration-tests/tests/multi-satellite-bv/customer-360-partitioned.hbv new file mode 100644 index 00000000..6c71bf86 --- /dev/null +++ b/integration-tests/tests/multi-satellite-bv/customer-360-partitioned.hbv @@ -0,0 +1,195 @@ + + + Y + Customer 360 business vault (4 hash-key partitions): four satellites merged into one functional SCD2 table + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360.hdv + business-vault-customer-360 + + Vault + + bv-scd2- + bv-pit- + bv-biz- + 1900-01-01 00:00:00 + 9999-12-31 23:59:59 + + + x_load_ts + x_from_ts + x_to_ts + 1000 + 1 + + +
+ x_load_ts + + + Y + 4 + + + sat_customer_demo + segment + cust_segment + + + sat_customer_demo + loyalty_tier + cust_loyalty_tier + + + sat_customer_demo + demo_score + cust_demo_score + + + sat_customer_contact + email + cust_email + + + sat_customer_contact + phone + cust_phone + + + sat_customer_address + address_line1 + cust_address + + + sat_customer_address + city + cust_city + + + sat_customer_address + postal_code + cust_postal_code + + + sat_customer_prefs + newsletter_opt_in + cust_newsletter + + + sat_customer_prefs + preferred_channel + cust_channel + + + sat_customer_prefs + language_code + cust_language + + + + + sat_customer_demo + + DEMO + + + sat_customer_contact + + CONTACT + + + sat_customer_address + + ADDRESS + + + sat_customer_prefs + + PREFS + + + customer_360_bv + Functional customer 360 view merged from four satellites + SCD2 + + + sat_customer_demo + SATELLITE + + + sat_customer_contact + SATELLITE + + + sat_customer_address + SATELLITE + + + sat_customer_prefs + SATELLITE + + + 496 + 208 + customer_360_bv +
+ + + + This SCD2 exercise uses the input from 4 different satellites to build +one "Customer 360" Type II Slowly Changing Dimension. + +The way it does this is by reading from all 4 sources at the same time and +using a Sorted Schema Merge (new transform) feeding into the [Repeat Fields](https://hop.apache.org/manual/latest/pipeline/transforms/repeatfields.html#_current_when_indicated) transform + +to build combined records of what is essentially a timeline per customer hash key. +What you get is a clear indication of which record version was valid at which point in time. + +Click on the [customer_360_bv](customer_360_bv) table and select "Show build pipeline" to see the full pipeline. + INFORMATION + 144 + 384 + 624 + 224 + + + + + sat_customer_demo + SATELLITE + 240 + 112 + + + sat_customer_contact + SATELLITE + 144 + 208 + + + sat_customer_address + SATELLITE + 144 + 304 + + + sat_customer_prefs + SATELLITE + 448 + 80 + + + customer-360-partitioned + diff --git a/integration-tests/tests/multi-satellite-bv/update-customer-360-partitioned.hwf b/integration-tests/tests/multi-satellite-bv/update-customer-360-partitioned.hwf new file mode 100644 index 00000000..4d212aa8 --- /dev/null +++ b/integration-tests/tests/multi-satellite-bv/update-customer-360-partitioned.hwf @@ -0,0 +1,558 @@ + + + update-customer-360-partitioned + Y + Load hub and four satellites from CSV waves, rebuild customer_360_bv with 4 hash-key partitions, and validate + + - + - + 2026/06/26 23:45:00.000 + 2026/06/26 23:45:00.000 + + + + DV_INITIAL_LOAD_DATE + Fixed load date for initial Data Vault Update actions + 2026/01/01 00:00:00.000 + + + OUTPUT_COPIES + Valid in the current workflow + 1 + + + DV_UPDATE2_LOAD_DATE + Load date for satellite update wave 2 + 2026/03/01 00:00:00.000 + + + METRICS_FOLDER + Valid in the current workflow + ${PROJECT_HOME}/metrics + + + PIPELINE_COPIES + Valid in the current workflow + 3 + + + DV_UPDATE4_LOAD_DATE + Load date for satellite update wave 4 + 2026/05/01 00:00:00.000 + + + DV_UPDATE1_LOAD_DATE + Load date for satellite update wave 1 + 2026/02/01 00:00:00.000 + + + DV_UPDATE3_LOAD_DATE + Load date for satellite update wave 3 + 2026/04/01 00:00:00.000 + + + + + N + 0 + 0 + 60 + 1 + 1 + 0 + 12 + N + Start + + SPECIAL + + 64 + 64 + N + + + + DROP TABLE IF EXISTS hub_customer; +DROP TABLE IF EXISTS sat_customer_demo; +DROP TABLE IF EXISTS sat_customer_contact; +DROP TABLE IF EXISTS sat_customer_address; +DROP TABLE IF EXISTS sat_customer_prefs; +DROP TABLE IF EXISTS customer_360_bv; + + Vault + N + N + + + N + Drop Vault tables + + SQL + + 240 + 64 + N + + + + + ${PROJECT_HOME} + + N + N + + + N + N + Basic + N + N + Y + + activate load1 sources + + SHELL + + 448 + 64 + N + + + + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360.hdv + local + Y + Y + Y + ${PIPELINE_COPIES} + + ${METRICS_FOLDER} + N + vault-catalog + Y + ${java.io.tmpdir}/ddl-customer-360.sql + N + ${DV_INITIAL_LOAD_DATE} + + Y + N + update customer-360 load1 + + DATA_VAULT_UPDATE + + 688 + 64 + N + + + + + ${PROJECT_HOME} + + N + N + + + N + N + Basic + N + N + Y + + activate update1 sources + + SHELL + + 64 + 160 + N + + + + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360.hdv + local + N + N + N + ${PIPELINE_COPIES} + + ${METRICS_FOLDER} + N + N + + N + ${DV_UPDATE1_LOAD_DATE} + + N + N + update customer-360 update1 + + DATA_VAULT_UPDATE + + 320 + 160 + N + + + + + ${PROJECT_HOME} + + N + N + + + N + N + Basic + N + N + Y + + activate update2 sources + + SHELL + + 608 + 160 + N + + + + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360.hdv + local + N + N + N + ${PIPELINE_COPIES} + + ${METRICS_FOLDER} + N + N + + N + ${DV_UPDATE2_LOAD_DATE} + + N + N + update customer-360 update2 + + DATA_VAULT_UPDATE + + 64 + 256 + N + + + + + ${PROJECT_HOME} + + N + N + + + N + N + Basic + N + N + Y + + activate update3 sources + + SHELL + + 352 + 256 + N + + + + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360.hdv + local + N + N + N + ${PIPELINE_COPIES} + + ${METRICS_FOLDER} + N + N + + N + ${DV_UPDATE3_LOAD_DATE} + + N + N + update customer-360 update3 + + DATA_VAULT_UPDATE + + 608 + 256 + N + + + + + ${PROJECT_HOME} + + N + N + + + N + N + Basic + N + N + Y + + activate update4 sources + + SHELL + + 64 + 352 + N + + + + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360.hdv + local + N + N + N + ${PIPELINE_COPIES} + + ${METRICS_FOLDER} + N + N + + N + ${DV_UPDATE4_LOAD_DATE} + + N + N + update customer-360 update4 + + DATA_VAULT_UPDATE + + 320 + 352 + N + + + + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360-partitioned.hbv + local + Y + Y + 1 + + + N + vault-catalog + + N + Y + + N + customer-360-partitioned.hbv + + BUSINESS_VAULT_UPDATE + + 608 + 352 + N + + + + + + validate-customer-360-bv UNIT + + + Validate customer 360 BV + + RunPipelineTests + + 848 + 352 + N + + + + Success + + SUCCESS + + 1056 + 464 + N + + + + ${PROJECT_HOME}/tests/multi-satellite-bv/customer-360-external.hbv + local + Y + Y + 1 + + + N + vault-catalog + + N + Y + + N + customer-360-external.hbv + + BUSINESS_VAULT_UPDATE + + 608 + 464 + N + + + + + + validate-customer-360-bv UNIT + + + Validate customer 360 BV + + RunPipelineTests + + 848 + 464 + N + + + + + + Start + Drop Vault tables + Y + Y + Y + + + Drop Vault tables + activate load1 sources + Y + N + Y + + + activate load1 sources + update customer-360 load1 + Y + N + Y + + + update customer-360 load1 + activate update1 sources + Y + N + Y + + + activate update1 sources + update customer-360 update1 + Y + N + Y + + + update customer-360 update1 + activate update2 sources + Y + N + Y + + + activate update2 sources + update customer-360 update2 + Y + N + Y + + + update customer-360 update2 + activate update3 sources + Y + N + Y + + + activate update3 sources + update customer-360 update3 + Y + N + Y + + + update customer-360 update3 + activate update4 sources + Y + N + Y + + + activate update4 sources + update customer-360 update4 + Y + N + Y + + + update customer-360 update4 + customer-360-partitioned.hbv + Y + N + Y + + + customer-360-partitioned.hbv + Validate customer 360 BV + Y + N + Y + + + customer-360-external.hbv + Validate customer 360 BV + Y + N + Y + + + Validate customer 360 BV + customer-360-external.hbv + Y + N + Y + + + Validate customer 360 BV + Success + Y + N + Y + + + + + diff --git a/integration-tests/tests/run-tests.hwf b/integration-tests/tests/run-tests.hwf index 32c0e68b..98c74625 100644 --- a/integration-tests/tests/run-tests.hwf +++ b/integration-tests/tests/run-tests.hwf @@ -321,6 +321,32 @@ limitations under the License. N + + ${PROJECT_HOME}/tests/multi-satellite-bv/update-customer-360-partitioned.hwf + N + N + N + + + N + N + Nothing + N + N + Y + + Y + + local + multi-satellite-bv-partitioned + Customer 360 SCD2 full rebuild split into 4 hash-key partitions + WORKFLOW + + 624 + 496 + N + + ${PROJECT_HOME}/tests/multi-source-hub/update-multi-source-hub.hwf N @@ -668,6 +694,13 @@ dbType == "" || dbType == "postgres"; multi-satellite-bv + multi-satellite-bv-partitioned + Y + N + Y + + + multi-satellite-bv-partitioned multi-satellite-bv-incremental Y N diff --git a/pom.xml b/pom.xml index 416bfa65..6ed37fe3 100644 --- a/pom.xml +++ b/pom.xml @@ -155,6 +155,11 @@ hop-transform-abort ${hop.version} + + org.apache.hop + hop-transform-addsequence + ${hop.version} + org.apache.hop hop-transform-analyticquery @@ -451,13 +456,6 @@ test - - org.apache.hop - hop-transform-addsequence - ${hop.version} - test - - org.junit.jupiter junit-jupiter diff --git a/src/assembly/assembly.xml b/src/assembly/assembly.xml index e42816f0..96afc789 100644 --- a/src/assembly/assembly.xml +++ b/src/assembly/assembly.xml @@ -135,6 +135,7 @@ org.apache.hop:hop-action-sql:jar org.apache.hop:hop-action-success:jar + org.apache.hop:hop-transform-addsequence:jar org.apache.hop:hop-transform-analyticquery:jar org.apache.hop:hop-transform-append:jar org.apache.hop:hop-transform-calculator:jar diff --git a/src/main/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilder.java b/src/main/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilder.java index ea16ef0d..d8715b06 100644 --- a/src/main/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilder.java +++ b/src/main/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilder.java @@ -137,6 +137,10 @@ public static String serializeModelStructure(BusinessVaultModel model) { if (table instanceof BvScd2Table scd2) { json.append(",\"buildMode\":") .append(DvAiContextBuilder.jsonString(String.valueOf(scd2.getBuildModeOrDefault()))); + json.append(",\"hashKeyPartitionCount\":") + .append( + DvAiContextBuilder.jsonString( + String.valueOf(scd2.getHashKeyPartitionCountOrDefault()))); json.append(",\"incrementalWatermarkField\":") .append(DvAiContextBuilder.jsonString(scd2.getIncrementalWatermarkField())); json.append(",\"satelliteConfigs\":["); diff --git a/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBusinessVaultGraph.java b/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBusinessVaultGraph.java index 52bb6231..aa504cf1 100644 --- a/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBusinessVaultGraph.java +++ b/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBusinessVaultGraph.java @@ -2224,6 +2224,17 @@ private void openBuildPipeline(IBvTable table, IVariables debugVariables) { ModelGeneratedArtifactOpenSupport.openGeneratedPipeline( hopGui, pipelineMeta, debugVariables); } + + List workflowMetas = + table.generateBuildWorkflows( + hopGui.getMetadataProvider(), debugVariables, model, dataVaultModel); + if (workflowMetas != null) { + for (WorkflowMeta workflowMeta : workflowMetas) { + if (workflowMeta != null) { + ModelGeneratedArtifactOpenSupport.openGeneratedWorkflow(workflowMeta); + } + } + } } catch (Exception e) { new ErrorDialog( hopGui.getShell(), diff --git a/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBvScd2TableDialog.java b/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBvScd2TableDialog.java index f297d676..be685dcf 100644 --- a/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBvScd2TableDialog.java +++ b/src/main/java/org/hopper/edw/datavault/hopgui/file/businessvault/HopGuiBvScd2TableDialog.java @@ -63,6 +63,7 @@ import org.hopper.edw.datavault.metadata.businessvault.BvScd2BuildMode; import org.hopper.edw.datavault.metadata.businessvault.BvScd2FieldMapping; import org.hopper.edw.datavault.metadata.businessvault.BvScd2FieldMappingDialogSupport; +import org.hopper.edw.datavault.metadata.businessvault.BvScd2HashPartitionCount; import org.hopper.edw.datavault.metadata.businessvault.BvScd2SatelliteConfig; import org.hopper.edw.datavault.metadata.businessvault.BvScd2Table; import org.hopper.edw.datavault.metadata.businessvault.IBvTable; @@ -84,6 +85,7 @@ public class HopGuiBvScd2TableDialog { private Text wTableName; private Combo wIncludeHashKey; private Combo wBuildMode; + private Combo wHashKeyPartitions; private Text wFunctionalTimestamp; private Text wIncrementalWatermark; private Text wValidFromField; @@ -260,17 +262,34 @@ private void addGeneralTab() { new FormDataBuilder().left(middle, 0).top(wIncludeHashKey, margin).right().result()); wBuildMode.addListener(SWT.Selection, e -> updateIncrementalFieldState()); + Label wlHashKeyPartitions = new Label(comp, SWT.RIGHT); + wlHashKeyPartitions.setText( + BaseMessages.getString(PKG, "HopGuiBvScd2TableDialog.HashKeyPartitions.Label")); + PropsUi.setLook(wlHashKeyPartitions); + wlHashKeyPartitions.setLayoutData( + new FormDataBuilder().left().top(wBuildMode, margin).right(middle, -margin).result()); + + wHashKeyPartitions = new Combo(comp, SWT.BORDER | SWT.READ_ONLY); + PropsUi.setLook(wHashKeyPartitions); + EnumDialogSupport.populateCombo(wHashKeyPartitions, BvScd2HashPartitionCount.class); + wHashKeyPartitions.setLayoutData( + new FormDataBuilder().left(middle, 0).top(wBuildMode, margin).right().result()); + Label wlFunctionalTimestamp = new Label(comp, SWT.RIGHT); wlFunctionalTimestamp.setText( BaseMessages.getString(PKG, "HopGuiBvScd2TableDialog.FunctionalTimestamp.Label")); PropsUi.setLook(wlFunctionalTimestamp); wlFunctionalTimestamp.setLayoutData( - new FormDataBuilder().left().top(wBuildMode, margin).right(middle, -margin).result()); + new FormDataBuilder() + .left() + .top(wHashKeyPartitions, margin) + .right(middle, -margin) + .result()); wFunctionalTimestamp = new Text(comp, SWT.SINGLE | SWT.LEFT | SWT.BORDER); PropsUi.setLook(wFunctionalTimestamp); wFunctionalTimestamp.setLayoutData( - new FormDataBuilder().left(middle, 0).top(wBuildMode, margin).right().result()); + new FormDataBuilder().left(middle, 0).top(wHashKeyPartitions, margin).right().result()); Label wlIncrementalWatermark = new Label(comp, SWT.RIGHT); wlIncrementalWatermark.setText( @@ -658,6 +677,7 @@ private void getData() { } wIncludeHashKey.select(input.isIncludeHashKey() ? 0 : 1); EnumDialogSupport.selectCombo(wBuildMode, input.getBuildModeOrDefault()); + EnumDialogSupport.selectCombo(wHashKeyPartitions, input.getHashKeyPartitionCountOrDefault()); if (!Utils.isEmpty(input.getFunctionalTimestampField())) { wFunctionalTimestamp.setText(input.getFunctionalTimestampField()); } @@ -818,6 +838,9 @@ private void applyWidgetsToTable(BvScd2Table target) { target.setBuildMode( EnumDialogSupport.readCombo( wBuildMode, BvScd2BuildMode.class, BvScd2BuildMode.FULL_REBUILD)); + target.setHashKeyPartitionCount( + EnumDialogSupport.readCombo( + wHashKeyPartitions, BvScd2HashPartitionCount.class, BvScd2HashPartitionCount.NONE)); target.setFunctionalTimestampField(wFunctionalTimestamp.getText()); target.setIncrementalWatermarkField(wIncrementalWatermark.getText()); target.setValidFromField(wValidFromField.getText()); @@ -873,6 +896,9 @@ private void updateIncrementalFieldState() { EnumDialogSupport.readCombo(wBuildMode, BvScd2BuildMode.class, BvScd2BuildMode.FULL_REBUILD) == BvScd2BuildMode.INCREMENTAL; wIncrementalWatermark.setEnabled(incremental); + if (wHashKeyPartitions != null && !wHashKeyPartitions.isDisposed()) { + wHashKeyPartitions.setEnabled(!incremental); + } } private void applyScd2FieldTooltips() { @@ -882,6 +908,8 @@ private void applyScd2FieldTooltips() { : new BusinessVaultConfiguration(); wBuildMode.setToolTipText( BaseMessages.getString(PKG, "HopGuiBvScd2TableDialog.BuildMode.Tooltip")); + wHashKeyPartitions.setToolTipText( + BaseMessages.getString(PKG, "HopGuiBvScd2TableDialog.HashKeyPartitions.Tooltip")); wIncrementalWatermark.setToolTipText( BaseMessages.getString(PKG, "HopGuiBvScd2TableDialog.IncrementalWatermark.Tooltip")); wFunctionalTimestamp.setToolTipText( diff --git a/src/main/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupport.java b/src/main/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupport.java index c5b95aae..b1f0dab5 100644 --- a/src/main/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupport.java +++ b/src/main/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupport.java @@ -58,6 +58,9 @@ public static final class TargetLoadContext { public final int locationX; public final int locationY; + /** Optional extra token in the staged CSV base (for example {@code ${PARTITION_NUMBER}}). */ + public final String stagingFileInfix; + public TargetLoadContext( IDvTargetLoadConfiguration config, IVariables variables, @@ -68,6 +71,30 @@ public TargetLoadContext( String modelName, int locationX, int locationY) { + this( + config, + variables, + targetDatabaseMeta, + targetDbName, + targetTableName, + pipelineName, + modelName, + locationX, + locationY, + null); + } + + public TargetLoadContext( + IDvTargetLoadConfiguration config, + IVariables variables, + DatabaseMeta targetDatabaseMeta, + String targetDbName, + String targetTableName, + String pipelineName, + String modelName, + int locationX, + int locationY, + String stagingFileInfix) { this.config = config; this.variables = variables; this.targetDatabaseMeta = targetDatabaseMeta; @@ -77,6 +104,7 @@ public TargetLoadContext( this.modelName = modelName; this.locationX = locationX; this.locationY = locationY; + this.stagingFileInfix = stagingFileInfix; } } @@ -141,8 +169,16 @@ public static TargetLoadResult addTargetLoad( */ public static String buildStagingFileBase( String stagingFolder, String pipelineName, boolean includeCopyVariable) { + return buildStagingFileBase(stagingFolder, pipelineName, includeCopyVariable, null); + } + + public static String buildStagingFileBase( + String stagingFolder, String pipelineName, boolean includeCopyVariable, String extraInfix) { String base = ensureTrailingSlash(stagingFolder) + stripStagedPipelineSequencePrefix(pipelineName); + if (!Utils.isEmpty(extraInfix)) { + base = base + "-" + extraInfix; + } if (includeCopyVariable) { base = base + "-" + STAGING_FILE_COPY_VARIABLE_PATTERN; } @@ -212,7 +248,8 @@ private static TargetLoadResult addStagingFileOutput( try { String stagingFolder = ctx.config.resolveBulkLoadStagingFolder(ctx.variables, ctx.modelName); - String fileBase = buildStagingFileBase(stagingFolder, ctx.pipelineName, true); + String fileBase = + buildStagingFileBase(stagingFolder, ctx.pipelineName, true, ctx.stagingFileInfix); String stagingFilePattern = fileBase + "." + STAGING_FILE_EXTENSION; TextFileOutputMeta textFileOutputMeta = new TextFileOutputMeta(); diff --git a/src/main/java/org/hopper/edw/datavault/metadata/DvUpdateWorkflowSupport.java b/src/main/java/org/hopper/edw/datavault/metadata/DvUpdateWorkflowSupport.java index 1d241619..9f342c8a 100644 --- a/src/main/java/org/hopper/edw/datavault/metadata/DvUpdateWorkflowSupport.java +++ b/src/main/java/org/hopper/edw/datavault/metadata/DvUpdateWorkflowSupport.java @@ -289,10 +289,8 @@ private static DvStagingLoadDescriptor inspectStagedPipeline( throws HopException { TransformMeta stagingTransform = findStagingTransform(pipelineMeta); if (stagingTransform == null) { - throw new HopException( - "Pipeline '" - + pipelineMeta.getName() - + "' does not contain a Text File Output staging transform"); + // Partition driver pipelines (and similar orchestrators) have no CSV writer. + return null; } String targetTableName = diff --git a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionCount.java b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionCount.java new file mode 100644 index 00000000..06f91f8c --- /dev/null +++ b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionCount.java @@ -0,0 +1,72 @@ +/* + * Copyright 2026 i-Bridge bv + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hopper.edw.datavault.metadata.businessvault; + +import lombok.Getter; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.metadata.api.IEnumHasCode; +import org.apache.hop.metadata.api.IEnumHasCodeAndDescription; + +/** + * Optional hash-key split for a full-rebuild Business Vault SCD2 load. {@link #NONE} keeps a single + * pipeline; 4 / 8 / 16 run sequential partition loads after a one-time truncate. + */ +@Getter +public enum BvScd2HashPartitionCount implements IEnumHasCodeAndDescription { + NONE( + "NONE", + BaseMessages.getString(BvScd2HashPartitionCount.class, "BvScd2HashPartitionCount.None"), + 1), + FOUR( + "4", + BaseMessages.getString(BvScd2HashPartitionCount.class, "BvScd2HashPartitionCount.Four"), + 4), + EIGHT( + "8", + BaseMessages.getString(BvScd2HashPartitionCount.class, "BvScd2HashPartitionCount.Eight"), + 8), + SIXTEEN( + "16", + BaseMessages.getString(BvScd2HashPartitionCount.class, "BvScd2HashPartitionCount.Sixteen"), + 16); + + private final String code; + private final String description; + private final int partitionCount; + + BvScd2HashPartitionCount(String code, String description, int partitionCount) { + this.code = code; + this.description = description; + this.partitionCount = partitionCount; + } + + public boolean isPartitioned() { + return partitionCount > 1; + } + + public static String[] getDescriptions() { + return IEnumHasCodeAndDescription.getDescriptions(BvScd2HashPartitionCount.class); + } + + public static BvScd2HashPartitionCount lookupDescription(String description) { + return IEnumHasCodeAndDescription.lookupDescription( + BvScd2HashPartitionCount.class, description, NONE); + } + + public static BvScd2HashPartitionCount lookupCode(String code) { + return IEnumHasCode.lookupCode(BvScd2HashPartitionCount.class, code, NONE); + } +} diff --git a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionSqlSupport.java b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionSqlSupport.java new file mode 100644 index 00000000..419a268a --- /dev/null +++ b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionSqlSupport.java @@ -0,0 +1,88 @@ +/* + * Copyright 2026 i-Bridge bv + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hopper.edw.datavault.metadata.businessvault; + +import org.apache.hop.core.database.DatabaseMeta; +import org.apache.hop.core.util.Utils; +import org.hopper.edw.datavault.metadata.HashKeyDataType; +import org.hopper.edw.datavault.metadata.businessvault.BvPitSnapshotSpineSupport.PitSqlDialect; + +/** + * Dialect-specific {@code first-byte(hash_key) % PARTITION_COUNT = PARTITION_NUMBER} predicates for + * partitioned SCD2 satellite reads. Values are Hop variables substituted by Table Input. + */ +public final class BvScd2HashPartitionSqlSupport { + + public static final String PARTITION_COUNT_VARIABLE = "PARTITION_COUNT"; + public static final String PARTITION_NUMBER_VARIABLE = "PARTITION_NUMBER"; + + static final String PARTITION_COUNT_REF = "${" + PARTITION_COUNT_VARIABLE + "}"; + static final String PARTITION_NUMBER_REF = "${" + PARTITION_NUMBER_VARIABLE + "}"; + + private BvScd2HashPartitionSqlSupport() {} + + /** + * Boolean SQL predicate using the quoted hash-key column. Empty column or metadata yields {@code + * null} (caller skips the filter). + */ + public static String buildPredicate( + DatabaseMeta databaseMeta, HashKeyDataType hashKeyDataType, String quotedHashKeyColumn) { + if (Utils.isEmpty(quotedHashKeyColumn)) { + return null; + } + HashKeyDataType type = hashKeyDataType != null ? hashKeyDataType : HashKeyDataType.HEX; + PitSqlDialect dialect = BvPitSnapshotSpineSupport.resolveDialect(databaseMeta); + String firstByte = firstByteExpression(dialect, type, quotedHashKeyColumn); + return firstByte + " % " + PARTITION_COUNT_REF + " = " + PARTITION_NUMBER_REF; + } + + static String firstByteExpression( + PitSqlDialect dialect, HashKeyDataType type, String quotedHashKeyColumn) { + return switch (type) { + case BINARY -> binaryFirstByte(dialect, quotedHashKeyColumn); + case STRING -> stringFirstByte(dialect, quotedHashKeyColumn); + default -> hexFirstByte(dialect, quotedHashKeyColumn); + }; + } + + private static String binaryFirstByte(PitSqlDialect dialect, String hk) { + return switch (dialect) { + case MYSQL, SINGLESTORE -> "CONV(HEX(SUBSTRING(" + hk + ", 1, 1)), 16, 10)"; + case SQL_SERVER -> "CONVERT(int, SUBSTRING(" + hk + ", 1, 1))"; + case SNOWFLAKE -> "TO_NUMBER(SUBSTR(HEX_ENCODE(" + hk + "), 1, 2), 'XX')"; + case POSTGRES -> "get_byte(" + hk + ", 0)"; + }; + } + + private static String hexFirstByte(PitSqlDialect dialect, String hk) { + return switch (dialect) { + case MYSQL, SINGLESTORE -> "CONV(SUBSTRING(" + hk + ", 1, 2), 16, 10)"; + case SQL_SERVER -> "CONVERT(int, CONVERT(varbinary(1), LEFT(" + hk + ", 2), 2))"; + case SNOWFLAKE -> "TO_NUMBER(SUBSTR(" + hk + ", 1, 2), 'XX')"; + case POSTGRES -> "('x' || substr(" + hk + ", 1, 2))::bit(8)::int"; + }; + } + + private static String stringFirstByte(PitSqlDialect dialect, String hk) { + return switch (dialect) { + case MYSQL, SINGLESTORE -> "SUBSTRING_INDEX(" + hk + ", '-', 1)"; + case SQL_SERVER -> + "TRY_CONVERT(int, LEFT(" + hk + " + '-', CHARINDEX('-', " + hk + " + '-') - 1))"; + case SNOWFLAKE -> "TRY_TO_NUMBER(SPLIT_PART(" + hk + ", '-', 1))"; + case POSTGRES -> "split_part(" + hk + ", '-', 1)::int"; + }; + } +} diff --git a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PartitionWorkflowSupport.java b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PartitionWorkflowSupport.java new file mode 100644 index 00000000..879af6b0 --- /dev/null +++ b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PartitionWorkflowSupport.java @@ -0,0 +1,475 @@ +/* + * Copyright 2026 i-Bridge bv + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hopper.edw.datavault.metadata.businessvault; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.apache.hop.core.database.DatabaseMeta; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.gui.Point; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.pipeline.PipelineHopMeta; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.TransformMeta; +import org.apache.hop.pipeline.transforms.addsequence.AddSequenceMeta; +import org.apache.hop.pipeline.transforms.pipelineexecutor.PipelineExecutorMeta; +import org.apache.hop.pipeline.transforms.pipelineexecutor.PipelineExecutorParameters; +import org.apache.hop.pipeline.transforms.rowgenerator.GeneratorField; +import org.apache.hop.pipeline.transforms.rowgenerator.RowGeneratorMeta; +import org.apache.hop.pipeline.transforms.textfileoutput.TextFileField; +import org.apache.hop.pipeline.transforms.textfileoutput.TextFileOutputMeta; +import org.apache.hop.workflow.WorkflowHopMeta; +import org.apache.hop.workflow.WorkflowMeta; +import org.apache.hop.workflow.action.ActionMeta; +import org.apache.hop.workflow.action.IAction; +import org.apache.hop.workflow.actions.sql.ActionSql; +import org.apache.hop.workflow.actions.start.ActionStart; +import org.hopper.edw.datavault.metadata.DataVaultConfiguration; +import org.hopper.edw.datavault.metadata.DvBulkLoadCommandSupport; +import org.hopper.edw.datavault.metadata.DvBulkLoadPluginSupport; +import org.hopper.edw.datavault.metadata.DvIntegerSettingValidationSupport; +import org.hopper.edw.datavault.metadata.DvMultiSourceUpdateWorkflowSupport; +import org.hopper.edw.datavault.metadata.DvMultiSourceUpdateWorkflowSupport.PipelineActionFactory; +import org.hopper.edw.datavault.metadata.DvStagingBulkLoadPipelineSupport; +import org.hopper.edw.datavault.metadata.DvTargetLoadMode; +import org.hopper.edw.datavault.metadata.DvTargetLoadSupport; +import org.hopper.edw.datavault.metadata.GeneratedPipelineMetadataSupport; +import org.hopper.edw.datavault.metadata.businessvault.BvScd2PipelineSupport.Scd2BuildContext; + +/** + * Driver pipeline and wrapper workflow for hash-key partitioned SCD2 full rebuilds: truncate once, + * then run the parameterized SCD2 pipeline once per partition. + */ +public final class BvScd2PartitionWorkflowSupport { + + public static final String GENERATE_PARTITIONS_TRANSFORM = "generate_partitions"; + public static final String PARTITION_NUMBER_TRANSFORM = "partition_number"; + public static final String EXECUTE_SCD2_TRANSFORM = "execute_scd2"; + public static final String DRIVER_NAME_SUFFIX = "-partitions"; + public static final String WORKFLOW_NAME_SUFFIX = "-partitioned"; + + private BvScd2PartitionWorkflowSupport() {} + + public static String driverPipelineName(String scd2PipelineName) { + String base = Utils.isEmpty(scd2PipelineName) ? "bv-scd2" : scd2PipelineName; + return base + DRIVER_NAME_SUFFIX; + } + + public static String workflowName(String scd2PipelineName) { + String base = Utils.isEmpty(scd2PipelineName) ? "bv-scd2" : scd2PipelineName; + return base + WORKFLOW_NAME_SUFFIX; + } + + public static PipelineMeta buildDriverPipeline(Scd2BuildContext ctx, PipelineMeta scd2Pipeline) + throws HopException { + if (ctx == null || ctx.scd2Table == null || !ctx.scd2Table.isHashKeyPartitioned()) { + throw new HopException( + "Hash-key partitioned SCD2 context is required for the driver pipeline"); + } + if (scd2Pipeline == null || Utils.isEmpty(scd2Pipeline.getName())) { + throw new HopException("SCD2 pipeline name is required for the partition driver"); + } + + int partitionCount = ctx.scd2Table.getHashKeyPartitionCountOrDefault().getPartitionCount(); + PipelineMeta driver = new PipelineMeta(); + driver.setName(driverPipelineName(scd2Pipeline.getName())); + GeneratedPipelineMetadataSupport.stampBvElementPipeline( + driver, ctx.bvModel, "scd2-partitions", ctx.scd2Table.getName(), ctx.bvTargetTableName); + + RowGeneratorMeta generateMeta = new RowGeneratorMeta(); + generateMeta.setNeverEnding(false); + generateMeta.setRowLimit(Integer.toString(partitionCount)); + generateMeta + .getFields() + .add( + new GeneratorField( + BvScd2HashPartitionSqlSupport.PARTITION_COUNT_VARIABLE, + "Integer", + null, + -1, + -1, + null, + null, + null, + Integer.toString(partitionCount), + false)); + TransformMeta generate = + new TransformMeta("RowGenerator", GENERATE_PARTITIONS_TRANSFORM, generateMeta); + generate.setLocation(new Point(160, 160)); + generate.setDistributes(false); + driver.addTransform(generate); + + AddSequenceMeta sequenceMeta = new AddSequenceMeta(); + sequenceMeta.setDefault(); + sequenceMeta.setDatabaseUsed(false); + sequenceMeta.setCounterUsed(true); + sequenceMeta.setValueName(BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_VARIABLE); + sequenceMeta.setStartAtByValue(0); + sequenceMeta.setIncrementByValue(1); + sequenceMeta.setMaxValueByValue(Math.max(0, partitionCount - 1)); + TransformMeta sequence = + new TransformMeta("Sequence", PARTITION_NUMBER_TRANSFORM, sequenceMeta); + sequence.setLocation(new Point(320, 160)); + driver.addTransform(sequence); + driver.addPipelineHop(new PipelineHopMeta(generate, sequence)); + + PipelineExecutorMeta executorMeta = new PipelineExecutorMeta(); + executorMeta.setDefault(); + executorMeta.setFilename(scd2Pipeline.getName() + PipelineMeta.PIPELINE_EXTENSION); + executorMeta.setFilenameInField(false); + executorMeta.setGroupSize("1"); + executorMeta.setInheritingAllVariables(true); + executorMeta + .getParameters() + .add(parameter(BvScd2HashPartitionSqlSupport.PARTITION_COUNT_VARIABLE)); + executorMeta + .getParameters() + .add(parameter(BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_VARIABLE)); + TransformMeta executor = + new TransformMeta("PipelineExecutor", EXECUTE_SCD2_TRANSFORM, executorMeta); + executor.setLocation(new Point(480, 160)); + executor.setCopiesString("1"); + driver.addTransform(executor); + driver.addPipelineHop(new PipelineHopMeta(sequence, executor)); + + return driver; + } + + public static WorkflowMeta buildWorkflow(Scd2BuildContext ctx, PipelineMeta driverPipeline) + throws HopException { + return buildWorkflow(ctx, driverPipeline, null, null); + } + + public static WorkflowMeta buildWorkflow( + Scd2BuildContext ctx, PipelineMeta driverPipeline, PipelineMeta scd2Pipeline) + throws HopException { + return buildWorkflow(ctx, driverPipeline, scd2Pipeline, null); + } + + public static WorkflowMeta buildWorkflow( + Scd2BuildContext ctx, PipelineMeta driverPipeline, PipelineActionFactory actionFactory) + throws HopException { + return buildWorkflow(ctx, driverPipeline, null, actionFactory); + } + + public static WorkflowMeta buildWorkflow( + Scd2BuildContext ctx, + PipelineMeta driverPipeline, + PipelineMeta scd2Pipeline, + PipelineActionFactory actionFactory) + throws HopException { + if (ctx == null || ctx.scd2Table == null || !ctx.scd2Table.isHashKeyPartitioned()) { + throw new HopException( + "Hash-key partitioned SCD2 context is required for the wrapper workflow"); + } + if (driverPipeline == null || Utils.isEmpty(driverPipeline.getName())) { + throw new HopException("Partition driver pipeline is required"); + } + + WorkflowMeta workflowMeta = new WorkflowMeta(); + workflowMeta.setName(workflowName(ctx.pipelineName)); + + ActionStart startAction = new ActionStart("Start"); + ActionMeta startMeta = new ActionMeta(startAction); + startMeta.setLocation(50, 50); + workflowMeta.addAction(startMeta); + + ActionSql sqlAction = new ActionSql("truncate_" + sanitize(ctx.bvTargetTableName)); + sqlAction.setConnection(ctx.targetDbName); + sqlAction.setSqlFromFile(false); + sqlAction.setSql( + buildTruncateSql(ctx.targetDatabaseMeta, ctx.variables, ctx.bvTargetTableName)); + sqlAction.setSendOneStatement(true); + sqlAction.setUseVariableSubstitution(true); + ActionMeta sqlMeta = new ActionMeta(sqlAction); + sqlMeta.setLocation(250, 50); + workflowMeta.addAction(sqlMeta); + workflowMeta.addWorkflowHop(new WorkflowHopMeta(startMeta, sqlMeta)); + + String placeholderFilename = driverPipeline.getName() + PipelineMeta.PIPELINE_EXTENSION; + ActionMeta pipelineAction = + DvMultiSourceUpdateWorkflowSupport.newPipelineActionMeta( + "run_" + sanitize(driverPipeline.getName()), placeholderFilename, null, actionFactory); + pipelineAction.setLocation(450, 50); + workflowMeta.addAction(pipelineAction); + workflowMeta.addWorkflowHop(new WorkflowHopMeta(sqlMeta, pipelineAction)); + + appendStagingBulkLoadActions(workflowMeta, pipelineAction, ctx, scd2Pipeline); + + return workflowMeta; + } + + /** + * After the partition driver finishes, bulk-load each {@code + * pipeline-${PARTITION_NUMBER}-${copy}.csv} shard. No-op unless target load mode is Staging file. + */ + static void appendStagingBulkLoadActions( + WorkflowMeta workflowMeta, + ActionMeta previousAction, + Scd2BuildContext ctx, + PipelineMeta scd2Pipeline) + throws HopException { + if (workflowMeta == null + || previousAction == null + || ctx == null + || ctx.bvConfig == null + || ctx.bvConfig.resolveTargetLoadMode() != DvTargetLoadMode.STAGING_FILE + || scd2Pipeline == null) { + return; + } + if (!DvBulkLoadPluginSupport.isModeAvailable( + ctx.targetDatabaseMeta, DvTargetLoadMode.STAGING_FILE)) { + throw new HopException( + "Staging file bulk loading is not available for the Business Vault target database of SCD2 table '" + + ctx.scd2Table.getName() + + "'"); + } + + TextFileOutputMeta textFileOutputMeta = findStagingFileOutput(scd2Pipeline); + if (textFileOutputMeta == null + || textFileOutputMeta.getFileSettings() == null + || Utils.isEmpty(textFileOutputMeta.getFileSettings().getFileName())) { + throw new HopException( + "Partitioned SCD2 pipeline '" + + scd2Pipeline.getName() + + "' is missing a Text File Output staging filename"); + } + + int partitionCount = ctx.scd2Table.getHashKeyPartitionCountOrDefault().getPartitionCount(); + int parallelCopies = + DvIntegerSettingValidationSupport.requirePositiveInteger( + ctx.bvConfig.resolveTargetTableParallelCopies(ctx.variables), + ctx.variables, + DataVaultConfiguration.DEFAULT_TARGET_TABLE_PARALLEL_COPIES, + "parallel copies"); + List columnNames = stagingColumnNames(textFileOutputMeta); + String fileBase = textFileOutputMeta.getFileSettings().getFileName(); + if (ctx.variables != null) { + fileBase = ctx.variables.resolve(fileBase); + } + + String bulkStagingFolder = + ctx.bvConfig.resolveBulkLoadStagingFolder( + ctx.variables, ctx.bvModel != null ? ctx.bvModel.getName() : "business-vault"); + int x = previousAction.getLocation() != null ? previousAction.getLocation().x + 200 : 650; + int y = previousAction.getLocation() != null ? previousAction.getLocation().y : 50; + ActionMeta previous = previousAction; + for (int partition = 0; partition < partitionCount; partition++) { + String partitionBase = resolvePartitionedStagingFileBase(fileBase, partition); + for (int copyIndex = 0; copyIndex < parallelCopies; copyIndex++) { + String stagedFilePath = + DvTargetLoadSupport.resolveStagedCsvFilePath(partitionBase, copyIndex); + ActionMeta bulkActionMeta = + newStagingBulkLoadAction( + ctx, bulkStagingFolder, columnNames, stagedFilePath, partition, copyIndex); + bulkActionMeta.setLocation(x, y); + workflowMeta.addAction(bulkActionMeta); + workflowMeta.addWorkflowHop(new WorkflowHopMeta(previous, bulkActionMeta)); + previous = bulkActionMeta; + x += 200; + } + } + } + + static String resolvePartitionedStagingFileBase(String stagingFileBase, int partitionNumber) { + if (Utils.isEmpty(stagingFileBase)) { + return stagingFileBase; + } + return stagingFileBase.replace( + BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_REF, Integer.toString(partitionNumber)); + } + + private static ActionMeta newStagingBulkLoadAction( + Scd2BuildContext ctx, + String bulkStagingFolder, + List columnNames, + String stagedFilePath, + int partition, + int copyIndex) + throws HopException { + String actionName = + "bulk_load_" + sanitize(ctx.bvTargetTableName) + "_p" + partition + "_" + copyIndex; + if (DvStagingBulkLoadPipelineSupport.usesClientSideBulkLoad(ctx.targetDatabaseMeta)) { + String bulkPipelinePath = + DvStagingBulkLoadPipelineSupport.buildAndStagePostgresBulkLoadPipeline( + bulkStagingFolder, + ctx.variables, + ctx.bvConfig, + ctx.targetDbName, + ctx.bvTargetTableName, + columnNames, + stagedFilePath, + copyIndex + partition * 100); + ActionMeta pipelineAction = + DvMultiSourceUpdateWorkflowSupport.newPipelineActionMeta( + actionName, bulkPipelinePath, null, null); + return pipelineAction; + } + IAction bulkAction = + DvBulkLoadCommandSupport.createStagingBulkLoadAction( + ctx.targetDatabaseMeta, + ctx.bvConfig, + ctx.variables, + ctx.targetDbName, + ctx.bvTargetTableName, + columnNames, + stagedFilePath, + copyIndex); + bulkAction.setName(actionName); + return new ActionMeta(bulkAction); + } + + private static TextFileOutputMeta findStagingFileOutput(PipelineMeta pipelineMeta) { + if (pipelineMeta == null || pipelineMeta.getTransforms() == null) { + return null; + } + for (TransformMeta transformMeta : pipelineMeta.getTransforms()) { + if (transformMeta != null + && transformMeta.getTransform() instanceof TextFileOutputMeta textFileOutputMeta) { + return textFileOutputMeta; + } + } + return null; + } + + private static List stagingColumnNames(TextFileOutputMeta textFileOutputMeta) { + List columnNames = new ArrayList<>(); + if (textFileOutputMeta.getOutputFields() == null) { + return columnNames; + } + for (TextFileField field : textFileOutputMeta.getOutputFields()) { + if (field != null && !Utils.isEmpty(field.getName())) { + columnNames.add(field.getName()); + } + } + return columnNames; + } + + static String buildTruncateSql( + DatabaseMeta databaseMeta, IVariables variables, String tableName) { + String quotedTable = + databaseMeta != null + ? databaseMeta.getQuotedSchemaTableCombination(variables, null, tableName) + : tableName; + if (databaseMeta != null) { + try { + String statement = databaseMeta.getTruncateTableStatement(variables, null, tableName); + if (!Utils.isEmpty(statement)) { + return statement; + } + } catch (Exception ignored) { + // Test stubs and unloaded database plugins have no iDatabase. + } + } + return "TRUNCATE TABLE " + quotedTable; + } + + /** + * Rewrites Pipeline Executor filenames after nested SCD2 pipelines have been staged to absolute + * paths. + */ + public static void applyStagedPipelineExecutorFilenames( + List pipelines, Map stagedPathByPipelineBasename) { + if (pipelines == null + || stagedPathByPipelineBasename == null + || stagedPathByPipelineBasename.isEmpty()) { + return; + } + for (PipelineMeta pipelineMeta : pipelines) { + if (pipelineMeta == null || pipelineMeta.getTransforms() == null) { + continue; + } + for (TransformMeta transformMeta : pipelineMeta.getTransforms()) { + if (transformMeta == null + || !(transformMeta.getTransform() instanceof PipelineExecutorMeta executorMeta)) { + continue; + } + String current = executorMeta.getFilename(); + if (Utils.isEmpty(current)) { + continue; + } + String staged = lookupStagedPath(stagedPathByPipelineBasename, basename(current)); + if (!Utils.isEmpty(staged)) { + executorMeta.setFilename(staged); + } + } + } + } + + public static void applyPipelineExecutorRunConfiguration( + List pipelines, String pipelineRunConfiguration) { + if (pipelines == null || Utils.isEmpty(pipelineRunConfiguration)) { + return; + } + for (PipelineMeta pipelineMeta : pipelines) { + if (pipelineMeta == null || pipelineMeta.getTransforms() == null) { + continue; + } + for (TransformMeta transformMeta : pipelineMeta.getTransforms()) { + if (transformMeta != null + && transformMeta.getTransform() instanceof PipelineExecutorMeta executorMeta) { + executorMeta.setRunConfigurationName(pipelineRunConfiguration); + } + } + } + } + + private static PipelineExecutorParameters parameter(String name) { + PipelineExecutorParameters mapping = new PipelineExecutorParameters(); + mapping.setVariable(name); + mapping.setField(name); + return mapping; + } + + private static String lookupStagedPath(Map map, String key) { + if (map.containsKey(key)) { + return map.get(key); + } + String withoutExt = stripPipelineExtension(key); + if (map.containsKey(withoutExt)) { + return map.get(withoutExt); + } + return map.get(withoutExt + PipelineMeta.PIPELINE_EXTENSION); + } + + private static String basename(String path) { + if (Utils.isEmpty(path)) { + return path; + } + int slash = Math.max(path.lastIndexOf('/'), path.lastIndexOf('\\')); + return slash >= 0 ? path.substring(slash + 1) : path; + } + + private static String stripPipelineExtension(String name) { + if (Utils.isEmpty(name)) { + return name; + } + String ext = PipelineMeta.PIPELINE_EXTENSION; + if (name.regionMatches(true, name.length() - ext.length(), ext, 0, ext.length())) { + return name.substring(0, name.length() - ext.length()); + } + return name; + } + + private static String sanitize(String name) { + if (Utils.isEmpty(name)) { + return "table"; + } + return name.replaceAll("[^A-Za-z0-9_\\-]", "_"); + } +} diff --git a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupport.java b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupport.java index 720f261e..1fb02e4f 100644 --- a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupport.java +++ b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupport.java @@ -75,6 +75,7 @@ import org.apache.hop.pipeline.transforms.update.UpdateKeyField; import org.apache.hop.pipeline.transforms.update.UpdateLookupField; import org.apache.hop.pipeline.transforms.update.UpdateMeta; +import org.apache.hop.workflow.WorkflowMeta; import org.hopper.edw.datavault.metadata.DataVaultConfiguration; import org.hopper.edw.datavault.metadata.DataVaultModel; import org.hopper.edw.datavault.metadata.DvHub; @@ -85,6 +86,7 @@ import org.hopper.edw.datavault.metadata.DvSpecialRecordSupport; import org.hopper.edw.datavault.metadata.DvSqlSupport; import org.hopper.edw.datavault.metadata.DvTableType; +import org.hopper.edw.datavault.metadata.DvTargetLoadMode; import org.hopper.edw.datavault.metadata.DvTargetLoadSupport; import org.hopper.edw.datavault.metadata.GeneratedPipelineMetadataSupport; import org.hopper.edw.datavault.metadata.HashAlgorithm; @@ -179,10 +181,12 @@ public static void validateTargetDatabases( } public static PipelineMeta generatePipeline(Scd2BuildContext ctx) throws HopException { - if (ctx.isMultiSatellite()) { - return generateMultiSatellitePipeline(ctx); - } - return generateSingleSatellitePipeline(ctx); + PipelineMeta pipelineMeta = + ctx.isMultiSatellite() + ? generateMultiSatellitePipeline(ctx) + : generateSingleSatellitePipeline(ctx); + applyPartitionParameters(pipelineMeta, ctx); + return pipelineMeta; } private static PipelineMeta generateSingleSatellitePipeline(Scd2BuildContext ctx) @@ -1049,6 +1053,7 @@ static String buildLegTableInputSql(Scd2BuildContext ctx, SatelliteLeg leg) { sql.append( ctx.sourceDatabaseMeta.getQuotedSchemaTableCombination( ctx.variables, null, leg.satelliteTableName)); + boolean hasWhere = false; if (ctx.scd2Table != null && ctx.scd2Table.isIncrementalBuild()) { String sourceTimestampField = ctx.isMultiSatellite() @@ -1057,6 +1062,19 @@ static String buildLegTableInputSql(Scd2BuildContext ctx, SatelliteLeg leg) { String sourceTimestampColumn = ctx.sourceDatabaseMeta.quoteField(sourceTimestampField); sql.append(" WHERE "); sql.append(buildIncrementalSatelliteFilterSql(sourceTimestampColumn)); + hasWhere = true; + } + if (isHashKeyPartitioned(ctx) && ctx.sourceDatabaseMeta != null) { + String quotedHashKey = ctx.sourceDatabaseMeta.quoteField(ctx.hashKeyFieldName); + String predicate = + BvScd2HashPartitionSqlSupport.buildPredicate( + ctx.sourceDatabaseMeta, + ctx.dvConfig != null ? ctx.dvConfig.resolveHashKeyDataType() : null, + quotedHashKey); + if (!Utils.isEmpty(predicate)) { + sql.append(hasWhere ? " AND " : " WHERE "); + sql.append(predicate); + } } sql.append(" ORDER BY "); if (ctx.includeHashKey) { @@ -1266,6 +1284,9 @@ private static TransformMeta addLegTableInput( TableInputMeta tableInputMeta = new TableInputMeta(); tableInputMeta.setConnection(ctx.sourceDbName); DvSqlSupport.assignDisplaySql(tableInputMeta, buildLegTableInputSql(ctx, leg)); + if (isHashKeyPartitioned(ctx)) { + tableInputMeta.setVariableReplacementActive(true); + } if (watermarkParam != null) { // Single ? for watermark — bound from param Constant info stream. tableInputMeta.setLookup(watermarkParam.getName()); @@ -1860,9 +1881,16 @@ private static TransformMeta addFullRebuildTableOutput( ? LOCATION_START.x + 9 * SPACING_WIDTH : LOCATION_START.x + 4 * SPACING_WIDTH; + boolean truncateTable = ctx.scd2Table == null || !ctx.scd2Table.isHashKeyPartitioned(); DvTargetLoadSupport.TargetLoadResult result = addScd2TargetLoad( - ctx, pipelineMeta, targetLayout, predecessor, tableOutputX, LOCATION_START.y, true); + ctx, + pipelineMeta, + targetLayout, + predecessor, + tableOutputX, + LOCATION_START.y, + truncateTable); return result.transformMeta; } @@ -2080,6 +2108,12 @@ private static DvTargetLoadSupport.TargetLoadResult addScd2TargetLoad( boolean truncateTable, Set excludeFields) throws HopException { + String stagingFileInfix = null; + if (isHashKeyPartitioned(ctx) + && ctx.bvConfig != null + && ctx.bvConfig.resolveTargetLoadMode() == DvTargetLoadMode.STAGING_FILE) { + stagingFileInfix = BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_REF; + } DvTargetLoadSupport.TargetLoadContext targetCtx = new DvTargetLoadSupport.TargetLoadContext( ctx.bvConfig, @@ -2090,7 +2124,8 @@ private static DvTargetLoadSupport.TargetLoadResult addScd2TargetLoad( ctx.pipelineName, ctx.bvModel.getName(), locationX, - locationY); + locationY, + stagingFileInfix); return DvTargetLoadSupport.addTargetLoad( targetCtx, pipelineMeta, targetLayout, predecessor, excludeFields, truncateTable); @@ -2110,7 +2145,13 @@ public static List generateBuildPipelines( if (ctx == null) { return List.of(); } - return List.of(generatePipeline(ctx)); + PipelineMeta scd2Pipeline = generatePipeline(ctx); + if (!isHashKeyPartitioned(ctx)) { + return List.of(scd2Pipeline); + } + PipelineMeta driverPipeline = + BvScd2PartitionWorkflowSupport.buildDriverPipeline(ctx, scd2Pipeline); + return List.of(scd2Pipeline, driverPipeline); } catch (Exception e) { throw new HopException( "Error generating SCD2 build pipeline for Business Vault table " + scd2Table.getName(), @@ -2118,6 +2159,56 @@ public static List generateBuildPipelines( } } + public static List generateBuildWorkflows( + IHopMetadataProvider metadataProvider, + IVariables variables, + BusinessVaultModel bvModel, + DataVaultModel dvModel, + BvScd2Table scd2Table) + throws HopException { + if (scd2Table == null || !scd2Table.isHashKeyPartitioned()) { + return List.of(); + } + try { + DbCache.clearAll(); + Scd2BuildContext ctx = + createContext(metadataProvider, variables, bvModel, dvModel, scd2Table); + if (ctx == null) { + return List.of(); + } + PipelineMeta scd2Pipeline = generatePipeline(ctx); + PipelineMeta driverPipeline = + BvScd2PartitionWorkflowSupport.buildDriverPipeline(ctx, scd2Pipeline); + return List.of( + BvScd2PartitionWorkflowSupport.buildWorkflow(ctx, driverPipeline, scd2Pipeline)); + } catch (Exception e) { + throw new HopException( + "Error generating SCD2 partition workflow for Business Vault table " + + scd2Table.getName(), + e); + } + } + + static boolean isHashKeyPartitioned(Scd2BuildContext ctx) { + return ctx != null && ctx.scd2Table != null && ctx.scd2Table.isHashKeyPartitioned(); + } + + static void applyPartitionParameters(PipelineMeta pipelineMeta, Scd2BuildContext ctx) + throws HopException { + if (pipelineMeta == null || !isHashKeyPartitioned(ctx)) { + return; + } + int count = ctx.scd2Table.getHashKeyPartitionCountOrDefault().getPartitionCount(); + pipelineMeta.addParameterDefinition( + BvScd2HashPartitionSqlSupport.PARTITION_COUNT_VARIABLE, + Integer.toString(count), + "Hash-key partition count for this SCD2 full rebuild"); + pipelineMeta.addParameterDefinition( + BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_VARIABLE, + "0", + "Zero-based hash-key partition number"); + } + /** Resolved inputs for one satellite branch in a generated SCD2 build pipeline. */ public static final class SatelliteLeg { final DvSatellite satellite; diff --git a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2Table.java b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2Table.java index 62b76bc7..a6adb423 100644 --- a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2Table.java +++ b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2Table.java @@ -21,6 +21,7 @@ import lombok.Setter; import org.apache.hop.core.CheckResult; import org.apache.hop.core.ICheckResult; +import org.apache.hop.core.database.DatabaseMeta; import org.apache.hop.core.exception.HopException; import org.apache.hop.core.row.IRowMeta; import org.apache.hop.core.util.Utils; @@ -29,9 +30,12 @@ import org.apache.hop.metadata.api.HopMetadataProperty; import org.apache.hop.metadata.api.IHopMetadataProvider; import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.workflow.WorkflowMeta; import org.hopper.edw.datavault.metadata.DataVaultConfiguration; import org.hopper.edw.datavault.metadata.DataVaultModel; +import org.hopper.edw.datavault.metadata.DvBulkLoadPluginSupport; import org.hopper.edw.datavault.metadata.DvTableType; +import org.hopper.edw.datavault.metadata.DvTargetLoadMode; /** Business Vault SCD2 table derived from one or more DV satellites. */ @Getter @@ -43,6 +47,9 @@ public class BvScd2Table extends BvTableBase { @HopMetadataProperty(storeWithCode = true) private BvScd2BuildMode buildMode = BvScd2BuildMode.FULL_REBUILD; + @HopMetadataProperty(storeWithCode = true) + private BvScd2HashPartitionCount hashKeyPartitionCount = BvScd2HashPartitionCount.NONE; + @HopMetadataProperty private String functionalTimestampField; @HopMetadataProperty private String incrementalWatermarkField; @@ -85,6 +92,14 @@ public boolean isIncrementalBuild() { return getBuildModeOrDefault() == BvScd2BuildMode.INCREMENTAL; } + public BvScd2HashPartitionCount getHashKeyPartitionCountOrDefault() { + return hashKeyPartitionCount != null ? hashKeyPartitionCount : BvScd2HashPartitionCount.NONE; + } + + public boolean isHashKeyPartitioned() { + return getHashKeyPartitionCountOrDefault().isPartitioned(); + } + public String resolveIncrementalWatermarkField( BusinessVaultConfiguration bvConfig, DataVaultConfiguration dvConfig, IVariables variables) { if (!Utils.isEmpty(incrementalWatermarkField)) { @@ -148,6 +163,36 @@ public void check( validateIncrementalMultiSatelliteHints(remarks, variables); } + if (isHashKeyPartitioned()) { + if (isIncrementalBuild()) { + remarks.add( + new CheckResult( + ICheckResult.TYPE_RESULT_ERROR, + BaseMessages.getString( + PKG, "BvScd2Table.CheckResult.HashKeyPartitionIncremental", getName()), + this)); + } + if (bvConfig.resolveTargetLoadMode() == DvTargetLoadMode.STAGING_FILE + && metadataProvider != null) { + try { + DatabaseMeta targetDatabase = + BvTargetDatabaseSupport.loadTargetDatabase(metadataProvider, bvConfig); + if (targetDatabase != null + && !DvBulkLoadPluginSupport.isModeAvailable( + targetDatabase, DvTargetLoadMode.STAGING_FILE)) { + remarks.add( + new CheckResult( + ICheckResult.TYPE_RESULT_ERROR, + BaseMessages.getString( + PKG, "BvScd2Table.CheckResult.HashKeyPartitionStagingFile", getName()), + this)); + } + } catch (HopException e) { + remarks.add(new CheckResult(ICheckResult.TYPE_RESULT_ERROR, e.getMessage(), this)); + } + } + } + if (dataVaultModel != null) { BvScd2FieldMappingValidationSupport.validate( remarks, this, bvConfig, dvConfig, dataVaultModel, variables); @@ -199,6 +244,17 @@ public List generateBuildPipelines( metadataProvider, variables, model, dataVaultModel, this); } + @Override + public List generateBuildWorkflows( + IHopMetadataProvider metadataProvider, + IVariables variables, + BusinessVaultModel model, + DataVaultModel dataVaultModel) + throws HopException { + return BvScd2PipelineSupport.generateBuildWorkflows( + metadataProvider, variables, model, dataVaultModel, this); + } + @Override public IRowMeta getTargetTableLayout( IHopMetadataProvider metadataProvider, diff --git a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvTableBase.java b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvTableBase.java index 51a5486e..73f49d6b 100644 --- a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvTableBase.java +++ b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/BvTableBase.java @@ -39,6 +39,7 @@ import org.apache.hop.metadata.api.IHopMetadata; import org.apache.hop.metadata.api.IHopMetadataProvider; import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.workflow.WorkflowMeta; import org.hopper.edw.datavault.metadata.DataVaultModel; import org.hopper.edw.datavault.metadata.DvConstraintDdlSupport; import org.hopper.edw.datavault.metadata.DvDdlSupport; @@ -154,6 +155,16 @@ public List generateBuildPipelines( return List.of(); } + @Override + public List generateBuildWorkflows( + IHopMetadataProvider metadataProvider, + IVariables variables, + BusinessVaultModel model, + DataVaultModel dataVaultModel) + throws HopException { + return List.of(); + } + @Override public List generateBuildDdl( IHopMetadataProvider metadataProvider, diff --git a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/IBvTable.java b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/IBvTable.java index 55f485f1..3aad6cce 100644 --- a/src/main/java/org/hopper/edw/datavault/metadata/businessvault/IBvTable.java +++ b/src/main/java/org/hopper/edw/datavault/metadata/businessvault/IBvTable.java @@ -29,6 +29,7 @@ import org.apache.hop.metadata.api.IHopMetadataObjectFactory; import org.apache.hop.metadata.api.IHopMetadataProvider; import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.workflow.WorkflowMeta; import org.hopper.edw.datavault.metadata.DataVaultModel; /** Common interface for Business Vault tables on a {@link BusinessVaultModel} canvas. */ @@ -65,6 +66,17 @@ List generateBuildPipelines( DataVaultModel dataVaultModel) throws HopException; + /** + * Wrapper workflows for a table's build pipelines (for example hash-key partitioned SCD2: + * truncate then sequential partition loads). Empty when the table is a single free pipeline. + */ + List generateBuildWorkflows( + IHopMetadataProvider metadataProvider, + IVariables variables, + BusinessVaultModel model, + DataVaultModel dataVaultModel) + throws HopException; + List generateBuildDdl( IHopMetadataProvider metadataProvider, IVariables variables, diff --git a/src/main/java/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/ActionBusinessVaultUpdate.java b/src/main/java/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/ActionBusinessVaultUpdate.java index 75225d42..28afa9b1 100644 --- a/src/main/java/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/ActionBusinessVaultUpdate.java +++ b/src/main/java/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/ActionBusinessVaultUpdate.java @@ -22,6 +22,7 @@ import java.util.HashSet; import java.util.List; import java.util.Locale; +import java.util.Map; import java.util.Set; import lombok.Getter; import lombok.Setter; @@ -51,6 +52,7 @@ import org.apache.hop.metadata.serializer.xml.XmlMetadataUtil; import org.apache.hop.pipeline.PipelineMeta; import org.apache.hop.pipeline.config.PipelineRunConfiguration; +import org.apache.hop.workflow.WorkflowMeta; import org.apache.hop.workflow.action.ActionBase; import org.apache.hop.workflow.action.IAction; import org.apache.hop.workflow.config.WorkflowRunConfiguration; @@ -64,7 +66,10 @@ import org.hopper.edw.datavault.metadata.DvIntegerSettingValidationSupport; import org.hopper.edw.datavault.metadata.DvLoadCycleSupport; import org.hopper.edw.datavault.metadata.DvModelBulkUpdateExecutionSupport; +import org.hopper.edw.datavault.metadata.DvMultiSourceUpdateWorkflowSupport; +import org.hopper.edw.datavault.metadata.DvPipelineOrchestratorSupport; import org.hopper.edw.datavault.metadata.DvTargetLoadMode; +import org.hopper.edw.datavault.metadata.DvUpdateWorkflowSupport; import org.hopper.edw.datavault.metadata.GeneratedPipelineMetadataConstants; import org.hopper.edw.datavault.metadata.ModelConfigurationResolver; import org.hopper.edw.datavault.metadata.businessvault.BusinessVaultConfiguration; @@ -72,6 +77,7 @@ import org.hopper.edw.datavault.metadata.businessvault.BusinessVaultModel; import org.hopper.edw.datavault.metadata.businessvault.BusinessVaultUpdateExecutionSupport; import org.hopper.edw.datavault.metadata.businessvault.BvGeneratedPipelineSupport; +import org.hopper.edw.datavault.metadata.businessvault.BvScd2PartitionWorkflowSupport; import org.hopper.edw.datavault.metadata.businessvault.BvTargetDatabaseSupport; import org.hopper.edw.datavault.metadata.businessvault.IBvTable; import org.hopper.edw.datavault.metadata.targettypemapping.TargetTypeMappingMeta; @@ -535,6 +541,7 @@ db, getVariables(), targetDatabase, ddl, createdInBatch)) { validatePipelineIntegerSettings(pipelineConfig); List allPipelineMetas = new ArrayList<>(); + List partitionedUnits = new ArrayList<>(); for (IBvTable table : tables) { if (table == null @@ -572,6 +579,7 @@ db, getVariables(), targetDatabase, ddl, createdInBatch)) { continue; } + List nestedPipelines = new ArrayList<>(); for (PipelineMeta pipelineMeta : pipelineMetas) { if (pipelineMeta == null) { logError( @@ -583,7 +591,7 @@ db, getVariables(), targetDatabase, ddl, createdInBatch)) { } pipelineMeta.lookupReferencesAfterLoading(); - allPipelineMetas.add(pipelineMeta); + nestedPipelines.add(pipelineMeta); String savedPipelineFile = BvGeneratedPipelineSupport.saveBeforeExecution( @@ -597,9 +605,34 @@ db, getVariables(), targetDatabase, ddl, createdInBatch)) { savedPipelineFile)); } } + + List workflowMetas = + table.generateBuildWorkflows(getMetadataProvider(), getVariables(), bvModel, dvModel); + if (workflowMetas != null && !workflowMetas.isEmpty()) { + for (WorkflowMeta workflowMeta : workflowMetas) { + if (workflowMeta == null) { + continue; + } + partitionedUnits.add( + new PartitionedScd2Unit(table.getName(), workflowMeta, nestedPipelines)); + String savedWorkflowFile = + BvGeneratedPipelineSupport.saveWorkflowBeforeExecution( + pipelineConfig, getVariables(), workflowMeta); + if (!Utils.isEmpty(savedWorkflowFile)) { + logBasic( + BaseMessages.getString( + PKG, + "ActionBusinessVaultUpdate.Log.SavedGeneratedWorkflow", + workflowMeta.getName(), + savedWorkflowFile)); + } + } + } else { + allPipelineMetas.addAll(nestedPipelines); + } } - if (!allPipelineMetas.isEmpty()) { + if (!partitionedUnits.isEmpty() || !allPipelineMetas.isEmpty()) { ResolvedExecutionMetrics executionMetrics = ExecutionMetricsProfileResolver.resolve( resolve(executionMetricsProfile), @@ -617,47 +650,93 @@ db, getVariables(), targetDatabase, ddl, createdInBatch)) { DvModelBulkUpdateExecutionSupport.ExecutionOutcome outcome; if (pipelineConfig.resolveTargetLoadMode() == DvTargetLoadMode.STAGING_FILE) { - DatabaseMeta targetDatabase = - BvTargetDatabaseSupport.loadTargetDatabase(getMetadataProvider(), pipelineConfig); - outcome = - DvModelBulkUpdateExecutionSupport.executeStagingFileUpdate( - result, - bvModel.getName(), - pipelineConfig, - allPipelineMetas, - realRunConfig, - realWorkflowRunConfig, - getLogLevel(), - pipelineStagingFolder, - targetDatabase, - pipelineConfig.getTargetDatabase(), - resolvedMetricsOutputFolder, - metricsPublishContext, - resolve(businessVaultModelFile), - getParentWorkflow(), - success, - totalErrors, - getVariables(), - this, - getMetadataProvider()); + if (!partitionedUnits.isEmpty()) { + DvUpdateWorkflowSupport.prepareBulkStagingFolder( + getVariables() + .resolve( + pipelineConfig.resolveBulkLoadStagingFolder( + getVariables(), bvModel.getName())), + getVariables()); + DvModelBulkUpdateExecutionSupport.ExecutionOutcome partitionedOutcome = + runPartitionedScd2Units( + result, + bvModel.getName(), + partitionedUnits, + realRunConfig, + realWorkflowRunConfig, + success, + totalErrors); + success = partitionedOutcome.success(); + totalErrors = partitionedOutcome.totalErrors(); + if (!success) { + return finishExecution(result, success, totalErrors, bvModel, dvModel); + } + } + if (allPipelineMetas.isEmpty()) { + outcome = new DvModelBulkUpdateExecutionSupport.ExecutionOutcome(success, totalErrors); + } else { + DatabaseMeta targetDatabase = + BvTargetDatabaseSupport.loadTargetDatabase(getMetadataProvider(), pipelineConfig); + outcome = + DvModelBulkUpdateExecutionSupport.executeStagingFileUpdate( + result, + bvModel.getName(), + pipelineConfig, + allPipelineMetas, + realRunConfig, + realWorkflowRunConfig, + getLogLevel(), + pipelineStagingFolder, + targetDatabase, + pipelineConfig.getTargetDatabase(), + resolvedMetricsOutputFolder, + metricsPublishContext, + resolve(businessVaultModelFile), + getParentWorkflow(), + success, + totalErrors, + getVariables(), + this, + getMetadataProvider()); + } } else { - outcome = - DvModelBulkUpdateExecutionSupport.executeOrchestratorUpdate( - result, - bvModel.getName(), - allPipelineMetas, - realRunConfig, - getLogLevel(), - pipelineStagingFolder, - parallelPipelineCopies, - resolvedMetricsOutputFolder, - metricsPublishContext, - success, - totalErrors, - getVariables(), - this, - getParentWorkflow(), - getMetadataProvider()); + if (!partitionedUnits.isEmpty()) { + DvModelBulkUpdateExecutionSupport.ExecutionOutcome partitionedOutcome = + runPartitionedScd2Units( + result, + bvModel.getName(), + partitionedUnits, + realRunConfig, + realWorkflowRunConfig, + success, + totalErrors); + success = partitionedOutcome.success(); + totalErrors = partitionedOutcome.totalErrors(); + if (!success) { + return finishExecution(result, success, totalErrors, bvModel, dvModel); + } + } + if (allPipelineMetas.isEmpty()) { + outcome = new DvModelBulkUpdateExecutionSupport.ExecutionOutcome(success, totalErrors); + } else { + outcome = + DvModelBulkUpdateExecutionSupport.executeOrchestratorUpdate( + result, + bvModel.getName(), + allPipelineMetas, + realRunConfig, + getLogLevel(), + pipelineStagingFolder, + parallelPipelineCopies, + resolvedMetricsOutputFolder, + metricsPublishContext, + success, + totalErrors, + getVariables(), + this, + getParentWorkflow(), + getMetadataProvider()); + } } success = outcome.success(); totalErrors = outcome.totalErrors(); @@ -673,6 +752,115 @@ db, getVariables(), targetDatabase, ddl, createdInBatch)) { } } + private record PartitionedScd2Unit( + String tableName, WorkflowMeta workflowMeta, List nestedPipelines) {} + + private DvModelBulkUpdateExecutionSupport.ExecutionOutcome runPartitionedScd2Units( + Result result, + String modelName, + List partitionedUnits, + String realRunConfig, + String realWorkflowRunConfig, + boolean success, + int totalErrors) + throws HopException { + String stagingRoot = + getVariables() + .resolve( + DvPipelineOrchestratorSupport.resolveStagingFolder( + pipelineStagingFolder, getVariables(), modelName)); + String workflowRunConfig = + !Utils.isEmpty(realWorkflowRunConfig) ? realWorkflowRunConfig : realRunConfig; + + for (PartitionedScd2Unit unit : partitionedUnits) { + if (unit == null || unit.workflowMeta() == null) { + continue; + } + List nested = + unit.nestedPipelines() != null ? unit.nestedPipelines() : List.of(); + String unitFolder = + DvPipelineOrchestratorSupport.resolveStagingFolder( + stagingRoot + "part-" + sanitizeModelName(unit.tableName()) + "/", + getVariables(), + unit.tableName()); + try { + DvPipelineOrchestratorSupport.prepareStagingFolder(unitFolder, getVariables()); + Map predictedPaths = predictedStagedPipelinePaths(unitFolder, nested); + BvScd2PartitionWorkflowSupport.applyStagedPipelineExecutorFilenames(nested, predictedPaths); + BvScd2PartitionWorkflowSupport.applyPipelineExecutorRunConfiguration(nested, realRunConfig); + DvPipelineOrchestratorSupport.stageNamedPipelines(unitFolder, getVariables(), nested); + Map stagedPaths = + DvMultiSourceUpdateWorkflowSupport.mapStagedPipelinePaths(nested); + DvMultiSourceUpdateWorkflowSupport.applyStagedPipelineFilenames( + unit.workflowMeta(), stagedPaths); + DvMultiSourceUpdateWorkflowSupport.applyPipelineRunConfiguration( + unit.workflowMeta(), realRunConfig); + DvPipelineOrchestratorSupport.stageWorkflow( + unitFolder, getVariables(), unit.workflowMeta()); + + logBasic( + BaseMessages.getString( + PKG, + "ActionBusinessVaultUpdate.Log.RunningPartitionedScd2", + unit.tableName(), + unit.workflowMeta().getName(), + nested.size())); + + Result workflowResult = + DvUpdateWorkflowSupport.runMasterWorkflow( + unit.workflowMeta(), + workflowRunConfig, + getLogLevel(), + this, + getVariables(), + getMetadataProvider()); + if (workflowResult == null) { + workflowResult = new Result(); + } + DvPipelineOrchestratorSupport.mergeResult(result, workflowResult); + if (workflowResult.getNrErrors() > 0 || !workflowResult.getResult()) { + logError( + BaseMessages.getString( + PKG, + "ActionBusinessVaultUpdate.Error.PartitionedScd2Failed", + unit.workflowMeta().getName(), + unit.tableName())); + success = false; + totalErrors += Math.max(1, (int) workflowResult.getNrErrors()); + return new DvModelBulkUpdateExecutionSupport.ExecutionOutcome(success, totalErrors); + } + } finally { + try { + DvPipelineOrchestratorSupport.cleanupStagingFolder(unitFolder, getVariables()); + } catch (HopException e) { + logError( + BaseMessages.getString( + PKG, "ActionBusinessVaultUpdate.Error.StagingCleanupFailed", unitFolder), + e); + } + } + } + return new DvModelBulkUpdateExecutionSupport.ExecutionOutcome(success, totalErrors); + } + + private static Map predictedStagedPipelinePaths( + String folder, List pipelines) { + Map map = new java.util.LinkedHashMap<>(); + if (Utils.isEmpty(folder) || pipelines == null) { + return map; + } + String prefix = folder.endsWith("/") || folder.endsWith("\\") ? folder : folder + "/"; + for (PipelineMeta pipelineMeta : pipelines) { + if (pipelineMeta == null || Utils.isEmpty(pipelineMeta.getName())) { + continue; + } + String path = prefix + pipelineMeta.getName() + PipelineMeta.PIPELINE_EXTENSION; + map.put(pipelineMeta.getName(), path); + map.put(pipelineMeta.getName() + PipelineMeta.PIPELINE_EXTENSION, path); + } + return map; + } + private Result finishExecution( Result result, boolean success, diff --git a/src/main/resources/org/hopper/edw/datavault/hopgui/file/businessvault/messages/messages_en_US.properties b/src/main/resources/org/hopper/edw/datavault/hopgui/file/businessvault/messages/messages_en_US.properties index af11bc7a..0de7e013 100644 --- a/src/main/resources/org/hopper/edw/datavault/hopgui/file/businessvault/messages/messages_en_US.properties +++ b/src/main/resources/org/hopper/edw/datavault/hopgui/file/businessvault/messages/messages_en_US.properties @@ -117,6 +117,8 @@ HopGuiBvScd2TableDialog.Description.Label=Description HopGuiBvScd2TableDialog.IncludeHashKey.Label=Include hash key HopGuiBvScd2TableDialog.BuildMode.Label=Build mode HopGuiBvScd2TableDialog.BuildMode.Tooltip=Full rebuild reads all satellite history and reloads the Business Vault table. Incremental reads only new satellite rows since the last target watermark and updates open SCD2 versions. +HopGuiBvScd2TableDialog.HashKeyPartitions.Label=Hash-key partitions +HopGuiBvScd2TableDialog.HashKeyPartitions.Tooltip=Split a Full rebuild into 4, 8, or 16 hash-key parts so each satellite read is a smaller ORDER BY. A workflow truncates the target once, then runs the SCD2 pipeline with '${PARTITION_COUNT}' and '${PARTITION_NUMBER}'. Table Output and Native bulk append; Staging file writes one CSV set per partition then bulk-loads. Incremental mode cannot be partitioned. HopGuiBvScd2TableDialog.FunctionalTimestamp.Label=Functional timestamp field HopGuiBvScd2TableDialog.IncrementalWatermark.Label=Incremental watermark field HopGuiBvScd2TableDialog.IncrementalWatermark.Tooltip=Leave empty to use the resolved functional timestamp field. When build mode is Incremental, satellite reads filter on rows newer than MAX(watermark field) in the target table. diff --git a/src/main/resources/org/hopper/edw/datavault/metadata/businessvault/messages/messages_en_US.properties b/src/main/resources/org/hopper/edw/datavault/metadata/businessvault/messages/messages_en_US.properties index e146fa62..fb18cfd0 100644 --- a/src/main/resources/org/hopper/edw/datavault/metadata/businessvault/messages/messages_en_US.properties +++ b/src/main/resources/org/hopper/edw/datavault/metadata/businessvault/messages/messages_en_US.properties @@ -108,11 +108,18 @@ BvTableBase.CheckResult.MissingTableName=Physical table name is required for Bus BvScd2BuildMode.FullRebuild=Full rebuild BvScd2BuildMode.Incremental=Incremental +BvScd2HashPartitionCount.None=None +BvScd2HashPartitionCount.Four=4 partitions +BvScd2HashPartitionCount.Eight=8 partitions +BvScd2HashPartitionCount.Sixteen=16 partitions + BvScd2Table.CheckResult.MissingSatelliteDerivative=SCD2 table ''{0}'' must reference at least one Data Vault satellite. BvScd2Table.CheckResult.MissingFunctionalTimestamp=SCD2 table ''{0}'' requires a functional timestamp field or a model load-date fallback. BvScd2Table.CheckResult.MissingIncrementalWatermark=SCD2 table ''{0}'' requires an incremental watermark field or a functional timestamp fallback when build mode is Incremental. BvScd2Table.CheckResult.MissingOpenEndSentinel=SCD2 table ''{0}'' requires an open-end sentinel in the Business Vault model configuration when build mode is Incremental. BvScd2Table.CheckResult.IncrementalMultiSatelliteSourceIndicators=SCD2 table ''{0}'' uses incremental multi-satellite merge; configure a source indicator value for every satellite on the Satellite settings tab. +BvScd2Table.CheckResult.HashKeyPartitionIncremental=SCD2 table ''{0}'' cannot combine hash-key partitions with Incremental build mode. Use Full rebuild. +BvScd2Table.CheckResult.HashKeyPartitionStagingFile=SCD2 table ''{0}'' uses hash-key partitions with Staging file load mode, but the Business Vault target database has no staged bulk-load action. Use Table Output or Native bulk, or install the matching bulk-load plugin. BvScd2PipelineSupport.CheckResult.MissingDvTargetDatabase=SCD2 table ''{0}'' requires a Data Vault target database in the linked .hdv model configuration. BvScd2PipelineSupport.CheckResult.MissingBvTargetDatabase=SCD2 table ''{0}'' requires a Business Vault target database in the model configuration. diff --git a/src/main/resources/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/messages/messages_en_US.properties b/src/main/resources/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/messages/messages_en_US.properties index 28fd85d3..6ec441e0 100644 --- a/src/main/resources/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/messages/messages_en_US.properties +++ b/src/main/resources/org/hopper/edw/datavault/workflow/actions/businessvaultupdate/messages/messages_en_US.properties @@ -85,6 +85,9 @@ ActionBusinessVaultUpdate.Log.SkippingDataUpdate=Skipping data load pipelines (d ActionBusinessVaultUpdate.Log.GeneratingForTable=Generating update pipeline for Business Vault table: {0} ({1}) ActionBusinessVaultUpdate.Log.SkippingUnsupportedTableType=Skipping Business Vault table ''{0}'' (type {1} is not supported in this release). ActionBusinessVaultUpdate.Log.SavedGeneratedPipeline=Saved generated pipeline ''{0}'' to {1} +ActionBusinessVaultUpdate.Log.SavedGeneratedWorkflow=Saved generated workflow ''{0}'' to {1} +ActionBusinessVaultUpdate.Log.RunningPartitionedScd2=Running hash-key partitioned SCD2 workflow ''{1}'' for table ''{0}'' ({2} nested pipeline(s)) +ActionBusinessVaultUpdate.Error.PartitionedScd2Failed=Hash-key partitioned SCD2 workflow ''{0}'' for table ''{1}'' completed with errors. ActionBusinessVaultUpdate.Log.StagingPipelines=Staging {1} generated update pipeline(s) to {0} ActionBusinessVaultUpdate.Log.ParallelCopies=Running staged pipelines with {0} parallel Pipeline Executor copy/copies ActionBusinessVaultUpdate.Log.RunningOrchestrator=Running orchestrator pipeline ''{0}'' using run configuration: {1} diff --git a/src/test/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilderTest.java b/src/test/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilderTest.java index 484c4475..cf12bfce 100644 --- a/src/test/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilderTest.java +++ b/src/test/java/org/hopper/edw/datavault/ai/businessvault/BvAiContextBuilderTest.java @@ -37,5 +37,8 @@ void serializeModelSummaryIncludesTargetLoadMode() { assertTrue(summary.contains("\"targetLoadMode\":\"STAGING_FILE\"")); assertTrue(summary.contains("\"name\":\"BV_TEST\"")); + assertTrue( + BvAiContextBuilder.serializeModelStructure(model) + .contains("\"hashKeyPartitionCount\":\"NONE\"")); } } diff --git a/src/test/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupportTest.java b/src/test/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupportTest.java index 89dc28d2..6e031647 100644 --- a/src/test/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupportTest.java +++ b/src/test/java/org/hopper/edw/datavault/metadata/DvTargetLoadSupportTest.java @@ -186,6 +186,14 @@ void stagingFileOutputClearsLengthAndPrecisionForWideFields() throws Exception { assertEquals(-1, textFileOutputMeta.getOutputFields().get(0).getPrecision()); } + @Test + void buildStagingFileBaseAppendsExtraInfixBeforeCopyVariable() { + assertEquals( + "/tmp/dv2/bulk/bv-scd2-customer-${PARTITION_NUMBER}-${Internal.Transform.CopyNr}", + DvTargetLoadSupport.buildStagingFileBase( + "/tmp/dv2/bulk/", "bv-scd2-customer", true, "${PARTITION_NUMBER}")); + } + @Test void buildStagingFileBaseStripsSequencedPipelinePrefix() { assertEquals( diff --git a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionSqlSupportTest.java b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionSqlSupportTest.java new file mode 100644 index 00000000..6a054ec3 --- /dev/null +++ b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2HashPartitionSqlSupportTest.java @@ -0,0 +1,114 @@ +/* + * Copyright 2026 i-Bridge bv + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hopper.edw.datavault.metadata.businessvault; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.hop.core.database.DatabaseMeta; +import org.hopper.edw.datavault.metadata.HashKeyDataType; +import org.junit.jupiter.api.Test; + +class BvScd2HashPartitionSqlSupportTest { + + @Test + void emptyColumnReturnsNull() { + assertNull( + BvScd2HashPartitionSqlSupport.buildPredicate( + database("POSTGRESQL"), HashKeyDataType.HEX, null)); + } + + @Test + void postgresHexUsesBitCastOfFirstTwoChars() { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database("POSTGRESQL"), HashKeyDataType.HEX, "customer_hk"); + assertEquals( + "('x' || substr(customer_hk, 1, 2))::bit(8)::int % ${PARTITION_COUNT} = ${PARTITION_NUMBER}", + sql); + } + + @Test + void postgresBinaryUsesGetByte() { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database("POSTGRESQL"), HashKeyDataType.BINARY, "customer_hk"); + assertEquals("get_byte(customer_hk, 0) % ${PARTITION_COUNT} = ${PARTITION_NUMBER}", sql); + } + + @Test + void postgresStringUsesSplitPart() { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database("POSTGRESQL"), HashKeyDataType.STRING, "customer_hk"); + assertEquals( + "split_part(customer_hk, '-', 1)::int % ${PARTITION_COUNT} = ${PARTITION_NUMBER}", sql); + } + + @Test + void mysqlAndSinglestoreBinaryMatchIssueSample() { + for (String pluginId : new String[] {"MYSQL", "SINGLESTORE"}) { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database(pluginId), HashKeyDataType.BINARY, "hash_key_field"); + assertEquals( + "CONV(HEX(SUBSTRING(hash_key_field, 1, 1)), 16, 10) % ${PARTITION_COUNT} = ${PARTITION_NUMBER}", + sql, pluginId); + } + } + + @Test + void mysqlHexUsesTwoCharacters() { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database("MYSQL"), HashKeyDataType.HEX, "customer_hk"); + assertEquals( + "CONV(SUBSTRING(customer_hk, 1, 2), 16, 10) % ${PARTITION_COUNT} = ${PARTITION_NUMBER}", + sql); + } + + @Test + void sqlServerNativeUsesMssqlDialect() { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database("MSSQLNATIVE"), HashKeyDataType.HEX, "customer_hk"); + assertTrue(sql.contains("CONVERT(varbinary(1), LEFT(customer_hk, 2), 2)")); + assertTrue(sql.contains("${PARTITION_COUNT}")); + } + + @Test + void snowflakeHexUsesToNumber() { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database("SNOWFLAKE"), HashKeyDataType.HEX, "\"CUSTOMER_HK\""); + assertEquals( + "TO_NUMBER(SUBSTR(\"CUSTOMER_HK\", 1, 2), 'XX') % ${PARTITION_COUNT} = ${PARTITION_NUMBER}", + sql); + } + + @Test + void unknownPluginIdDefaultsToPostgres() { + String sql = + BvScd2HashPartitionSqlSupport.buildPredicate( + database("ORACLE"), HashKeyDataType.BINARY, "hk"); + assertEquals("get_byte(hk, 0) % ${PARTITION_COUNT} = ${PARTITION_NUMBER}", sql); + } + + private static DatabaseMeta database(String pluginId) { + return new TestDatabaseMeta("Vault", pluginId); + } +} diff --git a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PartitionWorkflowSupportTest.java b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PartitionWorkflowSupportTest.java new file mode 100644 index 00000000..ae49cfc1 --- /dev/null +++ b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PartitionWorkflowSupportTest.java @@ -0,0 +1,200 @@ +/* + * Copyright 2026 i-Bridge bv + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.hopper.edw.datavault.metadata.businessvault; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.List; +import org.apache.hop.core.HopEnvironment; +import org.apache.hop.core.database.DatabaseMeta; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.variables.Variables; +import org.apache.hop.core.xml.XmlHandler; +import org.apache.hop.metadata.serializer.xml.XmlMetadataUtil; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.TransformMeta; +import org.apache.hop.pipeline.transforms.addsequence.AddSequenceMeta; +import org.apache.hop.pipeline.transforms.pipelineexecutor.PipelineExecutorMeta; +import org.apache.hop.pipeline.transforms.rowgenerator.RowGeneratorMeta; +import org.apache.hop.workflow.WorkflowMeta; +import org.apache.hop.workflow.action.ActionBase; +import org.apache.hop.workflow.action.ActionMeta; +import org.apache.hop.workflow.action.IAction; +import org.apache.hop.workflow.actions.sql.ActionSql; +import org.hopper.edw.datavault.metadata.DataVaultModel; +import org.hopper.edw.datavault.metadata.DvMultiSourceUpdateWorkflowSupport; +import org.hopper.edw.datavault.metadata.DvSatellite; +import org.hopper.edw.datavault.metadata.DvTableType; +import org.hopper.edw.datavault.metadata.businessvault.BvScd2PipelineSupport.Scd2BuildContext; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.w3c.dom.Document; +import org.w3c.dom.Node; + +class BvScd2PartitionWorkflowSupportTest { + + @BeforeAll + static void initHop() throws HopException { + HopEnvironment.init(); + } + + @Test + void driverPipelineGeneratesPartitionRowsAndExecutesScd2() throws Exception { + Scd2BuildContext ctx = partitionedContext(); + PipelineMeta scd2 = BvScd2PipelineSupport.generatePipeline(ctx); + PipelineMeta driver = BvScd2PartitionWorkflowSupport.buildDriverPipeline(ctx, scd2); + + assertEquals(scd2.getName() + "-partitions", driver.getName()); + TransformMeta generate = + driver.findTransform(BvScd2PartitionWorkflowSupport.GENERATE_PARTITIONS_TRANSFORM); + RowGeneratorMeta generateMeta = (RowGeneratorMeta) generate.getTransform(); + assertEquals("4", generateMeta.getRowLimit()); + assertEquals( + BvScd2HashPartitionSqlSupport.PARTITION_COUNT_VARIABLE, + generateMeta.getFields().get(0).getName()); + + TransformMeta sequence = + driver.findTransform(BvScd2PartitionWorkflowSupport.PARTITION_NUMBER_TRANSFORM); + AddSequenceMeta sequenceMeta = (AddSequenceMeta) sequence.getTransform(); + assertEquals( + BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_VARIABLE, sequenceMeta.getValueName()); + assertEquals("0", sequenceMeta.getStartAt()); + assertEquals("1", sequenceMeta.getIncrementBy()); + + TransformMeta executor = + driver.findTransform(BvScd2PartitionWorkflowSupport.EXECUTE_SCD2_TRANSFORM); + PipelineExecutorMeta executorMeta = (PipelineExecutorMeta) executor.getTransform(); + assertEquals(scd2.getName() + PipelineMeta.PIPELINE_EXTENSION, executorMeta.getFilename()); + assertEquals("1", executorMeta.getGroupSize()); + assertEquals(2, executorMeta.getParameters().size()); + assertEquals( + BvScd2HashPartitionSqlSupport.PARTITION_COUNT_VARIABLE, + executorMeta.getParameters().get(0).getVariable()); + assertEquals( + BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_VARIABLE, + executorMeta.getParameters().get(1).getField()); + } + + @Test + void wrapperWorkflowTruncatesThenRunsDriver() throws Exception { + Scd2BuildContext ctx = partitionedContext(); + PipelineMeta scd2 = BvScd2PipelineSupport.generatePipeline(ctx); + PipelineMeta driver = BvScd2PartitionWorkflowSupport.buildDriverPipeline(ctx, scd2); + WorkflowMeta workflow = + BvScd2PartitionWorkflowSupport.buildWorkflow(ctx, driver, StubPipelineAction::new); + + assertEquals(ctx.pipelineName + "-partitioned", workflow.getName()); + List actions = workflow.getActions(); + assertTrue(actions.size() >= 3); + + ActionSql sqlAction = + (ActionSql) + actions.stream() + .map(ActionMeta::getAction) + .filter(a -> a instanceof ActionSql) + .findFirst() + .orElseThrow(); + assertEquals("Vault", sqlAction.getConnection()); + assertTrue(sqlAction.getSql().contains("bv_customer_scd2"), sqlAction.getSql()); + assertTrue(sqlAction.getSql().toUpperCase().contains("TRUNCATE"), sqlAction.getSql()); + assertTrue(sqlAction.isSendOneStatement()); + assertTrue(sqlAction.isUseVariableSubstitution()); + } + + @Test + void resolvePartitionedStagingFileBaseSubstitutesPartitionNumber() { + assertEquals( + "/tmp/dv2/bulk/bv-scd2-sat-2-${Internal.Transform.CopyNr}", + BvScd2PartitionWorkflowSupport.resolvePartitionedStagingFileBase( + "/tmp/dv2/bulk/bv-scd2-sat-${PARTITION_NUMBER}-${Internal.Transform.CopyNr}", 2)); + } + + private static Scd2BuildContext partitionedContext() throws Exception { + DataVaultModel dvModel = loadVault1Model(); + DvSatellite satellite = (DvSatellite) dvModel.findTable("sat_customer"); + DatabaseMeta databaseMeta = new TestDatabaseMeta("Vault", "POSTGRESQL"); + + BvScd2Table scd2Table = new BvScd2Table(); + scd2Table.setName("bv_customer_scd2"); + scd2Table.setTableName("bv_customer_scd2"); + scd2Table.setFunctionalTimestampField("x_load_ts"); + scd2Table.setHashKeyPartitionCount(BvScd2HashPartitionCount.FOUR); + scd2Table.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); + + BusinessVaultModel bvModel = new BusinessVaultModel(); + bvModel.getConfigurationOrDefault().setTargetDatabase("Vault"); + + return new Scd2BuildContext( + scd2Table, + satellite, + bvModel, + dvModel, + bvModel.getConfigurationOrDefault(), + dvModel.getConfigurationOrDefault(), + null, + new Variables(), + databaseMeta, + "Vault", + databaseMeta, + "Vault", + "sat_customer", + "bv_customer_scd2", + "bv-scd2-bv_customer_scd2-sat_customer", + "customer_hk", + null, + BvScd2PipelineSupport.resolveAttributeFieldNames(satellite), + "x_load_ts", + "valid_from", + "valid_to", + "x_record_source", + BusinessVaultConfiguration.DEFAULT_OPEN_START_SENTINEL, + BusinessVaultConfiguration.DEFAULT_OPEN_END_SENTINEL, + true); + } + + private static DataVaultModel loadVault1Model() throws Exception { + Path path = Path.of("integration-tests/tests/basic/vault1.hdv").toAbsolutePath().normalize(); + Document document = XmlHandler.loadXmlFile(path.toFile()); + Node rootNode = XmlHandler.getSubNode(document, "data-vault-model"); + DataVaultModel model = new DataVaultModel(); + XmlMetadataUtil.deSerializeFromXml(rootNode, DataVaultModel.class, model, null); + return model; + } + + public static final class StubPipelineAction extends ActionBase implements IAction { + private String filename; + + StubPipelineAction(String name) { + super(name, ""); + setPluginId(DvMultiSourceUpdateWorkflowSupport.PIPELINE_ACTION_ID); + } + + public String getFilename() { + return filename; + } + + public void setFilename(String filename) { + this.filename = filename; + } + + @Override + public org.apache.hop.core.Result execute(org.apache.hop.core.Result prevResult, int nr) { + return prevResult != null ? prevResult : new org.apache.hop.core.Result(); + } + } +} diff --git a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupportTest.java b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupportTest.java index 6263458c..97139bc4 100644 --- a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupportTest.java +++ b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2PipelineSupportTest.java @@ -45,13 +45,17 @@ import org.apache.hop.pipeline.transforms.sort.SortRowsMeta; import org.apache.hop.pipeline.transforms.tableinput.TableInputMeta; import org.apache.hop.pipeline.transforms.tableoutput.TableOutputMeta; +import org.apache.hop.pipeline.transforms.textfileoutput.TextFileOutputMeta; import org.apache.hop.pipeline.transforms.update.UpdateMeta; import org.hopper.edw.datavault.metadata.DataVaultConfiguration; import org.hopper.edw.datavault.metadata.DataVaultModel; +import org.hopper.edw.datavault.metadata.DvBulkLoadPluginSupport; import org.hopper.edw.datavault.metadata.DvSatellite; import org.hopper.edw.datavault.metadata.DvTableType; +import org.hopper.edw.datavault.metadata.DvTargetLoadMode; import org.hopper.edw.datavault.metadata.GeneratedPipelineMetadataConstants; import org.hopper.edw.datavault.metadata.GeneratedPipelineMetadataSupport; +import org.hopper.edw.datavault.metadata.HashKeyDataType; import org.hopper.edw.datavault.metadata.SatelliteAttribute; import org.hopper.edw.datavault.metadata.businessvault.BvScd2PipelineSupport.SatelliteLeg; import org.hopper.edw.datavault.metadata.businessvault.BvScd2PipelineSupport.Scd2BuildContext; @@ -755,6 +759,206 @@ void generatedPipelineContainsExpectedTransformChain() throws Exception { assertEquals("bv_customer_scd2", tableOutputMeta.getTableName()); } + @Test + void partitionedSatelliteSqlFiltersOnHashKeyModAndDoesNotTruncate() throws Exception { + DataVaultModel dvModel = loadVault1Model(); + dvModel.getConfigurationOrDefault().setHashKeyDataType(HashKeyDataType.HEX.name()); + DvSatellite satellite = (DvSatellite) dvModel.findTable("sat_customer"); + DatabaseMeta databaseMeta = new TestDatabaseMeta("Vault", "POSTGRESQL"); + + BvScd2Table scd2Table = new BvScd2Table(); + scd2Table.setName("bv_customer_scd2"); + scd2Table.setTableName("bv_customer_scd2"); + scd2Table.setFunctionalTimestampField("x_load_ts"); + scd2Table.setHashKeyPartitionCount(BvScd2HashPartitionCount.FOUR); + scd2Table.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); + + BusinessVaultModel bvModel = new BusinessVaultModel(); + bvModel.getConfigurationOrDefault().setTargetDatabase("Vault"); + + Scd2BuildContext ctx = + new Scd2BuildContext( + scd2Table, + satellite, + bvModel, + dvModel, + bvModel.getConfigurationOrDefault(), + dvModel.getConfigurationOrDefault(), + null, + new Variables(), + databaseMeta, + "Vault", + databaseMeta, + "Vault", + "sat_customer", + "bv_customer_scd2", + "bv-scd2-bv_customer_scd2-sat_customer", + "customer_hk", + null, + BvScd2PipelineSupport.resolveAttributeFieldNames(satellite), + "x_load_ts", + "valid_from", + "valid_to", + "x_record_source", + BusinessVaultConfiguration.DEFAULT_OPEN_START_SENTINEL, + BusinessVaultConfiguration.DEFAULT_OPEN_END_SENTINEL, + true); + + String sql = BvScd2PipelineSupport.buildSatelliteTableInputSql(ctx); + assertTrue(sql.contains(" WHERE ")); + assertTrue(sql.contains("substr(customer_hk, 1, 2)")); + assertTrue(sql.contains("${PARTITION_COUNT}")); + assertTrue(sql.contains("${PARTITION_NUMBER}")); + assertTrue(sql.indexOf(" WHERE ") < sql.indexOf(" ORDER BY ")); + + PipelineMeta pipelineMeta = BvScd2PipelineSupport.generatePipeline(ctx); + TableInputMeta tableInputMeta = + (TableInputMeta) + pipelineMeta.getTransforms().stream() + .map(TransformMeta::getTransform) + .filter(t -> t instanceof TableInputMeta) + .findFirst() + .orElseThrow(); + assertTrue(tableInputMeta.isVariableReplacementActive()); + + TableOutputMeta tableOutputMeta = + (TableOutputMeta) + pipelineMeta.getTransforms().stream() + .map(TransformMeta::getTransform) + .filter(t -> t instanceof TableOutputMeta) + .findFirst() + .orElseThrow(); + assertFalse(tableOutputMeta.isTruncateTable()); + + List parameters = List.of(pipelineMeta.listParameters()); + assertTrue(parameters.contains(BvScd2HashPartitionSqlSupport.PARTITION_COUNT_VARIABLE)); + assertTrue(parameters.contains(BvScd2HashPartitionSqlSupport.PARTITION_NUMBER_VARIABLE)); + assertEquals( + "4", + pipelineMeta.getParameterDefault(BvScd2HashPartitionSqlSupport.PARTITION_COUNT_VARIABLE)); + } + + @Test + void partitionedStagingFileNameIncludesPartitionNumber() throws Exception { + DataVaultModel dvModel = loadVault1Model(); + DvSatellite satellite = (DvSatellite) dvModel.findTable("sat_customer"); + DatabaseMeta databaseMeta = new TestDatabaseMeta("Vault", "MYSQL"); + + BvScd2Table scd2Table = new BvScd2Table(); + scd2Table.setName("bv_customer_scd2"); + scd2Table.setTableName("bv_customer_scd2"); + scd2Table.setFunctionalTimestampField("x_load_ts"); + scd2Table.setHashKeyPartitionCount(BvScd2HashPartitionCount.FOUR); + scd2Table.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); + + BusinessVaultModel bvModel = new BusinessVaultModel(); + bvModel.getConfigurationOrDefault().setTargetDatabase("Vault"); + bvModel.getConfigurationOrDefault().setTargetLoadMode(DvTargetLoadMode.STAGING_FILE.getCode()); + bvModel.getConfigurationOrDefault().setBulkLoadStagingFolder("/tmp/dv2/bulk/"); + + Scd2BuildContext ctx = + new Scd2BuildContext( + scd2Table, + satellite, + bvModel, + dvModel, + bvModel.getConfigurationOrDefault(), + dvModel.getConfigurationOrDefault(), + null, + new Variables(), + databaseMeta, + "Vault", + databaseMeta, + "Vault", + "sat_customer", + "bv_customer_scd2", + "bv-scd2-bv_customer_scd2-sat_customer", + "customer_hk", + null, + BvScd2PipelineSupport.resolveAttributeFieldNames(satellite), + "x_load_ts", + "valid_from", + "valid_to", + "x_record_source", + BusinessVaultConfiguration.DEFAULT_OPEN_START_SENTINEL, + BusinessVaultConfiguration.DEFAULT_OPEN_END_SENTINEL, + true); + + PipelineMeta pipelineMeta = BvScd2PipelineSupport.generatePipeline(ctx); + TextFileOutputMeta textFileOutputMeta = + (TextFileOutputMeta) + pipelineMeta.getTransforms().stream() + .map(TransformMeta::getTransform) + .filter(t -> t instanceof TextFileOutputMeta) + .findFirst() + .orElseThrow(); + String fileName = textFileOutputMeta.getFileSettings().getFileName(); + assertTrue(fileName.contains("${PARTITION_NUMBER}"), fileName); + assertTrue(fileName.contains("${Internal.Transform.CopyNr}"), fileName); + assertEquals( + "/tmp/dv2/bulk/bv-scd2-bv_customer_scd2-sat_customer-0-${Internal.Transform.CopyNr}", + BvScd2PartitionWorkflowSupport.resolvePartitionedStagingFileBase(fileName, 0)); + } + + @Test + void partitionedNativeBulkUsesBulkLoaderWhenPluginInstalled() throws Exception { + if (!DvBulkLoadPluginSupport.isTransformPluginAvailable( + DvBulkLoadPluginSupport.MYSQL_BULK_LOADER_ID)) { + return; + } + DataVaultModel dvModel = loadVault1Model(); + DvSatellite satellite = (DvSatellite) dvModel.findTable("sat_customer"); + DatabaseMeta databaseMeta = + new DatabaseMeta("Vault", "MySQL", "Native", "", "localhost", "test", "root", ""); + + BvScd2Table scd2Table = new BvScd2Table(); + scd2Table.setName("bv_customer_scd2"); + scd2Table.setTableName("bv_customer_scd2"); + scd2Table.setFunctionalTimestampField("x_load_ts"); + scd2Table.setHashKeyPartitionCount(BvScd2HashPartitionCount.FOUR); + scd2Table.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); + + BusinessVaultModel bvModel = new BusinessVaultModel(); + bvModel.getConfigurationOrDefault().setTargetDatabase("Vault"); + bvModel.getConfigurationOrDefault().setTargetLoadMode(DvTargetLoadMode.NATIVE_BULK.getCode()); + + Scd2BuildContext ctx = + new Scd2BuildContext( + scd2Table, + satellite, + bvModel, + dvModel, + bvModel.getConfigurationOrDefault(), + dvModel.getConfigurationOrDefault(), + null, + new Variables(), + databaseMeta, + "Vault", + databaseMeta, + "Vault", + "sat_customer", + "bv_customer_scd2", + "bv-scd2-bv_customer_scd2-sat_customer", + "customer_hk", + null, + BvScd2PipelineSupport.resolveAttributeFieldNames(satellite), + "x_load_ts", + "valid_from", + "valid_to", + "x_record_source", + BusinessVaultConfiguration.DEFAULT_OPEN_START_SENTINEL, + BusinessVaultConfiguration.DEFAULT_OPEN_END_SENTINEL, + true); + + PipelineMeta pipelineMeta = BvScd2PipelineSupport.generatePipeline(ctx); + assertTrue( + pipelineMeta.getTransforms().stream() + .anyMatch(tm -> DvBulkLoadPluginSupport.MYSQL_BULK_LOADER_ID.equals(tm.getPluginId()))); + assertFalse( + pipelineMeta.getTransforms().stream() + .anyMatch(tm -> tm.getTransform() instanceof TableOutputMeta)); + } + @Test void targetTableLayoutIncludesHashKeyAttributesAndValidityColumns() throws Exception { DataVaultModel dvModel = loadVault1Model(); diff --git a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2TableTest.java b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2TableTest.java index 61ceed71..dc77d96f 100644 --- a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2TableTest.java +++ b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2TableTest.java @@ -30,6 +30,7 @@ import org.hopper.edw.datavault.metadata.DataVaultConfiguration; import org.hopper.edw.datavault.metadata.DataVaultModel; import org.hopper.edw.datavault.metadata.DvTableType; +import org.hopper.edw.datavault.metadata.DvTargetLoadMode; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.w3c.dom.Document; @@ -47,6 +48,8 @@ void defaultsToFullRebuild() { BvScd2Table table = new BvScd2Table(); assertEquals(BvScd2BuildMode.FULL_REBUILD, table.getBuildModeOrDefault()); assertFalse(table.isIncrementalBuild()); + assertEquals(BvScd2HashPartitionCount.NONE, table.getHashKeyPartitionCountOrDefault()); + assertFalse(table.isHashKeyPartitioned()); } @Test @@ -85,6 +88,7 @@ void xmlRoundTripPreservesBuildModeAndWatermarkField() throws Exception { original.setBuildMode(BvScd2BuildMode.INCREMENTAL); original.setFunctionalTimestampField("x_load_ts"); original.setIncrementalWatermarkField("event_ts"); + original.setHashKeyPartitionCount(BvScd2HashPartitionCount.EIGHT); original.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); String xml = XmlHandler.aroundTag("table", XmlMetadataUtil.serializeObjectToXml(original)); @@ -96,9 +100,54 @@ void xmlRoundTripPreservesBuildModeAndWatermarkField() throws Exception { assertEquals(BvScd2BuildMode.INCREMENTAL, restored.getBuildModeOrDefault()); assertEquals("event_ts", restored.getIncrementalWatermarkField()); + assertEquals(BvScd2HashPartitionCount.EIGHT, restored.getHashKeyPartitionCountOrDefault()); assertTrue(restored.isIncrementalBuild()); } + @Test + void hashKeyPartitionWithIncrementalIsError() { + BvScd2Table table = new BvScd2Table(); + table.setName("customer_bv"); + table.setTableName("customer_bv"); + table.setBuildMode(BvScd2BuildMode.INCREMENTAL); + table.setFunctionalTimestampField("x_load_ts"); + table.setHashKeyPartitionCount(BvScd2HashPartitionCount.FOUR); + table.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); + + List remarks = check(table, new DataVaultModel()); + + assertTrue( + remarks.stream() + .anyMatch( + r -> + r.getType() == ICheckResult.TYPE_RESULT_ERROR + && r.getText() != null + && r.getText().contains("hash-key partitions"))); + } + + @Test + void hashKeyPartitionAllowsStagingFileWhenCatalogIsUnavailable() { + BvScd2Table table = new BvScd2Table(); + table.setName("customer_bv"); + table.setTableName("customer_bv"); + table.setFunctionalTimestampField("x_load_ts"); + table.setHashKeyPartitionCount(BvScd2HashPartitionCount.SIXTEEN); + table.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); + + List remarks = new ArrayList<>(); + BusinessVaultModel bvModel = new BusinessVaultModel(); + bvModel.getConfigurationOrDefault().setTargetLoadMode(DvTargetLoadMode.STAGING_FILE.getCode()); + table.check(remarks, null, new Variables(), bvModel, new DataVaultModel()); + + assertFalse( + remarks.stream() + .anyMatch( + r -> + r.getText() != null + && r.getText().contains("Staging file") + && r.getType() == ICheckResult.TYPE_RESULT_ERROR)); + } + @Test void incrementalMultiSatelliteWithoutSourceIndicatorsProducesWarning() throws Exception { BvScd2Table table = new BvScd2Table();