Skip to content

Commit ed8d890

Browse files
zschmidzschmid
authored
Merge branch 'master' into klaviyoSourceAddSegmentsStream
2 parents c714107 + fe0fee4 commit ed8d890

58 files changed

Lines changed: 1015 additions & 781 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎airbyte-integrations/connectors/destination-snowflake/metadata.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ data:
55
connectorSubtype: database
66
connectorType: destination
77
definitionId: 424892c4-daac-4491-b35d-c6688ba547ba
8-
dockerImageTag: 4.0.16
8+
dockerImageTag: 4.0.17
99
dockerRepository: airbyte/destination-snowflake
1010
documentationUrl: https://docs.airbyte.com/integrations/destinations/snowflake
1111
githubIssueLabel: destination-snowflake

‎airbyte-integrations/connectors/destination-snowflake/src/main/kotlin/io/airbyte/integrations/destination/snowflake/client/SnowflakeAirbyteClient.kt‎

Lines changed: 109 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
package io.airbyte.integrations.destination.snowflake.client
66

77
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings
8+
import io.airbyte.cdk.ConfigErrorException
89
import io.airbyte.cdk.load.command.DestinationStream
910
import io.airbyte.cdk.load.component.TableOperationsClient
1011
import io.airbyte.cdk.load.component.TableSchemaEvolutionClient
@@ -64,39 +65,43 @@ class SnowflakeAirbyteClient(
6465
}
6566

6667
override suspend fun createNamespace(namespace: String) {
67-
// Check if the schema exists first
68-
val schemaExistsResult =
69-
dataSource.connection.use { connection ->
70-
val databaseName = snowflakeConfiguration.database.toSnowflakeCompatibleName()
71-
val statement =
72-
connection.prepareStatement(
73-
"""
74-
SELECT COUNT(*) > 0 AS SCHEMA_EXISTS
75-
FROM "$databaseName".INFORMATION_SCHEMA.SCHEMATA
76-
WHERE SCHEMA_NAME = ?
77-
""".andLog()
78-
)
79-
80-
// When querying information_schema, snowflake needs the "true" schema name,
81-
// so we unescape it here.
82-
val unescapedNamespace = namespace.replace("\"\"", "\"")
83-
statement.setString(1, unescapedNamespace)
84-
85-
statement.use {
86-
val resultSet = it.executeQuery()
87-
resultSet.use { rs ->
88-
if (rs.next()) {
89-
rs.getBoolean("SCHEMA_EXISTS")
90-
} else {
91-
false
68+
try {
69+
// Check if the schema exists first
70+
val schemaExistsResult =
71+
dataSource.connection.use { connection ->
72+
val databaseName = snowflakeConfiguration.database.toSnowflakeCompatibleName()
73+
val statement =
74+
connection.prepareStatement(
75+
"""
76+
SELECT COUNT(*) > 0 AS SCHEMA_EXISTS
77+
FROM "$databaseName".INFORMATION_SCHEMA.SCHEMATA
78+
WHERE SCHEMA_NAME = ?
79+
""".andLog()
80+
)
81+
82+
// When querying information_schema, snowflake needs the "true" schema name,
83+
// so we unescape it here.
84+
val unescapedNamespace = namespace.replace("\"\"", "\"")
85+
statement.setString(1, unescapedNamespace)
86+
87+
statement.use {
88+
val resultSet = it.executeQuery()
89+
resultSet.use { rs ->
90+
if (rs.next()) {
91+
rs.getBoolean("SCHEMA_EXISTS")
92+
} else {
93+
false
94+
}
9295
}
9396
}
9497
}
95-
}
9698

97-
if (!schemaExistsResult) {
98-
// Create the schema only if it doesn't exist
99-
execute(sqlGenerator.createNamespace(namespace))
99+
if (!schemaExistsResult) {
100+
// Create the schema only if it doesn't exist
101+
execute(sqlGenerator.createNamespace(namespace))
102+
}
103+
} catch (e: SnowflakeSQLException) {
104+
handleSnowflakePermissionError(e)
100105
}
101106
}
102107

@@ -194,28 +199,35 @@ class SnowflakeAirbyteClient(
194199
}
195200

196201
internal fun getColumnsFromDb(tableName: TableName): Set<ColumnDefinition> {
197-
val sql =
198-
sqlGenerator.describeTable(schemaName = tableName.namespace, tableName = tableName.name)
199-
dataSource.connection.use { connection ->
200-
val statement = connection.createStatement()
201-
return statement.use {
202-
val rs: ResultSet = it.executeQuery(sql)
203-
val columnsInDb: MutableSet<ColumnDefinition> = mutableSetOf()
204-
205-
while (rs.next()) {
206-
val columnName = escapeJsonIdentifier(rs.getString("name"))
207-
208-
// Filter out airbyte columns
209-
if (airbyteColumnNames.contains(columnName)) {
210-
continue
202+
try {
203+
val sql =
204+
sqlGenerator.describeTable(
205+
schemaName = tableName.namespace,
206+
tableName = tableName.name
207+
)
208+
dataSource.connection.use { connection ->
209+
val statement = connection.createStatement()
210+
return statement.use {
211+
val rs: ResultSet = it.executeQuery(sql)
212+
val columnsInDb: MutableSet<ColumnDefinition> = mutableSetOf()
213+
214+
while (rs.next()) {
215+
val columnName = escapeJsonIdentifier(rs.getString("name"))
216+
217+
// Filter out airbyte columns
218+
if (airbyteColumnNames.contains(columnName)) {
219+
continue
220+
}
221+
val dataType = rs.getString("type").takeWhile { char -> char != '(' }
222+
223+
columnsInDb.add(ColumnDefinition(columnName, dataType, false))
211224
}
212-
val dataType = rs.getString("type").takeWhile { char -> char != '(' }
213225

214-
columnsInDb.add(ColumnDefinition(columnName, dataType, false))
226+
columnsInDb
215227
}
216-
217-
columnsInDb
218228
}
229+
} catch (e: SnowflakeSQLException) {
230+
handleSnowflakePermissionError(e)
219231
}
220232
}
221233

@@ -302,32 +314,60 @@ class SnowflakeAirbyteClient(
302314
}
303315

304316
fun describeTable(tableName: TableName): LinkedHashMap<String, String> =
305-
dataSource.connection.use { connection ->
306-
val statement = connection.createStatement()
307-
return statement.use {
308-
val resultSet = it.executeQuery(sqlGenerator.showColumns(tableName))
309-
val columns = linkedMapOf<String, String>()
310-
while (resultSet.next()) {
311-
val columnName = resultSet.getString(DESCRIBE_TABLE_COLUMN_NAME_FIELD)
312-
// this is... incredibly annoying. The resultset will give us a string like
313-
// `{"type":"VARIANT","nullable":true}`.
314-
// So we need to parse that JSON, and then fetch the actual thing we care about.
315-
// Also, some of the type names aren't the ones we're familiar with (e.g.
316-
// `FIXED` for numeric columns),
317-
// so the output here is not particularly ergonomic.
318-
val columnType =
319-
resultSet
320-
.getString(DESCRIBE_TABLE_COLUMN_TYPE_FIELD)
321-
.deserializeToNode()["type"]
322-
.asText()
323-
columns[columnName] = columnType
317+
try {
318+
dataSource.connection.use { connection ->
319+
val statement = connection.createStatement()
320+
return statement.use {
321+
val resultSet = it.executeQuery(sqlGenerator.showColumns(tableName))
322+
val columns = linkedMapOf<String, String>()
323+
while (resultSet.next()) {
324+
val columnName = resultSet.getString(DESCRIBE_TABLE_COLUMN_NAME_FIELD)
325+
// this is... incredibly annoying. The resultset will give us a string like
326+
// `{"type":"VARIANT","nullable":true}`.
327+
// So we need to parse that JSON, and then fetch the actual thing we care
328+
// about.
329+
// Also, some of the type names aren't the ones we're familiar with (e.g.
330+
// `FIXED` for numeric columns),
331+
// so the output here is not particularly ergonomic.
332+
val columnType =
333+
resultSet
334+
.getString(DESCRIBE_TABLE_COLUMN_TYPE_FIELD)
335+
.deserializeToNode()["type"]
336+
.asText()
337+
columns[columnName] = columnType
338+
}
339+
columns
324340
}
325-
columns
326341
}
342+
} catch (e: SnowflakeSQLException) {
343+
handleSnowflakePermissionError(e)
327344
}
328345

329346
internal fun execute(query: String) =
330-
dataSource.connection.use { connection ->
331-
connection.createStatement().use { it.executeQuery(query) }
347+
try {
348+
dataSource.connection.use { connection ->
349+
connection.createStatement().use { it.executeQuery(query) }
350+
}
351+
} catch (e: SnowflakeSQLException) {
352+
handleSnowflakePermissionError(e)
332353
}
354+
355+
/**
356+
* Checks if a SnowflakeSQLException is related to permissions and wraps it as a
357+
* ConfigErrorException. Otherwise, rethrows the original exception.
358+
*/
359+
private fun handleSnowflakePermissionError(e: SnowflakeSQLException): Nothing {
360+
val errorMessage = e.message?.lowercase() ?: ""
361+
362+
// Check for known permission-related error patterns
363+
when {
364+
errorMessage.contains("current role has no privileges on it") -> {
365+
throw ConfigErrorException(e.message ?: "Permission error", e)
366+
}
367+
else -> {
368+
// Not a known permission error, rethrow as-is
369+
throw e
370+
}
371+
}
372+
}
333373
}

‎airbyte-integrations/connectors/destination-snowflake/src/test/kotlin/io/airbyte/integrations/destination/snowflake/client/SnowflakeAirbyteClientTest.kt‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
package io.airbyte.integrations.destination.snowflake.client
66

7+
import io.airbyte.cdk.ConfigErrorException
78
import io.airbyte.cdk.load.command.DestinationStream
89
import io.airbyte.cdk.load.command.NamespaceMapper
910
import io.airbyte.cdk.load.command.Overwrite
@@ -37,8 +38,10 @@ import javax.sql.DataSource
3738
import kotlinx.coroutines.runBlocking
3839
import net.snowflake.client.jdbc.SnowflakeSQLException
3940
import org.junit.jupiter.api.Assertions.assertEquals
41+
import org.junit.jupiter.api.Assertions.assertTrue
4042
import org.junit.jupiter.api.BeforeEach
4143
import org.junit.jupiter.api.Test
44+
import org.junit.jupiter.api.assertThrows
4245

4346
internal class SnowflakeAirbyteClientTest {
4447

@@ -750,4 +753,102 @@ internal class SnowflakeAirbyteClientTest {
750753
// In production, this would typically trigger a retry
751754
}
752755
}
756+
757+
@Test
758+
fun testExecuteWithNoPrivilegesError() {
759+
val connection = mockk<Connection>()
760+
val statement = mockk<Statement>()
761+
val sql = "CREATE TABLE test_table (id INT)"
762+
763+
every { dataSource.connection } returns connection
764+
every { connection.createStatement() } returns statement
765+
every { statement.close() } just Runs
766+
767+
// Simulate permission error matching the user's case
768+
every { statement.executeQuery(sql) } throws
769+
SnowflakeSQLException(
770+
"SQL compilation error:\n" +
771+
"Table 'APXT_REDLINING__CONTRACT_AGREEMENT__HISTORY' already exists, " +
772+
"but current role has no privileges on it. " +
773+
"If this is unexpected and you cannot resolve this problem, " +
774+
"contact your system administrator. " +
775+
"ACCOUNTADMIN role may be required to manage the privileges on the object."
776+
)
777+
every { connection.close() } just Runs
778+
779+
val exception = assertThrows<ConfigErrorException> { client.execute(sql) }
780+
781+
// Verify the error message was wrapped as ConfigErrorException with original message
782+
assertTrue(exception.message!!.contains("current role has no privileges on it"))
783+
// Verify the cause is the original SnowflakeSQLException
784+
assertTrue(exception.cause is SnowflakeSQLException)
785+
}
786+
787+
@Test
788+
fun testExecuteWithNonPermissionError() {
789+
val connection = mockk<Connection>()
790+
val statement = mockk<Statement>()
791+
val sql = "SELECT * FROM nonexistent_table"
792+
793+
every { dataSource.connection } returns connection
794+
every { connection.createStatement() } returns statement
795+
every { statement.close() } just Runs
796+
797+
// Simulate non-permission error (e.g., table not found)
798+
every { statement.executeQuery(sql) } throws
799+
SnowflakeSQLException("Table 'NONEXISTENT_TABLE' does not exist")
800+
every { connection.close() } just Runs
801+
802+
// Non-permission errors should be thrown as-is, not wrapped
803+
val exception = assertThrows<SnowflakeSQLException> { client.execute(sql) }
804+
805+
assertEquals("Table 'NONEXISTENT_TABLE' does not exist", exception.message)
806+
}
807+
808+
@Test
809+
fun testCreateNamespaceWithPermissionError() {
810+
val namespace = "test_namespace"
811+
val sql = "CREATE SCHEMA test_namespace"
812+
813+
every { sqlGenerator.createNamespace(namespace) } returns sql
814+
815+
// Mock for schema check - returns false (schema doesn't exist)
816+
val schemaCheckResultSet =
817+
mockk<ResultSet> {
818+
every { next() } returns true
819+
every { getBoolean("SCHEMA_EXISTS") } returns false
820+
every { close() } just Runs
821+
}
822+
823+
val preparedStatement =
824+
mockk<PreparedStatement>(relaxed = true) {
825+
every { executeQuery() } returns schemaCheckResultSet
826+
every { close() } just Runs
827+
}
828+
829+
val statement =
830+
mockk<Statement> {
831+
every { executeQuery(any()) } throws
832+
SnowflakeSQLException(
833+
"Schema 'TEST_NAMESPACE' already exists, but current role has no privileges on it"
834+
)
835+
every { close() } just Runs
836+
}
837+
838+
val connection =
839+
mockk<Connection> {
840+
every { createStatement() } returns statement
841+
every { prepareStatement(any()) } returns preparedStatement
842+
every { close() } just Runs
843+
}
844+
845+
every { dataSource.connection } returns connection
846+
847+
runBlocking {
848+
val exception = assertThrows<ConfigErrorException> { client.createNamespace(namespace) }
849+
850+
assertTrue(exception.message!!.contains("current role has no privileges on it"))
851+
assertTrue(exception.cause is SnowflakeSQLException)
852+
}
853+
}
753854
}

‎airbyte-integrations/connectors/source-freshdesk/metadata.yaml‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ data:
1010
connectorSubtype: api
1111
connectorType: source
1212
definitionId: ec4b9503-13cb-48ab-a4ab-6ade4be46567
13-
dockerImageTag: 3.2.1
13+
dockerImageTag: 3.2.2
1414
dockerRepository: airbyte/source-freshdesk
1515
documentationUrl: https://docs.airbyte.com/integrations/sources/freshdesk
1616
githubIssueLabel: source-freshdesk
@@ -31,6 +31,11 @@ data:
3131
rolloutConfiguration:
3232
enableProgressiveRollout: false
3333
releaseStage: generally_available
34+
suggestedStreams:
35+
streams:
36+
- tickets
37+
- agents
38+
- groups
3439
supportLevel: certified
3540
tags:
3641
- language:manifest-only

0 commit comments

Comments
 (0)