BigQuery: New BigQuery connector

build.sbt

@@ -173,7 +173,8 @@ lazy val googleCloudBigQuery = alpakkaProject(
   "google-cloud-bigquery",
   "google.cloud.bigquery",
   Dependencies.GoogleBigQuery,
-  Test / fork := true
+  Test / fork := true,
+  fatalWarnings := true
 ).disablePlugins(MimaPlugin).enablePlugins(spray.boilerplate.BoilerplatePlugin)
 
 lazy val googleCloudPubSub = alpakkaProject(
@@ -354,6 +355,10 @@ lazy val docs = project
         "javadoc.org.apache.kudu.base_url" -> s"https://kudu.apache.org/releases/${Dependencies.KuduVersion}/apidocs/",
         "javadoc.org.apache.hadoop.base_url" -> s"https://hadoop.apache.org/docs/r${Dependencies.HadoopVersion}/api/",
         "javadoc.software.amazon.awssdk.base_url" -> "https://sdk.amazonaws.com/java/api/latest/",
+        "javadoc.com.fasterxml.jackson.annotation.base_url" -> "https://javadoc.io/doc/com.fasterxml.jackson.core/jackson-annotations/latest/",
+        "javadoc.com.fasterxml.jackson.annotation.link_style" -> "direct",
+        // Scala
+        "scaladoc.spray.json.base_url" -> s"https://javadoc.io/doc/io.spray/spray-json_${scalaBinaryVersion.value}/latest/",
         // Eclipse Paho client for MQTT
         "javadoc.org.eclipse.paho.client.mqttv3.base_url" -> "https://www.eclipse.org/paho/files/javadoc/",
         "javadoc.org.bson.codecs.configuration.base_url" -> "https://mongodb.github.io/mongo-java-driver/3.7/javadoc/",

google-cloud-bigquery/src/main/boilerplate/akka/stream/alpakka/googlecloud/bigquery/scaladsl/schema/ProductSchemasInstances.scala.template

@@ -0,0 +1,23 @@
+/*
+ * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.googlecloud.bigquery.scaladsl.schema
+
+import akka.stream.alpakka.googlecloud.bigquery.model.TableJsonProtocol.{RequiredMode, TableFieldSchema}
+
+import scala.collection.immutable.Seq
+import scala.reflect.{classTag, ClassTag}
+
+trait ProductSchemasInstances { this: ProductSchemas with StandardSchemas =>
+[#  // Case classes with 1 parameters
+
+  def bigQuerySchema1[[#P1#], T <: Product: ClassTag](ev: ([#P1#]) => T)(implicit [#writer1: SchemaWriter[P1]#]): TableSchemaWriter[T] = new ProductSchemaWriter[T](schemaFields[[#P1#], T])
+
+  private def schemaFields[[#P1#], T <: Product: ClassTag](implicit [#writer1: SchemaWriter[P1]#]): Seq[TableFieldSchema] = {
+    val Array([#p1#]) = extractFieldNames(classTag[T])
+    Seq([#writer1.write(p1, RequiredMode)#])
+  }#
+
+]
+}

google-cloud-bigquery/src/main/boilerplate/akka/stream/alpakka/googlecloud/bigquery/scaladsl/BigQueryProductFormatsInstances.scala.template → google-cloud-bigquery/src/main/boilerplate/akka/stream/alpakka/googlecloud/bigquery/scaladsl/spray/BigQueryProductFormatsInstances.scala.template

@@ -2,21 +2,21 @@
  * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.stream.alpakka.googlecloud.bigquery.scaladsl
+package akka.stream.alpakka.googlecloud.bigquery.scaladsl.spray
 
-import spray.json.{JsObject, JsValue, JsonFormat, ProductFormats, RootJsonFormat, StandardFormats}
+import spray.json.{JsObject, JsValue, ProductFormats, StandardFormats}
 
 import scala.reflect.{classTag, ClassTag}
 
 trait BigQueryProductFormatsInstances { this: BigQueryProductFormats with ProductFormats with StandardFormats =>
 [#  // Case classes with 1 parameters
 
-  def bigQueryJsonFormat1[[#P1: JsonFormat#], T <: Product: ClassTag](construct: ([#P1#]) => T): RootJsonFormat[T] = {
+  def bigQueryJsonFormat1[[#P1: BigQueryJsonFormat#], T <: Product: ClassTag](construct: ([#P1#]) => T): BigQueryRootJsonFormat[T] = {
     val Array([#p1#]) = extractFieldNames(classTag[T])
     bigQueryJsonFormat(construct, [#p1#])
   }
-  def bigQueryJsonFormat[[#P1: JsonFormat#], T <: Product](construct: ([#P1#]) => T, [#fieldName1: String#]): RootJsonFormat[T] =
-    new RootJsonFormat[T] {
+  def bigQueryJsonFormat[[#P1: BigQueryJsonFormat#], T <: Product](construct: ([#P1#]) => T, [#fieldName1: String#]): BigQueryRootJsonFormat[T] =
+    new BigQueryRootJsonFormat[T] {
       def write(p: T) = {
         val fields = new collection.mutable.ListBuffer[(String, JsValue)]
         fields.sizeHint(1 * 2)
@@ -30,7 +30,8 @@ trait BigQueryProductFormatsInstances { this: BigQueryProductFormats with Produc
         ]
         construct([#p1V#])
       }
-    }#
+    }
+  implicit def bigQueryTuple1Reader[[#P1: BigQueryJsonFormat#]]: BigQueryRootJsonReader[Tuple1[[#P1#]]] = bigQueryJsonFormat1(Tuple1[[#P1#]])#
 
 ]
 }

google-cloud-bigquery/src/main/resources/reference.conf

@@ -1,16 +1,49 @@
 alpakka.google.bigquery {
 
+  credentials {
+    # Options: service-account, compute-engine
+    provider = service-account
+
+    service-account {
+      # If `path` is non-empty then read from file, otherwise fallback to config values
+      path = ""
+      path = ${?GOOGLE_APPLICATION_CREDENTIALS}
+      project-id = ""
+      client-email = ""
+      private-key = ""
+      scopes = ["https://www.googleapis.com/auth/bigquery"]
+    }
+
+    # Timeout for blocking call during settings initialization to compute engine metadata server
+    compute-engine.timeout = 1s
+  }
+
+  load-job {
+    # The minimum size of a chunk
+    chunk-size = 15 MB
+
+    # BigQuery has a hard limit of 1,500 load jobs per table per day (just over 1 job per minute)
+    # This sets the rate limit when loading data via BigQuery.insertAllAsync
+    per-table-quota = 1 minute
+  }
+
+  # The retry settings for requests to Google APIs
+  retry-settings {
+    max-retries = 6
+    min-backoff = 1 second
+    max-backoff = 1 minute
+    random-factor = 0.2
+  }
+
   # An address of a proxy that will be used for all connections using HTTP CONNECT tunnel.
   # forward-proxy {
+  #   scheme = "https"
   #   host = "proxy"
   #   port = 8080
   #   credentials {
   #     username = "username"
   #     password = "password"
   #   }
-  #   trustPem {
-  #     pemPath = "/path/to/file.pem"
-  #   }
+  #   trust-pem = "/path/to/file.pem"
   # }
-
-}
+}

google-cloud-bigquery/src/main/scala/akka/stream/alpakka/googlecloud/bigquery/BigQueryAttributes.scala

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.googlecloud.bigquery
+
+import akka.stream.Attributes.Attribute
+import akka.stream.alpakka.googlecloud.bigquery.impl.BigQueryExt
+import akka.stream.{Attributes, Materializer}
+
+/**
+ * Akka Stream [[Attributes]] that are used when materializing BigQuery stream blueprints.
+ */
+object BigQueryAttributes {
+
+  /**
+   * [[BigQuerySettings]] to use for the BigQuery stream
+   */
+  def settings(settings: BigQuerySettings): Attributes = Attributes(BigQuerySettingsValue(settings))
+
+  /**
+   * Config path which will be used to resolve [[BigQuerySettings]]
+   */
+  def settingsPath(path: String): Attributes = Attributes(BigQuerySettingsPath(path))
+
+  /**
+   * Resolves the most specific [[BigQuerySettings]] for some [[Attributes]]
+   */
+  def resolveSettings(attr: Attributes, mat: Materializer): BigQuerySettings =
+    attr.attributeList.collectFirst {
+      case BigQuerySettingsValue(settings) => settings
+      case BigQuerySettingsPath(path) => BigQueryExt(mat.system).settings(path)
+    } getOrElse {
+      BigQueryExt(mat.system).settings
+    }
+
+  private final case class BigQuerySettingsValue(settings: BigQuerySettings) extends Attribute
+  private final case class BigQuerySettingsPath(path: String) extends Attribute
+}