SQL blocks

​

Credentials

Before starting, you need to add credentials so Mage can execute your SQL commands.

Follow the steps for the database or data warehouse of your choice:

• BigQuery
• Druid
• MySQL
• PostgreSQL
• Redshift
• Snowflake

​

Add SQL block to pipeline

1. Create a new pipeline or open an existing pipeline.
2. Add a data loader, transformer, or data exporter block.
3. Select SQL.

​

Configure SQL block

There are 4 - 5 fields that must be configured for each SQL block:

​

Write policies

​

YAML configuration

You can also modify block configuration in pipeline’s metadata.yaml file. Each block has a configuration field.

Example configuration

  configuration:
    data_provider: sqlserver
    data_provider_profile: default
    data_provider_schema: ''
    export_write_policy: append
    limit: 1000
    limit_in_pipeline_run: 1
    use_raw_sql: false

In addition to the fields mentioned in the table above. Here are some extra fields that can be included in the configuration:

​

Automatically created tables

Each SQL block will create a table in the data provider of your choice.

When you run a block, it’ll execute your SQL command, then store the results in a table created in your database or data warehouse.

​

Using raw SQL

If you toggle this setting, you’re responsible for writing the CREATE TABLE command and the INSERT command.

For example, if a table already exists then you can write the INSERT statement:

INSERT INTO mage.users
SELECT 1 AS id, 'Urza' AS username;

If the table doesn’t exist yet, you can write both the CREATE TABLE statement and the INSERT statement:

CREATE TABLE IF NOT EXISTS mage.users (
    id BIGINT
    , username VARCHAR(255)
);

WITH users AS (
    SELECT
        1 AS id
        , 'Urza' AS username
)

INSERT INTO mage.users
SELECT
    *
FROM users;

This SQL query will create a table named mage.users with 2 columns: id as a BIGINT and username as a VARCHAR(255).

Then, it’ll insert a single row into that table.

​

Required SQL statements

When writing raw SQL, you must at least 1 of the following statements:

• SELECT
  Copy

  SELECT 1;
  

• INSERT
  Copy

  INSERT INTO some_table
  SELECT 1;
  

• CREATE TABLE
  Copy

  CREATE TABLE some_table (id BIGINT);
  

• DROP TABLE
  Copy

  DROP TABLE some_table;
  

• UPDATE
  Copy

  UPDATE some_table
  SET col1 = 'value1'
  WHERE col2 = 'value2'
  

​

Multiple SQL statements

You can execute multiple SQL statements in a SQL block. Separate your SQL statements using a semi-colon (;).

​

Automatic naming of tables

If you don’t choose the setting for using raw SQL, the name of this automatically created table follows these conventions:

• If Database field is configured: [database].[schema].[pipeline UUID]_[block UUID]
• If no Database field is configured: [schema].[pipeline UUID]_[block UUID]

Where pipeline UUID is the name of the current pipeline you’re editing.

Where block UUID is the name of the SQL block you are running.

​

Upstream blocks

If your SQL block depends on upstream blocks that aren’t SQL blocks (e.g. Python code blocks), then those blocks will also automatically create tables.

The name of those tables follows the same naming convention mentioned above.

​

Variables

All SQL blocks have the following variables they can access in their query:

​

{{ execution_date }}

The date and time the block is ran.

Example

SELECT '{{ execution_date }}' AS today

Result

If a SQL block has 1 or more upstream blocks, then they have access to their parent blocks’ output using the following variable:

​

{{ df_1 }}

Depending on how many upstream blocks there are, the variable name changes. For example, if there are 3 upstream blocks then there are 3 variables that can be accessed:

• {{ df_1 }}
• {{ df_2 }}
• {{ df_3 }}

The SQL block UI will display which variable maps to which upstream block. By convention, the 1st added upstream block will be {{ df_1 }}, and every upstream block added after that will have an incrementing number in the variable name after the prefix df_.

Example

SELECT a.id , b.username FROM {{ df_1 }} AS a LEFT JOIN {{ df_2 }} AS b ON
a.id = b.user_id LIMIT 1

Result

​

Variable interpolation

Interpolate values in the block’s code using data from:

1. Upstream block output
2. Variables
   1. Global variables
   2. Pipeline variables
   3. Runtime variables
3. Environment variables

​

Upstream block output

Use the data from 1 or more upstream block’s output by using the block_output function.

​

block_uuid

The UUID of the upstream block to get data from. If argument isn’t present, data from all upstream blocks will be fetched.

​

parse

A lambda function to parse the data from an upstream block’s output. If the parse argument isn’t present, then the fetched data from the upstream block’s output will be interpolated as is.

• data

  If the block_uuid argument isn’t present, then the 1st argument in the lambda function is a list of objects.

  The list of objects contain the data from an upstream block’s output. The positional order of the data in the list corresponds to the current block’s upstream blocks order.

  For example, if the current block has the following upstream blocks with the following output:

  1. load_api_data: [1, 2, 3]
  2. load_users_data: { 'mage': 'powerful' }

  Then the 1st argument in the lambda function will be the following list:

  Copy

  [
      [1, 2, 3],
      { 'mage': 'powerful' },
  ]
  

  Optional	❌
  Type	If block_uuid argument is present, then the type depends on the output from that block. If block_uuid isn’t present, then the type is list.
  Example	{ 'mage': 'powerful' }

• variables A dictionary containing pipeline variables and runtime variables.

  Optional	❌
  Type	dict
  Example	{ 'fire': 40 }

​

Example

With block_uuid

SELECT
    '{{ block_output("magic_config", parse=lambda data, _vars: data["power"]) }}' AS power
    , '{{ block_output("magic_config", parse=lambda data, _vars: data["spell"]) }}' AS spell
FROM mage.core_spells

SELECT
    '{{ block_output("data_loader_block", parse=lambda data, variables: data["runtime"] * variables["tries"]) }}' AS data
FROM mage.core_spells

SELECT
    '{{ block_output("load_recent_user", parse=lambda user, _variables: user["id"]) }}'
FROM mage.core_spells

Without block_uuid

SELECT
    '{{ block_output(parse=lambda data, _vars: data[0]["power"]) }}' AS power
    , '{{ block_output(parse=lambda data, _vars: data[0]["spell"]) }}' AS spell
FROM mage.core_spells

SELECT
    '{{ block_output(parse=lambda data, variables: data[0]["runtime"] * variables[0]["tries"]) }}' AS data
FROM mage.core_spells

SELECT
    '{{ block_output(parse=lambda user, _variables: user[0]["id"]) }}'
FROM mage.core_spells

​

Variables

Interpolate values from a dictionary containing keys and values from:

1. Global variables
2. Pipeline variables
3. Runtime variables

​

Example

SELECT
    '{{ variables("mana") }}' AS mana
FROM mage.dual_lands

​

Environment variables

Interpolate values from the environment variables.

​

Example

SELECT
    '{{ env_var("ALLOW_POWER_9") }}' AS vintage
FROM mage.mox

​

Executing pure SQL

Mage automatically pre-processes your SQL command before sending it to the final destination.

The pre-process extracts and structures the command so that Mage can build upstream tables and properly display preview data.

However, there are scenarios where the SQL command you write must be executed as is. You can disable Mage’s pre-processing by opening a SQL block and checking the box labeled Use raw SQL and Disable query preprocessing.