Feature/config job timeout

core/src/main/scala/no/nrk/bigquery/BQPoll.scala

@@ -73,7 +73,8 @@ object BQPoll {
 
     F.sleep(maxDuration).race(go(runningJob, Nil, Nil)).flatMap {
       case Left(_) =>
-        F.raiseError(BQExecutionException(runningJob.getJobId, None, Nil))
+        val error = new BigQueryError("job-wait-timeout", "", s"timeout after ${maxDuration.toMinutes} minutes")
+        F.raiseError(BQExecutionException(runningJob.getJobId, Some(error), Nil))
       case Right(finished) => F.pure(finished)
     }
   }

core/src/main/scala/no/nrk/bigquery/BigQueryClient.scala

@@ -44,7 +44,7 @@ class BigQueryClient[F[_]](
     bigQuery: BigQuery,
     val reader: BigQueryReadClient,
     val metricOps: MetricsOps[F],
-    defaults: Option[BQClientDefaults]
+    config: BigQueryClient.Config
 )(implicit F: Async[F], lf: LoggerFactory[F]) {
   private val logger = lf.getLogger
   private implicit def showJob[J <: JobInfo]: Show[J] = Show.show(Jsonify.job)
@@ -472,7 +472,7 @@ class BigQueryClient[F[_]](
               .poll[F](
                 runningJob,
                 baseDelay = 3.second,
-                maxDuration = 20.minutes,
+                maxDuration = config.jobTimeout,
                 maxErrorsTolerated = 10
               )(
                 retry = F.interruptible(bigQuery.getJob(runningJob.getJobId))
@@ -606,7 +606,7 @@ class BigQueryClient[F[_]](
     F.blocking(bigQuery.delete(RoutineId.of(udfId.dataset.project.value, udfId.dataset.id, udfId.name.value)))
 
   private def freshJobId(id: BQJobId): F[JobId] = {
-    val withDefaults = id.withDefaults(defaults)
+    val withDefaults = id.withDefaults(config.defaults)
 
     F.delay(
       JobId
@@ -623,6 +623,19 @@ object BigQueryClient {
   val readTimeoutSecs = 20L
   val connectTimeoutSecs = 60L
 
+  case class Config(
+      jobTimeout: FiniteDuration,
+      defaults: Option[BQClientDefaults]
+  )
+
+  object Config {
+    val default: Config =
+      Config(
+        jobTimeout = new FiniteDuration(20, TimeUnit.MINUTES),
+        defaults = None
+      )
+  }
+
   def defaultConfigure(
       builder: BigQueryOptions.Builder
   ): BigQueryOptions.Builder =
@@ -695,12 +708,12 @@ object BigQueryClient {
       credentials: Credentials,
       metricsOps: MetricsOps[F],
       configure: Option[BigQueryOptions.Builder => BigQueryOptions.Builder] = None,
-      clientDefaults: Option[BQClientDefaults] = None
+      clientConfig: Option[BigQueryClient.Config] = None
   ): Resource[F, BigQueryClient[F]] =
     for {
       bq <- Resource.eval(
-        BigQueryClient.fromCredentials(credentials, configure, clientDefaults)
+        BigQueryClient.fromCredentials(credentials, configure, clientConfig.flatMap(_.defaults))
       )
       bqRead <- BigQueryClient.readerResource(credentials)
-    } yield new BigQueryClient(bq, bqRead, metricsOps, clientDefaults)
+    } yield new BigQueryClient(bq, bqRead, metricsOps, clientConfig.getOrElse(BigQueryClient.Config.default))
 }

docs/example_query.md

@@ -15,7 +15,7 @@ object Schemas {
   object UserEventSchema {
     private val timestamp: BQField = BQField("timestamp", BQField.Type.TIMESTAMP, BQField.Mode.REQUIRED)
     val tableDef: BQTableDef.Table[LocalDate] = BQTableDef.Table(
-      BQTableId.unsafeOf(BQDataset.unsafeOf(ProjectId.unsafeFromString("my-gcp-project"), "prod", Some(LocationId.EU)), "user_log"),
+      BQTableId.unsafeOf(BQDataset.unsafeOf(ProjectId.unsafeFromString("my-gcp-project"), "prod", Some(LocationId.EU)).toRef, "user_log"),
       BQSchema.of(
         BQField("eventId", BQField.Type.STRING, BQField.Mode.REQUIRED),
         timestamp,
@@ -36,7 +36,7 @@ object Schemas {
       BQField("lastName", BQField.Type.STRING, BQField.Mode.REQUIRED)
     )
     val tableDef: BQTableDef.Table[Unit] = BQTableDef.Table(
-      BQTableId.unsafeOf(BQDataset.unsafeOf(ProjectId.unsafeFromString("my-gcp-project"), "prod").withLocation(LocationId.EU), "users"),
+      BQTableId.unsafeOf(BQDataset.unsafeOf(ProjectId.unsafeFromString("my-gcp-project"), "prod").withLocation(LocationId.EU).toRef, "users"),
       BQSchema.of(
         BQField("userId", BQField.Type.STRING, BQField.Mode.REQUIRED),
         namesStruct

testing/src/main/scala/no/nrk/bigquery/testing/BigQueryTestClient.scala

@@ -62,10 +62,15 @@ object BigQueryTestClient {
     } yield underlying
 
   def cachingClient(
-      cacheFrom: Resource[IO, BigQueryClient[IO]]
+      cacheFrom: Resource[IO, BigQueryClient[IO]],
+      config: Option[BigQueryClient.Config]
   ): Resource[IO, BigQueryClient[IO]] =
     cacheFrom.map(client =>
-      new BigQueryClient(client.underlying, client.reader, client.metricOps, None) {
+      new BigQueryClient(
+        client.underlying,
+        client.reader,
+        client.metricOps,
+        config.getOrElse(BigQueryClient.Config.default)) {
         override protected def synchronousQueryExecute(
             jobId: BQJobId,
             query: BQSqlFrag,

testing/src/test/scala/no/nrk/bigquery/RoundtripTest.scala

@@ -23,7 +23,7 @@ class RoundtripTest extends CatsEffectSuite {
 
   def roundtrip[P: BQShow: BQRead](expectedValues: P*): IO[Unit] =
     BigQueryTestClient
-      .cachingClient(BigQueryTestClient.testClient)
+      .cachingClient(BigQueryTestClient.testClient, None)
       .use(
         _.synchronousQuery(
           BQJobId.auto,