Hi, I'm testing StatementSets and I have two sources I want to insert into one sink, like so:

stmt_set \
        .add_insert_sql("INSERT INTO OFFER_ELIGIBILITY_CURRENT_STATE SELECT * FROM offer_elig_j_1")
    stmt_set \
        .add_insert_sql("INSERT INTO OFFER_ELIGIBILITY_CURRENT_STATE SELECT * FROM offer_elig_j_2")
    
    table_result = stmt_set.execute()

but I'm running into this error

Caused by: java.lang.IllegalArgumentException: Hash collision on user-specified ID "uid_hoodie_stream_writeOFFER_ELIGIBILITY_CURRENT_STATE". Most likely cause is a non-unique ID. Please check that all IDs specified via `uid(String)` are unique.

is how I'm using StatementSets to write into one destination from multiple sources supported?

This doesn't look like a problem with StatementSets, but more that something is wrong with your table or query prior to this.

@Martijn Visser hm.. I tested a SELECT UNION on the two source tables and didn't run into any errors

Which Flink version are you using?

1.15.3

And you're only using Table API? Or also DataStream API?

Using the table API only

I would to check because I’m not sure what’s the issue

I am using pyflink

Hmmm. The only statement set example I could find is at https://nightlies.apache.org/flink/flink-docs-master/api/python/_modules/pyflink/table/statement_set.html - We might need to ping one of the PyFlink maintainers for this

Do you know who to ping for that? I tried the

attach_as_datastream()

from https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/data_stream_api/#adding-table-api-pipelines-to-datastream-api but in the jar I'm using (

<https://repo1.maven.org/maven2/org/apache/hudi/hudi-flink1.14-bundle_2.12/0.11.1/hudi-flink1.14-bundle_2.12-0.11.1.jar>

) pyflink doesn't support that function yet I believe. Didn't try the updated jar because we've experienced other errors that made us use this downgraded jar instead

I’ve pinged them offline but it has been a holiday week so it might take a couple of days

thanks, much appreciated

I think this issue has nothing to do with PyFlink. From the exception stack, it seems that it has manually set the uid for the sink table in the hudi connector. I guess you need to use it as following:

INSERT INTO OFFER_ELIGIBILITY_CURRENT_STATE 
SELECT * FROM offer_elig_j_1
UNION ALL
SELECT * FROM offer_elig_j_2

@Dian Fu I'm not sure that's suitable for my case. I'm working with terabytes of data so I would like to split up the inserts into multiple steps to not have to load all the data at once. If I'm using

UNION ALL

, then I will be loading all the data at once and run into OOM unless I deploy a huge amount of resources, right?

As the table name is used to generate the uid, I guess you could try to use different sink table names (however, referring to the same hudi table) for separate

insert into

statements:

stmt_set \
        .add_insert_sql("INSERT INTO OFFER_ELIGIBILITY_CURRENT_STATE_1 SELECT * FROM offer_elig_j_1")
    stmt_set \
        .add_insert_sql("INSERT INTO OFFER_ELIGIBILITY_CURRENT_STATE_2 SELECT * FROM offer_elig_j_2")

this runs, but I was hoping that this would run in sequence rather than in parallel. When I ran this configured for 20 task managers, it created 20 for each pipeline, for 40 total for the two pipelines, and the pipelines ran in parallel