From 3d1304253a03a4f7ca34f17a4a95b5a233c6b06f Mon Sep 17 00:00:00 2001 From: MattPCollins Date: Tue, 20 Jan 2026 08:29:30 +0000 Subject: [PATCH 1/4] Update code for bugs related to ingest module. --- .../ingestpayload/CheckPayloadExecution.py | 8 ---- .../ingest/utils/ConfigurePayloadVariables.py | 31 ++++++------ .../utils/ConfigurePayloadVariables.py | 8 ++-- .../python/tests/test_CreateMergeQuery.py | 13 +++++ .../test_IngestConfigurePayloadVariables.py | 48 ++++++++++--------- .../Stored Procedures/GetDatasetPayload.sql | 4 +- .../Stored Procedures/GetMergePayload.sql | 18 +++---- .../Stored Procedures/GetNotebookPayload.sql | 8 ++-- 8 files changed, 72 insertions(+), 66 deletions(-) diff --git a/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py b/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py index 4ecd4356..860f3968 100644 --- a/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py +++ b/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py @@ -87,14 +87,6 @@ # COMMAND ---------- -check_container_name(container_name = raw_container_name) - -# COMMAND ---------- - -check_container_name(container_name = cleansed_container_name) - -# COMMAND ---------- - # MAGIC %md # MAGIC # Check: Storage accessibility diff --git a/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py b/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py index a2b740e5..daa21f5a 100644 --- a/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py +++ b/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py @@ -11,8 +11,11 @@ def get_merge_payload_variables(payload: dict()) -> list(): raw_secret = payload["RawStorageAccessKey"] raw_last_load_date = payload["RawLastLoadDate"] - raw_schema_name = payload["RawSchemaName"] - raw_file_type = payload["RawFileType"] + raw_schema_name = payload["RawSchemaName"] # connection name of data source + raw_file_type = payload["RawFileType"] + raw_source_path = payload["RawSourcePath"] # physical path in raw container + raw_source_name = payload["RawSourceName"] # file name in raw container + datetime_folder_hierarchy = payload["DateTimeFolderHierarchy"] cleansed_storage_name = payload["CleansedStorageName"] @@ -20,33 +23,27 @@ def get_merge_payload_variables(payload: dict()) -> list(): cleansed_secret = payload["CleansedStorageAccessKey"] cleansed_last_load_date = payload["CleansedLastLoadDate"] - cleansed_schema_name = payload["CleansedSchemaName"] + cleansed_schema_name = payload["CleansedSchemaName"] # connection name of data source + cleansed_source_path = payload["CleansedSourcePath"] # desired physical path in cleansed container + cleansed_source_name = payload["CleansedSourceName"] # desired file name in cleansed container filter_condition = payload["FilterCondition"] # Semantic checks for these required in the IngestChecks notebook? - pk_list = payload["CleansedPkList"].split(",") - partition_list = payload["CleansedPartitionFields"].split(",") if payload["CleansedPartitionFields"] != "" else [] + pk_list = payload["CleansedPkList"].split("|") + partition_list = payload["CleansedPartitionFields"].split("|") if payload["CleansedPartitionFields"] != "" else [] - columns_list = payload["CleansedColumnsList"].split(",") - columns_type_list = payload["CleansedColumnsTypeList"].split(",") - columns_format_list = payload["CleansedColumnsFormatList"].split(",") + columns_list = payload["CleansedColumnsList"].split("|") + columns_type_list = payload["CleansedColumnsTypeList"].split("|") + columns_format_list = payload["CleansedColumnsFormatList"].split("|") metadata_column_list = ["PipelineRunId","PipelineExecutionDateTime"] metadata_column_type_list = ["STRING","TIMESTAMP"] metadata_column_format_list = ["","yyyy-MM-dd HH:mm:ss"] - # metadata_column_list = payload["cleansedMetadataColumnList"].split(",") - # metadata_column_type_list = payload["cleansedMetadataColumnTypeList"].split(",") - # metadata_column_format_list = payload["cleansedMetadataColumnFormatList"].split(",") - total_column_list = columns_list + metadata_column_list total_column_type_list = columns_type_list + metadata_column_type_list total_column_format_list = columns_format_list + metadata_column_format_list - # totalColumnList = columns_list - # totalColumnTypeList = columnsTypeList - # totalColumnFormatList = columnsFormatList - - output = [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] + output = [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] return output diff --git a/src/azure.databricks/python/notebooks/transform/utils/ConfigurePayloadVariables.py b/src/azure.databricks/python/notebooks/transform/utils/ConfigurePayloadVariables.py index 004b62b4..66869482 100644 --- a/src/azure.databricks/python/notebooks/transform/utils/ConfigurePayloadVariables.py +++ b/src/azure.databricks/python/notebooks/transform/utils/ConfigurePayloadVariables.py @@ -9,11 +9,11 @@ def get_transform_payload_variables(payload: dict()) -> list(): curated_schema_name = payload["SchemaName"] curated_dataset_name = payload["DatasetName"] - columns_list = payload["ColumnsList"].split(",") - column_type_list = payload["ColumnTypeList"].split(",") + columns_list = payload["ColumnsList"].split("|") + column_type_list = payload["ColumnTypeList"].split("|") - bk_list = payload["BkAttributesList"].split(",") - partition_list = payload["PartitionByAttributesList"].split(",") if payload["PartitionByAttributesList"] != "" else [] + bk_list = payload["BkAttributesList"].split("|") + partition_list = payload["PartitionByAttributesList"].split("|") if payload["PartitionByAttributesList"] != "" else [] surrogate_key = payload["SurrogateKey"] diff --git a/src/azure.databricks/python/tests/test_CreateMergeQuery.py b/src/azure.databricks/python/tests/test_CreateMergeQuery.py index 4c55ff86..f006c508 100644 --- a/src/azure.databricks/python/tests/test_CreateMergeQuery.py +++ b/src/azure.databricks/python/tests/test_CreateMergeQuery.py @@ -67,7 +67,20 @@ def test_select_sql_columns_format_string_timestamp_format_type2(self): actual1 = select_sql_columns_format_string(['time'],['timestamp'], ['yyyy-MM-ddTHH:mm:ss.SSSSSSSZ']) expected1 = "to_timestamp(time) as time" assert actual1 == expected1 + + def test_select_sql_columns_format_string_decimal(self): + actual1 = select_sql_columns_format_string(['value'],['decimal(10,2)'], ['']) + expected1 = "cast(value as decimal(10,2)) as value" + actual2 = select_sql_columns_format_string(['value'],['decimal(10)'], ['']) + expected2 = "cast(value as decimal(10)) as value" + assert actual1 == expected1 + assert actual2 == expected2 + + def test_select_sql_columns_format_string_decimal2(self): + actual1 = select_sql_columns_format_string(['value', 'conversion_rate'],['decimal(10,2)', 'decimal(10,4)'], ['','']) + expected1 = "cast(value as decimal(10,2)) as value, cast(conversion_rate as decimal(10,4)) as conversion_rate" + assert actual1 == expected1 # def test_select_sql_columns_format_string_timestamp_long_format(self): # with self.assertRaises(Exception) as context: diff --git a/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py b/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py index ae3061d9..0078d911 100644 --- a/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py +++ b/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py @@ -29,108 +29,112 @@ def default_payload() -> dict: "RawFileType":"parquet", "VersionNumber":1, "CleansedSchemaName":"Oracle", - "CleansedTableName":"testdataset", "Enabled":True, "LoadType":"I", "LoadAction":"F", "RawLastLoadDate":"2025-05-23T10:18:36.33", "CleansedLastLoadDate":"null", - "CleansedColumnsList":"ID,NAME", - "CleansedColumnsTypeList":"INTEGER,STRING", - "CleansedColumnsFormatList":",", + "CleansedColumnsList":"ID|NAME", + "CleansedColumnsTypeList":"INTEGER|STRING", + "CleansedColumnsFormatList":"|", "CleansedPkList":"ID", "CleansedPartitionFields":"", "DateTimeFolderHierarchy":"year=2025/month=01/day=1/hour=10", - "FilterCondition": ""} + "FilterCondition": "", + "RawSourcePath": "folder", + "RawSourceName": "testdataset", + "CleansedSourcePath": "test_folder", + "CleansedSourceName": "cleanseddataset", + } return payload def test_get_merge_payload_variables_table_name(default_payload): - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'testdataset' assert table_name == expected def test_get_merge_payload_variables_load_type_default(default_payload): - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'I' assert load_type == expected def test_get_merge_payload_variables_load_type_override(default_payload): payload = default_payload.copy() payload['LoadType'] = 'F' - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = 'F' assert load_type == expected def test_get_merge_payload_variables_load_action_default(default_payload): - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'F' assert load_action == expected def test_get_merge_payload_variables_load_action_override(default_payload): payload = default_payload.copy() payload['LoadAction'] = 'I' - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = 'I' assert load_action == expected def test_get_merge_payload_variables_load_action_text(default_payload): - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'full' assert load_action_text == expected def test_get_merge_payload_variables_version_number(default_payload): - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = '0001' assert version_number == expected def test_get_merge_payload_variables_version_number_tens(default_payload): payload = default_payload.copy() payload['VersionNumber'] = 10 - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = '0010' assert version_number == expected def test_get_merge_payload_variables_version_number_hundreds(default_payload): payload = default_payload.copy() payload['VersionNumber'] = 151 - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = '0151' assert version_number == expected def test_get_merge_payload_variables_pk_list_empty(default_payload): payload = default_payload.copy() payload['CleansedPkList'] = "" - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = [''] assert pk_list == expected def test_get_merge_payload_variables_pk_list_single(default_payload): - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = ['ID'] assert pk_list == expected def test_get_merge_payload_variables_pk_list_multiple(default_payload): payload = default_payload.copy() - payload['CleansedPkList'] = "ID,NAME" - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + payload['CleansedPkList'] = "ID|NAME" + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID","NAME"] assert pk_list == expected def test_get_merge_payload_variables_partition_list_empty(default_payload): - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = [] assert partition_list == expected def test_get_merge_payload_variables_partition_list_single(default_payload): payload = default_payload.copy() payload['CleansedPartitionFields'] = "ID" - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID"] assert partition_list == expected def test_get_merge_payload_variables_partition_list_multiple(default_payload): payload = default_payload.copy() - payload['CleansedPartitionFields'] = "ID,NAME" - [table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload=payload) + payload['CleansedPartitionFields'] = "ID|NAME" + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID","NAME"] assert partition_list == expected \ No newline at end of file diff --git a/src/metadata.ingest/ingest/Stored Procedures/GetDatasetPayload.sql b/src/metadata.ingest/ingest/Stored Procedures/GetDatasetPayload.sql index c8fe582d..0f7b8706 100644 --- a/src/metadata.ingest/ingest/Stored Procedures/GetDatasetPayload.sql +++ b/src/metadata.ingest/ingest/Stored Procedures/GetDatasetPayload.sql @@ -15,7 +15,7 @@ BEGIN INNER JOIN [common].[Connections] cn1 ON ds.[ConnectionFK] = cn1.[ConnectionId] INNER JOIN [common].[Connections] cn2 - ON ds.[RawStorageConnectionFK] = cn2.ConnectionId AND cn2.[SourceLocation] IN ('raw','bronze') + ON ds.[RawStorageConnectionFK] = cn2.ConnectionId INNER JOIN [common].[Connections] cn3 ON cn3.[ConnectionDisplayName] = 'PrimaryKeyVault' WHERE @@ -414,7 +414,7 @@ BEGIN INNER JOIN [common].[Connections] cn1 ON ds.[ConnectionFK] = cn1.[ConnectionId] INNER JOIN [common].[Connections] cn2 - ON ds.[RawStorageConnectionFK] = cn2.ConnectionId AND cn2.[SourceLocation] IN ('raw','bronze') + ON ds.[RawStorageConnectionFK] = cn2.ConnectionId INNER JOIN [common].[Connections] cn3 ON cn3.[ConnectionDisplayName] = 'PrimaryKeyVault' WHERE diff --git a/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql b/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql index a29a3b84..9f3d4d2b 100644 --- a/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql +++ b/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql @@ -32,11 +32,11 @@ BEGIN INNER JOIN [common].Connections AS cn5 ON - ds.RawStorageConnectionFK = cn5.ConnectionId AND cn5.SourceLocation IN ('raw','bronze') + ds.RawStorageConnectionFK = cn5.ConnectionId INNER JOIN [common].Connections AS cn6 ON - ds.CleansedStorageConnectionFK = cn6.ConnectionId AND cn6.SourceLocation IN ('cleansed', 'silver') + ds.CleansedStorageConnectionFK = cn6.ConnectionId WHERE ds.DatasetId = @DatasetId @@ -85,9 +85,9 @@ BEGIN -- Get attribute data as comma separated string values for the dataset SELECT - @CleansedColumnsList = STRING_AGG(CAST(att.AttributeName AS NVARCHAR(MAX)),','), - @CleansedColumnsTypeList = STRING_AGG(CAST(att.AttributeTargetDataType AS NVARCHAR(MAX)),','), - @CleansedColumnsFormatList = STRING_AGG(CAST(att.AttributeTargetDataFormat AS NVARCHAR(MAX)), ',') + @CleansedColumnsList = STRING_AGG(CAST(att.AttributeName AS NVARCHAR(MAX)),'|'), + @CleansedColumnsTypeList = STRING_AGG(CAST(att.AttributeTargetDataType AS NVARCHAR(MAX)),'|'), + @CleansedColumnsFormatList = STRING_AGG(CAST(att.AttributeTargetDataFormat AS NVARCHAR(MAX)), '|') FROM [ingest].[Datasets] AS ds INNER JOIN @@ -105,7 +105,7 @@ BEGIN -- Get pk columns as comma separated string values for the dataset SELECT - @PkAttributesList = STRING_AGG(CAST(att.AttributeName AS NVARCHAR(MAX)),',') + @PkAttributesList = STRING_AGG(CAST(att.AttributeName AS NVARCHAR(MAX)),'|') FROM [ingest].[Datasets] AS ds INNER JOIN @@ -125,7 +125,7 @@ BEGIN -- Get pk columns as comma separated string values for the dataset SELECT - @PkAttributesList = STRING_AGG(CAST(att.AttributeName AS NVARCHAR(MAX)),',') + @PkAttributesList = STRING_AGG(CAST(att.AttributeName AS NVARCHAR(MAX)),'|') FROM [ingest].[Datasets] AS ds INNER JOIN @@ -251,11 +251,11 @@ BEGIN INNER JOIN [common].Connections AS cn5 ON - ds.RawStorageConnectionFK = cn5.ConnectionId AND cn5.SourceLocation IN ('raw','bronze') + ds.RawStorageConnectionFK = cn5.ConnectionId INNER JOIN [common].Connections AS cn6 ON - ds.CleansedStorageConnectionFK = cn6.ConnectionId AND cn6.SourceLocation IN ('cleansed', 'silver') + ds.CleansedStorageConnectionFK = cn6.ConnectionId INNER JOIN [common].Connections AS cn7 ON cn7.ConnectionDisplayName = 'PrimaryKeyVault' diff --git a/src/metadata.transform/transform/Stored Procedures/GetNotebookPayload.sql b/src/metadata.transform/transform/Stored Procedures/GetNotebookPayload.sql index 0466b245..1df6660c 100644 --- a/src/metadata.transform/transform/Stored Procedures/GetNotebookPayload.sql +++ b/src/metadata.transform/transform/Stored Procedures/GetNotebookPayload.sql @@ -70,8 +70,8 @@ BEGIN -- Get attribute data as comma separated string values for the dataset. This excludes the SurrogateKey. SELECT - @CuratedColumnsList = STRING_AGG(att.AttributeName,','), - @CuratedColumnsTypeList = STRING_AGG(att.AttributeTargetDataType,',') + @CuratedColumnsList = STRING_AGG(att.AttributeName,'|'), + @CuratedColumnsTypeList = STRING_AGG(att.AttributeTargetDataType,'|') FROM [transform].[Datasets] AS ds INNER JOIN @@ -107,7 +107,7 @@ BEGIN -- Get Bk columns as comma separated string values for the dataset SELECT - @BkAttributesList = STRING_AGG(att.AttributeName,',') + @BkAttributesList = STRING_AGG(att.AttributeName,'|') FROM [transform].[Datasets] AS ds INNER JOIN @@ -123,7 +123,7 @@ BEGIN -- Get partitionby columns as comma separated string values for the dataset SELECT - @PartitionByAttributesList = STRING_AGG(att.AttributeName,',') + @PartitionByAttributesList = STRING_AGG(att.AttributeName,'|') FROM [transform].[Datasets] AS ds INNER JOIN From 75f9771c58c07e99d5e3ba011778eaa50186006c Mon Sep 17 00:00:00 2001 From: MattPCollins Date: Tue, 20 Jan 2026 08:41:55 +0000 Subject: [PATCH 2/4] Update schema and connection references --- .../ingestpayload/CheckPayloadExecution.py | 6 ++-- .../ingest/ingestpayload/IngestExecution.py | 24 ++++++------- .../ingest/utils/ConfigurePayloadVariables.py | 6 ++-- .../test_IngestConfigurePayloadVariables.py | 34 +++++++++---------- .../Stored Procedures/GetMergePayload.sql | 4 +-- 5 files changed, 37 insertions(+), 37 deletions(-) diff --git a/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py b/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py index 860f3968..eb3925eb 100644 --- a/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py +++ b/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py @@ -48,7 +48,7 @@ # COMMAND ---------- -[table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload) +[table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload) # COMMAND ---------- @@ -105,7 +105,7 @@ # COMMAND ---------- -schema_exists = check_exists_delta_schema(schema_name = cleansed_schema_name) +schema_exists = check_exists_delta_schema(schema_name = cleansed_connection_name) # COMMAND ---------- @@ -114,7 +114,7 @@ # COMMAND ---------- -cleansed_table_path = set_table_path(schema_name =cleansed_schema_name, table_name =table_name) +cleansed_table_path = set_table_path(schema_name =cleansed_connection_name, table_name =table_name) print(cleansed_table_path) # COMMAND ---------- diff --git a/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py b/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py index 5433508f..5499985c 100644 --- a/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py +++ b/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py @@ -61,7 +61,7 @@ # COMMAND ---------- -[table_name, load_type, load_action, load_action_text, version_number, raw_storage_name, raw_container_name, raw_secret, raw_last_load_date, raw_schema_name, raw_file_type, datetime_folder_hierarchy, cleansed_storage_name, cleansed_container_name, cleansed_secret, cleansed_last_load_date, cleansed_schema_name, filter_condition, pk_list, partition_list, columns_list, columns_type_list, columns_format_list, metadata_column_list, metadata_column_type_list, metadata_column_format_list, total_column_list, total_column_type_list, total_column_format_list] = get_merge_payload_variables(payload) +[table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload) # COMMAND ---------- @@ -94,7 +94,7 @@ } #different options for specifying, based on how we save abfss folder hierarchy. - file_full_path = f"{raw_abfss_path}/{raw_schema_name}/{table_name}/version={version_number}/{load_action_text}/{datetime_folder_hierarchy}/{table_name}.{raw_file_type}" + file_full_path = f"{raw_abfss_path}/{raw_connection_name}/{table_name}/version={version_number}/{load_action_text}/{datetime_folder_hierarchy}/{table_name}.{raw_file_type}" print(file_full_path) # assuming json,csv, parquet @@ -104,7 +104,7 @@ .load(file_full_path) elif raw_file_type == "delta": - file_full_path = f"{raw_abfss_path}/{raw_schema_name}/{table_name}" + file_full_path = f"{raw_abfss_path}/{raw_connection_name}/{raw_source_path}/{raw_source_name}" df = spark.read.format("delta").load(file_full_path) # apply incremental loading filter @@ -185,20 +185,20 @@ # check Delta Objects exist (import check functions) # check schema exists -schema_exists = check_exists_delta_schema(schema_name=cleansed_schema_name) +schema_exists = check_exists_delta_schema(schema_name=cleansed_connection_name) # create schema, if required if schema_exists == False: - create_schema(container_name=cleansed_container_name, schema_name=cleansed_schema_name) + create_schema(container_name=cleansed_container_name, schema_name=cleansed_connection_name) # COMMAND ---------- # check Delta Objects exist (import check functions) # set Delta Table file path -location = set_delta_table_location(schema_name=cleansed_schema_name, table_name=table_name, abfss_path=cleansed_abfss_path) +location = set_delta_table_location(schema_name=cleansed_connection_name, table_name=table_name, abfss_path=cleansed_abfss_path) # check Delta table exists -cleansed_table_path = set_table_path(schema_name =cleansed_schema_name, table_name =table_name) +cleansed_table_path = set_table_path(schema_name =cleansed_connection_name, table_name =table_name) table_exists = check_exists_delta_table(table_path = cleansed_table_path, load_action = load_action, load_type = load_type) # Create Delta table, if required @@ -206,11 +206,11 @@ if table_exists == False: columns_string = format_columns_sql(total_column_list, total_column_type_list) - create_table(container_name=cleansed_container_name, schema_name=cleansed_schema_name, table_name=table_name,location=location, partition_fields_sql=partition_fields_sql, columns_string=columns_string) + create_table(container_name=cleansed_container_name, schema_name=cleansed_connection_name, table_name=table_name,location=location, partition_fields_sql=partition_fields_sql, columns_string=columns_string) table_created = True # get operations metrics - output = get_operation_metrics(schema_name=cleansed_schema_name, table_name=table_name, output=output) + output = get_operation_metrics(schema_name=cleansed_connection_name, table_name=table_name, output=output) # COMMAND ---------- @@ -224,13 +224,13 @@ else: raise Exception("load_action not supported.") -target_delta = get_target_delta_table(schema_name = cleansed_schema_name, table_name=table_name) +target_delta = get_target_delta_table(schema_name = cleansed_connection_name, table_name=table_name) -write_to_delta_executor(write_mode=write_mode, target_df=target_delta, df=df, schema_name=cleansed_schema_name, table_name=table_name, pk_fields=pk_list, columns_list=total_column_list, partition_fields=partition_list) +write_to_delta_executor(write_mode=write_mode, target_df=target_delta, df=df, schema_name=cleansed_connection_name, table_name=table_name, pk_fields=pk_list, columns_list=total_column_list, partition_fields=partition_list) # COMMAND ---------- -output = get_operation_metrics(schema_name=cleansed_schema_name, table_name=table_name, output=output) +output = get_operation_metrics(schema_name=cleansed_connection_name, table_name=table_name, output=output) print(output) # COMMAND ---------- diff --git a/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py b/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py index daa21f5a..944c928f 100644 --- a/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py +++ b/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py @@ -11,7 +11,7 @@ def get_merge_payload_variables(payload: dict()) -> list(): raw_secret = payload["RawStorageAccessKey"] raw_last_load_date = payload["RawLastLoadDate"] - raw_schema_name = payload["RawSchemaName"] # connection name of data source + raw_connection_name = payload["RawConnectionName"] # connection name of data source raw_file_type = payload["RawFileType"] raw_source_path = payload["RawSourcePath"] # physical path in raw container raw_source_name = payload["RawSourceName"] # file name in raw container @@ -23,7 +23,7 @@ def get_merge_payload_variables(payload: dict()) -> list(): cleansed_secret = payload["CleansedStorageAccessKey"] cleansed_last_load_date = payload["CleansedLastLoadDate"] - cleansed_schema_name = payload["CleansedSchemaName"] # connection name of data source + cleansed_connection_name = payload["CleansedConnectionName"] # connection name of data source cleansed_source_path = payload["CleansedSourcePath"] # desired physical path in cleansed container cleansed_source_name = payload["CleansedSourceName"] # desired file name in cleansed container @@ -44,6 +44,6 @@ def get_merge_payload_variables(payload: dict()) -> list(): total_column_type_list = columns_type_list + metadata_column_type_list total_column_format_list = columns_format_list + metadata_column_format_list - output = [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] + output = [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] return output diff --git a/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py b/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py index 0078d911..7b4be34a 100644 --- a/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py +++ b/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py @@ -5,7 +5,7 @@ @pytest.fixture def default_payload() -> dict: payload = { - "RawSchemaName":"Oracle", + "RawConnectionName":"Oracle", "ComputeName":"CF.Cumulus.Ingest.Compute", "ComputeWorkspaceURL":"https://devdatabricksguid.azuredatabricks.net/", "ComputeClusterId":"", @@ -28,7 +28,7 @@ def default_payload() -> dict: "SourceName":"testdataset", "RawFileType":"parquet", "VersionNumber":1, - "CleansedSchemaName":"Oracle", + "CleansedConnectionName":"Oracle", "Enabled":True, "LoadType":"I", "LoadAction":"F", @@ -49,92 +49,92 @@ def default_payload() -> dict: return payload def test_get_merge_payload_variables_table_name(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'testdataset' assert table_name == expected def test_get_merge_payload_variables_load_type_default(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'I' assert load_type == expected def test_get_merge_payload_variables_load_type_override(default_payload): payload = default_payload.copy() payload['LoadType'] = 'F' - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = 'F' assert load_type == expected def test_get_merge_payload_variables_load_action_default(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'F' assert load_action == expected def test_get_merge_payload_variables_load_action_override(default_payload): payload = default_payload.copy() payload['LoadAction'] = 'I' - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = 'I' assert load_action == expected def test_get_merge_payload_variables_load_action_text(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'full' assert load_action_text == expected def test_get_merge_payload_variables_version_number(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = '0001' assert version_number == expected def test_get_merge_payload_variables_version_number_tens(default_payload): payload = default_payload.copy() payload['VersionNumber'] = 10 - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = '0010' assert version_number == expected def test_get_merge_payload_variables_version_number_hundreds(default_payload): payload = default_payload.copy() payload['VersionNumber'] = 151 - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = '0151' assert version_number == expected def test_get_merge_payload_variables_pk_list_empty(default_payload): payload = default_payload.copy() payload['CleansedPkList'] = "" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = [''] assert pk_list == expected def test_get_merge_payload_variables_pk_list_single(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = ['ID'] assert pk_list == expected def test_get_merge_payload_variables_pk_list_multiple(default_payload): payload = default_payload.copy() payload['CleansedPkList'] = "ID|NAME" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID","NAME"] assert pk_list == expected def test_get_merge_payload_variables_partition_list_empty(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = [] assert partition_list == expected def test_get_merge_payload_variables_partition_list_single(default_payload): payload = default_payload.copy() payload['CleansedPartitionFields'] = "ID" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID"] assert partition_list == expected def test_get_merge_payload_variables_partition_list_multiple(default_payload): payload = default_payload.copy() payload['CleansedPartitionFields'] = "ID|NAME" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_schema_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_schema_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID","NAME"] assert partition_list == expected \ No newline at end of file diff --git a/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql b/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql index 9f3d4d2b..80f566ab 100644 --- a/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql +++ b/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql @@ -192,7 +192,7 @@ BEGIN ds.[Enabled] = 1 SELECT - [cn].[ConnectionDisplayName] AS 'RawSchemaName', + [cn].[ConnectionDisplayName] AS 'RawConnectionName', [cn2].[ConnectionDisplayName] AS 'ComputeName', [cn2].[ConnectionLocation] AS 'ComputeWorkspaceURL', [cn2].[ComputeLocation] AS 'ComputeClusterId', @@ -216,7 +216,7 @@ BEGIN ds.SourceName, ds.ExtensionType AS 'RawFileType', ds.VersionNumber, - [cn].[ConnectionDisplayName] AS 'CleansedSchemaName', + [cn].[ConnectionDisplayName] AS 'CleansedConnectionName', ds.CleansedName AS 'CleansedTableName', ds.Enabled, ds.LoadType, From 74b8370d60c4c4362728f487210499f2969f86e8 Mon Sep 17 00:00:00 2001 From: MattPCollins Date: Tue, 20 Jan 2026 08:43:20 +0000 Subject: [PATCH 3/4] Update delta path --- .../python/notebooks/ingest/ingestpayload/IngestExecution.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py b/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py index 5499985c..2d188e3a 100644 --- a/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py +++ b/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py @@ -104,7 +104,7 @@ .load(file_full_path) elif raw_file_type == "delta": - file_full_path = f"{raw_abfss_path}/{raw_connection_name}/{raw_source_path}/{raw_source_name}" + file_full_path = f"{raw_abfss_path}/{raw_source_path}/{raw_source_name}" df = spark.read.format("delta").load(file_full_path) # apply incremental loading filter From 291781d4c715de4f2d538b2b9294744e34307d24 Mon Sep 17 00:00:00 2001 From: MattPCollins Date: Tue, 20 Jan 2026 12:16:46 +0000 Subject: [PATCH 4/4] Update metadata for additional configurations in ingest metadata --- .../ingestpayload/CheckPayloadExecution.py | 6 +- .../ingest/ingestpayload/IngestExecution.py | 20 +++---- .../ingest/utils/ConfigurePayloadVariables.py | 12 ++-- .../test_IngestConfigurePayloadVariables.py | 59 ++++++++++++------- .../Stored Procedures/GetMergePayload.sql | 6 +- 5 files changed, 61 insertions(+), 42 deletions(-) diff --git a/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py b/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py index eb3925eb..2489fdd1 100644 --- a/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py +++ b/src/azure.databricks/python/notebooks/ingest/ingestpayload/CheckPayloadExecution.py @@ -48,7 +48,7 @@ # COMMAND ---------- -[table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload) +[table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload) # COMMAND ---------- @@ -105,7 +105,7 @@ # COMMAND ---------- -schema_exists = check_exists_delta_schema(schema_name = cleansed_connection_name) +schema_exists = check_exists_delta_schema(schema_name = cleansed_folder) # COMMAND ---------- @@ -114,7 +114,7 @@ # COMMAND ---------- -cleansed_table_path = set_table_path(schema_name =cleansed_connection_name, table_name =table_name) +cleansed_table_path = set_table_path(schema_name =cleansed_folder, table_name =table_name) print(cleansed_table_path) # COMMAND ---------- diff --git a/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py b/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py index 2d188e3a..4fcea907 100644 --- a/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py +++ b/src/azure.databricks/python/notebooks/ingest/ingestpayload/IngestExecution.py @@ -61,7 +61,7 @@ # COMMAND ---------- -[table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload) +[table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload) # COMMAND ---------- @@ -104,7 +104,7 @@ .load(file_full_path) elif raw_file_type == "delta": - file_full_path = f"{raw_abfss_path}/{raw_source_path}/{raw_source_name}" + file_full_path = f"{raw_abfss_path}/{raw_path}/{raw_name}" df = spark.read.format("delta").load(file_full_path) # apply incremental loading filter @@ -185,20 +185,20 @@ # check Delta Objects exist (import check functions) # check schema exists -schema_exists = check_exists_delta_schema(schema_name=cleansed_connection_name) +schema_exists = check_exists_delta_schema(schema_name=cleansed_folder) # create schema, if required if schema_exists == False: - create_schema(container_name=cleansed_container_name, schema_name=cleansed_connection_name) + create_schema(container_name=cleansed_container_name, schema_name=cleansed_folder) # COMMAND ---------- # check Delta Objects exist (import check functions) # set Delta Table file path -location = set_delta_table_location(schema_name=cleansed_connection_name, table_name=table_name, abfss_path=cleansed_abfss_path) +location = set_delta_table_location(schema_name=cleansed_folder, table_name=table_name, abfss_path=cleansed_abfss_path) # check Delta table exists -cleansed_table_path = set_table_path(schema_name =cleansed_connection_name, table_name =table_name) +cleansed_table_path = set_table_path(schema_name =cleansed_folder, table_name =table_name) table_exists = check_exists_delta_table(table_path = cleansed_table_path, load_action = load_action, load_type = load_type) # Create Delta table, if required @@ -206,11 +206,11 @@ if table_exists == False: columns_string = format_columns_sql(total_column_list, total_column_type_list) - create_table(container_name=cleansed_container_name, schema_name=cleansed_connection_name, table_name=table_name,location=location, partition_fields_sql=partition_fields_sql, columns_string=columns_string) + create_table(container_name=cleansed_container_name, schema_name=cleansed_folder, table_name=table_name,location=location, partition_fields_sql=partition_fields_sql, columns_string=columns_string) table_created = True # get operations metrics - output = get_operation_metrics(schema_name=cleansed_connection_name, table_name=table_name, output=output) + output = get_operation_metrics(schema_name=cleansed_folder, table_name=table_name, output=output) # COMMAND ---------- @@ -226,11 +226,11 @@ target_delta = get_target_delta_table(schema_name = cleansed_connection_name, table_name=table_name) -write_to_delta_executor(write_mode=write_mode, target_df=target_delta, df=df, schema_name=cleansed_connection_name, table_name=table_name, pk_fields=pk_list, columns_list=total_column_list, partition_fields=partition_list) +write_to_delta_executor(write_mode=write_mode, target_df=target_delta, df=df, schema_name=cleansed_folder, table_name=table_name, pk_fields=pk_list, columns_list=total_column_list, partition_fields=partition_list) # COMMAND ---------- -output = get_operation_metrics(schema_name=cleansed_connection_name, table_name=table_name, output=output) +output = get_operation_metrics(schema_name=cleansed_folder, table_name=table_name, output=output) print(output) # COMMAND ---------- diff --git a/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py b/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py index 944c928f..e38b6cd0 100644 --- a/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py +++ b/src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py @@ -13,8 +13,8 @@ def get_merge_payload_variables(payload: dict()) -> list(): raw_connection_name = payload["RawConnectionName"] # connection name of data source raw_file_type = payload["RawFileType"] - raw_source_path = payload["RawSourcePath"] # physical path in raw container - raw_source_name = payload["RawSourceName"] # file name in raw container + raw_path = payload["RawPath"] # physical path in raw container + raw_name = payload["RawName"] # file name in raw container datetime_folder_hierarchy = payload["DateTimeFolderHierarchy"] @@ -24,8 +24,10 @@ def get_merge_payload_variables(payload: dict()) -> list(): cleansed_last_load_date = payload["CleansedLastLoadDate"] cleansed_connection_name = payload["CleansedConnectionName"] # connection name of data source - cleansed_source_path = payload["CleansedSourcePath"] # desired physical path in cleansed container - cleansed_source_name = payload["CleansedSourceName"] # desired file name in cleansed container + cleansed_path = payload["CleansedPath"] # desired physical path in cleansed container + cleansed_name = payload["CleansedName"] # desired file name in cleansed container + + cleansed_folder = cleansed_path if raw_file_type == "delta" else cleansed_connection_name filter_condition = payload["FilterCondition"] @@ -44,6 +46,6 @@ def get_merge_payload_variables(payload: dict()) -> list(): total_column_type_list = columns_type_list + metadata_column_type_list total_column_format_list = columns_format_list + metadata_column_format_list - output = [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] + output = [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] return output diff --git a/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py b/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py index 7b4be34a..62ecebe8 100644 --- a/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py +++ b/src/azure.databricks/python/tests/test_IngestConfigurePayloadVariables.py @@ -24,8 +24,8 @@ def default_payload() -> dict: "CleansedStorageAccessKey":"devdatalakestoragecleansedaccesskey", "KeyVaultAddress":"https://devkeyvault.vault.azure.net/", "DatasetDisplayName":"testdataset", - "SourcePath":"dbo", - "SourceName":"testdataset", + "RawPath":"dbo", + "RawName":"testdataset", "RawFileType":"parquet", "VersionNumber":1, "CleansedConnectionName":"Oracle", @@ -41,100 +41,115 @@ def default_payload() -> dict: "CleansedPartitionFields":"", "DateTimeFolderHierarchy":"year=2025/month=01/day=1/hour=10", "FilterCondition": "", - "RawSourcePath": "folder", - "RawSourceName": "testdataset", - "CleansedSourcePath": "test_folder", - "CleansedSourceName": "cleanseddataset", + "RawPath": "folder", + "RawName": "testdataset", + "CleansedPath": "test_folder", + "CleansedName": "cleanseddataset", } return payload def test_get_merge_payload_variables_table_name(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'testdataset' assert table_name == expected def test_get_merge_payload_variables_load_type_default(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'I' assert load_type == expected def test_get_merge_payload_variables_load_type_override(default_payload): payload = default_payload.copy() payload['LoadType'] = 'F' - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = 'F' assert load_type == expected def test_get_merge_payload_variables_load_action_default(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'F' assert load_action == expected def test_get_merge_payload_variables_load_action_override(default_payload): payload = default_payload.copy() payload['LoadAction'] = 'I' - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = 'I' assert load_action == expected def test_get_merge_payload_variables_load_action_text(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = 'full' assert load_action_text == expected def test_get_merge_payload_variables_version_number(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = '0001' assert version_number == expected def test_get_merge_payload_variables_version_number_tens(default_payload): payload = default_payload.copy() payload['VersionNumber'] = 10 - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = '0010' assert version_number == expected def test_get_merge_payload_variables_version_number_hundreds(default_payload): payload = default_payload.copy() payload['VersionNumber'] = 151 - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = '0151' assert version_number == expected def test_get_merge_payload_variables_pk_list_empty(default_payload): payload = default_payload.copy() payload['CleansedPkList'] = "" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = [''] assert pk_list == expected def test_get_merge_payload_variables_pk_list_single(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = ['ID'] assert pk_list == expected def test_get_merge_payload_variables_pk_list_multiple(default_payload): payload = default_payload.copy() payload['CleansedPkList'] = "ID|NAME" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID","NAME"] assert pk_list == expected def test_get_merge_payload_variables_partition_list_empty(default_payload): - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=default_payload) expected = [] assert partition_list == expected def test_get_merge_payload_variables_partition_list_single(default_payload): payload = default_payload.copy() payload['CleansedPartitionFields'] = "ID" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID"] assert partition_list == expected def test_get_merge_payload_variables_partition_list_multiple(default_payload): payload = default_payload.copy() payload['CleansedPartitionFields'] = "ID|NAME" - [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_source_path,raw_source_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_source_path,cleansed_source_name,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) expected = ["ID","NAME"] - assert partition_list == expected \ No newline at end of file + assert partition_list == expected + + + +def test_get_merge_payload_variables_cleansed_folder_parquet(default_payload): + payload = default_payload.copy() + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + expected = "Oracle" + assert cleansed_folder == expected + +def test_get_merge_payload_variables_cleansed_folder_delta(default_payload): + payload = default_payload.copy() + payload['RawFileType'] = "delta" + [table_name,load_type,load_action,load_action_text,version_number,raw_storage_name,raw_container_name,raw_secret,raw_last_load_date,raw_connection_name,raw_file_type,raw_path,raw_name,datetime_folder_hierarchy,cleansed_storage_name,cleansed_container_name,cleansed_secret,cleansed_last_load_date,cleansed_connection_name,cleansed_path,cleansed_name,cleansed_folder,filter_condition,pk_list,partition_list,columns_list,columns_type_list,columns_format_list,metadata_column_list,metadata_column_type_list,metadata_column_format_list,total_column_list,total_column_type_list,total_column_format_list] = get_merge_payload_variables(payload=payload) + expected = "test_folder" + assert cleansed_folder == expected \ No newline at end of file diff --git a/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql b/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql index 80f566ab..2aeaeb1c 100644 --- a/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql +++ b/src/metadata.ingest/ingest/Stored Procedures/GetMergePayload.sql @@ -212,8 +212,10 @@ BEGIN [cn7].[ConnectionLocation] AS 'KeyVaultAddress', ds.DatasetDisplayName, - ds.SourcePath, - ds.SourceName, + ds.SourcePath AS 'RawPath', + ds.SourceName AS 'RawName', + ds.CleansedPath AS 'CleansedPath', + ds.CleansedName AS 'CleansedName', ds.ExtensionType AS 'RawFileType', ds.VersionNumber, [cn].[ConnectionDisplayName] AS 'CleansedConnectionName',