Snowflake destination: Report access to schema errors as config errors

[grishick]

Looking at the error in this connection attempt: https://cloud.airbyte.io/workspaces/69b7b606-b6e3-44cc-8a9b-0e083c93d204/connections/78e1ace0-d4b9-4d45-962a-5379cc3f5587/status#938512::2

CHECK passed, but right after that, sync failed with:

Schema 'PUBLIC' already exists, but current role has no privileges on it. If this is unexpected and you cannot resolve this problem, contact your system administrator. ACCOUNTADMIN role may be required to manage the privileges on the object.

[grishick]

The underlying problem is that when we configure the destination connector we check access to "default schema", but replication can be configured to mirror source structure, which will cause the connector to write to a different schema.

Quick fix: wrap the error in config error.
Long term fix: figure out a way to check access to the schema that we will be writing into (may need access to catalog in CHECK.

Pointers:
SnowflakeInternalStagingDestination::check gets schema from config:

      final String outputSchema = nameTransformer.getIdentifier(config.get("schema").asText());
      attemptTableOperations(outputSchema, database, nameTransformer,
          snowflakeInternalStagingSqlOperations, true);

Notes on testing: may need to create a Snowflake account that does not have access to a specific schema, but has permissions to create schemas.

[grishick]

Another example: https://cloud.airbyte.io/workspaces/760cbc70-1e15-495b-86fc-cddbce095e4d/connections/2db4fe8b-fafe-4463-9dc7-a7f1bf018259/status#908557::1

	at io.airbyte.db.jdbc.JdbcDatabase.lambda$execute$0(JdbcDatabase.java:47)
	at io.airbyte.db.jdbc.DefaultJdbcDatabase.execute(DefaultJdbcDatabase.java:46)
	at io.airbyte.db.jdbc.JdbcDatabase.execute(JdbcDatabase.java:47)
	at io.airbyte.integrations.destination.jdbc.JdbcSqlOperations.createSchemaIfNotExists(JdbcSqlOperations.java:47)
	at io.airbyte.integrations.destination.staging.StagingConsumerFactory.lambda$onStartFunction$1(StagingConsumerFactory.java:127)
	at io.airbyte.commons.concurrency.VoidCallable.call(VoidCallable.java:15)
	at io.airbyte.integrations.destination.buffered_stream_consumer.BufferedStreamConsumer.startTracked(BufferedStreamConsumer.java:118)
	at io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer.start(FailureTrackingAirbyteMessageConsumer.java:34)
	at io.airbyte.integrations.base.IntegrationRunner.consumeWriteStream(IntegrationRunner.java:233)
	at io.airbyte.integrations.base.IntegrationRunner.lambda$runConsumer$2(IntegrationRunner.java:241)
	at io.airbyte.integrations.base.IntegrationRunner.watchForOrphanThreads(IntegrationRunner.java:270)
	at io.airbyte.integrations.base.IntegrationRunner.runConsumer(IntegrationRunner.java:240)
	at io.airbyte.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.java:150)
	at io.airbyte.integrations.base.IntegrationRunner.run(IntegrationRunner.java:100)
	at io.airbyte.integrations.base.adaptive.AdaptiveDestinationRunner$Runner.run(AdaptiveDestinationRunner.java:87)
	at io.airbyte.integrations.destination.snowflake.SnowflakeDestinationRunner.main(SnowflakeDestinationRunner.java:17)

this error is caught by CHECK, but even when it is caught during Sync we should report it as config error.

[grishick]

Definition of done for this issue:
when Snowflake destination encounters Snowflake access errors, it emits a config error (AirbyteErrorTraceMessage.FailureType.CONFIG_ERROR)

This followup issue covers improving the UX: #20561

[jbfbell]

I was able to reproduce the issue locally with the following setup:

1. Running the platform locally with docker compose up
2. Setup the postgres source from the postgres to postgres tutorial make sure to add the data from the tutorial
3. Setup the snowflake destination using our snowflake test instance. Follow the instructions presented to the user on the right side of the web UI to set up a new warehouse, user, database, schema, etc
4. Set up a connection between the postgres source and the snowflake destination, and choose to "mirror source structure"
5. Run a sync

This sync will fail with the error

Schema 'PUBLIC' already exists, but current role has no privileges on it. If this is unexpected and you cannot resolve this problem, contact your system administrator.