feat: add integrationdb ksql scripts

.ksql/INTEGRATIONDB/INTEGRATIONDB_STREETNAME_OSLO_CONNECTOR.ksql

@@ -0,0 +1,28 @@
+CREATE SINK CONNECTOR `StreetNameIntegrationDbConnector` with (
+    "topics"= 'streetname.snapshot.oslo.flatten.integrationdb',
+    "input.data.format"= 'JSON_SR',
+    "input.key.format"= 'JSON_SR',
+    "delete.enabled"= false,
+    "connector.class"= 'PostgresSink',
+    "name"= 'StreetNameIntegrationDbConnector',
+    "kafka.auth.mode"= 'KAFKA_API_KEY',
+    "kafka.api.key"= '***',
+    "kafka.api.secret"= '***',
+    "connection.host"= '***',
+    "connection.port"= '5432',
+    "connection.user"= 'basisregisters',
+    "connection.password"= '***',
+    "db.name"= 'postgres',
+    "ssl.mode"= 'require',
+    "insert.mode"= 'UPSERT',
+    "table.name.format"= 'Integration.StreetNames',
+    "table.types"= 'TABLE',
+    "db.timezone"= 'UTC',
+    "pk.mode"= 'record_key',
+    "pk.fields"= 'PersistentLocalId',
+    "auto.create"= false,
+    "auto.evolve"= false,
+    "quote.sql.identifiers"= 'ALWAYS',
+    "batch.sizes"= 3000,
+    "tasks.max"= 1
+  );

.ksql/INTEGRATIONDB/INTEGRATIONDB_STREETNAME_SNAPSHOT_OSLO_STREAM_FLATTEN.ksql

@@ -0,0 +1,26 @@
+CREATE OR REPLACE STREAM IF NOT EXISTS STREETNAME_SNAPSHOT_OSLO_STREAM_FLATTEN_INTEGRATIONDB
+WITH (KAFKA_TOPIC='streetname.snapshot.oslo.flatten.integrationdb', PARTITIONS=1, VALUE_FORMAT='JSON_SR', KEY_FORMAT='JSON_SR')
+AS SELECT
+  CAST(REDUCE(SPLIT(URL_EXTRACT_PATH(MESSAGEKEY), '/'), '', (s,x) => x) AS INTEGER) PersistentLocalId,
+
+  STRAATNAAMSTATUS as "Status",
+  CAST(GEMEENTE->OBJECTID as INT) as "NisCode",
+
+  FILTER(STRAATNAMEN, (X) => (X->TAAL = 'nl'))[1]->SPELLING as "NameDutch",
+  FILTER(STRAATNAMEN, (X) => (X->TAAL = 'fr'))[1]->SPELLING as "NameFrench",
+  FILTER(STRAATNAMEN, (X) => (X->TAAL = 'de'))[1]->SPELLING as "NameGerman",
+  FILTER(STRAATNAMEN, (X) => (X->TAAL = 'en'))[1]->SPELLING as "NameEnglish",
+
+  FILTER(HOMONIEMTOEVOEGINGEN, (X) => (X->TAAL = 'nl'))[1]->SPELLING as "HomonymAdditionDutch",
+  FILTER(HOMONIEMTOEVOEGINGEN, (X) => (X->TAAL = 'fr'))[1]->SPELLING as "HomonymAdditionFrench",
+  FILTER(HOMONIEMTOEVOEGINGEN, (X) => (X->TAAL = 'de'))[1]->SPELLING as "HomonymAdditionGerman",
+  FILTER(HOMONIEMTOEVOEGINGEN, (X) => (X->TAAL = 'en'))[1]->SPELLING as "HomonymAdditionEnglish",
+
+  IDENTIFICATOR->ID as "PuriId",
+  IDENTIFICATOR->NAAMRUIMTE as "Namespace",
+  IDENTIFICATOR->VERSIEID as "VersionString",
+  PARSE_TIMESTAMP(IDENTIFICATOR->VERSIEID, 'yyyy-MM-dd''T''HH:mm:ssXXX', 'UTC') as "VersionTimestamp",
+  CASE WHEN IDENTIFICATOR->ID is null THEN TRUE ELSE FALSE END as "IsRemoved"
+
+FROM STREETNAME_SNAPSHOT_OSLO_STREAM
+PARTITION BY CAST(REDUCE(SPLIT(URL_EXTRACT_PATH(MESSAGEKEY), '/'), '', (s,x) => x) AS INTEGER);