Support Snowflake (#5500) #5502

turb · 2024-09-25T14:00:40Z

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 · 2024-09-25T16:10:54Z

Codecov Report

Attention: Patch coverage is 0% with 50 lines in your changes missing coverage. Please review.

Project coverage is 61.16%. Comparing base (14c7077) to head (25c5a76).
Report is 6 commits behind head on main.

Files with missing lines	Patch %	Lines
...scala/com/spotify/scio/snowflake/SnowflakeIO.scala	0.00%	38 Missing ⚠️
...tify/scio/snowflake/syntax/SCollectionSyntax.scala	0.00%	10 Missing ⚠️
...tify/scio/snowflake/syntax/ScioContextSyntax.scala	0.00%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #5502      +/-   ##
==========================================
- Coverage   61.41%   61.16%   -0.25%     
==========================================
  Files         312      315       +3     
  Lines       11106    11153      +47     
  Branches      771      771              
==========================================
+ Hits         6821     6822       +1     
- Misses       4285     4331      +46

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

turb · 2024-09-25T16:38:10Z

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).

turb · 2024-10-16T12:56:30Z

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

RustedBones · 2024-11-04T13:18:29Z

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

RustedBones · 2024-11-04T13:44:52Z

build.sbt

+  )
+  .settings(commonSettings)
+  .settings(
+    description := "Scio add-on for Neo4J",


Suggested change

description := "Scio add-on for Neo4J",

description := "Scio add-on for Snowflake",

fixup 42a4674

RustedBones · 2024-11-04T14:02:24Z

scio-snowflake/src/main/scala/com/spotify/scio/snowflake/SnowflakeOptions.scala

+
+package com.spotify.scio.snowflake
+
+trait SnowflakeAuthenticationOptions


Suggested change

trait SnowflakeAuthenticationOptions

sealed trait SnowflakeAuthenticationOptions

To avoid the match may not be exhaustive warning

RustedBones · 2024-11-04T14:10:27Z

scio-snowflake/src/main/scala/com/spotify/scio/snowflake/SnowflakeIO.scala

+      .withWarehouse(connectionOptions.warehouse)
+
+    connectionOptions.schema
+      .map(schema => datasourceBeforeSchema.withSchema(schema))
+      .getOrElse(datasourceBeforeSchema)


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))

TIL
fixup 25c5a76

RustedBones · 2024-11-04T14:12:32Z

scio-snowflake/src/main/scala/com/spotify/scio/snowflake/SnowflakeIO.scala

+    new CsvMapper[T] {
+      override def mapRow(parts: Array[String]): T = {
+        val unsnowedParts = parts.map {
+          case "\\N" => "" // needs to be mapped to an Option


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

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 · 2024-11-04T14:18:37Z

scio-snowflake/src/main/scala/com/spotify/scio/snowflake/SnowflakeOptions.scala

+final case class SnowflakeOptions(
+  connectionOptions: SnowflakeConnectionOptions,
+  stagingBucketName: String,
+  storageIntegrationName: String
+)


We need to make a distinction between Options and Params:

Options, passed in the IO constructor, are used to iodentify distinct sources

Params of an IO are some tuning parameters to consume the IO.

We need to split the models here to reflect that

I am unsure where each should stand:

connectionOptions are well, connection options

stagingBucketName is a bucket used by the IO to drop/store the CSV files used by COPY, so that's a temp storage not accessed by the user.

storageIntegrationName gives the name of a Snowflake abstraction linking to this storage

I have the feeling all that should go in Options, since they are linked to potentially distinct sources, and not tuning paramters, but maybe I'm mistaken?

RustedBones · 2024-11-04T14:20:36Z

scio-snowflake/src/main/scala/com/spotify/scio/snowflake/SnowflakeIO.scala

+          override def mapRow(element: T): Array[AnyRef] = rowEncoder.encode(element).toArray
+        })
+    )
+    EmptyTap


Isn't it possible to create a tap ?

TBH I now little about taps. The IO I played with (JDBC and Neo4J) are only using EmptyTap. If I understand it well, maybe it is possible to map it to the stored COPYed CSV file?

turb force-pushed the snowflake branch from dda5487 to 40e217c Compare September 25, 2024 15:23

turb mentioned this pull request Sep 25, 2024

Snowflake Support #5500

Open

turb force-pushed the snowflake branch 3 times, most recently from b53021e to 1cf2986 Compare September 25, 2024 15:48

turb force-pushed the snowflake branch 2 times, most recently from d179ff2 to 9f6010d Compare September 25, 2024 16:37

turb force-pushed the snowflake branch 3 times, most recently from 8c434ed to 74d15ae Compare October 16, 2024 12:11

Support Snowflake (spotify#5500)

901625d

turb force-pushed the snowflake branch from 74d15ae to 901625d Compare October 16, 2024 12:18

RustedBones reviewed Nov 4, 2024

View reviewed changes

turb added 2 commits November 4, 2024 16:01

fixup! Support Snowflake (spotify#5500)

42a4674

fixup! Support Snowflake (spotify#5500)

25c5a76

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support Snowflake (#5500) #5502

Support Snowflake (#5500) #5502

turb commented Sep 25, 2024

codecov bot commented Sep 25, 2024 •

edited

Loading

turb commented Sep 25, 2024

turb commented Oct 16, 2024

RustedBones commented Nov 4, 2024

RustedBones Nov 4, 2024

turb Nov 4, 2024

RustedBones Nov 4, 2024

RustedBones Nov 4, 2024

turb Nov 4, 2024

RustedBones Nov 4, 2024

turb Nov 4, 2024 •

edited

Loading

RustedBones Nov 4, 2024

turb Nov 4, 2024

RustedBones Nov 4, 2024

turb Nov 4, 2024

	description := "Scio add-on for Neo4J",
	description := "Scio add-on for Snowflake",


		package com.spotify.scio.snowflake

		trait SnowflakeAuthenticationOptions

	trait SnowflakeAuthenticationOptions
	sealed trait SnowflakeAuthenticationOptions

Support Snowflake (#5500) #5502

Are you sure you want to change the base?

Support Snowflake (#5500) #5502

Conversation

turb commented Sep 25, 2024

codecov bot commented Sep 25, 2024 • edited Loading

Codecov Report

turb commented Sep 25, 2024

turb commented Oct 16, 2024

RustedBones commented Nov 4, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

turb Nov 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Sep 25, 2024 •

edited

Loading

turb Nov 4, 2024 •

edited

Loading