issue #138 : better handling of source fields of sats in BV SCD2

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

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:

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
 

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<DvSatellite> 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));

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(

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("<storeRecordSource>N</storeRecordSource>"));
+
+    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", "<name>sat_customer</name><tableType>SATELLITE</tableType>");
+    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<TransformMeta> 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();