Add BigQuery BigLake Managed Tables as a Datastream destination (GoogleCloudPlatform#13189)

Author: m-habib

mmv1/products/datastream/Stream.yaml

@@ -197,6 +197,22 @@ examples:
     external_providers: ["random", "time"]
       # Random provider
     skip_vcr: true
+  - name: 'datastream_stream_bigquery_blmt'
+    primary_resource_id: 'default'
+    vars:
+      stream_id: 'blmt-stream'
+      database_instance_name: 'blmt-instance'
+      deletion_protection: 'true'
+      source_connection_profile_id: 'blmt-source-profile'
+      destination_connection_profile_id: 'blmt-destination-profile'
+      blmt_bucket_name: 'blmt-bucket'
+      blmt_connection_id: 'blmt-connection'
+    test_vars_overrides:
+      'deletion_protection': 'false'
+    external_providers: ["random"]
+    skip_vcr: true
+    # Involves complex dependency creation, which makes it impractical in this context
+    exclude_test: true
 virtual_fields:
   - name: 'desired_state'
     description: |
@@ -1224,6 +1240,40 @@ properties:
                       encryption key. i.e. projects/{project}/locations/{location}/keyRings/{key_ring}/cryptoKeys/{cryptoKey}.
                       See https://cloud.google.com/bigquery/docs/customer-managed-encryption for more information.
                     immutable: true
+          - name: 'blmtConfig'
+            type: NestedObject
+            description: 'BigLake Managed Tables configuration for BigQuery streams.'
+            properties:
+              - name: 'bucket'
+                type: String
+                description: |
+                      The Cloud Storage bucket name.
+                required: true
+              - name: 'connectionName'
+                type: String
+                description: |
+                      The bigquery connection. Format: `{project}.{location}.{name}`
+                required: true
+              - name: 'fileFormat'
+                type: String
+                description: |
+                      The file format.
+                required: true
+                enum_values:
+                  - FILE_FORMAT_UNSPECIFIED
+                  - PARQUET
+              - name: 'tableFormat'
+                type: String
+                description: |
+                      The table format.
+                required: true
+                enum_values:
+                  - TABLE_FORMAT_UNSPECIFIED
+                  - ICEBERG
+              - name: 'rootPath'
+                type: String
+                description: |
+                      The root path inside the Cloud Storage bucket.
           - name: 'merge'
             type: NestedObject
             description: |

mmv1/templates/terraform/examples/datastream_stream_bigquery_blmt.tf.tmpl

@@ -0,0 +1,126 @@
+data "google_project" "project" {
+}
+
+resource "google_sql_database_instance" "instance" {
+    name             = "{{index $.Vars "database_instance_name"}}"
+    database_version = "MYSQL_8_0"
+    region           = "us-central1"
+    settings {
+        tier = "db-f1-micro"
+        ip_configuration {
+
+            // Datastream IPs will vary by region.
+            authorized_networks {
+                value = "34.71.242.81"
+            }
+
+            authorized_networks {
+                value = "34.72.28.29"
+            }
+
+            authorized_networks {
+                value = "34.67.6.157"
+            }
+
+            authorized_networks {
+                value = "34.67.234.134"
+            }
+
+            authorized_networks {
+                value = "34.72.239.218"
+            }
+        }
+    }
+    deletion_protection = {{index $.Vars "deletion_protection"}}
+}
+
+resource "google_sql_database" "db" {
+    instance = google_sql_database_instance.instance.name
+    name     = "db"
+}
+
+resource "random_password" "pwd" {
+    length = 16
+    special = false
+}
+
+resource "google_sql_user" "user" {
+    name     = "user"
+    instance = google_sql_database_instance.instance.name
+    host     = "%"
+    password = random_password.pwd.result
+}
+
+resource "google_storage_bucket" "blmt_bucket" {
+  # Use variable from Stream.yaml for the name
+  name          = "{{index $.Vars "blmt_bucket_name"}}"
+  location      = "us-central1"
+  force_destroy = true
+}
+
+resource "google_bigquery_connection" "blmt_connection" {
+  project       = data.google_project.project.project_id
+  location      = "us-central1"
+  connection_id = "{{index $.Vars "blmt_connection_id"}}"
+  friendly_name = "Datastream BLMT Test Connection"
+  description   = "Connection for Datastream BLMT test"
+
+  cloud_resource {}
+}
+
+resource "google_storage_bucket_iam_member" "blmt_connection_bucket_admin" {
+  bucket = google_storage_bucket.blmt_bucket.name
+  role   = "roles/storage.admin"
+  member = "serviceAccount:${google_bigquery_connection.blmt_connection.cloud_resource[0].service_account_id}"
+}
+
+resource "google_datastream_connection_profile" "source_connection_profile" {
+    display_name          = "Source connection profile"
+    location              = "us-central1"
+    connection_profile_id = "{{index $.Vars "source_connection_profile_id"}}"
+
+    mysql_profile {
+        hostname = google_sql_database_instance.instance.public_ip_address
+        username = google_sql_user.user.name
+        password = google_sql_user.user.password
+    }
+}
+
+resource "google_datastream_connection_profile" "destination_connection_profile" {
+    display_name          = "Connection profile"
+    location              = "us-central1"
+    connection_profile_id = "{{index $.Vars "destination_connection_profile_id"}}"
+
+    bigquery_profile {}
+}
+
+resource "google_datastream_stream" "{{$.PrimaryResourceId}}" {
+    stream_id = "{{index $.Vars "stream_id"}}"
+    location = "us-central1"
+    display_name = "My BLMT stream"
+    source_config {
+        source_connection_profile = google_datastream_connection_profile.source_connection_profile.id
+        mysql_source_config {}
+    }
+    destination_config {
+        destination_connection_profile = google_datastream_connection_profile.destination_connection_profile.id
+        bigquery_destination_config {
+            source_hierarchy_datasets {
+                dataset_template {
+                    location = "us-central1"
+                }
+            }
+            blmt_config {
+                bucket          = google_storage_bucket.blmt_bucket.name
+                connection_name = "${google_bigquery_connection.blmt_connection.project}.${google_bigquery_connection.blmt_connection.location}.${google_bigquery_connection.blmt_connection.connection_id}"
+                file_format     = "PARQUET"
+                table_format    = "ICEBERG"
+                root_path       = "/"
+            }
+            append_only {}
+        }
+    }
+
+    backfill_none {
+    }
+}