sql: support query strings containing multiple statements

[ligfx]

Adds support for query strings containing multiple statements to the various SQL components. When a string contains multiple statements, they are all executed in the same database transaction.

Interpolated parameters (?, $1, etc.) may only be used in the first statement, similarly to the MySQL and Snowflake drivers which support this when not using the standard database/sql interfaces.

Affected components:

• The query argument to the sql_raw input, processor, and output
• The init_statement arg and init_files contents provided to sql_raw, sql_select, and sql_write components

This is incredibly useful for more complex use cases, and required in cases where you must execute multiple statements in the same transaction. Examples:

• Create Postgres tables on the fly based on topic names coming in:

  output:
   sql_raw:
     driver: postgres
     dsn: postgres://localhost/postgres
     unsafe_dynamic_query: true
     query: |
       CREATE TABLE IF NOT EXISTS ${!metadata("table_name")} (id varchar primary key, document jsonb);
       INSERT INTO ${!metadata("table_name")} (id, document) VALUES ($1, $2)
         ON CONFLICT (id) DO UPDATE SET document = EXCLUDED.document;
     args_mapping: |
       root = [ this.id, this.document.string() ]
  

• Upsert individual rows into Snowflake:

  output:
    sql_raw:
      driver: snowflake
      dsn: username:[email protected]/mydatabase/public
      args_mapping: root = [ this.id, this.data ]
      init_statement: |
        CREATE TABLE IF NOT EXISTS mytable (
          id STRING,
          data STRING
        );
        CREATE TABLE IF NOT EXISTS mytable_staging (
          id STRING,
          data STRING
        );
      query: |
        INSERT OVERWRITE INTO mytable_staging (id, data) VALUES (?, ?);
        MERGE INTO mytable USING mytable_staging AS incoming
          ON mytable.id = incoming.id
        WHEN MATCHED THEN UPDATE SET
          mytable.id = incoming.id,
          mytable.data = incoming.data
        WHEN NOT MATCHED THEN
          INSERT (id, data) VALUES (
            incoming.id,
            incoming.data
          )
        ;
  

• Put batches into a Snowflake staging table and then merge them into the main table:

  output:
    broker:
      pattern: fan_out_sequential
      outputs:
      - snowflake_put:
          init_statement: |
            CREATE TABLE IF NOT EXISTS mytable_staging (
              id string,
              data string,
              updated_at timestamptz
            );
           ...
      - sql_raw:
          driver: snowflake
          dsn: username:[email protected]/mydatabase/public
          init_statement: |
            CREATE TABLE IF NOT EXISTS mytable (
              id string,
              data string,
              updated_at timestamptz
            );
          query: |
            MERGE INTO mytable USING (
              SELECT * FROM mytable_staging
              QUALIFY 1 = RANK() OVER (PARTITION BY id ORDER BY updated_at DESC)
            ) mytable_staging
              ON mytable.id = mytable_staging.id
            WHEN MATCHED AND mytable.updated_at < mytable_staging.updated_at THEN UPDATE SET
              mytable.id = mytable_staging.id,
              mytable.data = mytable_staging.data,
              mytable.updated_at = mytable_staging.updated_at
            WHEN NOT MATCHED THEN
              INSERT (id, data, updated_at) VALUES (
                mytable_staging.id,
                mytable_staging.data,
                mytable_staging.updated_at
              )
            ;
            TRUNCATE mytable_staging
            ;