Skip to main content

AlloyDB

This connector materializes Estuary collections into tables in an AlloyDB database.

AlloyDB is a fully managed, PostgreSQL-compatible database available in the Google Cloud platform. This connector is derived from the PostgreSQL materialization connector, so the same configuration applies, but the setup steps look somewhat different.

Prerequisites

To use this connector, you'll need:

  • An AlloyDB database to which to materialize, and user credentials. The connector will create new tables in the database per your specification. Tables created manually in advance are not supported.
  • A virtual machine to connect securely to the instance via SSH tunneling. (AlloyDB doesn't support IP allowlisting.) Follow the instructions to create a virtual machine for SSH tunneling in the same Google Cloud project as your instance.

Authentication

This connector supports user/password authentication as well as IAM authentication. This allows you to use an IAM role with your cloud provider to manage access.

The connector will require credentials based on your chosen authentication method, such as a password or GCP workload identity pool audience.

See the GCP IAM authentication guide for additional setup details.

Configuration

To use this connector, begin with data in one or more Estuary collections. Use the below properties to configure an AlloyDB materialization, which will direct one or more of your Estuary collections to your desired tables, or views, in the database.

Properties

Endpoint

The SSH config section is required for this connector. You'll fill in the database address with a localhost IP address, and specify your VM's IP address as the SSH address. See the table below and the sample config.

PropertyTitleDescriptionTypeRequired/Default
/databaseDatabaseName of the logical database to materialize to.string
/addressAddressHost and port. Set to 127.0.0.1:5432 to enable SSH tunneling.stringRequired
/schemaDatabase SchemaDatabase schema to use for materialized tables (unless overridden within the binding resource configuration) as well as associated materialization metadata tablesstring"public"
/userUserDatabase user to connect as.stringRequired
/hardDeleteHard DeleteIf enabled, items deleted in the source will also be deleted from the destination. By default, deletions are tracked via _meta/op (soft delete).booleanfalse
/networkTunnelNetwork TunnelConnect to your system through an SSH server that acts as a bastion host for your network.Object
/networkTunnel/sshForwardingSSH ForwardingObject
/networkTunnel/sshForwarding/sshEndpointSSH EndpointEndpoint of the remote SSH server (in this case, your Google Cloud VM) that supports tunneling (in the form of ssh://user@address.String
/networkTunnel/sshForwarding/privateKeySSH Private KeyPrivate key to connect to the remote SSH server.String
Credentials
PropertyTitleDescriptionTypeRequired/Default
/credentialsAuthenticationAuthentication method and credentials that provide access to the database.objectRequired
/credentials/auth_typeAuth TypeThe authentication method to use. One of UserPassword or GCPIAM.string
/credentials/passwordPasswordPassword for the specified database user.stringRequired for UserPassword auth
/credentials/gcp_service_account_to_impersonateGCP Service AccountGCP service account email for Cloud SQL IAM authentication.stringRequired for GCPIAM auth
/credentials/gcp_workload_identity_pool_audienceWorkload Identity Pool AudienceGCP workload identity pool audience. The format should be similar to: //iam.googleapis.com/projects/123/locations/global/workloadIdentityPools/test-pool/providers/test-provider.stringRequired for GCPIAM auth

Bindings

PropertyTitleDescriptionTypeRequired/Default
/additional_table_create_sqlAdditional Table Create SQLAdditional SQL statement(s) to be run in the same transaction that creates the table. See Additional Table Create SQL for usage examples.string
/delta_updatesDelta UpdateShould updates to this table be done via delta updates.booleanfalse
/schemaAlternative SchemaAlternative schema for this table (optional). Overrides schema set in endpoint configuration.string
/tableTableTable name to materialize to. It will be created by the connector, unless the connector has previously created it.stringRequired

Sample

materializations:
${PREFIX}/${mat_name}:
endpoint:
connector:
image: ghcr.io/estuary/materialize-alloydb:v5
config:
database: postgres
address: host:port
user: flow
credentials:
auth_type: UserPassword
password: <secret>
networkTunnel:
sshForwarding:
sshEndpoint: ssh://sshUser@<vm-ip-address>
privateKey: |2
-----BEGIN RSA PRIVATE KEY-----
MIICXAIBAAKBgQCJO7G6R+kv2MMS8Suw21sk2twHg8Vog0fjimEWJEwyAfFM/Toi
EJ6r5RTaSvN++/+MPWUll7sUdOOBZr6ErLKLHEt7uXxusAzOjMxFKZpEARMcjwHY
v/tN1A2OYU0qay1DOwknEE0i+/Bvf8lMS7VDjHmwRaBtRed/+iAQHf128QIDAQAB
AoGAGoOUBP+byAjDN8esv1DCPU6jsDf/Tf//RbEYrOR6bDb/3fYW4zn+zgtGih5t
CR268+dwwWCdXohu5DNrn8qV/Awk7hWp18mlcNyO0skT84zvippe+juQMK4hDQNi
ywp8mDvKQwpOuzw6wNEitcGDuACx5U/1JEGGmuIRGx2ST5kCQQDsstfWDcYqbdhr
5KemOPpu80OtBYzlgpN0iVP/6XW1e5FCRp2ofQKZYXVwu5txKIakjYRruUiiZTza
QeXRPbp3AkEAlGx6wMe1l9UtAAlkgCFYbuxM+eRD4Gg5qLYFpKNsoINXTnlfDry5
+1NkuyiQDjzOSPiLZ4Abpf+a+myjOuNL1wJBAOwkdM6aCVT1J9BkW5mrCLY+PgtV
GT80KTY/d6091fBMKhxL5SheJ4SsRYVFtguL2eA7S5xJSpyxkadRzR0Wj3sCQAvA
bxO2fE1SRqbbF4cBnOPjd9DNXwZ0miQejWHUwrQO0inXeExNaxhYKQCcnJNUAy1J
6JfAT/AbxeSQF3iBKK8CQAt5r/LLEM1/8ekGOvBh8MAQpWBW771QzHUN84SiUd/q
xR9mfItngPwYJ9d/pTO7u9ZUPHEoat8Ave4waB08DsI=
-----END RSA PRIVATE KEY-----
bindings:
- resource:
table: ${TABLE_NAME}
source: ${PREFIX}/${COLLECTION_NAME}

Reserved words

PostgreSQL has a list of reserved words that must be quoted in order to be used as an identifier. Estuary considers all the reserved words that are marked as "reserved" in any of the columns in the official PostgreSQL documentation.

These reserved words are listed in the table below. Estuary automatically quotes fields that are in this list.

Reserved words
abscurrent_transform_group_for_typeindicatorordersqlexception
absolutecurrent_userinitialoutsqlstate
acoscursorinitiallyoutersqlwarning
actioncycleinneroutputsqrt
adddatalinkinoutoverstart
alldateinputoverlapsstatic
allocatedayinsensitiveoverlaystddev_pop
alterdeallocateinsertpadstddev_samp
analysedecintparametersubmultiset
analyzedecfloatintegerpartialsubset
anddecimalintersectpartitionsubstring
anydeclareintersectionpatternsubstring_regex
aredefaultintervalpersucceeds
arraydeferrableintopercentsum
array_aggdeferredispercentile_contsymmetric
array_max_cardinalitydefineisnullpercentile_discsystem
asdeleteisolationpercent_ranksystem_time
ascdense_rankjoinperiodsystem_user
asensitivederefjson_arraypermutetable
asindescjson_arrayaggplacingtablesample
assertiondescribejson_existsportiontan
asymmetricdescriptorjson_objectpositiontanh
atdeterministicjson_objectaggposition_regextemporary
atandiagnosticsjson_querypowerthen
atomicdisconnectjson_tableprecedestime
authorizationdistinctjson_table_primitiveprecisiontimestamp
avgdlnewcopyjson_valuepreparetimezone_hour
begindlpreviouscopykeypreservetimezone_minute
begin_framedlurlcompletelagprimaryto
begin_partitiondlurlcompleteonlylanguagepriortrailing
betweendlurlcompletewritelargeprivilegestransaction
bigintdlurlpathlastproceduretranslate
binarydlurlpathonlylast_valueptftranslate_regex
bitdlurlpathwritelateralpublictranslation
bit_lengthdlurlschemeleadrangetreat
blobdlurlserverleadingranktrigger
booleandlvalueleftreadtrim
bothdolevelreadstrim_array
bydomainlikerealtrue
calldoublelike_regexrecursivetruncate
calleddroplimitrefuescape
cardinalitydynamiclistaggreferencesunion
cascadeeachlnreferencingunique
cascadedelementlocalregr_avgxunknown
caseelselocaltimeregr_avgyunmatched
castemptylocaltimestampregr_countunnest
catalogendlogregr_interceptupdate
ceilend-execlog10regr_r2upper
ceilingend_framelowerregr_slopeusage
charend_partitionmatchregr_sxxuser
characterequalsmatchesregr_sxyusing
character_lengthescapematch_numberregr_syyvalue
char_lengtheverymatch_recognizerelativevalues
checkexceptmaxreleasevalue_of
classifierexceptionmeasuresrestrictvarbinary
clobexecmemberresultvarchar
closeexecutemergereturnvariadic
coalesceexistsmethodreturningvarying
collateexpminreturnsvar_pop
collationexternalminuterevokevar_samp
collectextractmodrightverbose
columnfalsemodifiesrollbackversioning
commitfetchmodulerollupview
concurrentlyfiltermonthrowwhen
conditionfirstmultisetrowswhenever
connectfirst_valuenamesrow_numberwhere
connectionfloatnationalrunningwidth_bucket
constraintfloornaturalsavepointwindow
constraintsforncharschemawith
containsforeignnclobscopewithin
continuefoundnewscrollwithout
convertframe_rownextsearchwork
copyfreenosecondwrite
corrfreezenonesectionxml
correspondingfromnormalizeseekxmlagg
cosfullnotselectxmlattributes
coshfunctionnotnullsensitivexmlbinary
countfusionnth_valuesessionxmlcast
covar_popgetntilesession_userxmlcomment
covar_sampglobalnullsetxmlconcat
creategonullifshowxmldocument
crossgotonumericsimilarxmlelement
cubegrantoccurrences_regexsinxmlexists
cume_distgroupoctet_lengthsinhxmlforest
currentgroupingofsizexmliterate
current_cataloggroupsoffsetskipxmlnamespaces
current_datehavingoldsmallintxmlparse
current_default_transform_groupholdomitsomexmlpi
current_pathhouronspacexmlquery
current_roleidentityonespecificxmlserialize
current_rowilikeonlyspecifictypexmltable
current_schemaimmediateopensqlxmltext
current_timeimportoptionsqlcodexmlvalidate
current_timestampinorsqlerroryear