Develop ingest bugs (#188)

* Update code for bugs related to ingest module.

* Update schema and connection references

* Update delta path

* Update metadata for additional configurations in ingest metadata

Author: MattPCollins

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_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 ----------
 
@@ -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
 
@@ -113,7 +105,7 @@
 
 # COMMAND ----------
 
-schema_exists = check_exists_delta_schema(schema_name = cleansed_schema_name)
+schema_exists = check_exists_delta_schema(schema_name = cleansed_folder)
 
 # COMMAND ----------
 
@@ -122,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_folder, table_name =table_name)
 print(cleansed_table_path)
 
 # COMMAND ----------

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_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 ----------
@@ -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_path}/{raw_name}"
     df = spark.read.format("delta").load(file_full_path)
 
     # apply incremental loading filter
@@ -185,32 +185,32 @@
 
 # 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_folder)
 
 # 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_folder)
 
 # 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_folder, 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_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
 table_created = False
 
 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_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_schema_name, table_name=table_name, output=output)
+    output = get_operation_metrics(schema_name=cleansed_folder, 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_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_schema_name, table_name=table_name, output=output)
+output = get_operation_metrics(schema_name=cleansed_folder, table_name=table_name, output=output)
 print(output)
 
 # COMMAND ----------

src/azure.databricks/python/notebooks/ingest/utils/ConfigurePayloadVariables.py

@@ -11,42 +11,41 @@ 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_connection_name = payload["RawConnectionName"] # connection name of data source
+    raw_file_type = payload["RawFileType"] 
+    raw_path = payload["RawPath"] # physical path in raw container
+    raw_name = payload["RawName"] # file name in raw container
+    
     datetime_folder_hierarchy = payload["DateTimeFolderHierarchy"]
 
     cleansed_storage_name = payload["CleansedStorageName"]
     cleansed_container_name = payload["CleansedContainerName"]
     cleansed_secret = payload["CleansedStorageAccessKey"]
     cleansed_last_load_date = payload["CleansedLastLoadDate"]
 
-    cleansed_schema_name = payload["CleansedSchemaName"] 
+    cleansed_connection_name = payload["CleansedConnectionName"] # connection name of data source
+    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"] 
 
     # 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_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

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"]
 

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: