DVLinkToSQL_IncrementalCopy Pipeline Issue

[sethbs]

The pipeline step Source_NewDataToCopy calls the stored procedure dvtosql.source_GetNewDataToCopy. The stored procedure selects the database tables to copy based on the tables listed in the various JobInfo_*.info files that are created in the storage account folder deltalake/conversionresults. The *.info files are selected based on the queuetime within the files being greater than when the last pipeline run ended (which is stored in the target database table dvtosql._controltableforcopy column lastdatetimemarker.).

The issue I see is that between the time when the pipeline starts and ends, additional *.info files are created and the queuetime within those files are less than the time the pipeline completes. Therefore the database tables listed in those new *.info files created will not be picked up during the next pipeline run and the changes in those database table will never make it into the downstream target database, until that database table is once again modified within FnO.

Has anyone using this pipeline come across this issue? I can provide additional information if needed.

As an example.

Table dvtosql._controltableforcopy shows the last time the pipeline completed was 2024-11-11 12:58:13 UTC

During the next pipeline run, the stored procedure dvtosql.source_GetNewDataToCopy should return the database tables within the following 3 *.info files. (note that these times are local UTC-5)

However the contents of the first info file shows the QueueTime = 2024-11-11 12:48:34 UTC which corresponds to the previous info file created at 7:48AM (12:48 UTC) and since the QueueTime is < the lastdatetimemarker in the _controltableforcopy, this info file is not parsed and the tables within it are not loaded into the downstream target database.

Thank you for any insight into this.

Seth

[gianluca-coletta]

Hello @sethbs,

I may have a problem similar to yours. I've noticed that my tables in the target db are not updated with the latest data from the same views in the source db, i.e. it can't capture the new data from the new synchronizations coming from the Synapse Link. it's as if it doesn't insert the latest data into the target tables. Did you solve it? did you have the same problem?

I used the following guide: https://github.com/microsoft/Dynamics-365-FastTrack-Implementation-Assets/tree/master/Analytics/DataverseLink/DataIntegration

Option: Synapse link with incremental CSV

Thank you in advance,
Gianluca

[sethbs]

Hi @gianluca-coletta

I am using the Delta Parquet version, so I have the Spark job running against the CSVs, converting them to parquet files, which I then ingest into my target database using the fast track pipeline. (I believe the Spark job creates the *.info, but I may be mistaken) I don't know if your issue is the same as what I had, but to correct the issue I was seeing I modified the stored procedure source_GetNewDataToCopy (on the Synapse source database) and changed the following, which is around line # 190.

where queuetime > dateadd(hour, -12, cast(''{lastdatetimemarker}'' as datetime2))';   

This change will look back in time to the previous 12 hours worth of *.info files and includes those database tables listed in those files to be loaded by the pipeline. It does not cause any problems with the target database since if there were no changes in any of the database tables, nothing will get updated in the target database.

Hope that helps. If you have specific questions, I can try to help further.

[gianluca-coletta]

First of all, thank you for answering me right away. I will explain the problem in more detail. I created the Synapse Link in Power App connected to an ADLS storage in Azure, created a source sql db in Synapse (servless) and a target sql db. Running the pipeline the number of rows in the tables is equal to the number of initial rows in Synapse Link. But several days have passed since I created the Synapse Link, and so in the meantime rows have been added to the d365 tables and the Synapse Link has performed synchronizations by increasing the number of rows in the tables. Only in my target db I do not see these new rows while in the equivalent source db views I do. When I run the pipeline it fails to fetch these new table rows even though the execution completes successfully.

Do you think your solution can solve my problem? I am stuck at this point.

Thank you very much @sethbs

[sethbs]

@gianluca-coletta

Not sure this is the same problem, since you indicated previously that you are using incremental CSVs. I am using delta format, so the Spark application is creating the *.info files that I was having problems with. If you are ingesting the CSV files directly, then the code I was referring to is not executed. If you are using delta files, then you can give the modification a try.

[gianluca-coletta]

@gianluca-coletta

Not sure this is the same problem, since you indicated previously that you are using incremental CSVs. I am using delta format, so the Spark application is creating the *.info files that I was having problems with. If you are ingesting the CSV files directly, then the code I was referring to is not executed. If you are using delta files, then you can give the modification a try.

@sethbs

Yes, I'm using CSV incrementale update. Do you have any proposal solution for my issue? How can I do It? I'm not so much skilled on this topics

[sethbs]

@gianluca-coletta
Not sure this is the same problem, since you indicated previously that you are using incremental CSVs. I am using delta format, so the Spark application is creating the *.info files that I was having problems with. If you are ingesting the CSV files directly, then the code I was referring to is not executed. If you are using delta files, then you can give the modification a try.

@sethbs

Yes, I'm using CSV incrementale update. Do you have any proposal solution for my issue? How can I do It? I'm not so much skilled on this topics

@gianluca-coletta

I'm not familiar with the incremental CSV sections of the included stored procedures so I won't be able to give any suggestions. The way I would go about it is to find a specific record that did not make it into the target SQL DB and then go back to the CSVs to find the dated folder where that record was created. Then determine if all of the records and/or tables in that folder are missing in the target database. If all of them are, then the stored procedure within the pipeline is skipping the folder. If only some of the tables/records are missing then it's something else.

Good luck.

[gianluca-coletta]

@sethbs

Okok, thank you. I try what you say

[gianluca-coletta]

Hello @sethbs,

I was able to notice that the views (corresponding to the tables) of the source DB within the Synapse Workspace update correctly. While the target sql DB does not. Despite the fact that the pipeline is running correctly. I wonder how come

[sethbs]

@gianluca-coletta

I'm not really able to help any further since I'm not very familiar with the pipeline and how it works with the incremental CSVs. As I indicated, I am running the delta version instead. It's going to take someone to dig into the data and stored procedures to determine what records were missed, what dated folder they were included in and then determine if all the data within that dated folder was missed or only some records.

[gianluca-coletta]

@gianluca-coletta

I'm not really able to help any further since I'm not very familiar with the pipeline and how it works with the incremental CSVs. As I indicated, I am running the delta version instead. It's going to take someone to dig into the data and stored procedures to determine what records were missed, what dated folder they were included in and then determine if all the data within that dated folder was missed or only some records.

Okok, thanks!