From 0f2d0c09dcb8234c617afbde51933ca960d3c30f Mon Sep 17 00:00:00 2001 From: mattcasters Date: Thu, 27 Aug 2026 15:35:25 +0200 Subject: [PATCH] issue #138 : better handling of source fields of sats in BV SCD2 --- CHANGELOG.md | 6 + docs/business-vault-scd2.adoc | 6 +- docs/help/bv-scd2-table-dialog.adoc | 5 +- .../businessvault/BvScd2PipelineSupport.java | 31 ++- .../BvScd2MultiSatelliteFixtureTest.java | 6 + .../BvScd2PipelineSupportTest.java | 188 ++++++++++++++++++ 6 files changed, 228 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c4fbdfb..d09ee13b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to Data Hopper EDW (formerly hop-datavault) are documented i ## Unreleased +### BV SCD2 does not read omitted satellite record-source columns (issue #138) + +- Generated SCD2 satellite `TableInput` SQL omits the vault record-source / source-indicator column when **Store record source indicator** is off on that satellite +- Multi-satellite SCD2 pipelines never select that physical column (BV record source comes from `_bv_source` / satellite config) +- When the column is omitted, a Constant still fills the BV SCD2 record-source field from the satellite config or satellite name + ### Dialog Help opens plugin HTML (issue #133) - Help buttons open the plugin-shipped AsciiDoc HTML (`plugins/misc/hopper-edw/docs/`), not classpath markdown stubs or hop.apache.org 404s diff --git a/docs/business-vault-scd2.adoc b/docs/business-vault-scd2.adoc index 79b636e2..ff4f46c2 100644 --- a/docs/business-vault-scd2.adoc +++ b/docs/business-vault-scd2.adoc @@ -42,8 +42,8 @@ Dialog help (Hop GUI **Help** on the SCD2 table dialog): link:help/bv-scd2-table For each SCD2 table the plugin builds a Hop pipeline that: -1. Reads satellite history via `TableInput` SQL (grain + functional timestamp + attributes) -2. Adds source indicators when multiple satellites feed one BV table +1. Reads satellite history via `TableInput` SQL (grain + functional timestamp + attributes). The vault record-source / source-indicator column is **not** selected when a satellite has **Store record source indicator** off (VaultSpeed-style mono-source sats). Multi-satellite pipelines never read that physical column; merge identity is the synthetic `_bv_source` value. +2. Adds source indicators when multiple satellites feed one BV table. When a satellite omits the physical column, a Constant supplies the BV record-source value from the satellite config (or the satellite name). 3. Merges streams (multi-satellite) with `SortedSchemaMerge` 4. Repeats sparse attribute columns across merged rows 5. Computes validity bounds with `Analytic Query` (LAG/LEAD) @@ -160,7 +160,7 @@ Open intervals use sentinels from configuration (default `1900-01-01 00:00:00` a |Required for multi-satellite tables. Maps satellite attribute → BV column. |Satellite configs -|Per-satellite functional timestamp and source indicator for multi-satellite merges. +|Per-satellite functional timestamp and source indicator for multi-satellite merges. The source indicator is the merge tag (`_bv_source`) and the BV record-source value when the satellite does not store a physical source-indicator column. Honor **Store record source indicator** on each DV satellite — SCD2 Table Input SQL never requests a column the satellite table does not have. |=== The Customer 360 sample uses all three SCD2 dialog areas: diff --git a/docs/help/bv-scd2-table-dialog.adoc b/docs/help/bv-scd2-table-dialog.adoc index f159bfd1..84882f96 100644 --- a/docs/help/bv-scd2-table-dialog.adoc +++ b/docs/help/bv-scd2-table-dialog.adoc @@ -83,7 +83,9 @@ Single-satellite tables can pass attributes through under DV names unless you ad Optional per-satellite settings for multi-satellite merges: - Functional timestamp override for that satellite -- Source indicator value when you need to tag which satellite contributed a version +- Source indicator value when you need to tag which satellite contributed a version (and to fill the BV record-source column when that satellite does not store a physical source-indicator column) + +If a derivative satellite has **Store record source indicator** unchecked, generated satellite `TableInput` SQL does not select the vault record-source column. The BV table still receives a record-source value from this config (or the satellite name). == Type 1 vs Type 2 (important) @@ -126,6 +128,7 @@ Longer guidance: project doc `docs/business-vault-scd2.adoc` (section *Type 1 vs - After changing the linked `.hdv` on disk, use **Reload DV model** on the Business Vault toolbar. - External read-only satellites are skipped by Data Vault Update but remain valid SCD2 sources if the vault DB can read them. - Multi-satellite example: `integration-tests/tests/multi-satellite-bv/customer-360.hbv`. +- Mono-source satellites often omit the physical source-indicator column. Uncheck **Store record source indicator** on the satellite; SCD2 loads still run. == Related docs 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 19efe694..a7fdc80c 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 @@ -52,6 +52,7 @@ import org.hopper.edw.datavault.metadata.DvLink; import org.hopper.edw.datavault.metadata.DvLoadCycleSupport; import org.hopper.edw.datavault.metadata.DvSatellite; +import org.hopper.edw.datavault.metadata.DvSourceFieldMappingSupport; import org.hopper.edw.datavault.metadata.DvSpecialRecordSupport; import org.hopper.edw.datavault.metadata.DvSqlSupport; import org.hopper.edw.datavault.metadata.DvTableType; @@ -750,6 +751,22 @@ public static List resolveSourceSatellites(BvScd2Table table, DataV return satellites; } + /** + * True when this SCD2 satellite leg should {@code SELECT} the physical vault record-source + * column. Multi-satellite pipelines never do: BV RS is {@link #SOURCE_INDICATOR_FIELD} renamed + * after Repeat. A missing satellite defaults to omit so JDBC cannot request a column the table + * may not have. + */ + static boolean shouldSelectPhysicalRecordSource(Scd2BuildContext ctx, SatelliteLeg leg) { + return ctx != null && !ctx.isMultiSatellite() && storesPhysicalRecordSource(leg); + } + + static boolean storesPhysicalRecordSource(SatelliteLeg leg) { + return leg != null + && leg.satellite != null + && DvSourceFieldMappingSupport.shouldStoreRecordSource(leg.satellite); + } + public static String resolveRecordSourceField( DataVaultConfiguration dvConfig, IVariables variables) { String rsFieldName = "RECORD_SOURCE"; @@ -1015,7 +1032,9 @@ static String buildLegTableInputSql(Scd2BuildContext ctx, SatelliteLeg leg) { selectFields.add(ctx.sourceDatabaseMeta.quoteField(attr)); } } - if (leg.satellite == null || leg.satellite.isStoreRecordSource()) { + // Multi-sat BV RS comes from _bv_source (post-repeat rename). Never read the physical + // satellite column — VaultSpeed-style sats omit it and JDBC would fail. + if (shouldSelectPhysicalRecordSource(ctx, leg)) { selectFields.add(ctx.sourceDatabaseMeta.quoteField(ctx.recordSourceField)); } selectFields.add( @@ -1465,13 +1484,6 @@ private static TransformMeta addLegSourceIndicatorConstant( ConstantField indicatorField = new ConstantField(SOURCE_INDICATOR_FIELD, "String", leg.sourceIndicatorValue); constantMeta.getFields().add(indicatorField); - // When the DV satellite does not store a record-source column, materialize the BV RS value - // from the configured leg indicator (or satellite name) so downstream collapse still has it. - if (leg.satellite != null && !leg.satellite.isStoreRecordSource()) { - constantMeta - .getFields() - .add(new ConstantField(ctx.recordSourceField, "String", leg.sourceIndicatorValue)); - } TransformMeta tm = new TransformMeta("Constant", "source_" + leg.satellite.getName(), constantMeta); @@ -1495,7 +1507,7 @@ private static TransformMeta injectRecordSourceConstantIfNeeded( if (predecessor == null || leg == null || leg.satellite == null - || leg.satellite.isStoreRecordSource()) { + || storesPhysicalRecordSource(leg)) { return predecessor; } ConstantMeta constantMeta = new ConstantMeta(); @@ -1535,7 +1547,6 @@ private static TransformMeta addLegSelectValues( String targetFieldName = ctx.variables.resolve(mapping.getTargetFieldName()); selectFields.add(selectField(sourceFieldName, targetFieldName)); } - selectFields.add(selectField(ctx.recordSourceField, null)); if (!leg.sourceFunctionalTimestampField.equals(ctx.functionalTimestampField)) { selectFields.add( selectField(leg.sourceFunctionalTimestampField, ctx.functionalTimestampField)); diff --git a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2MultiSatelliteFixtureTest.java b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2MultiSatelliteFixtureTest.java index b41b009f..595f5af2 100644 --- a/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2MultiSatelliteFixtureTest.java +++ b/src/test/java/org/hopper/edw/datavault/metadata/businessvault/BvScd2MultiSatelliteFixtureTest.java @@ -139,6 +139,12 @@ void customer360ModelsValidateAndGenerateFourSatellitePipeline() throws Exceptio assertEquals(19, transforms.size()); assertEquals( 4, transforms.stream().filter(t -> t.getTransform() instanceof TableInputMeta).count()); + for (TransformMeta transform : + transforms.stream().filter(t -> t.getTransform() instanceof TableInputMeta).toList()) { + assertFalse( + ((TableInputMeta) transform.getTransform()).getSql().contains("x_record_source"), + transform.getName()); + } assertEquals( 4, transforms.stream().filter(t -> t.getTransform() instanceof ConstantMeta).count()); assertEquals( 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 95a1e921..090cd690 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 @@ -185,6 +185,107 @@ void buildSatelliteTableInputSqlOrdersByGrainAndFunctionalTimestamp() throws Exc assertFalse(sql.contains(" WHERE ")); } + @Test + void satelliteTableInputSqlOmitsRecordSourceWhenSatelliteDoesNotStoreIt() throws Exception { + Scd2BuildContext ctx = singleSatelliteContext(BvScd2BuildMode.FULL_REBUILD, null); + ctx.getSatellite().setStoreRecordSource(false); + + String sql = BvScd2PipelineSupport.buildSatelliteTableInputSql(ctx); + + assertTrue(sql.contains("FROM sat_customer")); + assertTrue(sql.contains("customer_hk")); + assertTrue(sql.contains("x_load_ts")); + assertFalse(sql.contains("x_record_source")); + assertFalse(BvScd2PipelineSupport.shouldSelectPhysicalRecordSource(ctx, ctx.legs.get(0))); + } + + @Test + void singleSatellitePipelineInjectsRecordSourceConstantWhenNotStored() throws Exception { + Scd2BuildContext ctx = singleSatelliteContext(BvScd2BuildMode.FULL_REBUILD, null); + ctx.getSatellite().setStoreRecordSource(false); + + PipelineMeta pipelineMeta = BvScd2PipelineSupport.generatePipeline(ctx); + + TableInputMeta satInput = + (TableInputMeta) + pipelineMeta.getTransforms().stream() + .filter(t -> "read_sat_customer".equals(t.getName())) + .findFirst() + .orElseThrow() + .getTransform(); + assertFalse(satInput.getSql().contains("x_record_source")); + + TransformMeta constantTransform = + pipelineMeta.getTransforms().stream() + .filter(t -> "record_source_sat_customer".equals(t.getName())) + .findFirst() + .orElseThrow(); + ConstantMeta constantMeta = (ConstantMeta) constantTransform.getTransform(); + assertTrue( + constantMeta.getFields().stream() + .anyMatch( + field -> + "x_record_source".equals(field.getFieldName()) + && "sat_customer".equals(field.getValue()))); + + GroupByMeta groupByMeta = + (GroupByMeta) + pipelineMeta.getTransforms().stream() + .filter(t -> t.getTransform() instanceof GroupByMeta) + .findFirst() + .orElseThrow() + .getTransform(); + assertEquals("x_record_source", groupByMeta.getAggregations().get(2).getField()); + + var layout = + BvScd2PipelineSupport.buildTargetTableLayout( + ctx.scd2Table, ctx.bvConfig, ctx.dvModel, ctx.getSatellite(), new Variables()); + assertTrue( + layout.getValueMetaList().stream().anyMatch(vm -> "x_record_source".equals(vm.getName()))); + } + + @Test + void incrementalSatelliteSqlOmitsRecordSourceWhenNotStored() throws Exception { + Scd2BuildContext ctx = singleSatelliteContext(BvScd2BuildMode.INCREMENTAL, null); + ctx.getSatellite().setStoreRecordSource(false); + + String satSql = BvScd2PipelineSupport.buildSatelliteTableInputSql(ctx); + assertTrue(satSql.contains("x_load_ts > ?")); + assertFalse(satSql.contains("x_record_source")); + + String openSql = BvScd2PipelineSupport.buildOpenTargetTableInputSql(ctx); + assertTrue(openSql.contains("FROM bv_customer_scd2")); + assertTrue(openSql.contains("x_record_source")); + } + + @Test + void satelliteXmlRoundTripPreservesStoreRecordSourceOff() throws Exception { + DvSatellite original = new DvSatellite("sat_customer"); + original.setStoreRecordSource(false); + + String xml = XmlHandler.aroundTag("table", XmlMetadataUtil.serializeObjectToXml(original)); + assertTrue(xml.contains("N")); + + Document document = XmlHandler.loadXmlString(xml); + Node rootNode = XmlHandler.getSubNode(document, "table"); + DvSatellite restored = new DvSatellite(); + XmlMetadataUtil.deSerializeFromXml(rootNode, DvSatellite.class, restored, null); + + assertFalse(restored.isStoreRecordSource()); + } + + @Test + void satelliteXmlMissingStoreRecordSourceDefaultsToTrue() throws Exception { + String xml = + XmlHandler.aroundTag("table", "sat_customerSATELLITE"); + Document document = XmlHandler.loadXmlString(xml); + Node rootNode = XmlHandler.getSubNode(document, "table"); + DvSatellite restored = new DvSatellite(); + XmlMetadataUtil.deSerializeFromXml(rootNode, DvSatellite.class, restored, null); + + assertTrue(restored.isStoreRecordSource()); + } + @Test void buildIncrementalSatelliteFilterSqlUsesPositionalParameter() { assertEquals( @@ -745,9 +846,13 @@ void multiSatellitePipelineContainsMergeRepeatAndMappedCollapse() throws Excepti String customerSql = BvScd2PipelineSupport.buildLegTableInputSql(ctx, ctx.legs.get(0)); assertTrue(customerSql.contains("name")); assertFalse(customerSql.contains("demo_score")); + assertFalse(customerSql.contains("x_record_source")); assertTrue(customerSql.contains("ORDER BY")); assertTrue(customerSql.indexOf("ORDER BY") < customerSql.lastIndexOf("x_load_ts")); + String demoSql = BvScd2PipelineSupport.buildLegTableInputSql(ctx, ctx.legs.get(1)); + assertFalse(demoSql.contains("x_record_source")); + PipelineMeta pipelineMeta = BvScd2PipelineSupport.generatePipeline(ctx); List transforms = pipelineMeta.getTransforms(); @@ -760,6 +865,7 @@ void multiSatellitePipelineContainsMergeRepeatAndMappedCollapse() throws Excepti GeneratedPipelineMetadataConstants.ROLE_SOURCE_READ, GeneratedPipelineMetadataSupport.getTransformAttribute( transform, GeneratedPipelineMetadataConstants.LOGICAL_ROLE)); + assertFalse(((TableInputMeta) transform.getTransform()).getSql().contains("x_record_source")); } assertEquals( 2, transforms.stream().filter(t -> t.getTransform() instanceof ConstantMeta).count()); @@ -825,6 +931,16 @@ void multiSatellitePipelineContainsMergeRepeatAndMappedCollapse() throws Excepti assertEquals("x_load_ts", postRepeatFields.get(1).getName()); assertEquals(BvScd2PipelineSupport.SOURCE_INDICATOR_FIELD, postRepeatFields.get(2).getName()); assertEquals("x_record_source", postRepeatFields.get(2).getRename()); + SelectValuesMeta customerSelectMeta = + (SelectValuesMeta) + transforms.stream() + .filter(t -> "select_sat_customer".equals(t.getName())) + .findFirst() + .orElseThrow() + .getTransform(); + assertTrue( + customerSelectMeta.getSelectOption().getSelectFields().stream() + .noneMatch(field -> "x_record_source".equals(field.getName()))); assertEquals("_r_customer_name", postRepeatFields.get(3).getName()); assertEquals("customer_name", postRepeatFields.get(3).getRename()); assertEquals("_r_demo_score", postRepeatFields.get(4).getName()); @@ -854,6 +970,78 @@ void multiSatellitePipelineContainsMergeRepeatAndMappedCollapse() throws Excepti && hop.getToTransform().equals(mergeTransform))); } + @Test + void multiSatelliteTableInputOmitsRecordSourceEvenWhenOneSatelliteStoresIt() throws Exception { + DataVaultModel dvModel = loadVault1ModelWithDemoSatellite(); + DvSatellite customerSatellite = (DvSatellite) dvModel.findTable("sat_customer"); + DvSatellite demoSatellite = (DvSatellite) dvModel.findTable("sat_customer_demo"); + customerSatellite.setStoreRecordSource(true); + demoSatellite.setStoreRecordSource(false); + DatabaseMeta databaseMeta = new TestDatabaseMeta("Vault"); + + BvScd2Table scd2Table = new BvScd2Table(); + scd2Table.setName("customer_bv"); + scd2Table.setTableName("customer_bv"); + scd2Table.setFunctionalTimestampField("x_load_ts"); + scd2Table.getDerivatives().add(new BvDerivativeRef("sat_customer", DvTableType.SATELLITE)); + scd2Table.getDerivatives().add(new BvDerivativeRef("sat_customer_demo", DvTableType.SATELLITE)); + scd2Table + .getFieldMappings() + .add(new BvScd2FieldMapping("sat_customer", "name", "customer_name")); + scd2Table + .getFieldMappings() + .add(new BvScd2FieldMapping("sat_customer_demo", "demo_score", "demo_score")); + + Scd2BuildContext ctx = + new Scd2BuildContext( + scd2Table, + List.of( + new SatelliteLeg( + customerSatellite, + "sat_customer", + "sat_customer", + "x_load_ts", + List.of(scd2Table.getFieldMappings().get(0))), + new SatelliteLeg( + demoSatellite, + "sat_customer_demo", + "DEMO", + "x_load_ts", + List.of(scd2Table.getFieldMappings().get(1)))), + true, + List.of("customer_name", "demo_score"), + new BusinessVaultModel(), + dvModel, + new BusinessVaultConfiguration(), + dvModel.getConfigurationOrDefault(), + null, + new Variables(), + databaseMeta, + "Vault", + databaseMeta, + "Vault", + "sat_customer", + "customer_bv", + "bv-scd2-customer_bv-customer_bv", + "customer_hk", + null, + List.of("customer_name", "demo_score"), + "x_load_ts", + "valid_from", + "valid_to", + "x_record_source", + BusinessVaultConfiguration.DEFAULT_OPEN_START_SENTINEL, + BusinessVaultConfiguration.DEFAULT_OPEN_END_SENTINEL, + true); + + assertFalse( + BvScd2PipelineSupport.buildLegTableInputSql(ctx, ctx.legs.get(0)) + .contains("x_record_source")); + assertFalse( + BvScd2PipelineSupport.buildLegTableInputSql(ctx, ctx.legs.get(1)) + .contains("x_record_source")); + } + @Test void targetTableLayoutCanOmitHashKey() throws Exception { DataVaultModel dvModel = loadVault1Model();