Consuming and Storing Kafka Messages in Snowflake Using Kafka Connect (Part 2)

A story about adding a new column to a table with 600 million records The Challenge Picture this. You have a task to implement data loading from a large PostgreSQL database table to Snowflake through a data pipeline. Simple enough, right? Just do it incrementally, using the appropriate column as a bookmark to track how much data has already been processed. The problem is that the table has 600 million records in production and lacks a column suitable for the bookmark. This was exactly the situation we faced recently. We needed to: Add an updated_at column to track changes. Backfill existing records with updated_at = created_at Add an index on the updated_at column for efficient querying (essential for bookmark-based incremental loads) Do all of this without affecting production traffic. Why Traditional Approaches Don't Work Even though created_at was there, it didn’t fit. The updated_at timestamp was the most suitable choice for the bookmark because a single record can be updated over time, and we need to capture those updates.  The naive approach would be to create a new migration with an ALTER TABLE statement, followed by an UPDATE to populate the column, and then CREATE INDEX: ALTER TABLE tableADD COLUMN updated_at TIMESTAMP DEFAULT now();UPDATE tableSET updated_at = created_at;CREATE INDEX idx_table_updated_atON table(updated_at); The whole migration would be wrapped inside a transaction by default.  But what happens next? Adding a column with a default value updates each existing row. This can take hours and locks all rows until it’s done. An UPDATE without batching creates a large transaction that locks all rows for an extended period. Even without the transaction, the lock stays until the statement finishes. This can take a while. So, moving the statement to a separate migration won’t help much. A CREATE INDEX operation gets an exclusive lock on the whole table. This stops all writes and reads for hours.  Any of these operations could bring production to its knees. The table size is the real issue. With less data, all these operations would finish quickly and not disrupt the system. Locking becomes a problem when the operation takes too long. The Zero-Downtime Strategy Here’s how we addressed the issue using PostgreSQL's features and strategic batching. Step 1: Add the Column without a default value We began by adding the column without immediately populating it. In PostgreSQL, adding a nullable column without a default value is almost instantaneous: ALTER TABLE table ADD COLUMN updated_at TIMESTAMP; Step 2: Batch Update Existing Records Rather than updating all 600 million records in a single transaction, we processed them in smaller chunks of 10,000 records. This method: Keeps transactions small and efficient Allows other operations to continue between batches Prevents lock escalation DO 'DECLARE chunk_size INT := 10000; start_id INT; max_id INT;BEGIN SELECT MIN(id), MAX(id) INTO start_id, max_id FROM table; WHILE start_id <= max_id LOOP EXECUTE format( ''UPDATE table SET updated_at = created_at WHERE id >= %s AND id < %s'', start_id, start_id + chunk_size ); start_id := start_id + chunk_size; END LOOP;END'; Step 3: Create Index Concurrently PostgreSQL's CREATE INDEX CONCURRENTLY is a valuable option in these scenarios.  While index creation is slower than usual, it offers greater safety for production environments. Unlike regular index creation, it: Does not block reads and writes Performs multiple table scans to build the index Takes longer but ensures full availability CREATE INDEX CONCURRENTLY idx_table_updated_at ON table (updated_at); Important note: This command must be executed outside of a transaction block. Running it within a transaction would negate its benefits and cause blocking. The complete migration script now looks like this: ALTER TABLE table ADD COLUMN updated_at TIMESTAMP;DO 'DECLAREchunk_size INT := 10000;start_id INT;max_id INT;BEGINSELECT MIN(id), MAX(id) INTO start_id, max_id FROM table;WHILE start_id <= max_id LOOPEXECUTE format(''UPDATE table SET updated_at = created_at WHERE id >= %s AND id < %s'',start_id, start_id + chunk_size);start_id := start_id + chunk_size;END LOOP;END';CREATE INDEX CONCURRENTLY idx_table_updated_at ON table (updated_at); Implementation with Liquibase As our backend is built on Spring Boot, we utilize Liquibase for database migrations. Below is a complete migration file that reflects what we implemented in the production environment: <?xml version="1.1" encoding="UTF-8" standalone="no"?><databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.5.xsd"> <changeSet id="add-table-updated_at" author="amahovac" context="post-release"> <addColumn tableName="table"> <column name="updated_at" type="TIMESTAMP WITH TIME ZONE"/> </addColumn> </changeSet> <changeSet id="update-table-updated_at" author="amahovac" runInTransaction="false" context="post-release"> <!-- Batch update logic here --> </changeSet> <changeSet id="create-concurrent-index-table-updated_at" runInTransaction="false" author="amahovac" context="post-release"> <!-- Concurrent index creation here --> </changeSet></databaseChangeLog> Key Details Separate migrations We organized operations into distinct migrations (changesets) to improve control over each. Running migrations outside the transaction This is crucial for concurrent operations. If not managed, Liquibase would encompass everything in a transaction, leading to the locks we aim to avoid. Adding runInTransaction="false" prevents this issue. Skipping migrations during deployment We used context="post-release" and adjusted the Spring Boot configuration files to set spring.liquibase.contexts=!post-release before deployment.  This way, migrations won't run during deployment and potentially time out our CodeDeploy process. Running it manually After deployment, we executed these migrations manually during a low-traffic period to monitor performance impacts and minimize risk. The Result The entire operation was completed successfully with zero downtime. The production application continued serving requests throughout the 5+ hour process. Our data pipeline now efficiently syncs the table incrementally to Snowflake.  Lessons Learned Test with production-sized data: Our initial approach performed well in the development environment with 1 million rows, but required significant adjustments for the production database. Always consider data volume and test with realistic datasets. Monitor during execution: We kept an eye on CPU, I/O, and lock metrics throughout the process. Document everything: Future team members need to understand the nuances of these migrations. Our inline comments saved confusion later. Always include created_at and updated_at columns when creating a table: You never know when they might be needed. If we had included them from the start, we could have avoided all these complications. Avoid unnecessary locking: Consider concurrent solutions and batch large updates to minimize lock escalation. Run DDL operations with caution: If possible, execute long-running migrations outside of transactions and separate them from deployment.  Working with large production database tables requires patience, careful planning, and an awareness of the impact of each operation. The extra effort to ensure zero downtime is always worthwhile when users don't even notice you just modified 600 million records.

Introduction Building on our previous blog Building ETL Pipeline in Snowflake, we will cover different approaches to automated testing for ETL Pipelines in Snowflake. For these kinds of solutions, QA Engineers should use their creativity to implement useful test cases as much as possible. There are  various approaches, and we will go through the following: tasks verification, end-to-end testing, IQR analysis and data quality verification. Tasks verification This part of the automated testing helps the verification of E and T of ETL parts. If we examine the example from the previously mentioned blog, we can see the use of tasks in Snowflake ETL pipelines. This allows us to develop an automated test case that runs a Snowflake query to identify failed tasks within the last 24 hours or a specified time range, depending on your needs. Here is the example written in Node JS with Jest (or any other test runner that you prefer, such as Jasmine, etc.): 1. Initialise repository and install Jest with commands: npm init npm i jest 2. Create Snowflake Connector Util class: const Snowflake = require('snowflake-promise').Snowflake const connectionProperties = { account: 'TEST_ACCOUNT', username: 'TEST_USERNAME', role: 'TEST_ROLE', database: 'TEST_DB', schema: 'TEST_SCHEMA', warehouse: 'TEST_WAREHOUSE', authenticator: 'SNOWFLAKE_JWT', privateKeyPath: process.env.SF_TEST_PRIVATE_KEY, privateKeyPass: process.env.SF_TEST_PRIVATE_KEY_PASS, } const taskHistoryTable = `${connectionProperties.database}.INFORMATION_SCHEMA.TASK_HISTORY` class SnowflakeConnectorUtil { async snowflakePool() { return new Snowflake(connectionProperties) } async createPool() { const pool = await this.snowflakePool() await pool.connect() return pool } async getSnowflakeFailedTasksInLast24Hours() { return `SELECT * FROM ( SELECT distinct name, first_value(completed_time) ignore nulls over (partition by name order by completed_time desc) as last_completed_time, first_value(state) ignore nulls over (partition by name order by completed_time desc) as last_state FROM db.schema.task_history ) WHERE last_state IN ('FAILED', 'CANCELLED', 'FAILED_AND_AUTO_SUSPENDED') AND last_completed_time >= DATEADD(DAY, -1, GETDATE());` } } module.exports = SnowflakeConnectorUtil 3. Create a test file. For your test case use previously created Snowflake Connector Util: const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtil') const snowflakeConnectorUtil = new SnowflakeConnectorUtil() describe('Tasks verification', () => { let pool beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() }, 10000) it('Daily verification tasks', async () => { const query = await snowflakeConnectorUtil.getSnowflakeFailedTasksInLast24Hours() const records = await pool.execute(query) records.map(failedTask => { console.log( `Failed task + ${failedTask.NAME}, error code: ${failedTask.ERROR_CODE}, error message: ${failedTask.ERROR_MESSAGE}` ) }) expect(records).toEqual([]) }, 10000) afterAll(async () => { await pool.destroy() }) }) With this approach, we could have the following benefits: Early detection of ETL failures: issues are caught long before they affect downstream Improved data reliability: leads to higher confidence in the data across dashboards and reports Reduced manual effort: engineers do not have to waste time on a manual daily tasks check Proactive incident management: integration with the communication tools such as Slack, Teams, email, etc. allows proactive response and prevents cases in which users notify problems before the engineers Ability to have an overview picture of ETL health over time Improved SLA compliance: Early automated detection helps teams to maintain data availability and freshness commitments in line with business SLAs End-to-end In this part, we are covering the whole process of ETL by publishing some test data and verifying that the data is being processed as it is required. We will use an ETL pipeline example from a previous blog and develop our end-to-end test automation example based on that. In the previous blog, the main source of the data is the Kafka topic. Therefore, our first test step in some of the test environments would be to prepare the test data by publishing it to the Kafka topic. We will do that with Node.js, but of course, feel free to use any other programming language or tool with which you are comfortable. npm install kafka npm install jest or npm install jasmine Next, we need to produce the Data on our ETL data source that is in this example Kafka. So we will use the Kafka producer: const { Kafka } = require('kafkajs'); class KafkaProducer { kafka; producer; constructor() { this.kafka = new Kafka({ clientId: process.env.YOUR_CLIENT_ID, brokers: process.env.KAFKA_HOST }); this.producer = this.kafka.producer(); } async sendMessage(message) { try { await this.producer.connect(); await this.producer.send({ topic: process.env.YOUR_KAFKA_TOPIC, messages: [ { value: message }, ], }); await this.producer.disconnect() } catch (error) { console.log(error); } } } module.exports = KafkaProducer; Note: Although this data source issue could be resolved on the ETL side by various services, such as AWS S3 buckets and RabbitMQ, for this blog, we decided to test a data pipeline that uses Kafka as the data source. We will still need the Snowflake Connector Util class, but with getters for Snowflake queries for staging and the final table. It should look something as the following: const Snowflake = require('snowflake-promise').Snowflake const connectionProperties = { account: 'TEST_ACCOUNT', username: 'TEST_USERNAME', role: 'TEST_ROLE', database: 'TEST_DB', schema: 'TEST_SCHEMA', warehouse: 'TEST_WAREHOUSE', authenticator: 'SNOWFLAKE_JWT', privateKeyPath: process.env.SF_TEST_PRIVATE_KEY, privateKeyPass: process.env.SF_TEST_PRIVATE_KEY_PASS, } const taskHistoryTable = `${connectionProperties.database}.INFORMATION_SCHEMA.TASK_HISTORY` const packageStagingTable = `${connectionProperties.database}.PACKAGE_STAGING` const packageFinalTable = `${connectionProperties.database}.PACKAGE_FINAL` class SnowflakeConnectorUtil { async snowflakePool() { return new Snowflake(connectionProperties) } async createPool() { let pool = await this.snowflakePool() await pool.connect() return pool } async getSnowflakeFailedTasksInLast24Hours() { return `SELECT * FROM ( SELECT distinct name, first_value(completed_time) ignore nulls over (partition by name order by completed_time desc) as last_completed_time, first_value(state) ignore nulls over (partition by name order by completed_time desc) as last_state FROM ${taskHistoryTable} ) WHERE last_state IN ('FAILED', 'CANCELLED', 'FAILED_AND_AUTO_SUSPENDED') AND last_completed_time >= DATEADD(DAY, -1, GETDATE());` } checkNullValuesQueries = [ { description: 'Package Sender or Recipient ID', query: `select * from dbo.schema.packages_final where sender_id is null or getter_id is null`, expected: 0, }, { description: 'User ID', query: `select * from dbo.schema.user_request where user_id is null`, expected: 0, } ] getPackageStagingTableQuery(packageId) { return `select * from ${packageStagingTable} where package_id = '${packageId}'` } getPackageFinalTableQuery(packageId) { return `select * from ${packageFinalTable} where package_id = '${packageId}'` } } module.exports = SnowflakeConnectorUtil Then we just need to use everything mentioned in the test case: import { v6 as uuidv6 } from 'uuid'; const Kafka = require('./KafkaProducer') const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtill') const kafka = new Kafka() const snowflakeConnectorUtil = new SnowflakeConnectorUtil() describe('E2E test example', () => { let pool let packageId beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() packageId = uuidv6() // producing test data await kafka.sendMessage( `{ 'packageId', ${packageId}, 'senderId': '1', 'recipientId': '2', }`) }) it ('Verify staging table', async () => { const packagesStagingQuery = snowflakeConnectorUtil.getPackageStagingTableQuery(packageId) const packagesStaging = await pool.execute(packagesStagingQuery) expect(packagesStaging).toBeGreaterThan(0) expect(packagesStaging[0].senderFirstName).toBe('John') expect(packagesStaging[0].senderLastName).toBe('Doe') expect(packagesStaging[0].recipientFirstName).toBe('Jane') expect(packagesStaging[0].recipientLastName).toBe('Doe') }) it ('Verify final table', async () => { const packagesFinalQuery = snowflakeConnectorUtil.getPackageFinalTableQuery(packageId) const packagesFinal = await pool.execute(packagesFinalQuery) expect(packagesFinal).toBeGreaterThan(0) expect(packagesFinal[0].senderFirstName).toBe('John') expect(packagesFinal[0].senderLastName).toBe('Doe') expect(packagesFinal[0].recipientFirstName).toBe('Jane') expect(packagesFinal[0].recipientLastName).toBe('Doe') }) afterAll(async () => { await pool.destroy() }) }) By combining various data, we can create multiple test scenarios and achieve confidence in our pipeline after the team makes changes. This approach leads us to the following benefits: Ensuring data integrity across the pipeline Validates transformations and business logic Detects schema or contract breaks early Builds confidence in data freshness and accuracy Reduces manual verification effort IQR analysis You can learn more about the IQR on the following blog IQR in Automation Testing: Unleashing the Data Analytics Potential - Atlantbh Sarajevo. In the context of testing Snowflake ETL pipelines, we can implement this on the appropriate dataset. Test cases based on this approach will provide us more confidence in our ETL process. Significant changes in our data can point to potential issues. A significant increase might indicate that our user base is growing, so we should be prepared for that. A sudden drop could indicate an issue with data processing. For instance, let us say we want to find outliers in the data set for the final table names, which will be called “package_final”. A common use case would be to identify outliers, such as a significant drop or increase in data over the last 30 days. Since it is not valid in many cases to compare Mondays with Saturdays, we should create an appropriate query to retrieve the last 30 Mondays. We will add this query to the SnowflakeConnectorUtil class: getLast30Weeks() { return `WITH OrderedRows AS ( SELECT DATE(created_ts_pst) created_date, count(*) as count, ROW_NUMBER() OVER (ORDER BY created_date) AS row_num, FROM ${packageFinalTable} WHERE created_ts_pst BETWEEN DATEADD(DAY, -210, current_date) AND current_date GROUP BY created_date Order by created_date desc ) SELECT * FROM OrderedRows WHERE row_num % 7 = 0;` } We will wrap our IQR analyze logic into shared service: const iqrAnalyze = async data => { const dataSetValues = data.map(value => value.COUNT) const dataSetSorted = dataSetValues.sort((a, b) => a - b) const quartile1 = dataSetSorted[Math.floor(dataSetSorted.length / 4)] const quartile3 = dataSetSorted[Math.ceil(dataSetSorted.length * (3 / 4))] const iqr = quartile3 - quartile1 const lowerBound = quartile1 - 1.5 * iqr const upperBound = quartile3 + 1.5 * iqr const outliers = dataSetSorted.filter( value => (value < lowerBound || value > upperBound) && value == dataSet[0].COUNT ) expect(outliers.length).toBe(0) } module.exports = { iqrAnalyze: iqrAnalyze, } Next we test it by using data from Snowflake in our test file: const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtil') const snowflakeConnectorUtil = new SnowflakeConnectorUtil() const iqrAnalyzer = require('./services/iqrAnalyze') describe('IQR test suite', () => { let pool beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() }, 10000) it('Packages test', async () => { const query = await snowflakeConnectorUtil.getLast30Weeks() const data = await pool.execute(query) await iqrAnalyzer.iqrAnalyze(data) }, 10000) afterAll(async () => { await pool.destroy() }) }) The main advantages of incorporating IQR analysis into automated testing include: Early data anomaly detection Better confidence in data quality Data observability support that leads to better confidence in ETL Data Quality verification In this section, we should verify the quality of data that has undergone the pipeline transformations. One of the most common mistakes in ETL processes is invalid data in columns such as NULL values. To find these defects, we could use the following approach. First, let us define an array of queries, each with a description and the expected number of records that should be returned with the query: checkNullValuesQueries = [ { description: 'Package Sender or Recipient ID', query: `select * from ${packageFinalTable} where sender_id is null or getter_id is null`, expected: 0, }, { description: 'User ID', query: `select * from dbo.schema.user_request where user_id is null`, expected: 0, } ] Next, in the test case we go through the array of queries, execute each one and assert the expected records length: const SnowflakeConnectorUtil = require('/db/SnowflakeConnectorUtil') const snowflakeConnectorUtil = new SnowflakeConnectorUtil() describe('Data quality verification', () => { let pool beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() }, 10000) snowflakeConnectorUtil.checkNullValuesQueries.map(checkNullValueQuery => { it(`Verify ${checkNullValueQuery.description}`, async () => { const nullFieldsRecords = await pool.execute(checkNullValueQuery.query) expect(nullFieldsRecords.length) .withContext(JSON.stringify(nullFieldsRecords)) .toBe(checkNullValueQuery.expected) await sleep(10000) }, 100000) }) afterAll(async () => { await pool.destroy() }) }) This check strengthens the overall data governance framework by ensuring the standardized quality of the data, guarantees the data completeness, and supports compliance with business and regulatory requests. CI/CD For the CI/CD part, we will use Jenkins; therefore, the first step will be setting up the Jenkins server. The easiest way would be to use Docker and first pull the latest Jenkins Docker image, and run Jenkins on a Docker container by running the following scripts in the terminal: docker pull jenkins/jenkins:lts docker run -d --name jenkins-lts -p 8080:8080 -p 50000:50000 -v jenkins_home:/var/jenkins_home jenkins/jenkins:lts Next, go to your localhost:8080 (or any other port that you have configured) and configure the admin account. Since we are developing our test cases in Node JS, we need to install the appropriate Jenkins plugin by navigating to Manage Jenkins > Manage Plugins > Available plugins. Next, find “NodeJS Plugin” and click on the install button. After the plugin installation, we need to configure the Node JS version by going to Manage Jenkins > Tools, finding NodeJS installations, and clicking on Add “NodeJS”. This will open options for choosing the appropriate Node version that you need, as well as additional packages that you want to have on your Jenkins. Now we will create our Jenkins files in the git repository. The Jenkins file for an end-to-end test case could be such as the following: pipeline { agent any tools { nodejs 'NodeJS 18' } triggers { cron('H 0 * * *') // Run once a day at midnight } stages { stage('Checkout') { steps { checkout scm } } stage('Install Dependencies') { steps { sh 'npm install' } } stage('Run Test') { steps { sh 'npm test e2e.test.js' } } } } Now let us back to the Jenkins dashboard. Click on the “+ New Item” button and then enter the test name for instance: “E2E Snowflake test”, choose “Pipeline” and click the “Ok” button. Afterwards, add description, select the “Github project” checkbox, and paste your remote repository link. In the pipeline section, choose 'Pipeline script from SCM' and set your GitHub repository link, along with other details such as credentials, branch name, and script file path. Then, save the details and back to the main dashboard Jenkins where we should see our job: We can now manually trigger our test or wait for the cron schedule we set up earlier in the script by specifying the appropriate CRON. Next steps include adding other test cases and setting up Jenkins on a server of your choice. You could read more about Jenkins on our other blogs: Jenkins blogs. Reporting For the reporting part, we will introduce the Allure reporting library. First, we need to install the needed dependency in the Node JS repository with the command: npm i -D jest-allure2-reporter allure-commandline Then we should define jest.config.js as: module.exports = { reporters: [ 'default', [require.resolve('jest-allure2-reporter'), { resultsDir: 'allure-results' }], ], testEnvironment: require.resolve('jest-allure2-reporter/environment-node'), testTimeout: 30000, }; In package.json modify change scripts config to be such as: "scripts": { "test:ci": "jest --runInBand" }, We are using runInBand, which guarantees running multiple test cases as worker processes sequentially (one by one). If we do not use this, the test cases will run in parallel, and we could face race conditions while writing them to the allure reports and have a risk of corrupted reports. Then, inside your tests, add allure functions for attaching metadata to every test step that you will later need as an explanation in formatting your report. They could be a story, description, step, attachment, etc. Here is an example so you could use as a reference to your other steps: it ('Verify staging table', async () => { allure.story("Staging Table Check"); allure.description("Verify that staging table data matches expected schema"); allure.step("Querying and verifying staging table...", async () => { const packagesStagingQuery = snowflakeConnectorUtil.getPackageStagingTableQuery(packageId) const packagesStaging = await pool.execute(packagesStagingQuery) expect(packagesStaging).toBeGreaterThan(0) expect(packagesStaging[0].senderFirstName).toBe('John') expect(packagesStaging[0].senderLastName).toBe('Doe') expect(packagesStaging[0].recipientFirstName).toBe('Jane') expect(packagesStaging[0].recipientLastName).toBe('Doe') }); allure.attachment("Staging query result", "SELECT * FROM staging_table", "text/plain"); }) Note: Pay attention to the “allure.step”, it takes a string and also a callback as arguments, so here we are passing our test step logic as a callback. After running the test locally, we should get the allure-results folder in our repository with our report file with some additional details. Let us now include this in our CI/CD part on Jenkins. First, we need to install Allure Jenkins Plugin in Manage Jenkins > Plugins Then we need to install Allure Commandline in Manage Jenkins > Tools Once that is done, let us modify the Jenkins script in our Git repository so we could add the reporting part at the bottom. It should be such as the following: pipeline { agent any tools { nodejs 'NodeJS 18' } triggers { cron('H 0 * * *') // Run once a day at midnight } stages { stage('Checkout') { steps { checkout scm } } stage('Install Dependencies') { steps { sh 'npm install' } } stage('Run Test') { steps { sh 'npm run test:ci e2e.test.js' } } } post { always { archiveArtifacts artifacts: 'allure-results/**', allowEmptyArchive: true } success { allure([ includeProperties: true, jdk: '', reportBuildPolicy: 'ALWAYS', results: [[path: 'allure-results']] ]) } unsuccessful { allure([ includeProperties: true, jdk: '', reportBuildPolicy: 'ALWAYS', results: [[path: 'allure-results']] ]) } } } After running our Jenkins job we should get the following artifacts: If you download allure-report.zip you should find the index.html file as allure report. If you open it in a browser it should look like the following: Of course, you can modify everything by adopting the mentioned allure functions as you need. Note: Maybe your browser will block opening these metrics because of CORS but you could avoid that by running simple local static server from allure-report folder with python or some other technology by running command: python3 -m http.server 8081 If you open http://localhost:8081/ you should see the allure report without CORS errors. Finally, you could add a Jenkins step that sends an allure report as an email to the stakeholders or to make Slack notification. It could be done by shell script or Jenkins plugin “Email Extension. Conclusion We have explored various approaches to automated testing for ETL pipelines built in Snowflake, orchestrated through Jenkins CI/CD, and integrated with Allure reporting. All these approaches can help us detect issues early, maintain high data reliability, and continuously improve pipeline performance. Automated testing of the ETL pipeline in Snowflake has transformed our ETL validation from a manual process to a proactive and self-monitoring system.  

Introduction Snowflake is a cloud-based data platform that enables simple, fast, and scalable data storage, processing, and analysis. Unlike traditional solutions, Snowflake is not built on existing databases or Hadoop; instead uses its own SQL query engine and a unique architecture designed specifically for the cloud. It is used for analytical databases, processing large volumes of data, and building so-called "Data Cloud" solutions that connect data from various sources. The platform automatically scales resources based on workload, optimizing both performance and cost. Thanks to its support for standard SQL and compatibility with various analytics and visualization tools, Snowflake is suitable for a wide range of users, from data analysts to engineers and data scientists. In the following sections, we will focus on the technical aspect of integrating the Snowflake platform with the Java programming language, demonstrating how to establish a connection and work with data within Snowflake using Java. Snowflake Java Database Connectivity (JDBC) Snowflake Java Database Connectivity is the most commonly used method for integrating Snowflake with the Java programming language. It provides a standardized way to establish a connection, execute SQL queries, and manage data from within a Java application. Advantages and Disadvantages of Using Snowflake JDBC Using the Snowflake JDBC connection in Java offers key advantages. JDBC is a standard API that simplifies development and integrates well with existing Java tools. Snowflake’s official JDBC driver is regularly updated and allows direct SQL execution, transaction handling, and asynchronous queries, making it efficient for large data operations. However, this approach has some limitations. It requires manual management of connections and asynchronous logic, which can add complexity. JDBC also lacks native support for certain advanced Snowflake features and may involve extra overhead with driver version management in larger systems. Extending the JDBC Contract with Snowflake-Specific Interfaces While the standard JDBC API (java.sql) provides a well-defined contract for executing SQL statements and managing result sets, Snowflake extends this contract through vendor-specific interfaces that expose additional features tailored to its cloud data platform capabilities. These Snowflake-specific extensions are accessible by unwrapping the standard JDBC interfaces, such as Statement and Connection, into SnowflakeStatement and SnowflakeConnection. This unwrapping allows developers to leverage advanced functionalities not present in the core JDBC specification. Some of these features, some of which will be covered later, include: Asynchronous Query Execution: Using methods like executeAsyncQuery(), queries can be submitted without blocking the client thread, enabling improved application responsiveness and parallel processing. Query Identification and Tracking: Methods such as getQueryID() return a unique identifier for each query, allowing applications to track, monitor, or cancel queries as needed. Polling Query Status: Through functions like getQueryStatus(), clients can programmatically check the state of a running query, facilitating custom retry or timeout mechanisms. Session and Connection Metadata Access: SnowflakeConnection exposes session-level information, including session parameters, current warehouse, role, and database context, which are critical for auditing and dynamic query behavior. Retrieving Query Profile and Telemetry: Snowflake provides access to detailed query execution profiles, enabling performance tuning and diagnostics directly from the client side. Support for Query Cancellation and Abort: Advanced control allows clients to cancel running queries through the JDBC driver interface, an important feature for resource management in long-running or user-interrupted operations. File Transfer and Staging Utilities: Through extended connection interfaces, Snowflake JDBC supports operations for managing file uploads/downloads to internal or external stages, simplifying ETL workflows. By leveraging these vendor-specific extensions, developers can build richer integrations with Snowflake’s platform, improving both performance and control over query execution beyond the limitations of standard JDBC. Establishing a Connection to Snowflake Using JDBC There are several different methods for establishing a connection to Snowflake, such as using Single Sign-On (SSO) for authentication, leveraging OAuth 2.0 authorization, and employing key pair authentication, among others. In this section, we will focus specifically on using key pair authentication to connect to Snowflake through JDBC securely. This approach allows applications to authenticate without relying on usernames and passwords, instead using public-private key cryptography to enhance security and support automated, credential-free access. Below is a simple example of establishing this type of connection:   public static Connection getConnection() throws SQLException { String url = "jdbc:snowflake://<account_identifier>.snowflakecomputing.com"; Properties props = new Properties(); props.put("user", "<user>"); props.put("private_key_file", "/tmp/rsa_key.p8"); props.put("private_key_file_pwd", "dummyPassword"); props.put("db", "<database_name>"); props.put("schema", "<schema_name>"); props.put("warehouse", "<warehouse_name>"); props.put("role", "<role_name>"); return DriverManager.getConnection(url, props); } Snowflake Async JDBC Execution Snowflake enables asynchronous execution of SQL queries via the JDBC driver, which is particularly useful for long-running queries that should not block the main application thread. Instead of waiting for the query to complete, the application can immediately receive a queryId and then check the status in the background, fetching the results once the query has finished. The most commonly used asynchronous approach typically consists of the following three steps: Executing the Query and Retrieving the queryId  Checking the Query Execution Status Fetching the Results Once the Query Has Completed In the following sections, we will provide a more detailed explanation of each step in this approach. Executing the Query and Retrieving the queryId The first step in asynchronous query execution is submitting the SQL statement to Snowflake without waiting for the full result set. This method instantly returns a unique queryId, which serves as a reference to track the progress of the query and retrieve its results once completed.  Below is an example function that executes a query asynchronously and returns the corresponding queryId: public String executeAsyncQuery(final String query) throws SQLException { try (Connection conn = getConnection(); Statement stmt = conn.createStatement()) { SnowflakeStatement sfStatement = stmt.unwrap(SnowflakeStatement.class); sfStatement.executeAsyncQuery(query); String queryId = sfStatement.getQueryID(); return queryId; } catch (SQLException e) { throw e; } } Comparing JDBC Synchronous and Snowflake Asynchronous Execution When working with relational databases through JDBC, the standard approach to executing SQL queries is synchronous. In this model, the thread initiating the query blocks until the database returns the full result set. While this approach is straightforward and works well for short-running queries, it can become a bottleneck when handling complex or long-running operations. In contrast, Snowflake offers asynchronous query execution, allowing the client to submit a SQL statement and immediately continue with other tasks without waiting for the result. This is particularly useful for improving responsiveness in applications or for executing multiple queries in parallel. Having shown the asynchronous execution example above, the following demonstrates how the same query would be executed synchronously using standard JDBC: public void executeSyncQuery(final String query) throws SQLException { try (Connection conn = getConnection(); Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(query)) { while (rs.next()) { // Process each row System.out.println(rs.getString(1)); } } catch (SQLException e) { throw e; } } Checking the Query Execution Status After obtaining the queryId from the previous step, we can use the SnowflakeConnection.createResultSet(queryId) method to retrieve the query status using the Snowflake Java connector directly. This method returns a ResultSet object, which can be unwrapped into a SnowflakeResultSet, allowing us to access the query status. This approach provides a simple and efficient way to track the state of an asynchronously executed query. The function below implements the previously described logic, using the queryId to retrieve the query status directly via the Snowflake Java connector. public QueryStatus getQueryStatus(final String queryId) throws SQLException { try (Connection conn = getConnection()) { SnowflakeConnection sfConnection = conn.unwrap(SnowflakeConnection.class); QueryStatus sfStatus; try (ResultSet rs = sfConnection.createResultSet(queryId)) { sfStatus = rs.unwrap(SnowflakeResultSet.class).getStatus(); } return sfStatus; } } Fetching the Results Once the Query Has Completed This is the final step in the asynchronous query execution flow. Once the query has been submitted and its status successfully monitored, we can proceed to fetch the results, provided the query has completed successfully. In this step, we’ll reuse the previously defined getQueryStatus function to check the status of the query. If the status is SUCCEEDED, we will then retrieve the actual result set using SnowflakeConnection.createResultSet(queryId). public ResultSet fetchResults(final String queryId) throws SQLException { QueryStatus status = getQueryStatus(queryId); if (status == QueryStatus.SUCCEEDED) { Connection conn = getConnection(); SnowflakeConnection sfConnection = conn.unwrap(SnowflakeConnection.class); return sfConnection.createResultSet(queryId); } else { throw new IllegalStateException("Query has not completed successfully. Current status: " + status); } } Persistence and Retention of Asynchronous Query Results After handling all the previous steps, a natural question arises: How long can we retrieve the results of a given query? Results of asynchronously executed queries in Snowflake are retained for up to 24 hours after execution. Once this retention period expires, it is no longer possible to fetch the results using the queryId via the fetchResults() method. To ensure longer-term availability, it is recommended to explicitly save the query results into a permanent table immediately after the query finishes. This guarantees that the data remains accessible beyond the default 24-hour window and can be used for further analysis or processing as needed. "Integrating Snowflake with Java: JDBC and Async JDBC Execution" Tech Bite was brought to you by Almedin Pašalić, Junior Software Engineer at Atlantbh. Tech Bites are tips, tricks, snippets or explanations about various programming technologies and paradigms, which can help engineers with their everyday job.