Support Snowflake (#5500)

[turb]

Fix #5500

First draft of SnowflakeIO for scio, using the one from Beam.

It has:

• SnowflakeSelect (read only), allowing to read from a select
• SnowflakeTable (rw), allowing to read or write from/to a table
• ScioContext and SCollection implicits

Notes:

• I wanted to add some unit tests, but was unsure what could be done, since the only docker env I could find is LocalStack which requires signup
• I did test real-world scenarios for reads, but I ran into [Bug]: Fail to retrieve rowcount for first arrow chunk: null. apache/beam#24374 on write

[codecov]

Codecov Report

Attention: Patch coverage is 35.36585% with 53 lines in your changes missing coverage. Please review.

Project coverage is 61.25%. Comparing base (c45685a) to head (2a2ac46).
Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
...scala/com/spotify/scio/snowflake/SnowflakeIO.scala	27.39%	53 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #5502      +/-   ##
==========================================
- Coverage   61.43%   61.25%   -0.18%     
==========================================
  Files         312      315       +3     
  Lines       11103    11187      +84     
  Branches      762      765       +3     
==========================================
+ Hits         6821     6853      +32     
- Misses       4282     4334      +52     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

---

🚨 Try these New Features:

• Flaky Tests Detection - Detect and resolve failed and flaky tests

[turb]

The build wants me to have "com.nrinaudo" %% "kantan.csv" % kantanCsvVersion as a runtime dep, but then it cannot compile. I thought that a compile dep was also supposed to be in runtime (at least in the Maven model).

(edit: that has been fixed)

[turb]

@RustedBones I finally had the time to fix this build :)

[RustedBones]

Sorry for the delay. Will look at it this week !

[RustedBones]

Suggested change

	trait SnowflakeAuthenticationOptions
	sealed trait SnowflakeAuthenticationOptions

To avoid the match may not be exhaustive warning

[turb]

fixup 9ede3f3

[RustedBones]

Nit: to avoid using intermediate variable, you can use syntactic sugar from scala.util.chaining._.
(can be done above too)

Suggested change

	.withWarehouse(connectionOptions.warehouse)
	connectionOptions.schema
	.map(schema => datasourceBeforeSchema.withSchema(schema))
	.getOrElse(datasourceBeforeSchema)
	.withWarehouse(connectionOptions.warehouse)
	.pipe(ds => connectionOptions.schema.fold(ds)(ds.withSchema))

[turb]

TIL
fixup 25c5a76

[turb]

With fold: a9b56c4

[RustedBones]

I never used Snowflake. Can you give some context for this case ?

[turb]

SnowflakeIO from Apache Beam uses COPY from Snowflake, that simply exports to CSV, that is then read from storage. And this COPY uses \N to represent null.

Thing is, kantan maps empty string with None, so we provide it.

[RustedBones]

Hey @turb, can you test the latest change on a real Snowflake DB to make sure everything still works ?

[turb]

Hey @turb, can you test the latest change on a real Snowflake DB to make sure everything still works ?

Still works! (on read)

[clairemcginty]

Comparing this to other complex testId implementations in our Scio IOs (example 1, 2), I think a format like this would fit better:

SnowflakeIO(url, target, warehouse?, db?)

wdyt @RustedBones

[RustedBones]

This is the same as JdbcIO where we 'normalize' the connection url, hiding credentials. See https://docs.snowflake.com/en/developer-guide/jdbc/jdbc-parameters