From aeef7ff179dadc7f5d4138af387ccb504ff36403 Mon Sep 17 00:00:00 2001
From: "sourav.banerjee" <sourav.banerjee@databricks.com>
Date: Tue, 11 Jun 2024 17:59:51 +0530
Subject: [PATCH] Dropped Spec column from snapshot

---
 .../overwatch/pipeline/BronzeTransforms.scala | 35 ++++++++++---------
 1 file changed, 18 insertions(+), 17 deletions(-)

diff --git a/src/main/scala/com/databricks/labs/overwatch/pipeline/BronzeTransforms.scala b/src/main/scala/com/databricks/labs/overwatch/pipeline/BronzeTransforms.scala
index 6f24309f3..ad1350dc5 100644
--- a/src/main/scala/com/databricks/labs/overwatch/pipeline/BronzeTransforms.scala
+++ b/src/main/scala/com/databricks/labs/overwatch/pipeline/BronzeTransforms.scala
@@ -561,25 +561,26 @@ trait BronzeTransforms extends SparkSessionWrapper {
             .withColumn(s"azure_attributes", SchemaTools.structToMap(scrubbedDF, s"azure_attributes"))
             .withColumn(s"gcp_attributes", SchemaTools.structToMap(scrubbedDF, s"gcp_attributes"))
             .withColumn("organization_id", lit(config.organizationId))
+            .drop("spec")
             .verifyMinimumSchema(clusterSnapMinimumSchema)
 
-          val finalDF = df.schema.fields.find(_.name == "spec") match {
-            case Some(field) =>
-              field.dataType match {
-                case structType: StructType =>
-                  val newFields = structType.fields.map { f =>
-                    f.dataType match {
-                      case _: StructType => SchemaTools.structToMap(df, s"spec.${f.name}").as(f.name)
-                      case _ => col(s"spec.${f.name}").as(f.name)
-                    }
-                  }
-                  df.withColumn("spec", struct(newFields: _*))
-                case _ => df
-              }
-            case None => df
-          }
-
-          finalDF
+//          val finalDF = df.schema.fields.find(_.name == "spec") match {
+//            case Some(field) =>
+//              field.dataType match {
+//                case structType: StructType =>
+//                  val newFields = structType.fields.map { f =>
+//                    f.dataType match {
+//                      case _: StructType => SchemaTools.structToMap(df, s"spec.${f.name}").as(f.name)
+//                      case _ => col(s"spec.${f.name}").as(f.name)
+//                    }
+//                  }
+//                  df.withColumn("spec", struct(newFields: _*))
+//                case _ => df
+//              }
+//            case None => df
+//          }
+
+          df
             .verifyMinimumSchema(clusterSnapMinimumSchema)
 
         } else {