google_datastream_stream | max_staleness configuration for tables in BigQuery as output storage

[mr-manoj-dev]

Community Note

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request.
• Please do not leave +1 or me too comments, they generate extra noise for issue followers and do not help prioritize the request.
• If you are interested in working on this issue or have submitted a pull request, please leave a comment.
• If an issue is assigned to a user, that user is claiming responsibility for the issue.
• Customers working with a Google Technical Account Manager or Customer Engineer can ask them to reach out internally to expedite investigation and resolution of this issue.

Terraform Version & Provider Version(s)

Terraform v7.36.0
I have datastream which replicating CDC data from on-prem database into BigQuery.
I already have gone to production with the configuration where BigQuery tables are partitioned on ingestion timestamp

I am experiencing huge latency and trying to provide max_staleness on each table but I did not found this property in given provider.

Affected Resource(s)

The GCP Datastream provider (hashicorp/google ≥ v7.x) does not yet expose a

dedicated bigquery_managed_storage_settings block inside customization_rules.

The current provider schema only supports bigquery_partitioning and

bigquery_clustering inside customization_rules.

Terraform Configuration

resource "google_datastream_stream" "sql_to_bq" {

  project       = var.project_id
  location      = var.location
  stream_id     = var.stream_id
  display_name  = var.display_name
  desired_state = var.desired_state
  labels        = var.labels

  create_without_validation = var.create_without_validation

  source_config {
    source_connection_profile = var.src_connection_profile_name

    # Conditional SQL Server source configuration
    dynamic "sql_server_source_config" {
      for_each = var.database_type == "sqlserver" ? [1] : []
      content {
        max_concurrent_backfill_tasks = var.max_concurrent_backfill_tasks
        max_concurrent_cdc_tasks      = var.max_concurrent_cdc_tasks

        dynamic "change_tables" {
          for_each = trimspace(var.cdc_method) == "Change tables" ? [1] : []
          content {}
        }

        dynamic "transaction_logs" {
          for_each = trimspace(var.cdc_method) == "TRANSACTION_LOGS" ? [1] : []
          content {}
        }

        include_objects {
          dynamic "schemas" {
            for_each = length(var.source_tables) > 0 ? {
              for schema_name in distinct([for t in var.source_tables : t.schema]) :
              schema_name => [for t in var.source_tables : t if t.schema == schema_name]
            } : { (var.src_schema_name) = [{ schema = var.src_schema_name, table = var.src_table_pattern, columns = [] }] }

            content {
              schema = schemas.key
              dynamic "tables" {
                for_each = schemas.value
                content {
                  table = tables.value.table
                  dynamic "columns" {
                    for_each = try(tables.value.columns, [])
                    content {
                      column = columns.value
                    }
                  }
                }
              }
            }
          }
        }
      }
    }

    # Conditional MySQL source configuration
    dynamic "mysql_source_config" {
      for_each = var.database_type == "mysql" ? [1] : []
      content {
        max_concurrent_backfill_tasks = var.max_concurrent_backfill_tasks
        max_concurrent_cdc_tasks      = var.max_concurrent_cdc_tasks

        dynamic "binary_log_position" {
          for_each = contains(["Binary log", "BINARY_LOG_POSITION"], trimspace(var.cdc_method)) ? [1] : []
          content {}
        }

        dynamic "gtid" {
          for_each = trimspace(var.cdc_method) == "GTID" ? [1] : []
          content {}
        }

        include_objects {
          dynamic "mysql_databases" {
            for_each = length(var.source_tables) > 0 ? {
              for schema_name in distinct([for t in var.source_tables : t.schema]) :
              schema_name => [for t in var.source_tables : t if t.schema == schema_name]
            } : { (var.src_schema_name) = [{ schema = var.src_schema_name, table = var.src_table_pattern, columns = [] }] }

            content {
              database = mysql_databases.key
              dynamic "mysql_tables" {
                for_each = mysql_databases.value
                content {
                  table = mysql_tables.value.table
                  dynamic "mysql_columns" {
                    for_each = try(mysql_tables.value.columns, [])
                    content {
                      column = mysql_columns.value
                    }
                  }
                }
              }
            }
          }
        }
      }
    }

    # Conditional MongoDB source configuration
    dynamic "mongodb_source_config" {
      for_each = var.database_type == "mongodb" ? [1] : []
      content {
        max_concurrent_backfill_tasks = var.max_concurrent_backfill_tasks

        include_objects {
          dynamic "databases" {
            for_each = length(var.source_tables) > 0 ? {
              for database_name in distinct([for t in var.source_tables : t.schema]) :
              database_name => [for t in var.source_tables : t if t.schema == database_name]
            } : { (var.src_schema_name) = [] }

            content {
              database = databases.key
              dynamic "collections" {
                for_each = databases.value
                content {
                  collection = collections.value.table
                  dynamic "fields" {
                    for_each = coalesce(try(collections.value.columns, null), [])
                    content {
                      field = fields.value
                    }
                  }
                }
              }
            }
          }
        }

        dynamic "exclude_objects" {
          for_each = length(var.exclude_objects) > 0 ? [1] : []
          content {
            dynamic "databases" {
              for_each = {
                for database_name in distinct([for t in var.exclude_objects : t.schema]) :
                database_name => [for t in var.exclude_objects : t if t.schema == database_name]
              }

              content {
                database = databases.key
                dynamic "collections" {
                  for_each = databases.value
                  content {
                    collection = collections.value.table
                    dynamic "fields" {
                      for_each = coalesce(try(collections.value.columns, null), [])
                      content {
                        field = fields.value
                      }
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  }

  destination_config {
    destination_connection_profile = var.dst_connection_profile_name
    bigquery_destination_config {
      data_freshness = var.data_freshness
      single_target_dataset {
        dataset_id = "${var.project_id}:${var.dst_bigquery_dataset_id}"
      }
      dynamic "append_only" {
        for_each = var.bq_write_mode == "Append-only" ? [1] : []
        content {}
      }
      dynamic "merge" {
        for_each = var.bq_write_mode == "Merge" ? [1] : []
        content {}
      }
    }
  }

  dynamic "backfill_all" {
    for_each = var.backfill_all ? [1] : []
    content {}
  }
  dynamic "backfill_none" {
    for_each = var.backfill_none ? [1] : []
    content {}
  }

  # ── Ingestion-time partitioning – per-table (when source_tables specified) ──
  dynamic "rule_sets" {
    for_each = var.enable_partitioning && var.partition_field == null && length(var.source_tables) > 0 ? var.source_tables : []
    content {
      object_filter {
        source_object_identifier {
          dynamic "sql_server_identifier" {
            for_each = var.database_type == "sqlserver" ? [1] : []
            content {
              schema = rule_sets.value.schema
              table  = rule_sets.value.table
            }
          }
          dynamic "mysql_identifier" {
            for_each = var.database_type == "mysql" ? [1] : []
            content {
              database = rule_sets.value.schema
              table    = rule_sets.value.table
            }
          }
          dynamic "mongodb_identifier" {
            for_each = var.database_type == "mongodb" ? [1] : []
            content {
              database   = rule_sets.value.schema
              collection = rule_sets.value.table
            }
          }
        }
      }
      customization_rules {
        bigquery_partitioning {
          require_partition_filter = var.require_partition_filter
          ingestion_time_partition {
            partitioning_time_granularity = "PARTITIONING_TIME_GRANULARITY_${var.partition_type}"
          }
        }
      }
    }
  }

  # ── Ingestion-time partitioning – schema-level (when no source_tables) ───
  dynamic "rule_sets" {
    for_each = var.enable_partitioning && var.partition_field == null && length(var.source_tables) == 0 ? [1] : []
    content {
      object_filter {} # no source_object_identifier = applies to all objects
      customization_rules {
        bigquery_partitioning {
          require_partition_filter = var.require_partition_filter
          ingestion_time_partition {
            partitioning_time_granularity = "PARTITIONING_TIME_GRANULARITY_${var.partition_type}"
          }
        }
      }
    }
  }

  # ── Column-based partitioning – per-table (requires source_tables) ────────
  dynamic "rule_sets" {
    for_each = var.enable_partitioning && var.partition_field != null ? var.source_tables : []
    content {
      object_filter {
        source_object_identifier {
          dynamic "sql_server_identifier" {
            for_each = var.database_type == "sqlserver" ? [1] : []
            content {
              schema = rule_sets.value.schema
              table  = rule_sets.value.table
            }
          }
          dynamic "mysql_identifier" {
            for_each = var.database_type == "mysql" ? [1] : []
            content {
              database = rule_sets.value.schema
              table    = rule_sets.value.table
            }
          }
          dynamic "mongodb_identifier" {
            for_each = var.database_type == "mongodb" ? [1] : []
            content {
              database   = rule_sets.value.schema
              collection = rule_sets.value.table
            }
          }
        }
      }
      customization_rules {
        bigquery_partitioning {
          require_partition_filter = var.require_partition_filter
          time_unit_partition {
            column                        = var.partition_field
            partitioning_time_granularity = "PARTITIONING_TIME_GRANULARITY_${var.partition_type}"
          }
        }
      }
    }
  }
  # ── Max-staleness – per-table (when source_tables specified) ──
  dynamic "rule_sets" {
    for_each = var.max_staleness != null && length(var.source_tables) > 0 ? var.source_tables : []
    content {
      object_filter {
        source_object_identifier {
          dynamic "sql_server_identifier" {
            for_each = var.database_type == "sqlserver" ? [1] : []
            content {
              schema = rule_sets.value.schema
              table  = rule_sets.value.table
            }
          }
          dynamic "mysql_identifier" {
            for_each = var.database_type == "mysql" ? [1] : []
            content {
              database = rule_sets.value.schema
              table    = rule_sets.value.table
            }
          }
          dynamic "mongodb_identifier" {
            for_each = var.database_type == "mongodb" ? [1] : []
            content {
              database   = rule_sets.value.schema
              collection = rule_sets.value.table
            }
          }
        }
      }
      customization_rules {
        bigquery_managed_storage_settings {
          max_staleness = var.max_staleness
        }
      }
    }
  }

  # ── Max-staleness – schema-level (when no source_tables) ──
  dynamic "rule_sets" {
    for_each = var.max_staleness != null && length(var.source_tables) == 0 ? [1] : []
    content {
      object_filter {} # no source_object_identifier = applies to all objects
      customization_rules {
        bigquery_managed_storage_settings {
          max_staleness = var.max_staleness
        }
      }
    }
  }
}

Debug Output

SELECT distinct table_name, option_name, option_value
FROM gcp_project_id.bq_dataset_id.INFORMATION_SCHEMA.TABLE_OPTIONS

It returns:
[{
"table_name": "dbo_table_1",
"option_name": "require_partition_filter",
"option_value": "true"
}, {
"table_name": "dbo_table_2",
"option_name": "require_partition_filter",
"option_value": "true"
}, {
"table_name": "dbo_table_3",
"option_name": "require_partition_filter",
"option_value": "true"
}, {
"table_name": "dbo_table_4",
"option_name": "require_partition_filter",
"option_value": "true"
}, {
"table_name": "dbo_table_5",
"option_name": "require_partition_filter",
"option_value": "true"
}]

Only require_partition_filter can be seen.
I want max_staleness to be applied on tables.

Expected Behavior

No response

Actual Behavior

No response

Steps to reproduce

1. terraform apply

Important Factoids

No response

References

No response

[ggtisc]

Hi @mr-manoj-dev

Please share the terraform code without locals, labels and modules to replicate this issue. For sensitive data you could use examples like:

project = "my-project"
org_id = 1234567890
member = "user:my-user@my-domain.com"

We need this to replicate the issue