Skip to main content

AlloyDB

This connector materializes Flow 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.

It's available for use in the Flow web application. For local development or open-source workflows, ghcr.io/estuary/materialize-alloydb:dev provides the latest version of the connector as a Docker image. You can also follow the link in your browser to see past image versions.

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 whitelisting.) Follow the instructions to create a virtual machine for SSH tunneling in the same Google Cloud project as your instance.

Configuration

To use this connector, begin with data in one or more Flow collections. Use the below properties to configure a AlloyDB materialization, which will direct one or more of your Flow 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
/passwordPasswordPassword for the specified database user.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
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

Bindings

PropertyTitleDescriptionTypeRequired/Default
/additional_table_create_sqlAdditional Table Create SQLAdditional SQL statement(s) to be run in the same transaction that creates the table.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:dev
config:
database: postgres
address: 127.0.0.1:5432
password: flow
user: flow
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. Flow considers all the reserved words that are marked as "reserved" in any of the columns in the official PostgreSQL documentation.

These reserve words are listed in the table below. Flow 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