-
Notifications
You must be signed in to change notification settings - Fork 97
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
SNOW-1658905 Integration tests setup for Iceberg (#930)
- Loading branch information
1 parent
f0e5f23
commit 64e6f24
Showing
6 changed files
with
184 additions
and
36 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
35 changes: 35 additions & 0 deletions
35
...ain/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergTableSchemaValidator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,35 @@ | ||
package com.snowflake.kafka.connector.streaming.iceberg; | ||
|
||
import com.snowflake.kafka.connector.internal.SnowflakeConnectionService; | ||
|
||
/** Performs validations of Iceberg table schema on the connector startup. */ | ||
public class IcebergTableSchemaValidator { | ||
|
||
private final SnowflakeConnectionService snowflakeConnectionService; | ||
|
||
public IcebergTableSchemaValidator(SnowflakeConnectionService snowflakeConnectionService) { | ||
this.snowflakeConnectionService = snowflakeConnectionService; | ||
} | ||
|
||
/** | ||
* Ensure that table exists and record_metadata column is of type OBJECT(). | ||
* | ||
* <p>TODO SNOW-1658914 - write a test for table with record_metadata schema altered by the | ||
* connector | ||
*/ | ||
public void validateTable(String tableName, String role) { | ||
// TODO - plug into connector startup | ||
if (!snowflakeConnectionService.tableExist(tableName)) { | ||
// TODO - better errors | ||
throw new RuntimeException("TODO"); | ||
} | ||
|
||
// TODO - why is it so slow? | ||
if (!snowflakeConnectionService.hasSchemaEvolutionPermission(tableName, role)) { | ||
// TODO - better errors | ||
throw new RuntimeException("TODO"); | ||
} | ||
|
||
// TODO - call describe table and analyze record_metadata schema | ||
} | ||
} |
23 changes: 23 additions & 0 deletions
23
src/test/java/com/snowflake/kafka/connector/internal/TestSnowflakeConnection.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,23 @@ | ||
package com.snowflake.kafka.connector.internal; | ||
|
||
import static com.snowflake.kafka.connector.internal.TestUtils.getConfFromFileName; | ||
|
||
import com.snowflake.client.jdbc.SnowflakeDriver; | ||
import com.snowflake.kafka.connector.Utils; | ||
import java.sql.Connection; | ||
import java.util.Properties; | ||
|
||
/** Connection to test environment generated from a profile file stored locally. */ | ||
public class TestSnowflakeConnection { | ||
|
||
/** Given a profile file path name, generate a connection by constructing a snowflake driver. */ | ||
public static Connection getConnection() throws Exception { | ||
SnowflakeURL url = | ||
new SnowflakeURL(getConfFromFileName(TestUtils.PROFILE_PATH).get(Utils.SF_URL)); | ||
|
||
Properties properties = | ||
InternalUtils.createProperties(getConfFromFileName(TestUtils.PROFILE_PATH), url); | ||
|
||
return new SnowflakeDriver().connect(url.getJdbcUrl(), properties); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
42 changes: 42 additions & 0 deletions
42
src/test/java/com/snowflake/kafka/connector/streaming/iceberg/BaseIcebergIT.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,42 @@ | ||
package com.snowflake.kafka.connector.streaming.iceberg; | ||
|
||
import static com.snowflake.kafka.connector.internal.TestUtils.executeQueryWithParameter; | ||
|
||
import com.snowflake.kafka.connector.internal.SnowflakeConnectionService; | ||
import com.snowflake.kafka.connector.internal.TestUtils; | ||
import org.junit.jupiter.api.AfterAll; | ||
import org.junit.jupiter.api.BeforeAll; | ||
|
||
public class BaseIcebergIT { | ||
|
||
protected static SnowflakeConnectionService conn; | ||
|
||
@BeforeAll | ||
public static void setup() { | ||
conn = TestUtils.getConnectionServiceForStreaming(); | ||
} | ||
|
||
@AfterAll | ||
public static void teardown() { | ||
conn.close(); | ||
} | ||
|
||
protected static void createIcebergTable(String tableName) throws Exception { | ||
String query = | ||
"create or replace iceberg table identifier(?) (record_metadata object())" | ||
+ "external_volume = 'test_exvol'" | ||
+ "catalog = 'SNOWFLAKE'" | ||
+ "base_location = 'it'"; | ||
executeQueryWithParameter(query, tableName); | ||
} | ||
|
||
protected static void dropIcebergTable(String tableName) { | ||
String query = "drop iceberg table if exists identifier(?)"; | ||
executeQueryWithParameter(query, tableName); | ||
} | ||
|
||
protected static void enableSchemaEvolution(String tableName) throws Exception { | ||
String query = "alter iceberg table identifier(?) set enable_schema_evolution = true"; | ||
executeQueryWithParameter(query, tableName); | ||
} | ||
} |
60 changes: 60 additions & 0 deletions
60
...t/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergTableSchemaValidatorIT.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,60 @@ | ||
package com.snowflake.kafka.connector.streaming.iceberg; | ||
|
||
import com.snowflake.kafka.connector.internal.TestUtils; | ||
import org.junit.jupiter.api.AfterEach; | ||
import org.junit.jupiter.api.Assertions; | ||
import org.junit.jupiter.api.BeforeAll; | ||
import org.junit.jupiter.api.BeforeEach; | ||
import org.junit.jupiter.api.Test; | ||
|
||
public class IcebergTableSchemaValidatorIT extends BaseIcebergIT { | ||
|
||
private static final String TEST_ROLE = "testrole_kafka"; | ||
|
||
private static IcebergTableSchemaValidator schemaValidator; | ||
|
||
private String tableName; | ||
|
||
@BeforeAll | ||
// overrides the base class @BeforeAll | ||
public static void setup() { | ||
conn = TestUtils.getConnectionServiceForStreaming(); | ||
schemaValidator = new IcebergTableSchemaValidator(conn); | ||
} | ||
|
||
@BeforeEach | ||
public void setUp() { | ||
tableName = TestUtils.randomTableName(); | ||
} | ||
|
||
@AfterEach | ||
public void tearDown() { | ||
dropIcebergTable(tableName); | ||
} | ||
|
||
@Test | ||
public void shouldValidateExpectedIcebergTableSchema() throws Exception { | ||
// given | ||
createIcebergTable(tableName); | ||
enableSchemaEvolution(tableName); | ||
|
||
// when, then | ||
schemaValidator.validateTable(tableName, TEST_ROLE); | ||
} | ||
|
||
@Test | ||
public void shouldThrowExceptionWhenTableDoesNotExist() { | ||
Assertions.assertThrows( | ||
RuntimeException.class, () -> schemaValidator.validateTable(tableName, TEST_ROLE)); | ||
} | ||
|
||
@Test | ||
public void shouldThrowExceptionWhenRecordMetadataDoesNotExist() { | ||
// TODO | ||
} | ||
|
||
@Test | ||
public void shouldThrowExceptionWhenRecordMetadataHasInvalidType() { | ||
// TODO | ||
} | ||
} |