Update record from two different sources

[AhmedJoum]

I'm using ETL.net to load data using Entity Framework, I have a transaction Data warehouse table that I want to fill from two tables from two different databases but have a field whose value is common in both.

Transaction Model for a data warehouse:

public class Transaction
{
   public int Id {get; set;}
   public string? PaymentTransactionId {get; set;}
   public string? SaleTransactionId {get; set;}
// PaymentTransaction prop 
// SalesTransaction prop
}

From Payment DB I will have the Payment Transaction Model:

public class PaymentTransaction
{
   public int Id {get; set;}
   public string PaymentTransactionId {get; set;} // never null
// Another prop
}

From the sales database, I have Sales Transaction Model:

public class SalesTransaction
{
   public int Id {get; set;}
   public string? PaymentTransactionId {get; set;} // Could be null, the value is common between Sale & Payment
   public string SellTransactionId {get; set;} // Never null here
// Another props
}

I tried to make two processes for each table ETL seek on PaymentTransactionId, but the second one always overrides the first one properties and is set as null not updating only the associated fields.

[paillave]

Sorry for the late answer. Can you show me the code of the ETL process you built so far? Even an extract. It will help me to give you the correct hints and direction to follow.

[AhmedJoum]

Hi Stéphane

here is a sample:

 public async Task<EtlResult> DoETL()
    {
        var paymentTransactionPR = StreamProcessRunner.Create<string>(PaymentTransactionLoadProcess);
        var SalesTransactionPR = StreamProcessRunner.Create<string>(SalesTransactionLoadProcess);

        var executionOptions = new ExecutionOptions<string>
        {
            Resolver = new SimpleDependencyResolver()
                .Register(_dwhDBContext, ETLConsts.DwhSchema)
                .Register(_paymentDBContext, ETLConsts.StagingPaymentSchema)
                .Register(_salesDBContext, ETLConsts.StagingSalesSchema),
        };
        var paymentTransactionResult = await paymentTransactionPR.ExecuteAsync(
            PaymentExecuteMsg,
            executionOptions
        );

        var salesTransactionResult = await SalesTransactionPR.ExecuteAsync(
            SalesExecuteMsg,
            executionOptions
        );
    }
 
 
 
 public void PaymentLoadProcess(ISingleStream<string> contextStream)
    {
        contextStream
            .EfCoreSelect(
                GetPaymentTrxStep,
                (o, row) =>
                    o.Set<PaymentTransaction>(),
                connectionKey: ETLConsts.StagingPaymentSchema
            )
            .EfCoreSave(
                SavePaymentTrxStep,
                t =>
                    t.WithKeyedConnection(ETLConsts.DwhSchema)
                        .Entity(
                            t =>
                                new TransactionHeader
                                {
                                    AccountId = t.AccountId,
                                    CreatedBy = t.CreatedBy,
                                    PaymentTransactionId = t.Id,
                                    CreationDate = t.CreationDate,
                                    PaymentTransactionNo = t.TransactionNo,
                                    TotalAmount = t?.Amount,
                                    MerchantTransactionId = t?.TransactionNo, 
                                    MerchantName = "Internal Payment", 
                                }
                        )
                        .SeekOn(i => new { i.MerchantTransactionId, i.MerchantName })
            );
        _logger.LogInformation(GetTransactionDone);
    }


     public void SalesTransactionLoadProcess(ISingleStream<string> contextStream)
    {
        var defaultValues = _config.GetSection(DefaultValuesConfigSection);
        var vatRate = defaultValues.GetSection(VatRateConfigSection)?.Value?.ToString();

        contextStream
            .EfCoreSelect(
                GetSalesTrxStep,
                (o, row) =>
                    o.Set<SalesTransaction>(),
                connectionKey: ETLConsts.StagingSalesSchema
            )
            .EfCoreSave(
                SaveSalesTrxStep,
                t =>
                    t.WithKeyedConnection(ETLConsts.DwhSchema)
                        .Entity(
                            t =>
                                new TransactionHeader
                                {
                                    SalesTransactionId = t.Id,
                                    SalesTransactionNo = t.TransactionId,
                                    MerchantName = t.Merchant?.Name,
                                    MerchantTransactionId = t.MerchantTransactionId,
                                    SalesTransactionType =
                                        t.TransactionType?.ToString() ?? string.Empty,
                                    CreationDate = t.CreationDate,
                                    MaskedCard = t?.PaymentTransactionDetails?.MaskedCard,
                                    TotalAmount = t?.Amount,
                                    TerminalNumber = t?.TerminalNumber,
                                    MethodOfPayment = t?.Merchant?.Name,
                                    SAPReconSyncStatus = SAPSyncStatus.New.FastToString()
                                }
                        )
                        .SeekOn(i => new { i.MerchantTransactionId, i.MerchantName })
            );
    }

let me know if it's a clear example, what happens is some Payments and sales will have the same MerchantTransactionId and MerchantName, but when I run the process the latest one will override the first one properties with null.

[paillave]

It is clear, I come back to you asap