Merge branch 'master' of github.com:starlake-ai/starlake

Author: hayssams

CHANGELOG.md

@@ -27,6 +27,7 @@ __Bug fix__:
 - JSON type ingestion in bigquery are now created with JSON type instead of String.
 - excluded table during data extraction defined in jdbcSchema are now taken into account
 - if column is renamed, check pattern of renamed column instead of original name since it is the target table column's name during schema extraction
+- restore time expiration for temporary tables
 - **BREAKING CHANGE** when no fields could be inferred from input, inferred schema now fails
 
 

src/main/scala/ai/starlake/job/ingest/loaders/BigQueryNativeLoader.scala

@@ -17,6 +17,7 @@ import com.google.cloud.bigquery
 import com.google.cloud.bigquery.{Field, JobInfo, StandardSQLTypeName, TableId}
 import com.typesafe.scalalogging.StrictLogging
 
+import java.time.Duration
 import scala.jdk.CollectionConverters._
 import scala.util.{Failure, Success, Try}
 
@@ -89,8 +90,8 @@ class BigQueryNativeLoader(ingestionJob: IngestionJob, accessToken: Option[Strin
                   _.targetBqSchemaWithIgnoreAndScript(schemaHandler)
                 )
 
-                val enrichedTableInfo = firstStepTableInfo.copy(maybeSchema =
-                  firstStepTableInfo.maybeSchema.map((schema: bigquery.Schema) =>
+                val enrichedTableInfo = firstStepTableInfo.copy(
+                  maybeSchema = firstStepTableInfo.maybeSchema.map((schema: bigquery.Schema) =>
                     bigquery.Schema.of(
                       (schema.getFields.asScala :+
                       Field
@@ -101,7 +102,8 @@ class BigQueryNativeLoader(ingestionJob: IngestionJob, accessToken: Option[Strin
                         .setDefaultValueExpression(s"'$sourceUri'")
                         .build()).asJava
                     )
-                  )
+                  ),
+                  maybeDurationMs = Some(Duration.ofHours(24).toMillis)
                 )
 
                 // TODO What if type is changed by transform ? we need to use safe_cast to have the same behavior as in SPARK.

src/main/scala/ai/starlake/job/sink/bigquery/BigQueryJobBase.scala

@@ -509,15 +509,9 @@ trait BigQueryJobBase extends StrictLogging {
               .newBuilder(targetTableId, tableDefinition)
               .setDescription(tableInfo.maybeTableDescription.orNull)
 
-            if (tableInfo.maybePartition.isEmpty) {
-              cliConfig.days match {
-                case Some(days) =>
-                  bqTableInfoBuilder.setExpirationTime(
-                    System.currentTimeMillis() + days * 24 * 3600 * 1000
-                  )
-                case None =>
-              }
-            }
+            tableInfo.maybeDurationMs.foreach(d =>
+              bqTableInfoBuilder.setExpirationTime(System.currentTimeMillis() + d)
+            )
 
             val bqTableInfo = bqTableInfoBuilder.build
             logger.info(s"Creating table ${targetTableId.getDataset}.${targetTableId.getTable}")

src/main/scala/ai/starlake/schema/model/TableInfo.scala

@@ -8,5 +8,6 @@ case class TableInfo(
   maybeSchema: Option[BQSchema] = None,
   maybePartition: Option[FieldPartitionInfo] = None,
   maybeCluster: Option[ClusteringInfo] = None,
-  attributes: List[AttributeDesc] = Nil
+  attributes: List[AttributeDesc] = Nil,
+  maybeDurationMs: Option[Long] = None
 )