Closed dnxie12 closed 1 year ago
Hey @dnxie12 , how are yoursource
and target
created? Sorry I'm not able to tell what's wrong from this stackrace, could you provide a minimum repro for it?
Hey @sfc-gh-jdu, here is some more information:
Using the script below to generate target
and source
:
from snowflake.snowpark import Session
from snowflake.snowpark.functions import when_matched, when_not_matched
import os
from datetime import datetime
configuration = {
"account": os.getenv("SNOWFLAKE_ACCOUNT"),
"user": os.getenv("SNOWFLAKE_USER"),
"password": os.getenv("SNOWFLAKE_PASSWORD"),
"database": os.getenv("SNOWFLAKE_DATABASE"),
"role": os.getenv("SNOWFLAKE_ROLE"),
"warehouse": os.getenv("SNOWFLAKE_WAREHOUSE"),
"schema": "ANALYTICS_STAGING",
}
session = Session.builder.configs(configuration).create()
target_df = session.create_dataframe([[10, "old", "John", "John", "USA", "active", 45, datetime(2022, 1, 1)],
[11, "too_old", "John", "John", "USA", "active", 75, datetime(2022, 1, 1)],
[12, "old", "John", "John", "USA", "active", 45, datetime(2022, 1, 1)]],
schema=["id", "type", "name", "last_name", "country", "status", "age", "created_at"])
target_df.write.save_as_table("my_table", mode="overwrite")
target = session.table("my_table")
def generate_records(num_records: int):
results = []
for i in range(1, num_records + 1):
record = [i, "new", "john", "smith", "Canada", "active", 35, datetime(2022, 1, 1)]
results.append(record)
return results
source = session.create_dataframe(generate_records(num_records=100), schema=["id", "type", "name", "last_name", "country", "status", "age", "created_at"])
merge_mapping = {v: source[v] for v in source.columns}
target.merge(source, target["id"] == source["id"],
[when_matched().update(merge_mapping), when_not_matched().insert(merge_mapping )])
When I use num_records = 10
everything works and the translated Snowflake query is:
But when I use num_records>=70
, I get the SNOWPARK_TEMP_TABLE
error and the translated Snowflake query is:
Main difference I see between the 2 screenshots is that the second one uses a SELECT * FROM SNOWPARK_TEMP_TABLE
instead of listing out values individually.
hi @sfc-gh-jdu just a quick follow-up to confirm that the min repro provided also errors out on your side and/or if more context is needed? Thanks!
@dnxie12 ack it's a bug and we'll look into it next week. Thanks for bringing it up!
You got two different translated queries because the underlying data size of source
is different. When it's large, we will create a temp table first, insert data into it and load this table in Snowpark. One workaround is to set a larger value for the threshold:
from snowflake.snowpark._internal.analyzer import analyzer
analyzer.ARRAY_BIND_THRESHOLD = <a larger value>
The default value is 512. You can set a larger value though the performance might not be good in this way because we embed the value into SQL text.
That makes sense, thanks for looking into this and for this extra info!
Been experiencing this issue when trying to write a
MERGE
operation using Snowpark. Thanks for taking a look!pip freeze
)?Note: current version of
Snowflake-Snowpark
is 0.9.0 but have also been experiencing the same issue in 0.7.0 and 0.8.0I tried to run the following (
target
being a SnowparkTable
andsource
being a SnowparkDataFrame
)I expected the statement to merge
source
intotarget
, using an update + insert strategy based on whether a record was matched or not matched according to theid
field.What I got was an error that
SNOWPARK_TEMP_TABLE_X
does not exist or is not authorized (X being a random alphanumeric string of length 10). I noticed that the error only occurred when there were more than a few rows (e.g. > 10) in thesource
DataFrame. Running this statement in tests with <= 10 rows in thesource
DataFrame produced the expected result.Full stack trace below:
I can do this if necessary, can also provide the compiled SQL statements that Snowflake tried to run.